Mapa de impostos a partir do e-fatura

faturas.py

import requests
import logging
import json
import sqlite3
import math


search_params = {
    "dataInicioFilter": "2022-06-01",
    "dataFimFilter": "2022-08-31",
    "ambitoAquisicaoFilter": "TODOS",
    "tipoAmbitoAtividadeFilter": "0",
}


cookies = {}


con = sqlite3.connect("faturas.db")
cursor = con.cursor()

cursor.executescript(
    """
    create table documento (
        id integer primary key,
        numero text not null,
        nif integer not null,
        nome text not null,
        data text not null,
        valor_total integer not null,
        valor_iva integer not null
    );

    create table documento_linha (
        documento_id integer not null,
        taxa_iva integer,
        valor_total integer not null,
        valor_iva integer not null,
        valor_base_tributavel integer not null,
      
        unique (documento_id, taxa_iva) on conflict rollback,
        foreign key(documento_id) references documento(id)
    );

    create view mapa_impostos as
        select
            taxa_iva,
            sum(valor_total) as valor_total,
            sum(valor_iva) as valor_iva,
            sum(valor_base_tributavel) as valor_base_tributavel
        from documento_linha
        group by taxa_iva;
"""
)


s = requests.Session()
s.cookies.update(cookies)
s.headers.update({"Accept": "*/*"})

response = s.get(
    "https://faturas.portaldasfinancas.gov.pt/json/obterDocumentosAdquirente.action",
    params=search_params,
)
response_json = response.json()

try:
    documentos = response_json["linhas"]
except KeyError:
    logging.exception(response.text)
    raise


def criar_documento(documento_id, numero, nif, nome, data, valor_total, valor_iva):
    cursor.execute(
        "insert into documento values(?, ?, ?, ?, ?, ?, ?)",
        (documento_id, numero, nif, nome, data, valor_total, valor_iva),
    )


def adicionar_linha_documento(
    documento_id, taxa_iva, valor_total, valor_iva, valor_base_tributavel
):
    cursor.execute(
        "insert into documento_linha values(?, ?, ?, ?, ?)",
        (documento_id, taxa_iva, valor_total, valor_iva, valor_base_tributavel),
    )


def ensure_invoice_totals(documento_id):
    rows = cursor.execute(
        "select valor_total from documento where id = ?", (documento_id,)
    )
    (documento_total,) = rows.fetchone()

    rows = cursor.execute(
        "select sum(valor_total) from documento_linha where documento_id = ?",
        (documento_id,),
    )
    (total_sum,) = rows.fetchone()

    if not math.isclose(documento_total, total_sum, abs_tol=2):
        raise ValueError(
            f"total documento {documento_total} diferente da soma das linhas {total_sum}"
        )


for documento in documentos:
    documento_id = documento["idDocumento"]
    documento_data = documento["dataEmissaoDocumento"]

    criar_documento(
        documento_id,
        documento["numerodocumento"],
        documento["nifEmitente"],
        documento["nomeEmitente"],
        documento_data,
        documento["valorTotal"],
        documento["valorTotalIva"],
    )

    params = {
        "idDocumento": documento_id,
        "dataEmissaoDocumento": documento_data,
    }
    response = s.get(
        "https://faturas.portaldasfinancas.gov.pt/detalheDocumentoAdquirente.action",
        params=params,
    )
    response_text = response.text

    try:
        linhas = response_text.split("dadosLinhasDocumento = ")[1].split("];")[0] + "]"
        linhas_parsed = json.loads(linhas)

        for linha in linhas_parsed:
            adicionar_linha_documento(
                documento_id,
                linha["taxaIva"],
                linha["valorTotal"],
                linha["valorIva"],
                linha["valorBaseTributavel"],
            )

    except (IndexError, KeyError, json.decoder.JSONDecodeError):
        logging.exception(linhas)
        raise

    con.commit()

    ensure_invoice_totals(documento_id)