"""Human-readable values used by the operational control panel.""" from __future__ import annotations from datetime import datetime, timezone import os from zoneinfo import ZoneInfo from .config import settings def _as_utc(value: datetime | None) -> datetime | None: if value is None: return None if value.tzinfo is None: value = value.replace(tzinfo=timezone.utc) return value.astimezone(timezone.utc) def _ui_timezone() -> ZoneInfo: return ZoneInfo(os.getenv("UI_TIMEZONE", settings.ui_timezone)) def format_ui_datetime(value: datetime | None) -> str: """Format a stored timestamp in the configured UI timezone.""" value = _as_utc(value) if value is None: return "-" return value.astimezone(_ui_timezone()).strftime("%d/%m/%Y %H:%M") def format_age(value: datetime | None, now: datetime | None = None) -> str: value = _as_utc(value) if value is None: return "-" now = _as_utc(now or datetime.now(timezone.utc)) seconds = int((now - value).total_seconds()) if seconds <= 0: return "agora" minutes = seconds // 60 if minutes < 1: return "agora" if minutes < 60: return f"há {minutes} min" return f"há {minutes // 60} h" def format_duration(started_at: datetime | None, finished_at: datetime | None, now: datetime | None = None) -> str: del now started_at = _as_utc(started_at) if started_at is None: return "-" if finished_at is None: return "em andamento" elapsed = max(0, int((_as_utc(finished_at) - started_at).total_seconds())) hours, remainder = divmod(elapsed, 3600) minutes, seconds = divmod(remainder, 60) parts = [] if hours: parts.append(f"{hours} h") if minutes: parts.append(f"{minutes} min") if seconds or not parts: parts.append(f"{seconds} s") return " ".join(parts) def _blocker_value(blocker, name: str) -> str: if isinstance(blocker, dict): return str(blocker.get(name) or "").upper() return str(getattr(blocker, name, "") or "").upper() def operational_priority(blocker) -> str: blocker_type = _blocker_value(blocker, "blocker_type") field = _blocker_value(blocker, "field") reason = _blocker_value(blocker, "reason_code") operational_text = f"{blocker_type} {reason}" if blocker_type in {"TECHNICAL_FAILURE", "CONNECTION_ERROR", "AGENT_OFFLINE", "JOB_FAILED"} or any(token in operational_text for token in ("TECHNICAL", "CONNECTION", "TIMEOUT")): return "CRITICAL" if blocker_type in {"BLOCKED_MISSING_EVIDENCE", "MISSING_EVIDENCE"}: return "HIGH" if field in {"CFOP", "ST", "ICMS", "CSOSN", "NCM"} or reason in {"INSUFFICIENT_EVIDENCE", "NO_EVIDENCE"}: return "NORMAL" return "LOW" _HEALTH_LABELS = { "ONLINE": "Online", "DEGRADED": "Degradado", "OFFLINE": "Offline", "FAILED": "Falha", "UNKNOWN": "Desconhecido", } def health_label(value) -> str: if value is None or value == "": return "-" text = str(value) return _HEALTH_LABELS.get(text.upper(), text)