102 lines
3.1 KiB
Python
102 lines
3.1 KiB
Python
"""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)
|