ewerton.almeida
d8db2a60e5
All checks were successful
continuous-integration/drone/push Build is passing
72 lines
2.3 KiB
Python
Executable File
72 lines
2.3 KiB
Python
Executable File
import os
|
|
import fitz
|
|
import logging
|
|
import re
|
|
from datetime import datetime
|
|
from app.database import AsyncSessionLocal
|
|
from app.models import Fatura, LogProcessamento
|
|
from app.calculos import calcular_campos_dinamicos
|
|
from app.layouts.equatorial_go import extrair_dados as extrair_dados_equatorial
|
|
from sqlalchemy.future import select
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
def extrair_dados_pdf(caminho_pdf):
|
|
try:
|
|
with fitz.open(caminho_pdf) as doc:
|
|
texto_final = ""
|
|
for page in doc:
|
|
blocos = page.get_text("blocks")
|
|
blocos.sort(key=lambda b: (b[1], b[0]))
|
|
for b in blocos:
|
|
texto_final += b[4] + "\n"
|
|
|
|
if not texto_final.strip():
|
|
raise ValueError("PDF não contém texto legível")
|
|
|
|
dados_extraidos = extrair_dados_equatorial(texto_final)
|
|
return dados_extraidos
|
|
|
|
except Exception as e:
|
|
raise ValueError(f"Erro ao processar PDF: {str(e)}")
|
|
|
|
async def nota_ja_existente(nota_fiscal, uc):
|
|
async with AsyncSessionLocal() as session:
|
|
result = await session.execute(
|
|
select(Fatura).filter_by(nota_fiscal=nota_fiscal, unidade_consumidora=uc)
|
|
)
|
|
return result.scalar_one_or_none() is not None
|
|
|
|
async def adicionar_fatura(dados, caminho_pdf):
|
|
async with AsyncSessionLocal() as session:
|
|
try:
|
|
dados_calculados = await calcular_campos_dinamicos(dados, session)
|
|
fatura = Fatura(**dados_calculados)
|
|
fatura.arquivo = os.path.basename(caminho_pdf)
|
|
fatura.link_arquivo = os.path.abspath(caminho_pdf)
|
|
fatura.data_processamento = datetime.now()
|
|
|
|
session.add(fatura)
|
|
|
|
log = LogProcessamento(
|
|
status="PROCESSAMENTO",
|
|
mensagem=f"Fatura adicionada com sucesso: {fatura.nota_fiscal} - {fatura.nome}",
|
|
nome_arquivo=os.path.basename(caminho_pdf)
|
|
)
|
|
session.add(log)
|
|
|
|
await session.commit()
|
|
logger.info(log.mensagem)
|
|
return True
|
|
|
|
except Exception as e:
|
|
logger.error(f"Erro ao adicionar fatura no banco: {e}")
|
|
await session.rollback()
|
|
raise
|
|
|
|
def avaliar_formula(formula: str, contexto: dict):
|
|
try:
|
|
return eval(formula, {}, contexto)
|
|
except Exception as e:
|
|
return None
|