32 lines
1.4 KiB
Python
32 lines
1.4 KiB
Python
"""Orquestração do Dashboard Contábil: extrai o PDF e roda o motor de regras
|
|
sobre o resultado — só em memória (dataclasses), sem tocar no ORM diretamente.
|
|
Mesmo padrão de `portal_api.indicadores.pipeline`/`portal_api.planos_saude.
|
|
pipeline`; quem persiste é a view.
|
|
|
|
Diferente dos outros dois pacotes, aqui a chave de busca do histórico
|
|
(`codigo_empresa`) só é conhecida **depois** de extrair o cabeçalho do PDF —
|
|
por isso `processa_apuracao` recebe `busca_historico` como uma função
|
|
(injetada pela view, que sabe consultar o ORM) em vez de já receber o
|
|
histórico pronto."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import IO, Callable
|
|
|
|
from .modelos import CabecalhoExtraido, ResultadoProcessamento, SnapshotHistorico
|
|
from .parser import extrai_balancete_dre
|
|
from .regras import gera_achados
|
|
|
|
|
|
def processa_apuracao(
|
|
origem: str | IO[bytes],
|
|
busca_historico: Callable[[CabecalhoExtraido], list[SnapshotHistorico]],
|
|
nome_arquivo: str | None = None,
|
|
) -> ResultadoProcessamento:
|
|
"""`nome_arquivo` só é usado no leiaute Contabit, que tira dele o código
|
|
da empresa (ver `parser_contabit.codigo_empresa_do_nome_arquivo`)."""
|
|
extracao = extrai_balancete_dre(origem, nome_arquivo)
|
|
historico = busca_historico(extracao.cabecalho)
|
|
achados = gera_achados(extracao, historico)
|
|
return ResultadoProcessamento(extracao=extracao, achados=achados)
|