375 lines
18 KiB
Python
375 lines
18 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
|
|
import unicodedata
|
|
from datetime import date
|
|
from decimal import Decimal, InvalidOperation
|
|
from typing import IO
|
|
|
|
import pdfplumber
|
|
|
|
from .modelos import (
|
|
CabecalhoExtraido,
|
|
LinhaAnaliseVerticalExtraida,
|
|
LinhaBalanceteExtraida,
|
|
LinhaDreExtraida,
|
|
ResultadoExtracao,
|
|
ValorMensalAnaliseVertical,
|
|
)
|
|
|
|
# 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)"
|
|
|
|
|
|
def _normaliza_titulo(texto: str) -> str:
|
|
"""Remove acento antes de comparar título de seção — confirmado contra
|
|
`1751 - Balancete 07.2026.pdf` (cliente diferente do PDF de referência
|
|
original): a fonte embutida nesse relatório específico perde o til de
|
|
"ÇÃO" na extração ("DEMONSTRAÇÃO" vira "DEMONSTRAÇAO", só falta o til do
|
|
à — o resto do caractere sai certo, não é um replacement character nem
|
|
texto corrompido), o suficiente pra `texto.startswith(_TITULO_DRE)` nunca
|
|
bater e a seção DRE nunca ser detectada (`ExtracaoInvalidaError: Nenhuma
|
|
linha de DRE encontrada`). Como o relatório é sempre gerado pelo mesmo
|
|
sistema (Questor) mas cada cliente/instalação pode embutir uma fonte
|
|
ligeiramente diferente, comparar sem acento é mais robusto do que
|
|
calibrar um título fixo por cliente — mesmo espírito de `_RE_PERIODO`
|
|
já aceitar "Per[ií]odo" pelo mesmo tipo de variação."""
|
|
sem_acento = unicodedata.normalize("NFKD", texto).encode("ascii", "ignore").decode("ascii")
|
|
return sem_acento.upper()
|
|
|
|
|
|
_TITULO_DRE_NORM = _normaliza_titulo(_TITULO_DRE)
|
|
_TITULO_ANALISE_VERTICAL_NORM = _normaliza_titulo(_TITULO_ANALISE_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>.+)$")
|
|
|
|
# Uma linha da Análise Vertical repete N pares "Valor Variação" (um por mês
|
|
# mostrado) em vez do valor único da DRE. O valor às vezes traz um espaço
|
|
# residual antes do parêntese de fechamento (ex. "(552.338,51 )") — artefato
|
|
# do mesmo gerador de relatório, `Decimal()` já tolera esse espaço; a
|
|
# variação é sempre um percentual com "%" colado ao número. Calibrado contra
|
|
# `792 - balancete 072026.pdf` (mesmo arquivo de referência da DRE/Balancete).
|
|
_TOKEN_VALOR_ANALISE_VERTICAL = r"\(?-?[\d.]+,\d{2}\s?\)?"
|
|
_TOKEN_PERCENTUAL_ANALISE_VERTICAL = r"\(?-?[\d.]+,\d{2}\s?%\s?\)?"
|
|
_RE_PAR_VALOR_VARIACAO = re.compile(
|
|
rf"({_TOKEN_VALOR_ANALISE_VERTICAL})\s+({_TOKEN_PERCENTUAL_ANALISE_VERTICAL})"
|
|
)
|
|
# Cabeçalho de mês da Análise Vertical, ex. "mai - 2026 jun - 2026 jul - 2026"
|
|
# — capturado uma única vez (primeira página da seção), as páginas seguintes
|
|
# repetem o mesmo cabeçalho a cada quebra de página.
|
|
_RE_MES_ANALISE_VERTICAL = re.compile(r"([A-Za-z]{3})\s*-\s*(\d{4})")
|
|
|
|
# Trecho (não o texto exato) da linha "(+/-) Despesas/Receitas Financeiras",
|
|
# usado só pelo cálculo de EBIT/EBITDA em dashboard_contabil.indicadores —
|
|
# diferente das duas constantes acima, ainda não foi calibrado contra o
|
|
# texto exato de um PDF real, por isso casa por substring (ver
|
|
# `_busca_linha_por_trecho` em indicadores.py) em vez de igualdade exata.
|
|
LINHA_DRE_DESPESAS_RECEITAS_FINANCEIRAS = "DESPESAS/RECEITAS FINANCEIRAS"
|
|
|
|
|
|
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_percentual(texto: str) -> Decimal:
|
|
"""Mesma conversão de `_para_decimal`, só removendo o "%" antes — o sinal
|
|
negativo continua expresso só pelo parêntese (ex. "(7,33%)"), igual ao
|
|
valor monetário."""
|
|
return _para_decimal(texto.replace("%", ""))
|
|
|
|
|
|
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, também são extraídas (`linhas_analise_vertical`/
|
|
`meses_analise_vertical`) — é a mesma árvore da DRE, só que com um
|
|
valor+percentual por mês em vez de um valor único; **não** substitui o
|
|
histórico próprio do Portal usado pelas regras de variação mês a mês
|
|
(`regras.py` continua comparando contra apurações anteriores já
|
|
persistidas, não contra esta seção), só alimenta a aba/relatório
|
|
"Análise Vertical" com os mesmos meses que o próprio Questor já calcula."""
|
|
|
|
cabecalho: CabecalhoExtraido | None = None
|
|
contas: list[LinhaBalanceteExtraida] = []
|
|
linhas_dre: list[LinhaDreExtraida] = []
|
|
linhas_analise_vertical: list[LinhaAnaliseVerticalExtraida] = []
|
|
meses_analise_vertical: list[str] | None = None
|
|
|
|
secao = "cabecalho"
|
|
ordem_dre = 0
|
|
ordem_av = 0
|
|
base_x0_dre: float | None = None
|
|
base_x0_av: float | None = None
|
|
# Ver ResultadoExtracao.fonte_pdf_atipica — só vira True se a comparação
|
|
# SEM acento (texto_norm) bateu onde a comparação COM acento (texto) não
|
|
# bateria, ou seja, a normalização foi realmente necessária pra este
|
|
# arquivo, não só uma checagem redundante.
|
|
fonte_pdf_atipica = False
|
|
|
|
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
|
|
# Comparado sem acento (ver _normaliza_titulo) — a fonte
|
|
# embutida varia entre PDFs de clientes/instalações
|
|
# diferentes do Questor e pode perder um til/acento
|
|
# específico sem corromper o resto do texto.
|
|
texto_norm = _normaliza_titulo(texto)
|
|
|
|
if texto_norm.startswith(_TITULO_ANALISE_VERTICAL_NORM):
|
|
if not texto.startswith(_TITULO_ANALISE_VERTICAL):
|
|
fonte_pdf_atipica = True
|
|
if secao != "analise_vertical":
|
|
secao = "analise_vertical"
|
|
base_x0_av = None
|
|
continue
|
|
# Guarda só contra cabeçalho/balancete pra não disparar de
|
|
# novo quando o título da DRE se repete no topo de cada
|
|
# página já dentro da seção "dre" (comportamento de sempre)
|
|
# ou da Análise Vertical, que também reimprime esse título —
|
|
# sem essa restrição, `secao` voltaria de "analise_vertical"
|
|
# pra "dre" a cada quebra de página da Análise Vertical.
|
|
if secao in ("cabecalho", "balancete") and texto_norm.startswith(_TITULO_DRE_NORM):
|
|
if not texto.startswith(_TITULO_DRE):
|
|
fonte_pdf_atipica = True
|
|
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_norm.startswith(_TITULO_DRE_NORM):
|
|
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"],
|
|
)
|
|
)
|
|
elif secao == "analise_vertical":
|
|
# O cabeçalho de mês ("mai - 2026 jun - 2026 jul - 2026")
|
|
# só existe uma vez de verdade — repete a cada quebra de
|
|
# página, mas só a primeira ocorrência importa (captura
|
|
# os meses e não é tratada como linha de dado, já que
|
|
# não tem par valor+variação nenhum). Qualquer outra
|
|
# linha sem nenhum par (título repetido, cabeçalho de
|
|
# coluna "Descrição/Valor/Variação") é ignorada do mesmo
|
|
# jeito que a DRE ignora linhas sem valor monetário.
|
|
if meses_analise_vertical is None:
|
|
candidatos_mes = _RE_MES_ANALISE_VERTICAL.findall(texto)
|
|
if candidatos_mes:
|
|
meses_analise_vertical = [
|
|
f"{mes.lower()}/{ano}" for mes, ano in candidatos_mes
|
|
]
|
|
continue
|
|
pares = _RE_PAR_VALOR_VARIACAO.findall(texto)
|
|
if not pares:
|
|
continue
|
|
posicao = texto.find(pares[0][0])
|
|
if posicao == -1:
|
|
continue
|
|
descricao = texto[:posicao].strip()
|
|
if not descricao:
|
|
continue
|
|
if base_x0_av is None:
|
|
base_x0_av = linha["x0"]
|
|
nivel = max(0, round((linha["x0"] - base_x0_av) / 7.0))
|
|
ordem_av += 1
|
|
linhas_analise_vertical.append(
|
|
LinhaAnaliseVerticalExtraida(
|
|
ordem=ordem_av,
|
|
descricao=descricao,
|
|
nivel=nivel,
|
|
totalizador=linha["negrito"],
|
|
valores=[
|
|
ValorMensalAnaliseVertical(
|
|
valor=_para_decimal(valor_texto), percentual=_para_percentual(percentual_texto)
|
|
)
|
|
for valor_texto, percentual_texto in pares
|
|
],
|
|
)
|
|
)
|
|
|
|
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,
|
|
meses_analise_vertical=meses_analise_vertical or [],
|
|
linhas_analise_vertical=linhas_analise_vertical,
|
|
fonte_pdf_atipica=fonte_pdf_atipica,
|
|
)
|