482 lines
23 KiB
Python
482 lines
23 KiB
Python
"""Motor de regras de auditoria do Dashboard Contábil — cada `regra_*` é uma
|
|
função pura (`ResultadoExtracao` da apuração atual + histórico já persistido
|
|
da mesma empresa → `list[AchadoDetectado]`), sem tocar no ORM. Cobrem só o
|
|
que é derivável do próprio balancete/DRE anexado (ver CLAUDE.md do pacote) —
|
|
qualquer checagem que dependa de sistemas externos (Questor, extratos,
|
|
folha) fica fora de propósito, por decisão do usuário (ferramenta analítica,
|
|
não operacional)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from decimal import Decimal
|
|
|
|
from .modelos import (
|
|
LEIAUTE_CONTABIT,
|
|
AchadoDetectado,
|
|
LinhaBalanceteExtraida,
|
|
ResultadoExtracao,
|
|
SnapshotHistorico,
|
|
)
|
|
|
|
# Exclui "NUMERÁRIOS EM TRANSITO"/"... EM TRANSITO" (dinheiro em trânsito,
|
|
# conceito diferente de conta transitória/de compensação) do match de
|
|
# "TRANSIT" em regra_conta_transitoria_com_saldo, sem trocar o próprio
|
|
# critério por "TRANSITOR" — confirmado contra `1751 - Balancete 07.2026.pdf`
|
|
# que a fonte embutida corrompe o acento de "TRANSITÓRIA" num caractere
|
|
# ilegível (não dá pra recuperar a letra original, diferente do caso mais
|
|
# simples de `_normaliza_titulo` em parser.py, que só perde o til e mantém o
|
|
# resto do caractere certo) — "TRANSITOR" nunca bateria com essas contas.
|
|
# `\b` nas duas pontas garante que só a palavra isolada "TRANSITO" é
|
|
# excluída, nunca um prefixo de "TRANSITORIA"/"TRANSITORIO" sem acento.
|
|
_RE_TRANSITO_ISOLADO = re.compile(r"\bTRANSITO\b")
|
|
|
|
SEVERIDADE_ALTA = "alta"
|
|
SEVERIDADE_MEDIA = "media"
|
|
SEVERIDADE_BAIXA = "baixa"
|
|
|
|
# Trechos de descrição que, segundo o roteiro de conferência (ITD-FISCO-7513),
|
|
# devem ficar zeradas todo mês — ex.: "o saldo de adiantamento de salários e
|
|
# empréstimos a funcionários deve estar zerado todos os meses". Deliberadamente
|
|
# restrito a contas cujo saldo residual é sempre um sinal de pendência (não
|
|
# inclui, por exemplo, "ADIANTAMENTOS DE FÉRIAS"/"13º SALÁRIO", que legitimamente
|
|
# carregam saldo entre um mês e o pagamento seguinte).
|
|
TRECHOS_CONTA_DEVERIA_ZERAR = ["ADIANTAMENTOS DE SALÁRIOS", "ADIANTAMENTO DE SALÁRIOS"]
|
|
|
|
# Código de classificação fixo pra linha "LUCROS/PREJUÍZOS DO EXERCÍCIO" dentro
|
|
# do Patrimônio Líquido — calibrado contra os 2 balancetes reais de referência
|
|
# do usuário (`792`/`2017` e `1751`), mesmo padrão de risco já aceito em
|
|
# CODIGO_CAIXA/CODIGO_PATRIMONIO_LIQUIDO (indicadores.py). É sintética (tipo
|
|
# "S"): agrega "LUCROS DO EXERCÍCIO" (quando a empresa deu lucro) ou "(-)
|
|
# PREJUÍZOS DO EXERCÍCIO" (quando deu prejuízo) — por isso a checagem lê o
|
|
# saldo desta linha agregadora, não de uma das duas filhas específicas.
|
|
CODIGO_LUCRO_PREJUIZO_EXERCICIO = "2.04.13.002"
|
|
|
|
# Equivalentes no plano de contas do Contabit, calibrados contra `1512 -
|
|
# Balancete 082026.pdf` (único arquivo Contabit de referência até aqui):
|
|
# "2.40.40.20 LUCROS / PREJUIZOS DO EXERCÍCIO" e "1.10.10.01 BENS NUMERÁRIOS"
|
|
# (o grupo que contém "Caixa"). Mesmo risco dos códigos do Questor: se outro
|
|
# cliente Contabit usar numeração diferente, a regra deixa de disparar.
|
|
CODIGO_LUCRO_PREJUIZO_EXERCICIO_CONTABIT = "2.40.40.20"
|
|
PREFIXO_CAIXA_QUESTOR = "1.01.01.001"
|
|
PREFIXO_CAIXA_CONTABIT = "1.10.10.01"
|
|
|
|
VARIACAO_LIMIAR_PERCENTUAL = Decimal("0.65") # 65%
|
|
VARIACAO_VALOR_MINIMO = Decimal("1000") # ignora variações abaixo disso, mesmo que %-mente grandes
|
|
# Piso em pontos percentuais pra regra_variacao_atipica_dre (Análise Vertical)
|
|
# — mesmo espírito de VARIACAO_VALOR_MINIMO acima, adaptado pro domínio de
|
|
# percentual: evita disparar em saltos %-mente grandes só porque a base de
|
|
# comparação já era perto de zero (ex.: 0,05% → 0,2% é 300% de variação
|
|
# relativa, mas irrelevante em termos de composição da receita).
|
|
VARIACAO_AV_PONTOS_PERCENTUAIS_MINIMO = Decimal("1")
|
|
|
|
# Nome pedido pelo usuário pro apontamento da regra de variação: deixa claro
|
|
# que ele vem da Análise Vertical, não da DRE acumulada. A migração `0081`
|
|
# aplicou o mesmo título aos apontamentos já gravados.
|
|
TITULO_VARIACAO_ATIPICA_DRE = "Variação atípica na DRE - Análise Vertical"
|
|
|
|
|
|
def _moeda(valor: Decimal) -> str:
|
|
"""Formata no padrão brasileiro (ponto de milhar, vírgula decimal) —
|
|
mesmo espírito do helper `_moeda()`/`moeda()` já duplicado por arquivo em
|
|
`indicadores/recibo.py`, `custo_contratacao/pdf.py` e
|
|
`templatetags/contabil_extras.py`; duplicado aqui em vez de importado
|
|
porque este pacote é Python puro, sem depender do app registry do Django
|
|
(ver docstring do módulo). Interpolar `Decimal` direto num f-string (ex.:
|
|
`f"R$ {valor}"`) usa a formatação padrão do Python — ponto decimal, sem
|
|
separador de milhar — que é exatamente o bug que este helper evita."""
|
|
texto_us = f"{valor:,.2f}"
|
|
return f"R$ {texto_us.replace(',', '_').replace('.', ',').replace('_', '.')}"
|
|
|
|
|
|
def regra_balanceamento_ativo_passivo(atual: ResultadoExtracao, historico: list[SnapshotHistorico]) -> list[AchadoDetectado]:
|
|
ativo = next((c for c in atual.contas if c.codigo == "1"), None)
|
|
passivo = next((c for c in atual.contas if c.codigo == "2"), None)
|
|
if ativo is None or passivo is None:
|
|
return []
|
|
# Questor: o Passivo já vem negativo no relatório, então fechar é somar
|
|
# a zero. Contabit: os dois vêm positivos (saldo pela natureza).
|
|
saldo_passivo = passivo.saldo_atual if atual.leiaute == LEIAUTE_CONTABIT else -passivo.saldo_atual
|
|
diferenca = ativo.saldo_atual - saldo_passivo
|
|
if diferenca == 0:
|
|
return []
|
|
return [
|
|
AchadoDetectado(
|
|
regra="balanceamento_ativo_passivo",
|
|
severidade=SEVERIDADE_ALTA,
|
|
titulo="Ativo não bate com Passivo",
|
|
mensagem=(
|
|
f"Saldo do Ativo ({_moeda(ativo.saldo_atual)}) não coincide com o do Passivo "
|
|
f"({_moeda(saldo_passivo)}) — diferença de {_moeda(abs(diferenca))}."
|
|
),
|
|
valor_referencia=diferenca,
|
|
)
|
|
]
|
|
|
|
|
|
def regra_debito_credito_divergente(atual: ResultadoExtracao, historico: list[SnapshotHistorico]) -> list[AchadoDetectado]:
|
|
raizes = [c for c in atual.contas if "." not in c.codigo]
|
|
soma_debito = sum((c.debito for c in raizes), Decimal(0))
|
|
soma_credito = sum((c.credito for c in raizes), Decimal(0))
|
|
diferenca = soma_debito - soma_credito
|
|
if diferenca == 0:
|
|
return []
|
|
return [
|
|
AchadoDetectado(
|
|
regra="debito_credito_divergente",
|
|
severidade=SEVERIDADE_ALTA,
|
|
titulo="Débito total diferente do crédito total",
|
|
mensagem=(
|
|
f"Soma de débitos do período ({_moeda(soma_debito)}) não bate com a soma de "
|
|
f"créditos ({_moeda(soma_credito)}) — diferença de {_moeda(abs(diferenca))}."
|
|
),
|
|
valor_referencia=diferenca,
|
|
)
|
|
]
|
|
|
|
|
|
def regra_saldo_negativo_caixa(atual: ResultadoExtracao, historico: list[SnapshotHistorico]) -> list[AchadoDetectado]:
|
|
achados = []
|
|
prefixo = PREFIXO_CAIXA_CONTABIT if atual.leiaute == LEIAUTE_CONTABIT else PREFIXO_CAIXA_QUESTOR
|
|
for conta in atual.contas:
|
|
if conta.codigo.startswith(prefixo) and conta.saldo_atual < 0:
|
|
achados.append(
|
|
AchadoDetectado(
|
|
regra="saldo_negativo_caixa",
|
|
severidade=SEVERIDADE_ALTA,
|
|
titulo="Saldo de caixa negativo",
|
|
mensagem=f'Conta "{conta.descricao}" ({conta.codigo}) com saldo negativo de {_moeda(conta.saldo_atual)}.',
|
|
codigo_conta=conta.codigo,
|
|
valor_referencia=conta.saldo_atual,
|
|
)
|
|
)
|
|
return achados
|
|
|
|
|
|
def regra_conta_transitoria_com_saldo(atual: ResultadoExtracao, historico: list[SnapshotHistorico]) -> list[AchadoDetectado]:
|
|
achados = []
|
|
for conta in atual.contas:
|
|
descricao_normalizada = conta.descricao.upper()
|
|
tem_transit = "TRANSIT" in descricao_normalizada and not _RE_TRANSITO_ISOLADO.search(descricao_normalizada)
|
|
if tem_transit and conta.saldo_atual != 0:
|
|
achados.append(
|
|
AchadoDetectado(
|
|
regra="conta_transitoria_com_saldo",
|
|
severidade=SEVERIDADE_MEDIA,
|
|
titulo="Conta transitória com saldo",
|
|
mensagem=f'Conta transitória "{conta.descricao}" ({conta.codigo}) deveria estar zerada e está com {_moeda(conta.saldo_atual)}.',
|
|
codigo_conta=conta.codigo,
|
|
valor_referencia=conta.saldo_atual,
|
|
)
|
|
)
|
|
return achados
|
|
|
|
|
|
def regra_conta_deveria_zerar(atual: ResultadoExtracao, historico: list[SnapshotHistorico]) -> list[AchadoDetectado]:
|
|
achados = []
|
|
for conta in atual.contas:
|
|
descricao_normalizada = conta.descricao.upper()
|
|
if conta.saldo_atual == 0:
|
|
continue
|
|
if any(trecho in descricao_normalizada for trecho in TRECHOS_CONTA_DEVERIA_ZERAR):
|
|
achados.append(
|
|
AchadoDetectado(
|
|
regra="conta_deveria_zerar",
|
|
severidade=SEVERIDADE_MEDIA,
|
|
titulo="Conta que deveria estar zerada",
|
|
mensagem=f'Conta "{conta.descricao}" ({conta.codigo}) normalmente fica zerada todo mês e está com saldo de {_moeda(conta.saldo_atual)}.',
|
|
codigo_conta=conta.codigo,
|
|
valor_referencia=conta.saldo_atual,
|
|
)
|
|
)
|
|
return achados
|
|
|
|
|
|
def _eh_conta_redutora(conta: LinhaBalanceteExtraida) -> bool:
|
|
return conta.descricao.strip().startswith("(-)")
|
|
|
|
|
|
def _indices_descendentes_de_conta_redutora(contas: list[LinhaBalanceteExtraida]) -> set[int]:
|
|
"""Índices (posição em `contas`, mesma ordem de leitura do PDF) de toda
|
|
conta que tem algum ANCESTRAL sintético (não só o pai direto) com
|
|
descrição começando em "(-)" — pedido explícito do usuário: se a conta
|
|
"mãe" tem o sinal de redutora, o saldo "invertido" dos filhos é o
|
|
comportamento esperado, não uma inconsistência (ex.: "(-) LUCROS
|
|
DISTRIBUÍDOS" é uma conta do Passivo, mas devedora por natureza — as
|
|
contas analíticas dentro dela, um sócio por linha, herdam esse mesmo
|
|
sinal e não deveriam gerar achado de `regra_saldo_sinal_invertido`).
|
|
|
|
Usa o mesmo algoritmo de nível/hierarquia já usado em toda a aplicação
|
|
pra árvore de contas (`codigo.count(".")` + ordem de leitura do PDF, ver
|
|
`_contabil_arvore_contexto()`/`dcContaNivel()`), **não** comparação de
|
|
prefixo de código — o Questor reaproveita o mesmo código de classificação
|
|
entre contas analíticas irmãs (ver `ContabilObservacao.chave_conta()`),
|
|
então string matching por código não seria confiável pra identificar o
|
|
pai; a pilha de níveis, sim, já que segue estritamente a ordem/profundidade
|
|
real da árvore impressa no PDF."""
|
|
niveis = [conta.codigo.count(".") for conta in contas]
|
|
pilha: list[tuple[int, bool]] = [] # (nível, é redutora OU descende de uma)
|
|
descendentes: set[int] = set()
|
|
for i, conta in enumerate(contas):
|
|
nivel = niveis[i]
|
|
while pilha and pilha[-1][0] >= nivel:
|
|
pilha.pop()
|
|
heranca = pilha[-1][1] if pilha else False
|
|
if heranca:
|
|
descendentes.add(i)
|
|
pilha.append((nivel, heranca or _eh_conta_redutora(conta)))
|
|
return descendentes
|
|
|
|
|
|
def _saldo_sinal_invertido_contabit(atual: ResultadoExtracao) -> list[AchadoDetectado]:
|
|
"""No Contabit o saldo já sai pela natureza da própria conta (Passivo e
|
|
redutoras positivos), então saldo contrário à natureza é simplesmente
|
|
saldo negativo, em qualquer grupo. A natureza (devedora/credora) só
|
|
importa pro texto: grupo 1 é devedor, grupo 2 credor, invertido quando a
|
|
conta é redutora ou descende de uma (mesma pilha de níveis do Questor).
|
|
|
|
Premissa ainda não confirmada com arquivo real: o Contabit imprimir o
|
|
saldo contrário com "-". Se imprimir de outro jeito, a regra só deixa de
|
|
disparar (sem falso positivo)."""
|
|
achados = []
|
|
descendentes_de_redutora = _indices_descendentes_de_conta_redutora(atual.contas)
|
|
for i, conta in enumerate(atual.contas):
|
|
if conta.tipo != "A" or conta.saldo_atual >= 0 or not conta.codigo.startswith(("1.", "2.")):
|
|
continue
|
|
eh_ativo = conta.codigo.startswith("1.")
|
|
invertida = _eh_conta_redutora(conta) or i in descendentes_de_redutora
|
|
natureza_devedora = eh_ativo != invertida
|
|
grupo = "Ativo" if eh_ativo else "Passivo"
|
|
saldo = "credor" if natureza_devedora else "devedor"
|
|
achados.append(
|
|
AchadoDetectado(
|
|
regra="saldo_sinal_invertido",
|
|
severidade=SEVERIDADE_MEDIA,
|
|
titulo=f"Conta do {grupo} com saldo {saldo}",
|
|
mensagem=f'Conta do {grupo} "{conta.descricao}" ({conta.codigo}) está com saldo {saldo} de {_moeda(-conta.saldo_atual)}.',
|
|
codigo_conta=conta.codigo,
|
|
valor_referencia=conta.saldo_atual,
|
|
)
|
|
)
|
|
return achados
|
|
|
|
|
|
def regra_saldo_sinal_invertido(atual: ResultadoExtracao, historico: list[SnapshotHistorico]) -> list[AchadoDetectado]:
|
|
if atual.leiaute == LEIAUTE_CONTABIT:
|
|
return _saldo_sinal_invertido_contabit(atual)
|
|
achados = []
|
|
descendentes_de_redutora = _indices_descendentes_de_conta_redutora(atual.contas)
|
|
for i, conta in enumerate(atual.contas):
|
|
if conta.tipo != "A" or conta.saldo_atual == 0 or _eh_conta_redutora(conta) or i in descendentes_de_redutora:
|
|
continue
|
|
if conta.codigo.startswith("1.") and conta.saldo_atual < 0:
|
|
achados.append(
|
|
AchadoDetectado(
|
|
regra="saldo_sinal_invertido",
|
|
severidade=SEVERIDADE_MEDIA,
|
|
titulo="Conta do Ativo com saldo credor",
|
|
mensagem=f'Conta do Ativo "{conta.descricao}" ({conta.codigo}) está com saldo credor de {_moeda(-conta.saldo_atual)}.',
|
|
codigo_conta=conta.codigo,
|
|
valor_referencia=conta.saldo_atual,
|
|
)
|
|
)
|
|
elif conta.codigo.startswith("2.") and conta.saldo_atual > 0:
|
|
achados.append(
|
|
AchadoDetectado(
|
|
regra="saldo_sinal_invertido",
|
|
severidade=SEVERIDADE_MEDIA,
|
|
titulo="Conta do Passivo com saldo devedor",
|
|
mensagem=f'Conta do Passivo "{conta.descricao}" ({conta.codigo}) está com saldo devedor de {_moeda(conta.saldo_atual)}.',
|
|
codigo_conta=conta.codigo,
|
|
valor_referencia=conta.saldo_atual,
|
|
)
|
|
)
|
|
return achados
|
|
|
|
|
|
def regra_lucro_balancete_diverge_dre(atual: ResultadoExtracao, historico: list[SnapshotHistorico]) -> list[AchadoDetectado]:
|
|
"""O resultado do exercício precisa ser o mesmo número nas duas
|
|
demonstrações — só que, por convenção, o Passivo/PL vem com o sinal
|
|
invertido no relatório do Questor (ver `regra_balanceamento_ativo_passivo`),
|
|
então o saldo da linha do Balancete precisa ser negado antes de comparar
|
|
com a última linha da DRE. No Contabit o saldo já sai pela natureza
|
|
(lucro positivo), sem negar."""
|
|
contabit = atual.leiaute == LEIAUTE_CONTABIT
|
|
codigo_lucro = CODIGO_LUCRO_PREJUIZO_EXERCICIO_CONTABIT if contabit else CODIGO_LUCRO_PREJUIZO_EXERCICIO
|
|
conta_lucro = next((c for c in atual.contas if c.codigo == codigo_lucro), None)
|
|
if conta_lucro is None or not atual.linhas_dre:
|
|
return []
|
|
lucro_balancete = conta_lucro.saldo_atual if contabit else -conta_lucro.saldo_atual
|
|
lucro_dre = atual.linhas_dre[-1].valor
|
|
diferenca = lucro_balancete - lucro_dre
|
|
if diferenca == 0:
|
|
return []
|
|
return [
|
|
AchadoDetectado(
|
|
regra="lucro_balancete_diverge_dre",
|
|
severidade=SEVERIDADE_ALTA,
|
|
titulo="Lucro do balancete diferente do lucro da DRE",
|
|
mensagem=(
|
|
f'Resultado do exercício no Balancete ("{conta_lucro.descricao}", {conta_lucro.codigo}) é de '
|
|
f"{_moeda(lucro_balancete)}, mas a DRE aponta {_moeda(lucro_dre)} — diferença de {_moeda(abs(diferenca))}."
|
|
),
|
|
codigo_conta=conta_lucro.codigo,
|
|
valor_referencia=diferenca,
|
|
)
|
|
]
|
|
|
|
|
|
def regra_descricao_generica(atual: ResultadoExtracao, historico: list[SnapshotHistorico]) -> list[AchadoDetectado]:
|
|
achados = []
|
|
for conta in atual.contas:
|
|
if conta.descricao.strip().upper() == "DIVERSOS" and conta.saldo_atual != 0:
|
|
achados.append(
|
|
AchadoDetectado(
|
|
regra="descricao_generica",
|
|
severidade=SEVERIDADE_BAIXA,
|
|
titulo='Conta com descrição genérica ("DIVERSOS")',
|
|
mensagem=f'Conta {conta.codigo} está descrita apenas como "DIVERSOS", com saldo de {_moeda(conta.saldo_atual)} — recomenda-se detalhar em conta analítica própria.',
|
|
codigo_conta=conta.codigo,
|
|
valor_referencia=conta.saldo_atual,
|
|
)
|
|
)
|
|
return achados
|
|
|
|
|
|
def regra_variacao_atipica_dre(atual: ResultadoExtracao, historico: list[SnapshotHistorico]) -> list[AchadoDetectado]:
|
|
"""Usa a própria seção "Demonstração Mensal (Análise Vertical)" do PDF
|
|
(quando presente) em vez do histórico de apurações anteriores do Portal
|
|
— cada linha já vem com o percentual sobre a Receita Operacional Bruta
|
|
em cada um dos últimos meses (calculado pelo próprio Questor, valor já
|
|
isolado por mês, não acumulado), então a regra só compara os 2 meses
|
|
mais recentes dessa própria tabela. Roda mesmo na primeira apuração de
|
|
uma empresa nova, desde que o PDF traga essa seção — não depende de
|
|
`historico` (recebido só pra manter a assinatura comum a toda regra)."""
|
|
achados = []
|
|
for linha in atual.linhas_analise_vertical:
|
|
if len(linha.valores) < 2:
|
|
continue
|
|
mes_anterior, mes_atual = linha.valores[-2], linha.valores[-1]
|
|
diferenca_pp = mes_atual.percentual - mes_anterior.percentual
|
|
if abs(diferenca_pp) < VARIACAO_AV_PONTOS_PERCENTUAIS_MINIMO:
|
|
continue
|
|
if mes_anterior.percentual != 0:
|
|
variacao_relativa = abs(diferenca_pp) / abs(mes_anterior.percentual)
|
|
if variacao_relativa < VARIACAO_LIMIAR_PERCENTUAL:
|
|
continue
|
|
rotulo_anterior = atual.meses_analise_vertical[-2] if len(atual.meses_analise_vertical) >= 2 else "mês anterior"
|
|
rotulo_atual = atual.meses_analise_vertical[-1] if atual.meses_analise_vertical else "mês atual"
|
|
achados.append(
|
|
AchadoDetectado(
|
|
regra="variacao_atipica_dre",
|
|
severidade=SEVERIDADE_BAIXA,
|
|
titulo=TITULO_VARIACAO_ATIPICA_DRE,
|
|
mensagem=(
|
|
f'Linha "{linha.descricao}" da DRE foi de {mes_anterior.percentual:.2f}% da Receita Bruta em '
|
|
f"{rotulo_anterior} para {mes_atual.percentual:.2f}% em {rotulo_atual} "
|
|
f"({diferenca_pp:+.2f} pontos percentuais)."
|
|
),
|
|
valor_referencia=diferenca_pp,
|
|
ordem_linha_analise_vertical=linha.ordem,
|
|
)
|
|
)
|
|
return achados
|
|
|
|
|
|
# Chave da única "regra" que não roda sobre o PDF, e sim sobre a diferença
|
|
# entre o que estava salvo e o que veio no reprocessamento (ver
|
|
# `achados_itens_removidos()` abaixo). Fica fora de `REGRAS` de propósito:
|
|
# `gera_achados()` só enxerga a extração atual, nunca o estado anterior.
|
|
REGRA_ITEM_REMOVIDO = "item_removido_reprocessamento"
|
|
|
|
ORIGEM_BALANCETE = "Balancete"
|
|
ORIGEM_DRE = "D.R.E."
|
|
ORIGEM_ANALISE_VERTICAL = "Análise Vertical"
|
|
|
|
# A mesma origem, com artigo, pra caber na frase do apontamento.
|
|
_ORIGEM_COM_ARTIGO = {
|
|
ORIGEM_BALANCETE: "do Balancete",
|
|
ORIGEM_DRE: "da D.R.E.",
|
|
ORIGEM_ANALISE_VERTICAL: "da Análise Vertical",
|
|
}
|
|
|
|
|
|
def _lista_origens(origens: list[str]) -> str:
|
|
"""`"da D.R.E. e da Análise Vertical"` — uma linha da DRE quase sempre
|
|
some das duas tabelas ao mesmo tempo (é a mesma árvore), e dois
|
|
apontamentos com o mesmo texto seriam ruído."""
|
|
partes = [_ORIGEM_COM_ARTIGO.get(origem, origem) for origem in origens]
|
|
if len(partes) == 1:
|
|
return partes[0]
|
|
return f"{', '.join(partes[:-1])} e {partes[-1]}"
|
|
|
|
|
|
def achados_itens_removidos(removidos: list[tuple[str, str]]) -> list[AchadoDetectado]:
|
|
"""Um apontamento por conta/linha que existia na versão anterior da
|
|
apuração e não veio no arquivo do reprocessamento.
|
|
|
|
`removidos` é uma lista de `(origem, rótulo)` — a origem é uma das
|
|
constantes `ORIGEM_*` e o rótulo já vem pronto de quem detectou a remoção
|
|
(`_contabil_sincroniza_*()` em `views.py`): `"1.01.01.001 CAIXA"` no
|
|
Balancete, o caminho na árvore na DRE/Análise Vertical. Rótulos iguais
|
|
viram **um** apontamento citando as duas tabelas: DRE e Análise Vertical
|
|
são a mesma árvore, então uma linha retirada do PDF some das duas.
|
|
|
|
Por que nasce aqui e não numa `regra_*`: as regras são funções puras
|
|
sobre a extração do PDF atual, e "sumiu" só é visível comparando com o
|
|
que estava salvo, que é justamente o que a sincronização do
|
|
reprocessamento faz. A construção do achado fica neste módulo mesmo
|
|
assim, junto de todos os outros textos/severidades, para não espalhar a
|
|
definição de apontamento por dois arquivos.
|
|
|
|
**O aviso vale para o reprocessamento em que a remoção aconteceu.** Como
|
|
todo achado é recriado do zero a cada reprocessamento e a linha já não
|
|
está no banco, um reprocessamento seguinte (com o mesmo arquivo sem ela)
|
|
não repete o aviso: naquele ponto não há mais nada sendo removido.
|
|
"""
|
|
agrupados: dict[str, list[str]] = {}
|
|
for origem, rotulo in removidos:
|
|
origens = agrupados.setdefault(rotulo, [])
|
|
if origem not in origens:
|
|
origens.append(origem)
|
|
|
|
achados: list[AchadoDetectado] = []
|
|
for rotulo, origens in agrupados.items():
|
|
eh_conta = origens == [ORIGEM_BALANCETE]
|
|
achados.append(
|
|
AchadoDetectado(
|
|
regra=REGRA_ITEM_REMOVIDO,
|
|
severidade=SEVERIDADE_MEDIA,
|
|
titulo=("Conta removida no reprocessamento" if eh_conta else "Linha removida no reprocessamento"),
|
|
mensagem=(
|
|
f'{"A conta" if eh_conta else "A linha"} "{rotulo}" constava na versão anterior '
|
|
f"{_lista_origens(origens)} desta apuração e não veio no arquivo reprocessado, por isso "
|
|
"foi excluída. Confira se a exclusão era esperada. As observações registradas nela "
|
|
"continuam no histórico da empresa."
|
|
),
|
|
)
|
|
)
|
|
return achados
|
|
|
|
|
|
REGRAS = [
|
|
regra_balanceamento_ativo_passivo,
|
|
regra_debito_credito_divergente,
|
|
regra_saldo_negativo_caixa,
|
|
regra_saldo_sinal_invertido,
|
|
regra_lucro_balancete_diverge_dre,
|
|
regra_conta_transitoria_com_saldo,
|
|
regra_conta_deveria_zerar,
|
|
regra_descricao_generica,
|
|
regra_variacao_atipica_dre,
|
|
]
|
|
|
|
|
|
def gera_achados(atual: ResultadoExtracao, historico: list[SnapshotHistorico]) -> list[AchadoDetectado]:
|
|
"""`historico` deve vir ordenado da apuração mais recente pra mais antiga
|
|
(mesma empresa, competências anteriores à de `atual`)."""
|
|
achados: list[AchadoDetectado] = []
|
|
for regra in REGRAS:
|
|
achados.extend(regra(atual, historico))
|
|
return achados
|