"""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'
{esc(empty_text)}
' items = "".join(f"
  • {esc(item)}
  • " for item in blockers) return f'
    Bloqueios atuais
    ' def readiness_checklist_html( *, title: str, missing: Iterable[str], ok_text: str = "Pronto para avançar.", blocked_text: str = "Ação bloqueada até corrigir os dados em falta.", ) -> str: missing_items = [str(item) for item in missing if str(item or "").strip()] if missing_items: pills = "".join(f'⚠ {esc(item)}' for item in missing_items) state = f'
    {esc(blocked_text)}
    ' else: pills = '✓ Dados mínimos completos' state = f'
    {esc(ok_text)}
    ' return f'''

    {esc(title)}

    {pills}
    {state}
    ''' def fiscal_contact_panel_html( *, fiscal_customer: dict[str, Any] | None, contact_name: Any = "", contact_email: Any = "", contact_phone: Any = "", conversation_id: Any = "", contact_id: Any = "", customer_href: str = "", ) -> str: customer = fiscal_customer or {} customer_name = _field(customer, "name", "linked_customer_name") or "Cliente fiscal por associar" tax_id = _field(customer, "tax_id", "linked_customer_tax_id") or "—" email = _field(customer, "email", "linked_customer_email") or "—" address_bits = [ _field(customer, "street_name"), " ".join(bit for bit in [_field(customer, "postal_zone"), _field(customer, "city_name")] if bit), ] address = " · ".join(bit for bit in address_bits if bit) or "—" contact_name = str(contact_name or "Contacto por confirmar") contact_email = str(contact_email or "—") contact_phone = str(contact_phone or "") conversation = str(conversation_id or "") contact_ref = str(contact_id or "") customer_button = f'Ver cliente fiscal' if customer_href else "" return f'''

    Cliente fiscal e contacto Chatwoot

    Cliente fiscal
    {esc(customer_name)}
    NIF {esc(tax_id)}
    {esc(email)}
    {esc(address)}
    {customer_button}
    Contacto Chatwoot
    {esc(contact_name)}
    {esc(contact_email)}
    {esc(contact_phone)}
    {esc('Conversa #' + conversation if conversation else 'Sem conversa ligada')}
    {esc('Contacto #' + contact_ref if contact_ref else '')}
    ''' def fiscal_contact_inline_html( *, fiscal_name: Any = "", contact_ref: Any = "", conversation_id: Any = "", show_fiscal: bool = True, fiscal_required: bool = False, ) -> str: contact = str(contact_ref or "").strip() conversation = str(conversation_id or "").strip() contact_bits = [] if contact: contact_bits.append(f"contacto {contact}") if conversation: contact_bits.append(f"conversa #{conversation}") contact_line = " · ".join(contact_bits) or "contacto por confirmar" fiscal_value = str(fiscal_name or "").strip() if not show_fiscal and not fiscal_required: return f'''
    Contacto Chatwoot{esc(contact_line)}
    ''' fiscal_label = fiscal_value or "Por associar" return f'''
    Cliente fiscal{esc(fiscal_label)}
    Contacto Chatwoot{esc(contact_line)}
    ''' def outbox_operator_message(item: dict[str, Any]) -> dict[str, str]: target = str(item.get("target_system") or "integração").strip() action = str(item.get("action_type") or "ação").strip() error = str(item.get("last_error") or "").strip() lower = error.casefold() title = f"Falha em {target}.{action}" if error else f"Ação {target}.{action}" probable = "Ver o detalhe técnico do erro antes de reprocessar." fix_label = "Ver detalhe" if "tax" in lower or "nif" in lower or "vat" in lower or "fiscal" in lower: probable = "Cliente fiscal sem NIF válido ou dados fiscais incompletos." fix_label = "Corrigir cliente" elif "address" in lower or "morada" in lower or "postal" in lower or "city" in lower: probable = "Morada fiscal ou morada de entrega incompleta." fix_label = "Corrigir morada" elif "product" in lower or "item" in lower or "sales_item" in lower or "artigo" in lower: probable = "Produto sem código externo/Jasmin ou artigo inválido." fix_label = "Corrigir produto" elif "payment" in lower or "pagamento" in lower: probable = "Pagamento ainda não validado ou estado financeiro inconsistente." fix_label = "Ver pagamento" elif "timeout" in lower or "connection" in lower or "tempor" in lower: probable = "Falha temporária de ligação à integração externa." fix_label = "Reprocessar depois" return {"title": title, "probable": probable, "fix_label": fix_label, "technical": error}