Correção arredondamento dasalíquotas e valor taxa
All checks were successful
continuous-integration/drone/push Build is passing

This commit is contained in:
2025-08-14 08:44:41 -03:00
parent f6c8943d4e
commit 98c6cf2363
3 changed files with 82 additions and 23 deletions

View File

@@ -31,6 +31,7 @@ from fastapi import Query
from sqlalchemy import select as sqla_select from sqlalchemy import select as sqla_select
from app.models import AliquotaUF from app.models import AliquotaUF
import pandas as pd import pandas as pd
from fastapi.responses import Response
app = FastAPI() app = FastAPI()
@@ -499,31 +500,54 @@ async def export_excel(
cols = [c for c in df.columns if c != "Arquivo PDF"] + ["Arquivo PDF"] cols = [c for c in df.columns if c != "Arquivo PDF"] + ["Arquivo PDF"]
df = df[cols] df = df[cols]
# garante que colunas percentuais estejam numéricas (se existirem) # converte colunas numéricas (percentuais, R$, etc.)
for col in ["ICMS (%)", "ICMS (%) (UF/Ref)", "Dif. ICMS (pp)", "PIS (%)", "COFINS (%)"]: percent_cols = ["ICMS (%)", "ICMS (%) (UF/Ref)", "Dif. ICMS (pp)", "PIS (%)", "COFINS (%)"]
money_cols = ["Valor Total", "ICMS (R$)", "PIS (R$)", "COFINS (R$)",
"Base ICMS (R$)", "Base PIS (R$)", "Base COFINS (R$)"]
other_dec6 = ["Tarifa", "Consumo (kWh)"]
from decimal import Decimal
for col in percent_cols + money_cols + other_dec6:
if col in df.columns: if col in df.columns:
df[col] = df[col].map(lambda x: float(x) if isinstance(x, Decimal) else x)
df[col] = pd.to_numeric(df[col], errors="coerce") df[col] = pd.to_numeric(df[col], errors="coerce")
# --- gera o XLSX ---
with pd.ExcelWriter(output, engine="xlsxwriter") as writer: with pd.ExcelWriter(output, engine="xlsxwriter") as writer:
df.to_excel(writer, index=False, sheet_name="Relatório") df.to_excel(writer, index=False, sheet_name="Relatório")
# aplica formatação 6 casas decimais
wb = writer.book wb = writer.book
ws = writer.sheets["Relatório"] ws = writer.sheets["Relatório"]
fmt_4dec = wb.add_format({"num_format": "0.000000"})
for col in ["ICMS (%)", "ICMS (%) (UF/Ref)", "Dif. ICMS (pp)", "PIS (%)", "COFINS (%)", "Consumo (kWh)"]: fmt_dec6 = wb.add_format({"num_format": "0.000000"})
fmt_money6 = wb.add_format({"num_format": "#,##0.000000"})
fmt_money2 = wb.add_format({"num_format": "#,##0.00"})
for col in percent_cols:
if col in df.columns: if col in df.columns:
i = df.columns.get_loc(col) i = df.columns.get_loc(col)
# largura automática básica + formato; ajuste a largura se quiser (ex.: 12) ws.set_column(i, i, 14, fmt_dec6)
ws.set_column(i, i, None, fmt_4dec)
for col in money_cols:
if col in df.columns:
i = df.columns.get_loc(col)
ws.set_column(i, i, 14, fmt_money6) # ou fmt_money2 se quiser 2 casas
for col in other_dec6:
if col in df.columns:
i = df.columns.get_loc(col)
ws.set_column(i, i, 14, fmt_dec6)
# IMPORTANTE: só aqui, FORA do with
output.seek(0) output.seek(0)
data = output.getvalue()
return StreamingResponse( return Response(
output, content=data,
media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
headers={"Content-Disposition": f'attachment; filename="{filename}"'} headers={
"Content-Disposition": f'attachment; filename="{filename}"',
"Content-Length": str(len(data)),
},
) )

View File

@@ -26,22 +26,22 @@ class Fatura(Base):
classificacao_tarifaria = Column("classificacao_tarifaria", String) classificacao_tarifaria = Column("classificacao_tarifaria", String)
unidade_consumidora = Column("unidade_consumidora", String) unidade_consumidora = Column("unidade_consumidora", String)
referencia = Column(String) referencia = Column(String)
valor_total = Column(Float) valor_total = Column(Numeric(18, 6, asdecimal=True))
pis_aliq = Column("pis_aliq", Float) pis_aliq = Column(Numeric(8, 6, asdecimal=True))
pis_valor = Column("pis_valor", Float) pis_valor = Column(Numeric(18, 6, asdecimal=True))
pis_base = Column("pis_base", Float) pis_base = Column(Numeric(18, 6, asdecimal=True))
icms_aliq = Column("icms_aliq", Float) icms_aliq = Column(Numeric(8, 6, asdecimal=True))
icms_valor = Column("icms_valor", Float) icms_valor = Column(Numeric(18, 6, asdecimal=True))
icms_base = Column("icms_base", Float) icms_base = Column(Numeric(18, 6, asdecimal=True))
cofins_aliq = Column("cofins_aliq", Float) cofins_aliq = Column(Numeric(8, 6, asdecimal=True))
cofins_valor = Column("cofins_valor", Float) cofins_valor = Column(Numeric(18, 6, asdecimal=True))
cofins_base = Column("cofins_base", Float) cofins_base = Column(Numeric(18, 6, asdecimal=True))
consumo = Column("consumo", Float) consumo = Column(Numeric(14, 6, asdecimal=True))
tarifa = Column("tarifa", Float) tarifa = Column(Numeric(12, 6, asdecimal=True))
nota_fiscal = Column(String) nota_fiscal = Column(String)
data_processamento = Column(DateTime, default=datetime.utcnow) data_processamento = Column(DateTime, default=datetime.utcnow)

View File

@@ -64,6 +64,41 @@ async def process_single_file(caminho_pdf_temp: str, nome_original: str, cliente
dados = extrair_dados_pdf(caminho_pdf_temp) dados = extrair_dados_pdf(caminho_pdf_temp)
dados['arquivo_pdf'] = nome_original dados['arquivo_pdf'] = nome_original
from decimal import Decimal, ROUND_HALF_UP
_Q6 = Decimal("0.000000")
def _to_percent_6(x):
"""Converte para percent (se vier em fração) e quantiza em 6 casas."""
if x is None:
return None
try:
v = Decimal(str(x))
except Exception:
return None
# se vier em fração (ex.: 0.012872), vira 1.2872… (percentual)
if Decimal("0") < v <= Decimal("1"):
v = v * Decimal("100")
return v.quantize(_Q6, rounding=ROUND_HALF_UP)
def _to_dec6(x):
"""Apenas 6 casas, sem % (use para tarifa, bases, etc.)."""
if x is None:
return None
try:
v = Decimal(str(x))
except Exception:
return None
return v.quantize(_Q6, rounding=ROUND_HALF_UP)
dados['icms_aliq'] = _to_percent_6(dados.get('icms_aliq'))
dados['pis_aliq'] = _to_percent_6(dados.get('pis_aliq'))
dados['cofins_aliq'] = _to_percent_6(dados.get('cofins_aliq'))
# tarifa NÃO é percentual: apenas 6 casas
dados['tarifa'] = _to_dec6(dados.get('tarifa'))
# Verifica se a fatura já existe # Verifica se a fatura já existe
existente_result = await session.execute( existente_result = await session.execute(
select(Fatura).filter_by( select(Fatura).filter_by(