portal_publico/portal_api/dashboard_contabil/regras.py

"""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 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"

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")


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 []
    diferenca = ativo.saldo_atual + passivo.saldo_atual  # passivo já vem negativo no relatório
    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(-passivo.saldo_atual)}) — 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 = []
    for conta in atual.contas:
        if conta.codigo.startswith("1.01.01.001") 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 regra_saldo_sinal_invertido(atual: ResultadoExtracao, historico: list[SnapshotHistorico]) -> list[AchadoDetectado]:
    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 (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."""
    conta_lucro = next((c for c in atual.contas if c.codigo == CODIGO_LUCRO_PREJUIZO_EXERCICIO), None)
    if conta_lucro is None or not atual.linhas_dre:
        return []
    lucro_balancete = -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="Variação atípica na 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,
            )
        )
    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