358 lines
14 KiB
Python
358 lines
14 KiB
Python
"""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'<div class="alert alert-success border-0 mb-0"><strong>{esc(empty_text)}</strong></div>'
|
|
items = "".join(f"<li>{esc(item)}</li>" for item in blockers)
|
|
return f'<div class="alert alert-warning border-0 mb-0"><strong>Bloqueios atuais</strong><ul class="mb-0 mt-2">{items}</ul></div>'
|
|
|
|
|
|
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'<span class="cf-readiness-pill missing">⚠ {esc(item)}</span>' for item in missing_items)
|
|
state = f'<div class="small text-danger fw-bold mt-2">{esc(blocked_text)}</div>'
|
|
else:
|
|
pills = '<span class="cf-readiness-pill ok">✓ Dados mínimos completos</span>'
|
|
state = f'<div class="small text-success fw-bold mt-2">{esc(ok_text)}</div>'
|
|
return f'''
|
|
<section class="card cf-card cf-readiness-card">
|
|
<div class="card-body p-4">
|
|
<h2 class="cf-section-title mb-2">{esc(title)}</h2>
|
|
<div class="d-flex flex-wrap gap-2">{pills}</div>
|
|
{state}
|
|
</div>
|
|
</section>
|
|
'''
|
|
|
|
|
|
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'<a class="btn btn-sm btn-outline-secondary mt-2" href="{esc(customer_href)}">Ver cliente fiscal</a>' if customer_href else ""
|
|
return f'''
|
|
<section class="card cf-card cf-fiscal-contact-panel">
|
|
<div class="card-body p-4">
|
|
<h2 class="cf-section-title mb-3">Cliente fiscal e contacto Chatwoot</h2>
|
|
<div class="cf-fiscal-contact-grid">
|
|
<div class="cf-fiscal-box fiscal">
|
|
<div class="cf-fiscal-label">Cliente fiscal</div>
|
|
<strong>{esc(customer_name)}</strong>
|
|
<div class="small text-secondary">NIF {esc(tax_id)}</div>
|
|
<div class="small text-secondary text-break">{esc(email)}</div>
|
|
<div class="small text-secondary text-break">{esc(address)}</div>
|
|
{customer_button}
|
|
</div>
|
|
<div class="cf-fiscal-box contact">
|
|
<div class="cf-fiscal-label">Contacto Chatwoot</div>
|
|
<strong>{esc(contact_name)}</strong>
|
|
<div class="small text-secondary text-break">{esc(contact_email)}</div>
|
|
<div class="small text-secondary">{esc(contact_phone)}</div>
|
|
<div class="small text-secondary">{esc('Conversa #' + conversation if conversation else 'Sem conversa ligada')}</div>
|
|
<div class="small text-secondary">{esc('Contacto #' + contact_ref if contact_ref else '')}</div>
|
|
</div>
|
|
</div>
|
|
</div>
|
|
</section>
|
|
'''
|
|
|
|
|
|
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'''
|
|
<div class="cf-fiscal-mini compact">
|
|
<div><span>Contacto Chatwoot</span><strong>{esc(contact_line)}</strong></div>
|
|
</div>
|
|
'''
|
|
|
|
fiscal_label = fiscal_value or "Por associar"
|
|
return f'''
|
|
<div class="cf-fiscal-mini">
|
|
<div><span>Cliente fiscal</span><strong>{esc(fiscal_label)}</strong></div>
|
|
<div><span>Contacto Chatwoot</span><strong>{esc(contact_line)}</strong></div>
|
|
</div>
|
|
'''
|
|
|
|
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}
|