248 lines
11 KiB
Python
248 lines
11 KiB
Python
"""Extração do PDF de Balancete + DRE (leiaute Questor, único leiaute-fonte —
|
|
ao contrário de `portal_api.planos_saude`, que precisa de um parser por
|
|
operadora, aqui o relatório é sempre gerado pelo mesmo sistema contábil).
|
|
|
|
O texto deste relatório é desenhado caractere a caractere (cada letra/número
|
|
tem sua própria posição X, sem depender de kerning do PDF) e o próprio PDF
|
|
inclui, por baixo do texto real, uma grade densa de caracteres de espaço
|
|
cobrindo a largura inteira de cada linha — isso confunde tanto
|
|
`page.extract_words()` quanto `page.extract_text()` do pdfplumber, que
|
|
tratam esses espaços "de fundo" como separadores de palavra reais e acabam
|
|
quebrando números em dígitos isolados. `_reconstroi_linhas()` contorna isso
|
|
ignorando todo caractere de espaço literal do PDF e reconstruindo a linha
|
|
a partir da posição real dos caracteres não-espaço, reinserindo um único
|
|
espaço só quando o vão horizontal entre dois caracteres consecutivos indica
|
|
uma quebra de campo/palavra de verdade (validado contra
|
|
`792 - balancete 072026.pdf`, ver `plano.md`)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from datetime import date
|
|
from decimal import Decimal, InvalidOperation
|
|
from typing import IO
|
|
|
|
import pdfplumber
|
|
|
|
from .modelos import CabecalhoExtraido, LinhaBalanceteExtraida, LinhaDreExtraida, ResultadoExtracao
|
|
|
|
# Vão horizontal (em pontos) acima do qual dois caracteres consecutivos são
|
|
# tratados como pertencendo a campos/palavras diferentes. Calibrado contra o
|
|
# relatório real: o vão dentro de uma palavra/número é ~0, entre duas
|
|
# palavras da mesma descrição é ~1.7-1.9, e entre campos da tabela (conta →
|
|
# flag S/A → código → descrição, ou entre colunas de valor) é sempre >= 5.
|
|
_GAP_ESPACO = 0.8
|
|
|
|
_TITULO_DRE = "DEMONSTRAÇÃO DO RESULTADO DO EXERCÍCIO"
|
|
_TITULO_ANALISE_VERTICAL = "Demonstração Mensal (Análise Vertical)"
|
|
|
|
_RE_MONETARIO = re.compile(r"\(?-?[\d.]+,\d{2}\)?")
|
|
_RE_LINHA_BALANCETE = re.compile(r"^(?P<conta>\d+)\s+(?P<tipo>S)?\s*(?P<codigo>[\d.]+)\s+(?P<resto>.+)$")
|
|
_RE_CNPJ = re.compile(r"CNPJ:\s*([\d./-]+)")
|
|
_RE_PERIODO = re.compile(r"Per[ií]odo:\s*(\d{2}/\d{2}/\d{4})\s*a\s*(\d{2}/\d{2}/\d{4})")
|
|
_RE_CODIGO_NOME_EMPRESA = re.compile(r"^(?P<codigo>\d+)\s+(?P<nome>.+)$")
|
|
|
|
# Linhas totalizadoras conhecidas da DRE (usadas por regras que precisam de
|
|
# uma linha específica, ex. razão custo/receita) — além do negrito (que já
|
|
# identifica visualmente todo totalizador pro frontend), casar por texto é
|
|
# mais robusto a pequenas variações de nível/indentação entre empresas. O
|
|
# texto inclui o prefixo "(=)"/"(-)" tal como o Questor imprime (ver
|
|
# `792 - balancete 072026.pdf`).
|
|
LINHA_DRE_RECEITA_LIQUIDA = "(=) RECEITA OPERACIONAL LIQUIDA"
|
|
LINHA_DRE_CUSTOS_TOTAIS = "(-) CUSTOS TOTAIS"
|
|
|
|
|
|
class ExtracaoInvalidaError(Exception):
|
|
"""PDF não tem o formato esperado (cabeçalho/tabela do Balancete/DRE não
|
|
encontrados) — a view converte isso numa resposta 400 genérica, mesmo
|
|
padrão de `planos_saude`/`indicadores`."""
|
|
|
|
|
|
def _para_decimal(texto: str) -> Decimal:
|
|
texto = texto.strip()
|
|
# Algumas empresas têm linhas de DRE negativas no PDF com só o parêntese
|
|
# de abertura ("(1.114.989,29", sem o "()" de fechamento) — confirmado
|
|
# direto nos caracteres brutos do PDF (não é um artefato de
|
|
# _reconstroi_linhas), então o "(" sozinho já basta pra marcar negativo.
|
|
negativo = texto.startswith("(")
|
|
if negativo:
|
|
texto = texto[1:]
|
|
if texto.endswith(")"):
|
|
texto = texto[:-1]
|
|
texto = texto.replace(".", "").replace(",", ".")
|
|
try:
|
|
valor = Decimal(texto)
|
|
except InvalidOperation as exc:
|
|
raise ExtracaoInvalidaError(f"Valor monetário inválido: {texto!r}") from exc
|
|
return -valor if negativo else valor
|
|
|
|
|
|
def _para_data(texto: str) -> date:
|
|
dia, mes, ano = texto.split("/")
|
|
return date(int(ano), int(mes), int(dia))
|
|
|
|
|
|
def _reconstroi_linhas(pagina) -> list[dict]:
|
|
caracteres = [c for c in pagina.chars if c["text"] != " "]
|
|
linhas: dict[float, list] = {}
|
|
for c in caracteres:
|
|
chave = round(c["top"], 0)
|
|
linhas.setdefault(chave, []).append(c)
|
|
|
|
resultado = []
|
|
for topo in sorted(linhas.keys()):
|
|
fileira = sorted(linhas[topo], key=lambda c: c["x0"])
|
|
texto = ""
|
|
anterior_x1 = None
|
|
negrito = False
|
|
for c in fileira:
|
|
if anterior_x1 is not None and (c["x0"] - anterior_x1) > _GAP_ESPACO:
|
|
texto += " "
|
|
texto += c["text"]
|
|
anterior_x1 = c["x1"]
|
|
if "bold" in c["fontname"].lower():
|
|
negrito = True
|
|
resultado.append({"top": topo, "x0": fileira[0]["x0"], "texto": texto, "negrito": negrito})
|
|
return resultado
|
|
|
|
|
|
_RE_CARIMBO_DATA_PAGINA = re.compile(r"^\d{2}/\d{2}/\d{4}\s+\d{2}:\d{2}")
|
|
|
|
|
|
def _extrai_cabecalho(primeiras_linhas: list[str]) -> CabecalhoExtraido:
|
|
texto_completo = " ".join(primeiras_linhas)
|
|
|
|
match_cnpj = _RE_CNPJ.search(texto_completo)
|
|
match_periodo = _RE_PERIODO.search(texto_completo)
|
|
if not match_cnpj or not match_periodo:
|
|
raise ExtracaoInvalidaError("Não foi possível localizar CNPJ/período no cabeçalho do PDF.")
|
|
|
|
# As linhas reconstruídas vêm ordenadas por posição vertical (`top`), não
|
|
# por ordem de leitura — nesta página, o carimbo "data/hora Pág:NNNN" (que
|
|
# fica alinhado à direita) cai *entre* as duas linhas do nome da empresa
|
|
# (que pode quebrar em duas linhas quando é longo). Só as linhas antes de
|
|
# "CNPJ:" que não são esse carimbo nem "Período:" compõem o nome.
|
|
linhas_nome = []
|
|
for linha in primeiras_linhas:
|
|
if linha.startswith("CNPJ:"):
|
|
break
|
|
if _RE_CARIMBO_DATA_PAGINA.match(linha) or linha.startswith("Período:"):
|
|
continue
|
|
linhas_nome.append(linha)
|
|
match_nome = _RE_CODIGO_NOME_EMPRESA.match(" ".join(linhas_nome).strip())
|
|
if not match_nome:
|
|
raise ExtracaoInvalidaError("Não foi possível localizar código/nome da empresa no cabeçalho do PDF.")
|
|
|
|
return CabecalhoExtraido(
|
|
codigo_empresa=match_nome.group("codigo"),
|
|
nome_empresa=match_nome.group("nome").strip(),
|
|
cnpj=match_cnpj.group(1),
|
|
periodo_inicio=_para_data(match_periodo.group(1)),
|
|
periodo_fim=_para_data(match_periodo.group(2)),
|
|
)
|
|
|
|
|
|
def extrai_balancete_dre(origem: str | IO[bytes]) -> ResultadoExtracao:
|
|
"""Lê o PDF combinado de Balancete + DRE (modelo Questor) e devolve tudo
|
|
já estruturado. `origem` aceita tanto um caminho em disco quanto um
|
|
arquivo já aberto em memória (`io.BytesIO`) — a view chama isto direto
|
|
sobre o upload, antes de saber `codigo_empresa`/`competencia` (extraídos
|
|
daqui), então ainda não há um caminho em disco definitivo nesse momento.
|
|
As páginas finais de "Demonstração Mensal (Análise Vertical)", quando
|
|
presentes, são ignoradas de propósito — o Dashboard Contábil calcula
|
|
variação mês a mês a partir do próprio histórico de apurações já
|
|
processadas no Portal, não a partir desse relatório complementar (ver
|
|
CLAUDE.md/plano.md)."""
|
|
|
|
cabecalho: CabecalhoExtraido | None = None
|
|
contas: list[LinhaBalanceteExtraida] = []
|
|
linhas_dre: list[LinhaDreExtraida] = []
|
|
|
|
secao = "cabecalho"
|
|
ordem_dre = 0
|
|
base_x0_dre: float | None = None
|
|
|
|
with pdfplumber.open(origem) as pdf:
|
|
if not pdf.pages:
|
|
raise ExtracaoInvalidaError("O PDF não tem páginas.")
|
|
|
|
primeiras_linhas_pagina1 = [
|
|
linha["texto"] for linha in _reconstroi_linhas(pdf.pages[0]) if linha["texto"].strip()
|
|
][:8]
|
|
cabecalho = _extrai_cabecalho(primeiras_linhas_pagina1)
|
|
|
|
for pagina in pdf.pages:
|
|
for linha in _reconstroi_linhas(pagina):
|
|
texto = linha["texto"].strip()
|
|
if not texto:
|
|
continue
|
|
|
|
if texto.startswith(_TITULO_ANALISE_VERTICAL):
|
|
secao = "ignorar"
|
|
break
|
|
if secao != "dre" and texto.startswith(_TITULO_DRE):
|
|
secao = "dre"
|
|
base_x0_dre = None
|
|
continue
|
|
|
|
if secao in ("cabecalho", "balancete"):
|
|
match = _RE_LINHA_BALANCETE.match(texto)
|
|
if not match:
|
|
continue
|
|
secao = "balancete"
|
|
valores = _RE_MONETARIO.findall(match.group("resto"))
|
|
if len(valores) < 4:
|
|
continue
|
|
saldo_anterior, debito, credito, saldo_atual = valores[-4:]
|
|
# Descrição = tudo antes do início do bloco de 4 valores
|
|
# (sempre exatamente 4 colunas por linha de balancete).
|
|
posicao_valores = match.group("resto").find(saldo_anterior)
|
|
if posicao_valores == -1:
|
|
continue
|
|
descricao = match.group("resto")[:posicao_valores].strip()
|
|
contas.append(
|
|
LinhaBalanceteExtraida(
|
|
conta_numero=int(match.group("conta")),
|
|
codigo=match.group("codigo"),
|
|
descricao=descricao,
|
|
tipo=match.group("tipo") or "A",
|
|
saldo_anterior=_para_decimal(saldo_anterior),
|
|
debito=_para_decimal(debito),
|
|
credito=_para_decimal(credito),
|
|
saldo_atual=_para_decimal(saldo_atual),
|
|
)
|
|
)
|
|
elif secao == "dre":
|
|
if texto.startswith("Valores expressos") or texto.startswith(_TITULO_DRE):
|
|
continue
|
|
valores = _RE_MONETARIO.findall(texto)
|
|
if not valores:
|
|
continue
|
|
valor_texto = valores[-1]
|
|
posicao_valor = texto.rfind(valor_texto)
|
|
descricao = texto[:posicao_valor].strip()
|
|
if not descricao:
|
|
continue
|
|
if base_x0_dre is None:
|
|
base_x0_dre = linha["x0"]
|
|
nivel = max(0, round((linha["x0"] - base_x0_dre) / 7.0))
|
|
ordem_dre += 1
|
|
linhas_dre.append(
|
|
LinhaDreExtraida(
|
|
ordem=ordem_dre,
|
|
descricao=descricao,
|
|
nivel=nivel,
|
|
valor=_para_decimal(valor_texto),
|
|
totalizador=linha["negrito"],
|
|
)
|
|
)
|
|
if secao == "ignorar":
|
|
break
|
|
|
|
if not contas:
|
|
raise ExtracaoInvalidaError("Nenhuma linha de balancete encontrada no PDF.")
|
|
if not linhas_dre:
|
|
raise ExtracaoInvalidaError("Nenhuma linha de DRE encontrada no PDF.")
|
|
|
|
return ResultadoExtracao(cabecalho=cabecalho, contas=contas, linhas_dre=linhas_dre)
|