diff --git a/app/main.py b/app/main.py index 4df182b..7f531e7 100644 --- a/app/main.py +++ b/app/main.py @@ -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"
Falha no dashboard:\n{e}",
+ 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))
diff --git a/app/routes/dashboard_unused.py b/app/routes/dashboard_unused.py
new file mode 100644
index 0000000..6bcb44b
--- /dev/null
+++ b/app/routes/dashboard_unused.py
@@ -0,0 +1,140 @@
+from fastapi import APIRouter, Request
+from fastapi.templating import Jinja2Templates
+from sqlalchemy import create_engine, text
+import os
+from datetime import date
+
+# Usa o avaliador de fórmulas já existente
+from app.utils import avaliar_formula
+
+router = APIRouter()
+templates = Jinja2Templates(directory="app/templates")
+
+# Conexão com o banco (use a mesma DATABASE_URL do restante do app)
+DATABASE_URL = os.getenv("DATABASE_URL")
+engine = create_engine(DATABASE_URL)
+
+def _parse_referencia(ref: str):
+ """Aceita 'JAN/2024', '01/2024' ou '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 = ref.split("/")
+ if a.isdigit():
+ mes, ano = int(a), int(b)
+ else:
+ mes, ano = meses.get(a, 1), int(b)
+ else:
+ ano, mes = int(ref[:4]), int(ref[4:]) if len(ref) >= 6 else 1
+ return ano, mes
+
+def _fator_selic_acumulado(conn, ano_inicio, mes_inicio, hoje):
+ selic = conn.execute(text("""
+ SELECT ano, mes, percentual
+ FROM faturas.selic_mensal
+ """)).mappings().all()
+ selic_map = {(r["ano"], r["mes"]): float(r["percentual"]) for r in selic}
+
+ fator = 1.0
+ ano, mes = int(ano_inicio), int(mes_inicio)
+ 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
+
+@router.get("/dashboard")
+def dashboard(request: Request, cliente: str | None = None):
+ with engine.begin() as conn:
+ # Lista de clientes (distinct nome)
+ clientes = [r[0] for r in conn.execute(text("""
+ SELECT DISTINCT nome FROM faturas.faturas ORDER BY nome
+ """)).fetchall()]
+
+ # Carrega fórmulas (ativas)
+ formula_pis = conn.execute(text("""
+ SELECT formula FROM faturas.parametros_formula
+ WHERE nome = 'Cálculo PIS sobre ICMS' AND ativo = TRUE
+ LIMIT 1
+ """)).scalar_one_or_none()
+
+ formula_cofins = conn.execute(text("""
+ SELECT formula FROM faturas.parametros_formula
+ WHERE nome = 'Cálculo COFINS sobre ICMS' AND ativo = TRUE
+ LIMIT 1
+ """)).scalar_one_or_none()
+
+ # Carrega faturas (com filtro opcional de cliente)
+ params = {}
+ sql = "SELECT * FROM faturas.faturas"
+ if cliente:
+ sql += " WHERE nome = :cliente"
+ params["cliente"] = cliente
+
+ faturas = conn.execute(text(sql), params).mappings().all()
+
+
+ total_faturas = len(faturas)
+
+ # Cálculos de restituição e % ICMS na base
+ hoje = date.today()
+ soma_corrigida = 0.0
+ qtd_icms_na_base = 0
+
+ for f in faturas:
+ contexto = dict(f) # usa colunas como variáveis da fórmula
+ # PIS sobre ICMS
+ v_pis_icms = avaliar_formula(formula_pis, contexto) if formula_pis else None
+ # COFINS sobre ICMS
+ v_cofins_icms = avaliar_formula(formula_cofins, contexto) if formula_cofins else None
+
+ # Contagem para % ICMS na base: considera PIS_sobre_ICMS > 0
+ if v_pis_icms and float(v_pis_icms) > 0:
+ qtd_icms_na_base += 1
+
+ # Corrigir pela SELIC desde a referência da fatura
+ try:
+ ano, mes = _parse_referencia(f.get("referencia"))
+ fator = _fator_selic_acumulado(conn, ano, mes, hoje)
+ except Exception:
+ fator = 1.0
+
+ valor_bruto = (float(v_pis_icms) if v_pis_icms else 0.0) + (float(v_cofins_icms) if v_cofins_icms else 0.0)
+ soma_corrigida += valor_bruto * fator
+
+ percentual_icms_base = (qtd_icms_na_base / total_faturas * 100.0) if total_faturas else 0.0
+ valor_restituicao_corrigida = soma_corrigida
+
+ # --- Análise STF (mantida) ---
+ def media_percentual_icms(inicio: str, fim: str):
+ # Aproximação: base PIS = base ICMS => configurado como proxy “com ICMS na base”
+ q = text(f"""
+ SELECT
+ ROUND(AVG(CASE WHEN icms_base IS NOT NULL AND pis_base = icms_base THEN 100.0 ELSE 0.0 END), 2) AS percentual_com_icms,
+ ROUND(AVG(COALESCE(pis_valor,0) + COALESCE(cofins_valor,0)), 2) AS media_valor
+ FROM faturas.faturas
+ WHERE data_processamento::date BETWEEN :inicio AND :fim
+ { "AND nome = :cliente" if cliente else "" }
+ """)
+ params = {"inicio": inicio, "fim": fim}
+ if cliente: params["cliente"] = cliente
+ r = conn.execute(q, params).mappings().first() or {}
+ return {"percentual_com_icms": r.get("percentual_com_icms", 0), "media_valor": r.get("media_valor", 0)}
+
+ analise_stf = {
+ "antes": media_percentual_icms("2000-01-01", "2017-03-15"),
+ "depois": media_percentual_icms("2017-03-16", "2099-12-31")
+ }
+
+ 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,
+ "analise_stf": analise_stf
+ })
diff --git a/app/templates/dashboard.html b/app/templates/dashboard.html
index e67eaa4..0332d4f 100755
--- a/app/templates/dashboard.html
+++ b/app/templates/dashboard.html
@@ -2,106 +2,254 @@
{% block title %}Dashboard{% endblock %}
{% block content %}
-