gustavosantos/GTI_GERENCIADOR_FISCAL
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: reconcile final control panel coverage UX

Browse Source
This commit is contained in:
gti
2026-09-22 23:04:53 -03:00
parent 283749e860
commit ce606b04cc
12 changed files with 312 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
central/app/control_plane.py
+35 -1
View File
@@ -76,6 +76,24 @@ def _fiscal_status(snapshot) -> str:
return "INCOMPLETE"
def _coverage_partition(snapshot) -> dict:
"""Expose only explicit snapshot states and label the remaining count honestly."""
total = int(getattr(snapshot, "total_products", 0) or 0)
complete = int(getattr(snapshot, "fiscally_complete", 0) or 0)
safe_blocked = int(getattr(snapshot, "safe_blocked", 0) or 0)
failed = int(getattr(snapshot, "failed_products", 0) or 0)
explicit = complete + safe_blocked + failed
residual = total - explicit
invalid_counts = min(total, complete, safe_blocked, failed) < 0
overcount = max(0, explicit - total)
return {
"unclassified_products": max(0, residual) if not invalid_counts else 0,
"coverage_overcount_products": overcount,
"coverage_invalid_counts": invalid_counts,
"coverage_invariant_pass": not invalid_counts and overcount == 0,
}
def _technical_timestamp(item):
if item["technical_health"] in {"FAILED", "DEGRADED"}:
return item.get("last_job_at") or item.get("last_seen_at")
@@ -105,6 +123,9 @@ def installation_view(installation, agent, tenant, snapshot=None, *, now=None):
if version_drift == "UNKNOWN":
version_drift = compute_version_drift(agent.agent_version)
fiscal_status = _fiscal_status(snapshot)
coverage_partition = _coverage_partition(snapshot)
version_drift_label = {"CURRENT": "Atualizada", "OUTDATED": "Desatualizada", "UNKNOWN": "Desconhecida"}.get(version_drift, str(version_drift))
agent_version_display = f"{agent.agent_version} · {version_drift_label}" if agent.agent_version else "Desconhecida"
return {
"agent_id": agent.id,
"installation_id": installation.installation_id,
@@ -119,7 +140,9 @@ def installation_view(installation, agent, tenant, snapshot=None, *, now=None):
"company_uf": installation.company_uf,
"company_crt": installation.company_crt,
"agent_version": agent.agent_version,
"agent_version_display": agent_version_display,
"version_drift": version_drift,
"version_drift_label": version_drift_label,
"health": health,
"health_reason": reason,
"technical_health": health,
@@ -134,6 +157,7 @@ def installation_view(installation, agent, tenant, snapshot=None, *, now=None):
"fiscally_complete": snapshot.fiscally_complete,
"safe_blocked": snapshot.safe_blocked,
"failed_products": snapshot.failed_products,
**coverage_partition,
"coverage_percent": snapshot.coverage_percent,
"fiscal_completion_percent": _fiscal_completion(snapshot),
"fiscal_status": fiscal_status,
@@ -362,15 +386,25 @@ def dashboard_summary(session, now=None):
for state, count in health_rows:
counts[state] = int(count)
snapshot_total = func.coalesce(AgentSnapshot.total_products, 0)
snapshot_complete = func.coalesce(AgentSnapshot.fiscally_complete, 0)
snapshot_blocked = func.coalesce(AgentSnapshot.safe_blocked, 0)
snapshot_failed = func.coalesce(AgentSnapshot.failed_products, 0)
snapshot_explicit = snapshot_complete + snapshot_blocked + snapshot_failed
snapshot_has_negative = or_(snapshot_total < 0, snapshot_complete < 0, snapshot_blocked < 0, snapshot_failed < 0)
total_row = session.execute(
select(
func.coalesce(func.sum(AgentSnapshot.total_products), 0),
func.coalesce(func.sum(AgentSnapshot.fiscally_complete), 0),
func.coalesce(func.sum(AgentSnapshot.safe_blocked), 0),
func.coalesce(func.sum(AgentSnapshot.failed_products), 0),
func.coalesce(func.sum(case((and_(~snapshot_has_negative, snapshot_explicit <= snapshot_total), snapshot_total - snapshot_explicit), else_=0)), 0),
func.coalesce(func.sum(case((and_(~snapshot_has_negative, snapshot_explicit > snapshot_total), snapshot_explicit - snapshot_total), else_=0)), 0),
func.coalesce(func.sum(case((snapshot_has_negative, 1), else_=0)), 0),
).select_from(Installation).outerjoin(AgentSnapshot, AgentSnapshot.installation_id == Installation.id)
).one()
totals = dict(zip(("total_products", "fiscally_complete", "safe_blocked", "failed_products"), (int(value or 0) for value in total_row)))
totals = dict(zip(("total_products", "fiscally_complete", "safe_blocked", "failed_products", "unclassified_products", "coverage_overcount_products", "coverage_invalid_snapshots"), (int(value or 0) for value in total_row)))
totals["coverage_invariant_pass"] = totals["coverage_overcount_products"] == 0 and totals["coverage_invalid_snapshots"] == 0
open_filter = BlockedProduct.resolved_at.is_(None)
open_blockers = session.scalar(select(func.count(BlockedProduct.id)).where(open_filter)) or 0