ewerton.almeida/app_faturas
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(dashboard): reorganiza cards, remove indicadores antigos e adiciona 'Valor médio por fatura' junto aos demais; ajusta gráfico mensal para seguir padrão de design

Browse Source
This commit is contained in:
Ewerton Almeida
2025-08-09 16:06:20 -03:00
parent 291eec35a8
commit 7f659a0058
4 changed files with 593 additions and 109 deletions
Download Patch File Download Diff File Expand all files Collapse all files

229
app/main.py
View File

@@ -6,9 +6,10 @@ from fastapi.responses import HTMLResponse, JSONResponse
from fastapi.staticfiles import StaticFiles
import os, shutil
from sqlalchemy import text
from datetime import date
import re
from fastapi.responses import StreamingResponse
from io import BytesIO
import pandas as pd
from app.models import ParametrosFormula
from sqlalchemy.future import select
from app.database import AsyncSessionLocal
@@ -19,14 +20,12 @@ from app.processor import (
status_arquivos,
limpar_arquivos_processados
)
from app.parametros import router as parametros_router
from fastapi.responses import FileResponse
from app.models import Fatura, SelicMensal, ParametrosFormula
from datetime import date
from app.utils import avaliar_formula
app = FastAPI()
templates = Jinja2Templates(directory="app/templates")
app.mount("/static", StaticFiles(directory="app/static"), name="static")
@@ -34,26 +33,215 @@ app.mount("/static", StaticFiles(directory="app/static"), name="static")
UPLOAD_DIR = "uploads/temp"
os.makedirs(UPLOAD_DIR, exist_ok=True)
def _parse_referencia(ref: str):
"""Aceita 'JAN/2024', 'JAN/24', '01/2024', '01/24', '202401'. Retorna (ano, mes)."""
meses = {'JAN':1,'FEV':2,'MAR':3,'ABR':4,'MAI':5,'JUN':6,'JUL':7,'AGO':8,'SET':9,'OUT':10,'NOV':11,'DEZ':12}
ref = (ref or "").strip().upper()
if "/" in ref:
a, b = [p.strip() for p in ref.split("/", 1)]
# mês pode vir 'JAN' ou '01'
mes = meses.get(a, None)
if mes is None:
mes = int(re.sub(r"\D", "", a) or 1)
ano = int(re.sub(r"\D", "", b) or 0)
# ano 2 dígitos -> 2000+
if ano < 100:
ano += 2000
else:
# '202401' ou '2024-01'
num = re.sub(r"\D", "", ref)
if len(num) >= 6:
ano, mes = int(num[:4]), int(num[4:6])
elif len(num) == 4: # '2024'
ano, mes = int(num), 1
else:
ano, mes = date.today().year, 1
return ano, mes
async def _carregar_selic_map(session):
res = await session.execute(text("SELECT ano, mes, percentual FROM faturas.selic_mensal"))
rows = res.mappings().all()
return {(int(r["ano"]), int(r["mes"])): float(r["percentual"]) for r in rows}
def _fator_selic_from_map(selic_map: dict, ano_inicio: int, mes_inicio: int, hoje: date) -> float:
try:
ano, mes = int(ano_inicio), int(mes_inicio)
except Exception:
return 1.0
if ano > hoje.year or (ano == hoje.year and mes > hoje.month):
return 1.0
fator = 1.0
while (ano < hoje.year) or (ano == hoje.year and mes <= hoje.month):
perc = selic_map.get((ano, mes))
if perc is not None:
fator *= (1 + (perc / 100.0))
mes += 1
if mes > 12:
mes = 1
ano += 1
return fator
def _avaliar_formula(texto_formula: str | None, contexto: dict) -> float:
if not texto_formula:
return 0.0
expr = str(texto_formula)
# Substitui nomes de campos por valores numéricos (None -> 0)
for campo, valor in contexto.items():
v = valor
if v is None or v == "":
v = 0
# aceita vírgula como decimal vindo do banco
if isinstance(v, str):
v = v.replace(".", "").replace(",", ".") if re.search(r"[0-9],[0-9]", v) else v
expr = re.sub(rf'\b{re.escape(str(campo))}\b', str(v), expr)
try:
return float(eval(expr, {"__builtins__": {}}, {}))
except Exception:
return 0.0
@app.get("/", response_class=HTMLResponse)
def dashboard(request: Request):
indicadores = [
{"titulo": "Total de Faturas", "valor": 124},
{"titulo": "Faturas com ICMS", "valor": "63%"},
{"titulo": "Valor Total", "valor": "R$ 280.000,00"},
]
async def dashboard(request: Request, cliente: str | None = None):
print("DBG /: inicio", flush=True)
try:
async with AsyncSessionLocal() as session:
print("DBG /: abrindo sessão", flush=True)
analise_stf = {
"antes": {"percentual_com_icms": 80, "media_valor": 1200},
"depois": {"percentual_com_icms": 20, "media_valor": 800},
}
r = await session.execute(text(
"SELECT DISTINCT nome FROM faturas.faturas ORDER BY nome"
))
clientes = [c for c, in r.fetchall()]
print(f"DBG /: clientes={len(clientes)}", flush=True)
# Fórmulas
fp = await session.execute(text("""
SELECT formula FROM faturas.parametros_formula
WHERE nome = 'Cálculo PIS sobre ICMS' AND ativo = TRUE LIMIT 1
"""))
formula_pis = fp.scalar_one_or_none()
fc = await session.execute(text("""
SELECT formula FROM faturas.parametros_formula
WHERE nome = 'Cálculo COFINS sobre ICMS' AND ativo = TRUE LIMIT 1
"""))
formula_cofins = fc.scalar_one_or_none()
print(f"DBG /: tem_formulas pis={bool(formula_pis)} cofins={bool(formula_cofins)}", flush=True)
sql = "SELECT * FROM faturas.faturas"
params = {}
if cliente:
sql += " WHERE nome = :cliente"
params["cliente"] = cliente
print("DBG /: SQL faturas ->", sql, params, flush=True)
ftrs = (await session.execute(text(sql), params)).mappings().all()
print(f"DBG /: total_faturas={len(ftrs)}", flush=True)
# ===== KPIs e Séries para o dashboard =====
from collections import defaultdict
total_faturas = len(ftrs)
qtd_icms_na_base = 0
soma_corrigida = 0.0
hoje = date.today()
selic_map = await _carregar_selic_map(session)
# Séries e somatórios comerciais
serie_mensal = defaultdict(float) # {(ano, mes): valor_corrigido}
sum_por_dist = defaultdict(float) # {"distribuidora": valor_corrigido}
somatorio_v_total = 0.0
contagem_com_icms = 0
for f in ftrs:
ctx = dict(f)
# PIS/COFINS sobre ICMS
v_pis = _avaliar_formula(formula_pis, ctx)
v_cof = _avaliar_formula(formula_cofins, ctx)
v_total = max(0.0, float(v_pis or 0) + float(v_cof or 0))
# % de faturas com ICMS na base
if (v_pis or 0) > 0:
qtd_icms_na_base += 1
contagem_com_icms += 1
# referência -> (ano,mes)
try:
ano, mes = _parse_referencia(f.get("referencia"))
except Exception:
ano, mes = hoje.year, hoje.month
# SELIC
fator = _fator_selic_from_map(selic_map, ano, mes, hoje)
valor_corrigido = v_total * fator
soma_corrigida += valor_corrigido
somatorio_v_total += v_total
# séries
serie_mensal[(ano, mes)] += valor_corrigido
dist = (f.get("distribuidora") or "").strip() or "Não informado"
sum_por_dist[dist] += valor_corrigido
percentual_icms_base = (qtd_icms_na_base / total_faturas * 100.0) if total_faturas else 0.0
valor_restituicao_corrigida = soma_corrigida
valor_medio_com_icms = (somatorio_v_total / contagem_com_icms) if contagem_com_icms else 0.0
# total de clientes (distinct já carregado)
total_clientes = len(clientes)
# Série mensal – últimos 12 meses
ultimos = []
a, m = hoje.year, hoje.month
for _ in range(12):
ultimos.append((a, m))
m -= 1
if m == 0:
m = 12; a -= 1
ultimos.reverse()
serie_mensal_labels = [f"{mes:02d}/{ano}" for (ano, mes) in ultimos]
serie_mensal_valores = [round(serie_mensal.get((ano, mes), 0.0), 2) for (ano, mes) in ultimos]
# Top 5 distribuidoras
top5 = sorted(sum_por_dist.items(), key=lambda kv: kv[1], reverse=True)[:5]
top5_labels = [k for k, _ in top5]
top5_valores = [round(v, 2) for _, v in top5]
print("DBG /: calculos OK", flush=True)
print("DBG /: render template", flush=True)
return templates.TemplateResponse("dashboard.html", {
"request": request,
"clientes": clientes,
"cliente_atual": cliente or "",
"total_faturas": total_faturas,
"valor_restituicao_corrigida": valor_restituicao_corrigida,
"percentual_icms_base": percentual_icms_base,
# Novos dados para o template
"total_clientes": total_clientes,
"valor_medio_com_icms": valor_medio_com_icms,
"situacao_atual_percent": percentual_icms_base, # para gráfico de alerta
"serie_mensal_labels": serie_mensal_labels,
"serie_mensal_valores": serie_mensal_valores,
"top5_labels": top5_labels,
"top5_valores": top5_valores,
})
except Exception as e:
import traceback
print("ERR /:", e, flush=True)
traceback.print_exc()
# Página de erro amigável (sem derrubar servidor)
return HTMLResponse(
f"<pre style='padding:16px;color:#b91c1c;background:#fff1f2'>Falha no dashboard:\n{e}</pre>",
status_code=500
)
return templates.TemplateResponse("dashboard.html", {
"request": request,
"cliente_atual": "",
"clientes": ["Cliente A", "Cliente B"],
"indicadores": indicadores,
"analise_stf": analise_stf
})
@app.get("/upload", response_class=HTMLResponse)
def upload_page(request: Request):
@@ -118,6 +306,7 @@ async def clear_all():
@app.get("/export-excel")
async def export_excel():
import pandas as pd
async with AsyncSessionLocal() as session:
# 1. Coletar faturas e tabela SELIC
faturas_result = await session.execute(select(Fatura))