"""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\d+(?:\.\d+)*)\s+(?P.+)$") _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, )