"""Small operator-facing UI helpers for guided operations. v4.8.2 intentionally keeps business rules unchanged. Prontidão is shown as UI guidance only. These helpers only turn existing opportunity/task/outbox data into clearer labels, blockers and checklists for the operator. """ from __future__ import annotations import html from typing import Any, Iterable def esc(value: Any) -> str: return html.escape("" if value is None else str(value)) def _present(value: Any) -> bool: return bool(str(value or "").strip()) def _field(data: dict[str, Any] | None, *keys: str) -> str: data = data or {} for key in keys: value = data.get(key) if _present(value): return str(value).strip() return "" DOCUMENT_ACTION_CODES = { "SEND_PROFORMA", "SEND_INVOICE", } # Pagamento e envio têm checklists próprias. Não devem bloquear uma oportunidade # antiga/reconstruída só porque faltam campos fiscais no ClientFlow depois de o # documento oficial já ter sido emitido. FULFILMENT_ACTION_CODES = { "CONFIRM_PAYMENT_AND_PREPARE_SHIPMENT", "PREPARE_ORDER", "CREATE_SHIPMENT", } DOCUMENT_STAGES = { "PROFORMA_REQUESTED", "PROFORMA_SENT", "INVOICE_REQUESTED", "INVOICE_SENT", "WAITING_PAYMENT", "PAYMENT_CONFIRMED", "ORDER_PREPARATION", "READY_TO_SHIP", "INVOICED", "SHIPMENT_CREATED", "SHIPPED", "TRACKING_SENT", "DELIVERED", } SHIPMENT_ACTION_CODES = {"CONFIRM_PAYMENT_AND_PREPARE_SHIPMENT", "PREPARE_ORDER", "VALIDATE_PHYSICAL_ORDER", "CREATE_SHIPMENT"} SHIPMENT_STAGES = {"PAYMENT_CONFIRMED", "ORDER_PREPARATION", "READY_TO_SHIP", "SHIPMENT_CREATED", "SHIPPED", "TRACKING_SENT", "DELIVERED"} def normalized_action_code(value: Any) -> str: return str(value or "").strip().upper() def action_requires_fiscal_customer(action_code: Any) -> bool: return normalized_action_code(action_code) in DOCUMENT_ACTION_CODES def action_requires_shipment_readiness(action_code: Any) -> bool: return normalized_action_code(action_code) in SHIPMENT_ACTION_CODES def stage_requires_fiscal_customer(stage: Any) -> bool: return str(stage or "").strip().upper() in DOCUMENT_STAGES def stage_requires_shipment_readiness(stage: Any) -> bool: return str(stage or "").strip().upper() in SHIPMENT_STAGES def item_requires_fiscal_customer(item: dict[str, Any] | None) -> bool: item = item or {} return action_requires_fiscal_customer(item.get("action_code")) def item_requires_shipment_readiness(item: dict[str, Any] | None) -> bool: item = item or {} return action_requires_shipment_readiness(item.get("action_code")) def fiscal_customer_missing_fields(customer: dict[str, Any] | None) -> list[str]: """Minimum fields needed before fiscal documents should be issued.""" if not customer: return ["cliente fiscal associado"] missing: list[str] = [] if not _field(customer, "name", "linked_customer_name"): missing.append("nome fiscal") if not _field(customer, "tax_id", "linked_customer_tax_id"): missing.append("NIF") if not _field(customer, "email", "linked_customer_email", "customer_email"): missing.append("email de faturação") if not _field(customer, "street_name"): missing.append("morada fiscal") if not _field(customer, "postal_zone"): missing.append("código postal") if not _field(customer, "city_name"): missing.append("localidade") return missing def shipment_missing_fields(customer: dict[str, Any] | None, opportunity: dict[str, Any] | None = None) -> list[str]: missing: list[str] = [] if not customer: return ["cliente fiscal associado", "morada de entrega", "telefone"] if not _field(customer, "street_name"): missing.append("morada de entrega") if not _field(customer, "postal_zone"): missing.append("código postal de entrega") if not _field(customer, "city_name"): missing.append("localidade de entrega") if not (_field(customer, "phone") or _field(opportunity or {}, "customer_phone")): missing.append("telefone") return missing def opportunity_context_customer(opportunity: dict[str, Any], linked_customer: dict[str, Any] | None = None) -> dict[str, Any] | None: if linked_customer: return linked_customer if opportunity.get("linked_customer_id") or opportunity.get("linked_customer_name"): return { "id": opportunity.get("linked_customer_id"), "name": opportunity.get("linked_customer_name"), "email": opportunity.get("linked_customer_email"), "tax_id": opportunity.get("linked_customer_tax_id"), "street_name": opportunity.get("linked_customer_street_name"), "postal_zone": opportunity.get("linked_customer_postal_zone"), "city_name": opportunity.get("linked_customer_city_name"), "phone": opportunity.get("linked_customer_phone"), } return None def opportunity_blockers( opportunity: dict[str, Any], linked_customer: dict[str, Any] | None = None, *, action_code: Any = None, ) -> list[str]: """Return only blockers that matter for the current journey stage. Missing fiscal data is not automatically a blocker at the first contact. It becomes a current blocker only when the next action/stage needs fiscal documents, payment or fulfilment. """ customer = opportunity_context_customer(opportunity, linked_customer) blockers: list[str] = [] linking_status = str(opportunity.get("opportunity_linking_status") or opportunity.get("linking_status") or "").lower() if linking_status == "ambiguous": blockers.append("Associação de oportunidade por confirmar") needs_fiscal = action_requires_fiscal_customer(action_code) if action_code else stage_requires_fiscal_customer(opportunity.get("stage")) if needs_fiscal: if not customer: blockers.append("Cliente fiscal por associar") else: for item in fiscal_customer_missing_fields(customer): blockers.append(f"Cliente fiscal sem {item}") needs_product = needs_fiscal or str(opportunity.get("stage") or "").upper() in {"QUOTE_REQUESTED", "QUOTE_SENT"} if needs_product and not _field(opportunity, "product_interest", "title"): blockers.append("Produto/interesse por definir") return blockers def work_item_fiscal_customer(item: dict[str, Any]) -> dict[str, Any] | None: """Return the fiscal customer carried by an Operations item, if any. The work queue must not infer a fiscal customer from Chatwoot contact_id. It only uses the opportunity/customer links already resolved by the backend. """ item = item or {} if not _field(item, "fiscal_customer_name"): return None return { "name": item.get("fiscal_customer_name"), "email": item.get("fiscal_customer_email"), "tax_id": item.get("fiscal_customer_tax_id"), "street_name": item.get("fiscal_customer_street_name"), "postal_zone": item.get("fiscal_customer_postal_zone"), "city_name": item.get("fiscal_customer_city_name"), } def work_item_blockers(item: dict[str, Any]) -> list[str]: """Compact blockers for Operations cards, based on the next action.""" blockers: list[str] = [] linking_status = str(item.get("opportunity_linking_status") or "").lower() if linking_status == "ambiguous": blockers.append("Associação de oportunidade por confirmar") action_code = normalized_action_code(item.get("action_code")) if action_code == "REVIEW_RECONSTRUCTED_PROCESS": blockers.append("Processo reconstruído por validar") if item_requires_fiscal_customer(item): customer = work_item_fiscal_customer(item) if not customer: blockers.append("Cliente fiscal por associar") else: for missing in fiscal_customer_missing_fields(customer): blockers.append(f"Cliente fiscal sem {missing}") return blockers def blocker_alert_html(blockers: Iterable[str], *, empty_text: str = "Sem bloqueios críticos visíveis.") -> str: blockers = [str(item) for item in blockers if str(item or "").strip()] if not blockers: return f'