335 lines
14 KiB
Python
335 lines
14 KiB
Python
"""View model for the operator workbench.
|
|
|
|
This module is intentionally UI-focused: it does not change workflow rules, it
|
|
only prepares the data returned by operations_service for full-page and HTMX
|
|
partial rendering.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime, timezone
|
|
from typing import Any
|
|
|
|
from app.operations_service import get_operations_summary
|
|
from app.admin_ui.labels import primary_action_label, queue_label
|
|
from app.admin_ui.guidance import item_requires_fiscal_customer, work_item_blockers
|
|
|
|
# Legacy label kept for regression context: Ambíguas.
|
|
OPERATION_FILTERS = [
|
|
("all", "A fazer", "/operations?scope=all", "/operations/partials/work-items?scope=all"),
|
|
("bloqueadas", "Bloqueadas", "/operations?scope=bloqueadas", "/operations/partials/work-items?scope=bloqueadas"),
|
|
("ambiguas", "Associações por confirmar", "/operations?scope=ambiguas", "/operations/partials/work-items?scope=ambiguas"),
|
|
("atrasadas", "Atrasadas", "/operations?scope=atrasadas", "/operations/partials/work-items?scope=atrasadas"),
|
|
("vendas", "Vendas", "/operations?scope=vendas", "/operations/partials/work-items?scope=vendas"),
|
|
("financeiro", "Financeiro", "/operations?scope=financeiro", "/operations/partials/work-items?scope=financeiro"),
|
|
("logistica", "Logística", "/operations?scope=logistica", "/operations/partials/work-items?scope=logistica"),
|
|
("revisao", "Revisão", "/operations?scope=revisao", "/operations/partials/work-items?scope=revisao"),
|
|
("done", "Concluídas hoje", "/tasks?status=done", "/tasks/partials/list?status=done"),
|
|
]
|
|
|
|
|
|
|
|
|
|
def _is_numeric_reference(value: Any) -> bool:
|
|
value = str(value or "").strip()
|
|
return bool(value) and value.isdigit()
|
|
|
|
|
|
def is_classification_failed(item: dict[str, Any]) -> bool:
|
|
detail = str(item.get("detail") or "").casefold()
|
|
return "classificação da mensagem falhou" in detail or "classificacao da mensagem falhou" in detail
|
|
|
|
|
|
def has_no_commercial_opportunity(item: dict[str, Any]) -> bool:
|
|
reason = str(item.get("no_opportunity_reason") or "").strip()
|
|
action_code = str(item.get("action_code") or "").upper()
|
|
return bool(reason) or action_code in {"REVIEW_MANUALLY", "REMOVE_FROM_LIST", "IGNORE_SPAM", "NO_ACTION"}
|
|
|
|
|
|
def _looks_like_generic_label(value: str) -> bool:
|
|
value = " ".join(str(value or "").strip().casefold().split())
|
|
return value in {"", "geral", "cliente", "contacto", "sem nome"}
|
|
|
|
|
|
# Legacy label intentionally no longer used as primary title: "Contacto Chatwoot #".
|
|
# It is only suitable as technical metadata/fallback, not as operator identity.
|
|
def operation_card_title(item: dict[str, Any]) -> str:
|
|
if str(item.get("source") or "") == "outbox":
|
|
return str(item.get("customer_name") or item.get("title") or "Integração")
|
|
fiscal = str(item.get("fiscal_customer_name") or "").strip()
|
|
contact = str(item.get("contact_display_name") or "").strip()
|
|
customer = str(item.get("customer_name") or "").strip()
|
|
process_hint = str(item.get("process_customer_hint") or "").strip()
|
|
identity_status = str(item.get("contact_identity_status") or "").strip().lower()
|
|
if fiscal and not _is_numeric_reference(fiscal) and not _looks_like_generic_label(fiscal):
|
|
return fiscal
|
|
if identity_status == "unsafe" and process_hint and not _is_numeric_reference(process_hint):
|
|
return process_hint
|
|
if contact and not _is_numeric_reference(contact) and not _looks_like_generic_label(contact):
|
|
return contact
|
|
if customer and not _is_numeric_reference(customer) and not _looks_like_generic_label(customer):
|
|
return customer
|
|
if process_hint and not _is_numeric_reference(process_hint):
|
|
return process_hint
|
|
return "Contacto sem identificação"
|
|
|
|
|
|
def operation_card_subtitle(item: dict[str, Any]) -> str:
|
|
if is_classification_failed(item):
|
|
return "Classificação por rever"
|
|
reason = str(item.get("no_opportunity_reason") or "").strip()
|
|
if reason in {"system_or_bounce_message", "bounce_ignored"}:
|
|
return "Mensagem automática / bounce"
|
|
if reason:
|
|
return "Sem oportunidade comercial"
|
|
opportunity = str(item.get("opportunity_title") or "").strip()
|
|
if opportunity:
|
|
return opportunity
|
|
if str(item.get("source") or "") == "outbox":
|
|
return str(item.get("title") or "Integração")
|
|
if str(item.get("action_code") or "").upper() in {"REMOVE_FROM_LIST", "IGNORE_SPAM", "NO_ACTION"}:
|
|
return "Sem oportunidade comercial"
|
|
conv = str(item.get("conversation_id") or "").strip()
|
|
if operation_card_title(item) == "Contacto sem identificação" and conv:
|
|
return f"Conversa Chatwoot #{conv}"
|
|
return "Sem oportunidade associada"
|
|
|
|
|
|
def operation_has_useful_details(item: dict[str, Any]) -> bool:
|
|
if str(item.get("source") or "") == "outbox":
|
|
return True
|
|
if str(item.get("opportunity_id") or "").strip():
|
|
return True
|
|
if str(item.get("fiscal_customer_name") or "").strip():
|
|
return True
|
|
if item_requires_fiscal_customer(item):
|
|
return True
|
|
if work_item_blockers(item):
|
|
return True
|
|
return False
|
|
|
|
|
|
def operation_card_detail(item: dict[str, Any]) -> str:
|
|
if is_classification_failed(item):
|
|
return "O sistema não conseguiu classificar com segurança. Abrir no Chatwoot e escolher a ação correta."
|
|
reason = str(item.get("no_opportunity_reason") or "").strip()
|
|
if reason in {"system_or_bounce_message", "bounce_ignored"}:
|
|
return "Mensagem automática detetada. Não entra no fluxo operacional."
|
|
if reason == "send_info_without_clear_commercial_intent":
|
|
return "Responder sem criar oportunidade até existir intenção comercial clara."
|
|
return str(item.get("detail") or item.get("request_text") or "").strip()
|
|
|
|
|
|
def operation_status_chip(item: dict[str, Any]) -> str:
|
|
if is_association_review(item):
|
|
return "associação por confirmar"
|
|
if str(item.get("action_code") or "").upper() == "REVIEW_RECONSTRUCTED_PROCESS":
|
|
return "validação obrigatória"
|
|
blockers = work_item_blockers(item)
|
|
if blockers:
|
|
return "bloqueada"
|
|
if is_high_priority(item):
|
|
return "prioridade alta"
|
|
if has_no_commercial_opportunity(item):
|
|
return "sem oportunidade comercial"
|
|
return "normal"
|
|
|
|
|
|
def is_association_review(item: dict[str, Any]) -> bool:
|
|
linking = str(item.get("opportunity_linking_status") or "").lower()
|
|
code = str(item.get("action_code") or "").upper()
|
|
return linking in {"ambiguous", "review_required"} or code == "ASSOCIATE_OPPORTUNITY"
|
|
|
|
|
|
def is_ambiguous(item: dict[str, Any]) -> bool:
|
|
return is_association_review(item)
|
|
|
|
|
|
def _parse_dt(value: Any) -> datetime | None:
|
|
if not value:
|
|
return None
|
|
if isinstance(value, str):
|
|
try:
|
|
dt = datetime.fromisoformat(value.replace("Z", "+00:00"))
|
|
except ValueError:
|
|
return None
|
|
else:
|
|
dt = value
|
|
if getattr(dt, "tzinfo", None) is None:
|
|
dt = dt.replace(tzinfo=timezone.utc)
|
|
return dt
|
|
|
|
|
|
def is_scheduled_for_future(item: dict[str, Any]) -> bool:
|
|
action_code = str(item.get("action_code") or "").upper()
|
|
if not action_code.startswith("FOLLOW_UP_"):
|
|
return False
|
|
due_dt = _parse_dt(item.get("due_at"))
|
|
return bool(due_dt and due_dt > datetime.now(timezone.utc))
|
|
|
|
|
|
def is_overdue(item: dict[str, Any]) -> bool:
|
|
due_dt = _parse_dt(item.get("due_at"))
|
|
if due_dt is not None:
|
|
return due_dt < datetime.now(timezone.utc)
|
|
created_dt = _parse_dt(item.get("created_at"))
|
|
if not created_dt:
|
|
return False
|
|
queue = str(item.get("queue") or "").lower()
|
|
hours = {"suporte": 2, "vendas": 4, "financeiro": 8, "operacoes": 24, "logistica": 24, "rever": 24}.get(queue, 24)
|
|
return (datetime.now(timezone.utc) - created_dt).total_seconds() > hours * 3600
|
|
|
|
def operation_due_label(item: dict[str, Any], *, now: datetime | None = None) -> str:
|
|
now = now or datetime.now(timezone.utc)
|
|
due_dt = _parse_dt(item.get("due_at"))
|
|
if due_dt is not None:
|
|
seconds = (due_dt - now).total_seconds()
|
|
days = int(abs(seconds) // 86400)
|
|
hours = max(1, int(abs(seconds) // 3600))
|
|
if seconds < 0:
|
|
if days >= 1:
|
|
return f"Vencida há {days} dia(s)"
|
|
return f"Vencida há {hours} h"
|
|
if days >= 1:
|
|
return f"Vence em {days} dia(s)"
|
|
return f"Vence em {hours} h"
|
|
created_dt = _parse_dt(item.get("created_at"))
|
|
if created_dt is None:
|
|
return "Sem prazo"
|
|
age_seconds = max((now - created_dt).total_seconds(), 0)
|
|
age_days = int(age_seconds // 86400)
|
|
if age_days >= 1:
|
|
return f"Criada há {age_days} dia(s)"
|
|
return f"Criada há {max(1, int(age_seconds // 3600))} h"
|
|
|
|
|
|
def operation_value(item: dict[str, Any]) -> float:
|
|
for key in ("opportunity_value_amount", "amount", "total_amount"):
|
|
try:
|
|
value = float(item.get(key) or 0)
|
|
except (TypeError, ValueError):
|
|
continue
|
|
if value > 0:
|
|
return value
|
|
return 0.0
|
|
|
|
|
|
def operation_queue_label(item: dict[str, Any]) -> str:
|
|
return queue_label(item.get("queue"))
|
|
|
|
|
|
def operation_priority_reason(item: dict[str, Any]) -> str:
|
|
if is_association_review(item):
|
|
return "Bloqueia a ligação segura do processo"
|
|
if str(item.get("action_code") or "").upper() == "REVIEW_RECONSTRUCTED_PROCESS":
|
|
return "Validação necessária antes de ação sensível"
|
|
if is_overdue(item):
|
|
return "Prazo ultrapassado"
|
|
if str(item.get("source") or "").lower() == "outbox":
|
|
return "Integração exige intervenção"
|
|
if str(item.get("queue") or "").lower() == "suporte":
|
|
return "Cliente aguarda resposta de suporte"
|
|
value = operation_value(item)
|
|
if value > 0 and str(item.get("queue") or "").lower() in {"financeiro", "vendas"}:
|
|
return f"Impacto comercial/financeiro de {value:.2f} EUR"
|
|
return "Prioridade definida pela fila operacional"
|
|
|
|
|
|
def is_high_priority(item: dict[str, Any]) -> bool:
|
|
status = str(item.get("status") or "").lower()
|
|
source = str(item.get("source") or "").lower()
|
|
queue = str(item.get("queue") or "").lower()
|
|
action_code = str(item.get("action_code") or "").upper()
|
|
priority = str(item.get("priority") or "").lower()
|
|
|
|
if is_ambiguous(item):
|
|
return True
|
|
if source == "outbox" and status in {"failed", "blocked", "stale"}:
|
|
return True
|
|
if is_overdue(item):
|
|
return True
|
|
if is_classification_failed(item):
|
|
return False
|
|
if queue == "marketing" or action_code in {"REMOVE_FROM_LIST", "IGNORE_SPAM", "NO_ACTION", "REVIEW_MANUALLY"}:
|
|
return False
|
|
if action_code in {"SEND_PROFORMA", "SEND_INVOICE", "CONFIRM_PAYMENT", "PREPARE_ORDER", "VALIDATE_PHYSICAL_ORDER", "CREATE_SHIPMENT"}:
|
|
return priority in {"alta", "high", "urgente", "critical"} or status in {"failed", "blocked"}
|
|
return priority in {"alta", "high", "urgente"} or status in {"failed", "blocked"}
|
|
|
|
|
|
def is_blocked(item: dict[str, Any]) -> bool:
|
|
status = str(item.get("status") or "").lower()
|
|
source = str(item.get("source") or "").lower()
|
|
return status in {"failed", "blocked", "stale"} or source == "outbox" or bool(work_item_blockers(item))
|
|
|
|
|
|
def operation_item_matches_scope(item: dict[str, Any], scope: str) -> bool:
|
|
queue = str(item.get("queue") or "").lower()
|
|
action_code = str(item.get("action_code") or "").upper()
|
|
scope = str(scope or "all").strip().lower()
|
|
if scope in {"", "all", "todas"}:
|
|
return True
|
|
if scope == "bloqueadas":
|
|
return is_blocked(item)
|
|
if scope in {"ambiguas", "ambíguas"}:
|
|
return is_ambiguous(item)
|
|
if scope == "atrasadas":
|
|
return is_overdue(item)
|
|
if scope == "logistica":
|
|
return queue in {"operacoes", "logistica", "logística"}
|
|
if scope == "revisao":
|
|
return queue in {"rever", "revisao", "revisão"} or action_code == "REVIEW_MANUALLY"
|
|
return queue == scope
|
|
|
|
|
|
def operation_primary_label(item: dict[str, Any]) -> str:
|
|
if is_classification_failed(item):
|
|
return "Rever mensagem"
|
|
if str(item.get("source") or "") == "outbox":
|
|
status = str(item.get("status") or "").lower()
|
|
return "Reprocessar" if status == "failed" else "Ver bloqueio"
|
|
if is_association_review(item):
|
|
source = str(item.get("source_system") or "").strip().lower()
|
|
if source == "jasmin":
|
|
return "Validar associação do documento"
|
|
if source == "odoo":
|
|
return "Validar associação da venda"
|
|
return "Validar associação"
|
|
return primary_action_label(item.get("action_code") or item.get("action_label"), fallback="Abrir")
|
|
|
|
|
|
def operation_origin_label(item: dict[str, Any]) -> str:
|
|
source = str(item.get("source_system") or item.get("source") or "").strip()
|
|
conv = str(item.get("conversation_id") or "").strip()
|
|
if conv:
|
|
return f"Chatwoot #{conv}"
|
|
if source:
|
|
return source
|
|
return "ClientFlow"
|
|
|
|
|
|
def build_operations_view_model(scope: str = "all", *, limit: int = 30) -> dict[str, Any]:
|
|
data = get_operations_summary(limit=limit)
|
|
counts = data.get("counts") or {}
|
|
scope = str(scope or "all").strip().lower()
|
|
work_items = list(data.get("work_items") or [])
|
|
visible_items = [item for item in work_items if operation_item_matches_scope(item, scope)]
|
|
high_items = [item for item in visible_items if is_high_priority(item)]
|
|
normal_items = [item for item in visible_items if not is_high_priority(item)]
|
|
review_total = int(counts.get("review_tasks", 0) or 0) + int(counts.get("communications_needs_review", 0) or 0)
|
|
blocked_total = int(counts.get("blocked_outbox", 0) or 0) + sum(1 for item in work_items if is_blocked(item))
|
|
ambiguous_total = sum(1 for item in work_items if is_association_review(item))
|
|
overdue_total = int(counts.get("overdue_tasks", 0) or 0) + sum(1 for item in work_items if item.get("source") == "outbox" and str(item.get("status") or "").lower() in {"failed", "stale"})
|
|
return {
|
|
"scope": scope,
|
|
"counts": counts,
|
|
"review_total": review_total,
|
|
"blocked_total": blocked_total,
|
|
"ambiguous_total": ambiguous_total,
|
|
"overdue_total": overdue_total,
|
|
"today_label": datetime.now(timezone.utc).strftime("%d/%m/%Y"),
|
|
"filters": OPERATION_FILTERS,
|
|
"visible_items": visible_items,
|
|
"high_items": high_items,
|
|
"normal_items": normal_items,
|
|
}
|