343 lines
14 KiB
Python
343 lines
14 KiB
Python
"""Extração do PDF de Balancete + DRE gerado pelo Contabit (o segundo sistema
|
|
contábil usado pelo escritório). `parser.extrai_balancete_dre()` detecta o
|
|
leiaute pela primeira página (`eh_leiaute_contabit()`) e despacha pra cá;
|
|
o resultado sai no mesmo `ResultadoExtracao` do Questor, com
|
|
`leiaute="contabit"`.
|
|
|
|
Calibrado contra `1512 - Balancete 082026.pdf` (arquivo de referência, fora
|
|
do repositório, em `Projetos\\Balancetes\\Contabit`). Diferenças de leiaute
|
|
que moldam este módulo:
|
|
|
|
- **Sem código da empresa no PDF.** O cabeçalho só tem nome e CNPJ; o código
|
|
vem do prefixo numérico do nome do arquivo (decisão explícita do usuário).
|
|
- **Balancete: classificação, descrição, nº da conta, 4 valores.** Não existe
|
|
flag S/A: conta sintética é a que não tem nº de conta.
|
|
- **Descrição longa "vaza" por cima das colunas seguintes.** O PDF corta a
|
|
descrição na tela (clip), mas o texto inteiro continua no fluxo e seus
|
|
caracteres ficam na mesma faixa horizontal do nº da conta e até do Saldo
|
|
Anterior. Ordenar por `x0` intercala os dois ("GRAFICOS2 2L9T3D5A" =
|
|
"GRAFICOS LTDA" + "22935"). Por isso o Balancete é lido na **ordem do
|
|
fluxo do PDF** (descrição, depois os valores, depois o nº da conta, cada um
|
|
um trecho contínuo), e só a posição dos 4 valores entre si usa `x0`.
|
|
- **Saldos pela natureza da conta.** Passivo e contas redutoras aparecem
|
|
positivos (não negativos como no Questor); o grupo pai já faz a subtração.
|
|
- **DRE com negativo em "-"** (não entre parênteses) e, nas linhas em negrito,
|
|
o valor desenhado 1pt abaixo do texto — por isso as linhas são agrupadas
|
|
com tolerância vertical, não por `round(top)`.
|
|
- **Demonstração mensal** com meses por extenso ("JUNHO/2026"), AV% sem "%",
|
|
uma coluna TOTAL a mais (soma dos meses mostrados, descartada) e linhas
|
|
espaçadoras só com zeros, sem descrição (ignoradas)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import re
|
|
|
|
from .conversao import CodigoEmpresaAusenteError, ExtracaoInvalidaError, para_data, para_decimal, para_percentual
|
|
from .modelos import (
|
|
LEIAUTE_CONTABIT,
|
|
CabecalhoExtraido,
|
|
LinhaAnaliseVerticalExtraida,
|
|
LinhaBalanceteExtraida,
|
|
LinhaDreExtraida,
|
|
ResultadoExtracao,
|
|
ValorMensalAnaliseVertical,
|
|
)
|
|
|
|
# Tolerância vertical (pt) pra considerar dois caracteres da mesma linha
|
|
# visual. Cobre o valor da linha em negrito da DRE (1pt abaixo do texto) e a
|
|
# linha de AV% da demonstração mensal (1pt abaixo do valor); duas linhas
|
|
# reais de tabela ficam sempre ~11pt uma da outra.
|
|
_TOLERANCIA_LINHA = 1.6
|
|
|
|
# Vão horizontal (pt) acima do qual dois caracteres consecutivos (ordenados
|
|
# por x) viram campos diferentes. Diferente do Questor, aqui o próprio PDF
|
|
# desenha os espaços entre palavras como caractere, então o vão só precisa
|
|
# separar colunas.
|
|
_GAP_CAMPO = 1.0
|
|
|
|
# Leitura em ordem de fluxo (Balancete): um trecho novo começa quando o
|
|
# próximo caractere volta pra trás (além do que o kerning explica, ex. "LT"
|
|
# em "LTDA") ou pula pra frente além de um espaço normal.
|
|
_RECUO_MAXIMO_KERNING = 2.0
|
|
_SALTO_MAXIMO_TRECHO = 4.0
|
|
|
|
# Recuo (pt) de cada nível da árvore. DRE: 67.7 → 75.2 → 82.6 → 90.1;
|
|
# demonstração mensal: 30.7 → 36.7 → 42.6 → 48.6.
|
|
_PASSO_NIVEL_DRE = 7.45
|
|
_PASSO_NIVEL_MENSAL = 5.95
|
|
|
|
_RE_MONETARIO_EXATO = re.compile(r"^-?[\d.]+,\d{2}$")
|
|
_RE_MONETARIO = re.compile(r"-?[\d.]+,\d{2}")
|
|
_RE_CLASSIFICACAO = re.compile(r"^(?P<codigo>\d+(?:\.\d+)*)\s+(?P<descricao>.+)$")
|
|
_RE_NUMERO_CONTA = re.compile(r"^\d+$")
|
|
_RE_CNPJ = re.compile(r"CNPJ:\s*([\d./-]+)")
|
|
_RE_PERIODO = re.compile(r"PER[IÍ]ODO DE\s*(\d{2}/\d{2}/\d{4})\s*[AÀ]\s*(\d{2}/\d{2}/\d{4})", re.IGNORECASE)
|
|
_RE_MES_MENSAL = re.compile(r"([A-ZÇ]+)/(\d{4})")
|
|
_RE_CODIGO_NO_NOME_ARQUIVO = re.compile(r"^\s*(\d+)")
|
|
|
|
_MESES_ABREVIADOS = {
|
|
"JANEIRO": "jan",
|
|
"FEVEREIRO": "fev",
|
|
"MARCO": "mar",
|
|
"MARÇO": "mar",
|
|
"ABRIL": "abr",
|
|
"MAIO": "mai",
|
|
"JUNHO": "jun",
|
|
"JULHO": "jul",
|
|
"AGOSTO": "ago",
|
|
"SETEMBRO": "set",
|
|
"OUTUBRO": "out",
|
|
"NOVEMBRO": "nov",
|
|
"DEZEMBRO": "dez",
|
|
}
|
|
|
|
_SECAO_BALANCETE = "balancete"
|
|
_SECAO_DRE = "dre"
|
|
_SECAO_MENSAL = "mensal"
|
|
|
|
|
|
def _agrupa_linhas(pagina) -> list[list[dict]]:
|
|
"""Caracteres da página agrupados por linha visual, **preservando a ordem
|
|
do fluxo do PDF** dentro de cada linha (é o que `pagina.chars` já
|
|
entrega). Linhas em ordem de `top`."""
|
|
grupos: list[tuple[float, list[dict]]] = []
|
|
for c in sorted(pagina.chars, key=lambda c: c["top"]):
|
|
if grupos and c["top"] - grupos[-1][0] <= _TOLERANCIA_LINHA:
|
|
grupos[-1][1].append(c)
|
|
else:
|
|
grupos.append((c["top"], [c]))
|
|
ordem_fluxo = {id(c): i for i, c in enumerate(pagina.chars)}
|
|
return [sorted(chars, key=lambda c: ordem_fluxo[id(c)]) for _, chars in grupos]
|
|
|
|
|
|
def _texto_por_posicao(chars: list[dict]) -> str:
|
|
"""Texto da linha lido da esquerda pra direita, espaços colapsados."""
|
|
texto = ""
|
|
anterior_x1 = None
|
|
for c in sorted(chars, key=lambda c: c["x0"]):
|
|
if anterior_x1 is not None and c["x0"] - anterior_x1 > _GAP_CAMPO:
|
|
texto += " "
|
|
texto += c["text"]
|
|
anterior_x1 = c["x1"]
|
|
return " ".join(texto.split())
|
|
|
|
|
|
def _trechos_por_fluxo(chars: list[dict]) -> list[tuple[str, float]]:
|
|
"""Trechos contínuos da linha na ordem do fluxo do PDF, cada um com o
|
|
`x0` do primeiro caractere não-espaço (ver docstring do módulo)."""
|
|
trechos: list[tuple[str, float]] = []
|
|
atual = ""
|
|
x0_atual: float | None = None
|
|
anterior = None
|
|
for c in chars:
|
|
if anterior is not None and (
|
|
c["x0"] < anterior["x1"] - _RECUO_MAXIMO_KERNING or c["x0"] - anterior["x1"] > _SALTO_MAXIMO_TRECHO
|
|
):
|
|
if atual.strip():
|
|
trechos.append((" ".join(atual.split()), x0_atual))
|
|
atual, x0_atual = "", None
|
|
atual += c["text"]
|
|
if x0_atual is None and c["text"].strip():
|
|
x0_atual = c["x0"]
|
|
anterior = c
|
|
if atual.strip():
|
|
trechos.append((" ".join(atual.split()), x0_atual))
|
|
return trechos
|
|
|
|
|
|
def _x0_texto(chars: list[dict]) -> float:
|
|
return min(c["x0"] for c in chars if c["text"].strip())
|
|
|
|
|
|
def _negrito(chars: list[dict]) -> bool:
|
|
return any("bold" in c["fontname"].lower() for c in chars if c["text"].strip())
|
|
|
|
|
|
def eh_leiaute_contabit(primeira_pagina) -> bool:
|
|
"""O cabeçalho do Contabit tem "Classificação/Conta" (coluna única) e o
|
|
período no formato "PERÍODO DE ... À ..."; o Questor usa "Período:" e
|
|
colunas separadas de conta/classificação."""
|
|
textos = [_texto_por_posicao(chars) for chars in _agrupa_linhas(primeira_pagina)[:12]]
|
|
return any("Classificação/Conta" in t for t in textos) and any(_RE_PERIODO.search(t) for t in textos)
|
|
|
|
|
|
def codigo_empresa_do_nome_arquivo(nome_arquivo: str | None) -> str:
|
|
"""Prefixo numérico do nome do arquivo ("1512 - Balancete 082026.pdf" →
|
|
"1512"). Sem prefixo numérico, a análise não tem como ser identificada
|
|
(é a chave natural junto da competência) e o upload é recusado."""
|
|
base = os.path.basename(nome_arquivo or "")
|
|
match = _RE_CODIGO_NO_NOME_ARQUIVO.match(base)
|
|
if not match:
|
|
raise CodigoEmpresaAusenteError(
|
|
"No modelo Contabit o código da empresa vem do nome do arquivo, que precisa começar "
|
|
'pelo código (ex.: "1512 - Balancete 082026.pdf").'
|
|
)
|
|
return match.group(1)
|
|
|
|
|
|
def _extrai_cabecalho(primeira_pagina, nome_arquivo: str | None) -> CabecalhoExtraido:
|
|
textos = [_texto_por_posicao(chars) for chars in _agrupa_linhas(primeira_pagina)[:12]]
|
|
texto_completo = " ".join(textos)
|
|
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.")
|
|
|
|
# O nome da empresa é a linha logo acima do "CNPJ:".
|
|
indice_cnpj = next(i for i, t in enumerate(textos) if t.startswith("CNPJ:"))
|
|
nome_empresa = textos[indice_cnpj - 1].strip() if indice_cnpj > 0 else ""
|
|
if not nome_empresa or _RE_PERIODO.search(nome_empresa):
|
|
raise ExtracaoInvalidaError("Não foi possível localizar o nome da empresa no cabeçalho do PDF.")
|
|
|
|
return CabecalhoExtraido(
|
|
codigo_empresa=codigo_empresa_do_nome_arquivo(nome_arquivo),
|
|
nome_empresa=nome_empresa,
|
|
cnpj=match_cnpj.group(1),
|
|
periodo_inicio=para_data(match_periodo.group(1)),
|
|
periodo_fim=para_data(match_periodo.group(2)),
|
|
)
|
|
|
|
|
|
def _secao_da_pagina(pagina) -> str | None:
|
|
textos = [_texto_por_posicao(chars) for chars in _agrupa_linhas(pagina)[:8]]
|
|
cabecalho = " ".join(textos).upper()
|
|
if "CLASSIFICAÇÃO/CONTA" in cabecalho:
|
|
return _SECAO_BALANCETE
|
|
if "RESULTADO ACUMULADO" in cabecalho:
|
|
return _SECAO_DRE
|
|
if "MENSAL" in cabecalho and "DEMONSTRAÇÃO DO RESULTADO" in cabecalho:
|
|
return _SECAO_MENSAL
|
|
return None
|
|
|
|
|
|
def _linha_balancete(chars: list[dict]) -> LinhaBalanceteExtraida | None:
|
|
trechos = _trechos_por_fluxo(chars)
|
|
if not trechos:
|
|
return None
|
|
valores = [(texto, x0) for texto, x0 in trechos if _RE_MONETARIO_EXATO.match(texto)]
|
|
numeros_conta = [texto for texto, _ in trechos if _RE_NUMERO_CONTA.match(texto)]
|
|
descricao_trechos = [
|
|
texto for texto, _ in trechos if not _RE_MONETARIO_EXATO.match(texto) and not _RE_NUMERO_CONTA.match(texto)
|
|
]
|
|
match = _RE_CLASSIFICACAO.match(" ".join(descricao_trechos))
|
|
if not match:
|
|
return None
|
|
if len(valores) != 4 or len(numeros_conta) > 1:
|
|
raise ExtracaoInvalidaError(f"Linha de balancete com colunas inesperadas: {match.group('codigo')}.")
|
|
saldo_anterior, debito, credito, saldo_atual = (texto for texto, _ in sorted(valores, key=lambda v: v[1]))
|
|
return LinhaBalanceteExtraida(
|
|
conta_numero=int(numeros_conta[0]) if numeros_conta else None,
|
|
codigo=match.group("codigo"),
|
|
descricao=match.group("descricao").strip(),
|
|
tipo="A" if numeros_conta else "S",
|
|
saldo_anterior=para_decimal(saldo_anterior),
|
|
debito=para_decimal(debito),
|
|
credito=para_decimal(credito),
|
|
saldo_atual=para_decimal(saldo_atual),
|
|
)
|
|
|
|
|
|
def _separa_descricao_valores(chars: list[dict]) -> tuple[str, list[str]]:
|
|
texto = _texto_por_posicao(chars)
|
|
primeiro = _RE_MONETARIO.search(texto)
|
|
if not primeiro:
|
|
return texto, []
|
|
return texto[: primeiro.start()].strip(), _RE_MONETARIO.findall(texto[primeiro.start():])
|
|
|
|
|
|
def _meses_do_cabecalho(texto: str) -> list[str]:
|
|
meses = []
|
|
for mes, ano in _RE_MES_MENSAL.findall(texto.upper()):
|
|
abreviado = _MESES_ABREVIADOS.get(mes)
|
|
if abreviado is None:
|
|
raise ExtracaoInvalidaError(f"Mês não reconhecido no cabeçalho da demonstração mensal: {mes!r}.")
|
|
meses.append(f"{abreviado}/{ano}")
|
|
return meses
|
|
|
|
|
|
def extrai(pdf, nome_arquivo: str | None) -> ResultadoExtracao:
|
|
"""Lê um PDF Contabit já aberto (`pdfplumber.PDF`) — ver
|
|
`parser.extrai_balancete_dre()`, que abre o arquivo e despacha pra cá."""
|
|
cabecalho = _extrai_cabecalho(pdf.pages[0], nome_arquivo)
|
|
contas: list[LinhaBalanceteExtraida] = []
|
|
linhas_dre: list[LinhaDreExtraida] = []
|
|
linhas_mensal: list[LinhaAnaliseVerticalExtraida] = []
|
|
meses: list[str] | None = None
|
|
base_x0_dre: float | None = None
|
|
base_x0_mensal: float | None = None
|
|
|
|
for pagina in pdf.pages:
|
|
secao = _secao_da_pagina(pagina)
|
|
if secao is None:
|
|
continue
|
|
for chars in _agrupa_linhas(pagina):
|
|
if not any(c["text"].strip() for c in chars):
|
|
continue
|
|
if secao == _SECAO_BALANCETE:
|
|
conta = _linha_balancete(chars)
|
|
if conta is not None:
|
|
contas.append(conta)
|
|
continue
|
|
|
|
if secao == _SECAO_MENSAL and meses is None:
|
|
candidatos = _meses_do_cabecalho(_texto_por_posicao(chars))
|
|
if candidatos:
|
|
meses = candidatos
|
|
continue
|
|
|
|
descricao, valores = _separa_descricao_valores(chars)
|
|
# Linha sem descrição (espaçadora "0,00 0,00", valor solto) ou
|
|
# sem valor (título, cabeçalho, assinatura) não é dado.
|
|
if not descricao or not valores:
|
|
continue
|
|
x0 = _x0_texto(chars)
|
|
if secao == _SECAO_DRE:
|
|
if base_x0_dre is None:
|
|
base_x0_dre = x0
|
|
linhas_dre.append(
|
|
LinhaDreExtraida(
|
|
ordem=len(linhas_dre) + 1,
|
|
descricao=descricao,
|
|
nivel=max(0, round((x0 - base_x0_dre) / _PASSO_NIVEL_DRE)),
|
|
valor=para_decimal(valores[-1]),
|
|
totalizador=_negrito(chars),
|
|
)
|
|
)
|
|
else:
|
|
if meses is None:
|
|
raise ExtracaoInvalidaError("Cabeçalho de meses da demonstração mensal não encontrado.")
|
|
# Pares valor + AV% por mês, mais o par da coluna TOTAL no fim.
|
|
if len(valores) != 2 * (len(meses) + 1):
|
|
raise ExtracaoInvalidaError(f"Linha da demonstração mensal com colunas inesperadas: {descricao!r}.")
|
|
if base_x0_mensal is None:
|
|
base_x0_mensal = x0
|
|
linhas_mensal.append(
|
|
LinhaAnaliseVerticalExtraida(
|
|
ordem=len(linhas_mensal) + 1,
|
|
descricao=descricao,
|
|
nivel=max(0, round((x0 - base_x0_mensal) / _PASSO_NIVEL_MENSAL)),
|
|
totalizador=_negrito(chars),
|
|
valores=[
|
|
ValorMensalAnaliseVertical(
|
|
valor=para_decimal(valores[2 * i]), percentual=para_percentual(valores[2 * i + 1])
|
|
)
|
|
for i in range(len(meses))
|
|
],
|
|
)
|
|
)
|
|
|
|
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 or [],
|
|
linhas_analise_vertical=linhas_mensal,
|
|
leiaute=LEIAUTE_CONTABIT,
|
|
)
|