"""Bounded control-panel queries backed by installation snapshots.""" from __future__ import annotations from datetime import datetime, timedelta, timezone from sqlalchemy import and_, case, func, or_, select from .config import settings from .events import list_events from .installation_identity import compute_agent_health, compute_version_drift from .models import Agent, AgentEvent, AgentSnapshot, AuditEvent, BlockedProduct, Installation, Job, Tenant from .panel_presentation import format_age, format_duration, format_ui_datetime, health_label, operational_priority HEALTH_STATES = ("ONLINE", "DEGRADED", "OFFLINE", "FAILED", "UNKNOWN") BLOCKER_FIELDS = ("CFOP", "ST", "ICMS", "CSOSN", "NCM", "OTHER") ATTENTION_CANDIDATE_LIMIT = 10 def _page(page: int, page_size: int): return max(1, page), min(100, max(1, page_size)) def _iso(value): return value.isoformat() if value else None def _utc_now(value=None): value = value or datetime.now(timezone.utc) return value.replace(tzinfo=timezone.utc) if value.tzinfo is None else value def _health_expression(now): offline_after = settings.agent_offline_after_seconds or settings.agent_heartbeat_interval_seconds * 3 cutoff = _utc_now(now) - timedelta(seconds=offline_after) return case( (Installation.last_seen_at.is_(None), "UNKNOWN"), (or_(AgentSnapshot.agent_health == "FAILED", Installation.last_status == "FAILED"), "FAILED"), (Installation.last_seen_at < cutoff, "OFFLINE"), (or_(AgentSnapshot.failed_products > 0, AgentSnapshot.last_job_status == "FAILED"), "DEGRADED"), else_="ONLINE", ) def _fiscal_completion_expression(): total = func.coalesce(AgentSnapshot.total_products, 0) complete = func.coalesce(AgentSnapshot.fiscally_complete, 0) raw = case((total > 0, complete * 100.0 / total), else_=func.coalesce(AgentSnapshot.coverage_percent, 0)) return func.round(case((raw < 0, 0), (raw > 100, 100), else_=raw), 1) def _fiscal_completion(snapshot) -> float: total = int(getattr(snapshot, "total_products", 0) or 0) complete = int(getattr(snapshot, "fiscally_complete", 0) or 0) if total: return round(max(0, min(total, complete)) * 100 / total, 1) return round(max(0, min(100, float(getattr(snapshot, "coverage_percent", 0) or 0))), 1) def _fiscal_status(snapshot) -> str: if int(getattr(snapshot, "safe_blocked", 0) or 0): return "BLOCKED" total = int(getattr(snapshot, "total_products", 0) or 0) complete = int(getattr(snapshot, "fiscally_complete", 0) or 0) if total and complete >= total: return "COMPLETE" return "INCOMPLETE" def _technical_timestamp(item): if item["technical_health"] in {"FAILED", "DEGRADED"}: return item.get("last_job_at") or item.get("last_seen_at") return item.get("last_seen_at") def _attention_item(item, problem, timestamp, *, rank, link=None, now=None): timestamp_value = datetime.fromisoformat(timestamp) if isinstance(timestamp, str) else timestamp return { "client": item.get("client"), "display_name": item.get("display_name"), "hostname": item.get("hostname"), "problem": problem, "timestamp": timestamp if isinstance(timestamp, str) else _iso(timestamp_value), "timestamp_display": format_ui_datetime(timestamp_value), "age": format_age(timestamp_value, now), "link": link or f"/agents/{item['agent_id']}", "technical_health": item["technical_health"], "rank": rank, } def installation_view(installation, agent, tenant, snapshot=None, *, now=None): snapshot = snapshot or AgentSnapshot(installation_id=installation.id) health, reason = compute_agent_health(installation, now=now, supplemental=snapshot) version_drift = snapshot.version_drift if version_drift == "UNKNOWN": version_drift = compute_version_drift(agent.agent_version) fiscal_status = _fiscal_status(snapshot) return { "agent_id": agent.id, "installation_id": installation.installation_id, "installation_db_id": installation.id, "client": tenant.name if tenant else None, "tenant_id": installation.tenant_id, "display_name": installation.display_name, "hostname": installation.hostname, "company_name": installation.company_name, "company_code": installation.company_code, "company_cnpj": installation.company_cnpj_masked, "company_uf": installation.company_uf, "company_crt": installation.company_crt, "agent_version": agent.agent_version, "version_drift": version_drift, "health": health, "health_reason": reason, "technical_health": health, "health_label": health_label(health), "technical_health_label": health_label(health), "last_seen_at": _iso(installation.last_seen_at), "last_seen_display": format_ui_datetime(installation.last_seen_at), "last_seen_age": format_age(installation.last_seen_at, now), "started_at": _iso(installation.started_at), "last_status": installation.last_status, "total_products": snapshot.total_products, "fiscally_complete": snapshot.fiscally_complete, "safe_blocked": snapshot.safe_blocked, "failed_products": snapshot.failed_products, "coverage_percent": snapshot.coverage_percent, "fiscal_completion_percent": _fiscal_completion(snapshot), "fiscal_status": fiscal_status, "fiscal_status_label": fiscal_status.title(), "blockers": snapshot.blockers or {}, "last_job_status": snapshot.last_job_status, "last_job_at": _iso(snapshot.last_job_at), "last_successful_job_at": _iso(snapshot.last_successful_job_at), "firebird_version": installation.firebird_version, "firebird_ods": installation.firebird_ods, "database_identifier": installation.database_identifier, "capabilities": installation.capabilities or {}, "fiscal_engine_version": installation.fiscal_engine_version, "dataset_versions": installation.dataset_versions or {}, "technical_ids": {"agent_id": agent.id, "installation_db_id": installation.id, "tenant_id": installation.tenant_id}, "link": f"/agents/{agent.id}", } def _installation_query_filters(query, *, client=None, version=None, search=None, last_seen_before=None): if client: query = query.where(or_(Tenant.name.ilike(f"%{client}%"), Installation.customer_code.ilike(f"%{client}%"))) if version: query = query.where(Agent.agent_version == version) if search: query = query.where(or_(Installation.display_name.ilike(f"%{search}%"), Installation.hostname.ilike(f"%{search}%"), Tenant.name.ilike(f"%{search}%"))) if last_seen_before: query = query.where(or_(Installation.last_seen_at.is_(None), Installation.last_seen_at < last_seen_before)) return query def list_installations(session, *, page=1, page_size=50, client=None, health=None, version=None, search=None, last_seen_before=None, sort="display_name", now=None): page, page_size = _page(page, page_size) now = _utc_now(now) query = select(Installation, Agent, Tenant, AgentSnapshot).join(Agent, Agent.id == Installation.agent_id).join(Tenant, Tenant.id == Installation.tenant_id).outerjoin(AgentSnapshot, AgentSnapshot.installation_id == Installation.id) query = _installation_query_filters(query, client=client, version=version, search=search, last_seen_before=last_seen_before) health_expression = _health_expression(now) if health: query = query.where(health_expression == health) if sort in {"worst_coverage", "coverage", "coverage_asc", "fiscal_completion_percent"}: ordered = query.order_by(_fiscal_completion_expression().asc(), Installation.display_name.asc(), Installation.id.asc()) else: ordered = query.order_by(Installation.display_name.asc(), Installation.id.asc()) rows = session.execute(ordered.offset((page - 1) * page_size).limit(page_size)).all() items = [installation_view(i, a, t, s, now=now) for i, a, t, s in rows] count_query = select(func.count()).select_from(query.with_only_columns(Installation.id).order_by(None).subquery()) total = session.scalar(count_query) or 0 return {"page": page, "page_size": page_size, "items": items, "total": total, "health_filter_applied": health} def _blocker_item(row, tenant, installation=None, agent=None, *, now=None): return { "id": row.id, "tenant_id": row.tenant_id, "client": tenant.name if tenant else row.company, "product_code": row.product_code, "description": row.description, "field": row.field, "blocker_type": row.blocker_type, "reason": row.reason_code, "reason_code": row.reason_code, "evidence_summary": row.evidence_summary, "first_seen_at": _iso(row.first_seen_at), "last_seen_at": _iso(row.last_seen_at), "first_seen_display": format_ui_datetime(row.first_seen_at), "last_seen_display": format_ui_datetime(row.last_seen_at), "age": format_age(row.last_seen_at, now), "retry_count": row.retry_count, "resolved_at": _iso(row.resolved_at), "resolution_source": row.resolution_source, "priority": operational_priority(row), "installation_id": installation.installation_id if installation else None, "installation_db_id": installation.id if installation else None, "display_name": installation.display_name if installation else None, "hostname": installation.hostname if installation else None, "agent_id": agent.id if agent else None, "link": f"/blockers/{row.id}", } def _parse_age(value): if value is None: return None if isinstance(value, (int, float)): return timedelta(days=float(value)) text = str(value).strip().lower() try: if text.endswith("h"): return timedelta(hours=float(text[:-1])) if text.endswith("d"): return timedelta(days=float(text[:-1])) return timedelta(days=float(text)) except ValueError: return None def _blockers_query(session, *, tenant_id=None, agent_id=None, blocker_type=None, field=None, status="OPEN", client=None, installation=None, search=None, age=None): query = select(BlockedProduct, Tenant, Installation, Agent).join(Tenant, Tenant.id == BlockedProduct.tenant_id).outerjoin(Installation, Installation.id == BlockedProduct.installation_id).outerjoin(Agent, Agent.id == Installation.agent_id) status = (status or "OPEN").upper() if status == "OPEN": query = query.where(BlockedProduct.resolved_at.is_(None)) elif status in {"RESOLVED", "CLOSED"}: query = query.where(BlockedProduct.resolved_at.is_not(None)) if tenant_id is not None: query = query.where(BlockedProduct.tenant_id == tenant_id) if agent_id is not None: query = query.where(Agent.id == agent_id) if blocker_type: query = query.where(BlockedProduct.blocker_type == blocker_type) if field: query = query.where(BlockedProduct.field == field) if client: query = query.where(or_(Tenant.name.ilike(f"%{client}%"), BlockedProduct.company.ilike(f"%{client}%"))) if installation: query = query.where(or_(Installation.installation_id.ilike(f"%{installation}%"), Installation.display_name.ilike(f"%{installation}%"), Installation.hostname.ilike(f"%{installation}%"))) if search: term = f"%{search}%" query = query.where(or_(BlockedProduct.product_code.ilike(term), BlockedProduct.description.ilike(term), BlockedProduct.field.ilike(term), BlockedProduct.blocker_type.ilike(term), BlockedProduct.reason_code.ilike(term), Tenant.name.ilike(term))) age_delta = _parse_age(age) if age_delta: query = query.where(BlockedProduct.last_seen_at <= datetime.now(timezone.utc) - age_delta) return query def list_blockers(session, *, tenant_id=None, agent_id=None, page=1, page_size=50, blocker_type=None, field=None, resolved=False, client=None, installation=None, status="OPEN", age=None, search=None): page, page_size = _page(page, page_size) if resolved and status == "OPEN": status = "RESOLVED" query = _blockers_query(session, tenant_id=tenant_id, agent_id=agent_id, blocker_type=blocker_type, field=field, status=status, client=client, installation=installation, search=search, age=age) ordered = query.order_by(BlockedProduct.last_seen_at.desc(), BlockedProduct.id.desc()) rows = session.execute(ordered.offset((page - 1) * page_size).limit(page_size)).all() items = [_blocker_item(blocker, tenant, installation_row, agent, now=datetime.now(timezone.utc)) for blocker, tenant, installation_row, agent in rows] count_query = query.with_only_columns(func.count(BlockedProduct.id)).order_by(None) total = session.scalar(count_query) or 0 return {"page": page, "page_size": page_size, "items": items, "total": total} def _job_item(row, agent, installation, tenant, *, now=None): return { "job_id": row.id, "agent_id": row.agent_id, "client": tenant.name if tenant else None, "display_name": installation.display_name if installation else None, "hostname": installation.hostname if installation else None, "kind": row.kind, "status": row.status, "created_at": _iso(row.created_at), "started_at": _iso(row.started_at), "finished_at": _iso(row.finished_at), "created_display": format_ui_datetime(row.created_at), "started_display": format_ui_datetime(row.started_at), "finished_display": format_ui_datetime(row.finished_at), "duration": format_duration(row.started_at, row.finished_at, now), "attempts": row.attempts, "result_summary": row.result_summary, "error": row.error, } def list_jobs(session, *, agent_id=None, page=1, page_size=50, status=None, kind=None, since=None, until=None, client=None, search=None): page, page_size = _page(page, page_size) query = select(Job, Agent, Installation, Tenant).join(Agent, Agent.id == Job.agent_id).join(Installation, Installation.agent_id == Agent.id).join(Tenant, Tenant.id == Installation.tenant_id) if agent_id is not None: query = query.where(Job.agent_id == agent_id) if status: query = query.where(Job.status == status) if kind: query = query.where(Job.kind == kind) if since: query = query.where(Job.created_at >= since) if until: query = query.where(Job.created_at <= until) if client: query = query.where(or_(Tenant.name.ilike(f"%{client}%"), Installation.customer_code.ilike(f"%{client}%"))) if search: term = f"%{search}%" query = query.where(or_(Tenant.name.ilike(term), Installation.display_name.ilike(term), Installation.hostname.ilike(term), Job.kind.ilike(term), Job.status.ilike(term))) ordered = query.order_by(Job.created_at.desc(), Job.id.desc()) rows = session.execute(ordered.offset((page - 1) * page_size).limit(page_size)).all() count_query = query.with_only_columns(func.count(Job.id)).order_by(None) total = session.scalar(count_query) or 0 return {"page": page, "page_size": page_size, "items": [_job_item(row, agent, installation, tenant) for row, agent, installation, tenant in rows], "total": total} def dashboard_summary(session, now=None): now = _utc_now(now) installation_query = select(Installation, Agent, Tenant, AgentSnapshot).join(Agent, Agent.id == Installation.agent_id).join(Tenant, Tenant.id == Installation.tenant_id).outerjoin(AgentSnapshot, AgentSnapshot.installation_id == Installation.id) health_expression = _health_expression(now) health_rows = session.execute( select(health_expression.label("health"), func.count(Installation.id)) .select_from(Installation) .join(Agent, Agent.id == Installation.agent_id) .join(Tenant, Tenant.id == Installation.tenant_id) .outerjoin(AgentSnapshot, AgentSnapshot.installation_id == Installation.id) .group_by(health_expression) ).all() counts = {state: 0 for state in HEALTH_STATES} for state, count in health_rows: counts[state] = int(count) 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), ).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))) open_filter = BlockedProduct.resolved_at.is_(None) open_blockers = session.scalar(select(func.count(BlockedProduct.id)).where(open_filter)) or 0 breakdown_expression = func.coalesce(BlockedProduct.field, "OTHER") breakdown_rows = session.execute(select(breakdown_expression, func.count(BlockedProduct.id)).where(open_filter).group_by(breakdown_expression)).all() open_blocker_breakdown = {field: 0 for field in BLOCKER_FIELDS} for field, count in breakdown_rows: key = str(field or "").upper() open_blocker_breakdown[key if key in open_blocker_breakdown else "OTHER"] += int(count) product_pairs = select(BlockedProduct.tenant_id, BlockedProduct.product_code).where(open_filter).distinct().subquery() safe_blocked_products = session.scalar(select(func.count()).select_from(product_pairs)) or 0 blocker_counts = {field: count for field, count in open_blocker_breakdown.items() if count} def installation_candidates(health): query = installation_query.where(health_expression == health).order_by(Installation.id.asc()).limit(ATTENTION_CANDIDATE_LIMIT) return [installation_view(i, a, t, s, now=now) for i, a, t, s in session.execute(query).all()] health_candidates = {health: installation_candidates(health) for health in ("FAILED", "DEGRADED", "OFFLINE", "UNKNOWN")} version_filter = or_(AgentSnapshot.version_drift == "OUTDATED", and_(Agent.agent_version.is_not(None), Agent.agent_version != settings.agent_server_version)) outdated_query = installation_query.where(version_filter).order_by(Installation.id.asc()).limit(ATTENTION_CANDIDATE_LIMIT) outdated_agents = [installation_view(i, a, t, s, now=now) for i, a, t, s in session.execute(outdated_query).all()] blocker_query = select(BlockedProduct, Tenant, Installation, Agent, AgentSnapshot).join(Tenant, Tenant.id == BlockedProduct.tenant_id).outerjoin(Installation, Installation.id == BlockedProduct.installation_id).outerjoin(Agent, Agent.id == Installation.agent_id).outerjoin(AgentSnapshot, AgentSnapshot.installation_id == Installation.id).where(open_filter, BlockedProduct.field.in_(BLOCKER_FIELDS[:-1])).order_by(BlockedProduct.last_seen_at.desc(), BlockedProduct.id.desc()).limit(ATTENTION_CANDIDATE_LIMIT) blocker_candidates = session.execute(blocker_query).all() attention = [] for health in ("FAILED", "DEGRADED", "OFFLINE", "UNKNOWN"): rank = {"FAILED": 0, "DEGRADED": 1, "OFFLINE": 2, "UNKNOWN": 3}[health] attention.extend((_attention_item(item, item["health_reason"], _technical_timestamp(item), rank=rank, now=now), rank, item["agent_id"]) for item in health_candidates[health]) for blocker, tenant, installation, agent, snapshot in blocker_candidates: if installation and agent: item = installation_view(installation, agent, tenant, snapshot, now=now) else: item = {"client": tenant.name, "display_name": None, "hostname": None, "agent_id": None, "technical_health": "UNKNOWN", "health_reason": "tenant-level blocker", "last_seen_at": None, "last_job_at": None} attention.append((_attention_item(item, f"{blocker.field or blocker.blocker_type}: {blocker.reason_code}", blocker.last_seen_at, rank=4, link=f"/blockers/{blocker.id}", now=now), 4, blocker.id)) for item in outdated_agents: attention.append((_attention_item(item, "Versão do agent desatualizada", item.get("last_seen_at"), rank=5, now=now), 5, item["agent_id"])) attention.sort(key=lambda entry: (entry[1], -(entry[0]["timestamp"] is not None), entry[2])) failed_jobs = session.scalar(select(func.count(Job.id)).where(Job.status == "FAILED")) or 0 return { "agents_total": sum(counts.values()), **{key.lower(): value for key, value in counts.items()}, "health_counts": counts, **totals, "blockers": blocker_counts, "stale_agents": health_candidates["OFFLINE"], "outdated_agents": outdated_agents, "safe_blocked_products": int(safe_blocked_products), "open_blockers": int(open_blockers), "open_blocker_breakdown": open_blocker_breakdown, "attention_items": [entry[0] for entry in attention[:10]], "failed_jobs": failed_jobs, } def blocker_detail(session, blocker_id): row = session.execute(_blockers_query(session, status="ALL").where(BlockedProduct.id == blocker_id)).first() if not row: return None blocker, tenant, installation, agent = row item = _blocker_item(blocker, tenant, installation, agent) events = session.scalars(select(AgentEvent).where(AgentEvent.blocker_id == blocker_id).order_by(AgentEvent.occurred_at.desc(), AgentEvent.id.desc()).limit(100)).all() event_views = [{"id": event.id, "severity": event.severity, "event_type": event.event_type, "message": event.message, "context": event.structured_context, "job_id": event.job_id, "product_code": event.product_code, "occurred_at": _iso(event.occurred_at)} for event in events] job_ids = {event.job_id for event in events if event.job_id is not None} related_jobs = [] if job_ids: job_query = select(Job, Agent, Installation, Tenant).join(Agent, Agent.id == Job.agent_id).join(Installation, Installation.agent_id == Agent.id).join(Tenant, Tenant.id == Installation.tenant_id).where(Job.id.in_(job_ids)).order_by(Job.created_at.desc(), Job.id.desc()).limit(100) related_jobs = [_job_item(job, related_agent, related_installation, related_tenant) for job, related_agent, related_installation, related_tenant in session.execute(job_query).all()] audits = session.scalars(select(AuditEvent).where(AuditEvent.target_type == "blocker", AuditEvent.target_id == str(blocker_id)).order_by(AuditEvent.occurred_at.desc(), AuditEvent.id.desc()).limit(100)).all() audit_views = [{"id": audit.id, "admin_user_id": audit.admin_user_id, "action": audit.action, "target_id": audit.target_id, "context": audit.context, "severity": audit.severity, "occurred_at": _iso(audit.occurred_at)} for audit in audits] server = { "hostname": installation.hostname if installation else None, "company_name": installation.company_name if installation else None, "company_code": installation.company_code if installation else None, "company_uf": installation.company_uf if installation else None, "company_crt": installation.company_crt if installation else None, "firebird_version": installation.firebird_version if installation else None, "firebird_ods": installation.firebird_ods if installation else None, "database_identifier": installation.database_identifier if installation else None, } return { **item, "evidence": blocker.evidence_summary or {}, "installation": {"installation_id": item["installation_id"], "installation_db_id": item["installation_db_id"], "display_name": item["display_name"], "hostname": item["hostname"], "agent_id": item["agent_id"]}, "server": server, "related_events": event_views, "related_jobs": related_jobs, "resolution_audit": audit_views, } def list_panel_events(session, installation_db_id, **filters): rows = list_events(session, installation_db_id, **filters) return [{"id": row.id, "severity": row.severity, "event_type": row.event_type, "message": row.message, "context": row.structured_context, "job_id": row.job_id, "product_code": row.product_code, "occurred_at": _iso(row.occurred_at)} for row in rows]