"""Operational summaries for ClientFlow. This module is deliberately query-oriented: it builds the daily operation view, health checks and unified opportunity timeline without changing business state. """ from __future__ import annotations import os import re import subprocess from typing import Any, Dict, List, Optional from sqlalchemy import text from app.config import settings from app.db import engine from app.operation_noise import is_low_value_no_opportunity_item, is_noise_operation_item from app.work_center_action_policy import ( RECONSTRUCTED_SENSITIVE_ACTIONS, association_review_required, canonical_action_code, reconstructed_review_required, ) from app.canonical_operations import ( canonicalize_operations, opportunity_ids_from_work_seeds, partition_canonical_items, ) from app.opportunity_next_action_service import get_opportunity_next_actions from app.document_reconciliation_service import active_document_link_exclusion_sql def _int(value: Any) -> int: try: return int(value or 0) except Exception: return 0 def _is_manually_resolved_error(value: Any) -> bool: text_value = str(value or "").strip().casefold() return "limpo manualmente" in text_value or "resolvido manualmente" in text_value def _humanize_operation_detail(value: Any) -> str: detail = str(value or "").strip() lower = detail.casefold() if "resposta llm inválida" in lower or "invalid" in lower and "action_code" in lower: return "Classificação da mensagem falhou. Rever no Chatwoot e escolher a ação correta." if _is_manually_resolved_error(detail): return "Item já limpo manualmente. Deve ficar no histórico/outbox, não na fila diária." return detail def _systemd_state(unit: str) -> Dict[str, Any]: """Return a best-effort systemd unit status. Works on the production server and degrades gracefully elsewhere. """ try: result = subprocess.run( ["systemctl", "is-active", unit], text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=3, check=False, ) return { "unit": unit, "active_state": result.stdout.strip() or result.stderr.strip() or "unknown", "ok": result.returncode == 0, } except Exception as exc: return {"unit": unit, "active_state": "unknown", "ok": False, "error": str(exc)} def build_chatwoot_conversation_url(conversation_id: Optional[str]) -> str: """Build a public Chatwoot URL when configuration is available.""" conversation_id = str(conversation_id or "").strip() if not conversation_id: return "" base_url = ( getattr(settings, "chatwoot_public_url", "") or getattr(settings, "chatwoot_base_url", "") or "" ).rstrip("/") account_id = str(getattr(settings, "chatwoot_account_id", "") or "").strip() if not base_url or not account_id: return "" return f"{base_url}/app/accounts/{account_id}/conversations/{conversation_id}" _IDENTITY_STOPWORDS = { "da", "de", "do", "dos", "das", "e", "lda", "ltda", "unipessoal", "sa", "s.a", "s.a.", "email", "mail", "geral", "info", "office", "frontoffice", "comercial", "vendas", "admin", "contacto", "contact", } def _norm_identity(value: Any) -> str: return " ".join(re.sub(r"[^0-9a-zA-ZÀ-ÿ]+", " ", str(value or "").casefold()).split()) def _identity_tokens(value: Any) -> set[str]: return { token for token in _norm_identity(value).split() if len(token) >= 3 and token not in _IDENTITY_STOPWORDS } def _email_tokens(value: Any) -> set[str]: email = str(value or "").strip().casefold() local = email.split("@", 1)[0] return { token for token in re.split(r"[^0-9a-zA-ZÀ-ÿ]+", local) if len(token) >= 3 and token not in _IDENTITY_STOPWORDS } def _name_matches_email(name: Any, email: Any) -> bool: name_tokens = _identity_tokens(name) email_tokens = _email_tokens(email) if not name_tokens or not email_tokens: return False if name_tokens & email_tokens: return True return any(nt in et or et in nt for nt in name_tokens for et in email_tokens) def _looks_like_campaign_process(title: Any) -> bool: title_norm = _norm_identity(title) return "carregadores" in title_norm and "para" in title_norm def _process_customer_hint(title: Any) -> str: """Extract the intended customer/company from common campaign/process titles. This is only a UI safety hint. It must never create/link customers. It helps avoid displaying a contaminated Chatwoot contact name as the card owner. """ value = str(title or "").strip() if not value: return "" # Examples: "Re: Carregadores ... para a Nova Maquiambiente" and # "Processo reconstruído · F.S. Motors". if "·" in value: tail = value.rsplit("·", 1)[-1].strip() if tail and not tail.upper().startswith(("ORC.", "S0")): return tail match = re.search(r"\bpara\s+(?:a|o|as|os|à|ao)?\s*(.+)$", value, re.IGNORECASE) if match: candidate = re.sub(r"\s+", " ", match.group(1)).strip(" .:-–—") # Avoid returning a long quoted email/thread suffix. candidate = re.split(r"\s+(?:de:|from:|enviada:|sent:)", candidate, maxsplit=1, flags=re.IGNORECASE)[0].strip() if 2 <= len(candidate) <= 120: return candidate return "" def _sanitize_operation_identities(items: List[Dict[str, Any]]) -> List[Dict[str, Any]]: """Prevent polluted Chatwoot sender names from becoming card titles. Chatwoot contact names can be reused or contaminated after campaign replies. The workbench must show the real process/customer context, not blindly trust ``payload.sender.name``. This function is deliberately UI-only: it does not change tasks, contacts, customers or opportunities. """ name_to_emails: Dict[str, set[str]] = {} for item in items: name = str(item.get("sender_name") or item.get("contact_display_name") or "").strip() email = str(item.get("sender_email") or item.get("customer_email") or "").strip().casefold() key = _norm_identity(name) if key and email: name_to_emails.setdefault(key, set()).add(email) repeated_names = {key for key, emails in name_to_emails.items() if len(emails) > 1} for item in items: fiscal = str(item.get("fiscal_customer_name") or "").strip() if fiscal: item["contact_identity_status"] = "fiscal_customer" continue sender_name = str(item.get("sender_name") or item.get("contact_display_name") or "").strip() sender_email = str(item.get("sender_email") or item.get("customer_email") or "").strip() opportunity_title = str(item.get("opportunity_title") or "").strip() hint = _process_customer_hint(opportunity_title) if hint: item["process_customer_hint"] = hint key = _norm_identity(sender_name) reused = bool(key and key in repeated_names) email_match = _name_matches_email(sender_name, sender_email) campaign_context = _looks_like_campaign_process(opportunity_title) unsafe = False reason = "" if sender_name and reused: unsafe = True reason = "sender_name_reused_across_emails" elif sender_name and campaign_context and hint and not email_match: unsafe = True reason = "sender_name_not_supported_by_email_or_process" if unsafe: item["contact_identity_status"] = "unsafe" item["contact_identity_reason"] = reason # Force operation_card_title to prefer the process/company hint or # email instead of a possibly contaminated person name. item["contact_display_name"] = "" item["customer_name"] = hint or sender_email or "Contacto sem identificação" else: item["contact_identity_status"] = "trusted" if sender_name else "missing" if not sender_name and hint: item["customer_name"] = hint return items def _is_reconstructed_mode(value: Any) -> bool: return str(value or "").strip().lower() in { "reconstructed_invoice_review", "historical_reconstructed", "legacy_review", } def _looks_like_association_review(item: Dict[str, Any]) -> bool: """Return True only for an explicitly persisted association blocker. v132 deliberately ignores opportunity titles such as "sem oportunidade". Those titles are historical provenance and must not invent a current blocker. """ return association_review_required( action_code=item.get("action_code"), linking_status=item.get("opportunity_linking_status"), ) def _normalise_work_item_intent(items: List[Dict[str, Any]]) -> List[Dict[str, Any]]: """Expose one first safe operator action from persisted blockers/tasks. No blocker is inferred from titles or free text. Association review and reconstructed review must be explicit in persisted state. The underlying task remains available through ``original_action_code`` for auditability. """ sensitive_codes = {canonical_action_code(code) for code in RECONSTRUCTED_SENSITIVE_ACTIONS} for item in items: original_code = canonical_action_code(item.get("action_code")) item["original_action_code"] = original_code if _looks_like_association_review(item): item["action_code"] = "ASSOCIATE_OPPORTUNITY" item["queue"] = "rever" item["priority"] = "alta" item["opportunity_linking_status"] = item.get("opportunity_linking_status") or "review_required" source = str(item.get("source_system") or "").strip().lower() item["title"] = "Validar associação do documento" if source == "jasmin" else "Validar associação operacional" item["detail"] = ( "Confirmar primeiro a que processo pertence a evidência. " "Ligar a uma oportunidade existente, criar uma nova ou ignorar; " "só depois executar ações comerciais, fiscais ou financeiras." ) continue metadata = item.get("opportunity_metadata") or {} if not isinstance(metadata, dict): metadata = {} if reconstructed_review_required(metadata) and original_code in sensitive_codes: item["action_code"] = "REVIEW_RECONSTRUCTED_PROCESS" item["queue"] = "rever" item["priority"] = "alta" item["title"] = "Validar processo reconstruído" item["detail"] = ( "Confirmar cliente, documento principal, valor e evidência de pagamento " "antes de executar a ação financeira, fiscal ou logística sugerida." ) item["reconstructed_review_required"] = True return items def _attach_operation_urls(items: List[Dict[str, Any]]) -> List[Dict[str, Any]]: cleaned: List[Dict[str, Any]] = [] for item in items: original_detail = item.get("detail") if str(item.get("source") or "") == "outbox" and _is_manually_resolved_error(original_detail): continue item["detail"] = _humanize_operation_detail(original_detail) # v4.9.0: Operations is not a mailbox. Delivery bounces, postmaster # notifications, and technical no-action items must stay in # Chatwoot/Thunderbird/logs, not in the daily work queue. if is_noise_operation_item(item) or is_low_value_no_opportunity_item(item): continue if str(item.get("source_system") or "") == "chatwoot" or item.get("conversation_id"): item["chatwoot_url"] = build_chatwoot_conversation_url(item.get("conversation_id")) else: item["chatwoot_url"] = "" cleaned.append(item) return _sanitize_operation_identities(cleaned) def get_operations_summary(limit: int = 24) -> Dict[str, Any]: """Build the /operations work queue summary. /operations is intentionally not a mini-dashboard. It returns a compact set of counters and a single prioritized queue of human work items. Technical lists remain available in their own pages and should only appear here when they block an operator action. """ display_limit = max(1, min(int(limit), 200)) # Bound each entity source independently, then apply the user-facing limit # after canonicalization. This prevents duplicate rows from consuming the # display limit while still avoiding an unbounded history scan. candidate_limit = max(200, min(display_limit * 20, 1000)) with engine.begin() as conn: counts = conn.execute(text(""" SELECT (SELECT COUNT(*) FROM opportunities WHERE status = 'open')::int AS open_opportunities, (SELECT COUNT(*) FROM tasks WHERE status = 'pending')::int AS pending_tasks, (SELECT COUNT(*) FROM tasks WHERE status = 'pending' AND action_code LIKE 'FOLLOW_UP_%' AND due_at IS NOT NULL AND due_at > now())::int AS scheduled_followups, (SELECT COUNT(*) FROM integration_outbox WHERE status = 'pending')::int AS outbox_pending, (SELECT COUNT(*) FROM integration_outbox WHERE status = 'failed' AND COALESCE(last_error,'') NOT ILIKE '%limpo manualmente%' AND COALESCE(last_error,'') NOT ILIKE '%resolvido manualmente%')::int AS outbox_failed, (SELECT COUNT(*) FROM commercial_documents WHERE document_kind = 'quotation' AND status NOT IN ('failed','cancelled','converted'))::int AS open_quotations, (SELECT COUNT(*) FROM commercial_documents WHERE document_kind = 'invoice' AND status IN ('issued','created'))::int AS active_invoices, (SELECT COUNT(*) FROM commercial_documents WHERE document_kind = 'invoice' AND created_at::date = CURRENT_DATE)::int AS invoices_today, (SELECT COUNT(*) FROM shipments WHERE status NOT IN ('cancelled','failed','delivered','shipped'))::int AS shipments_pending, (SELECT COUNT(*) FROM customers WHERE COALESCE(tax_id,'') = '' OR COALESCE(street_name,'') = '' OR COALESCE(postal_zone,'') = '' OR COALESCE(city_name,'') = '')::int AS customers_incomplete, (SELECT COUNT(*) FROM products WHERE active = TRUE AND COALESCE(jasmin_sales_item,'') = '')::int AS products_missing_jasmin, (SELECT COUNT(*) FROM communications WHERE status IN ('new','classified','needs_review'))::int AS communications_open, (SELECT COUNT(*) FROM communications WHERE status = 'needs_review')::int AS communications_needs_review, (SELECT COUNT(*) FROM communications WHERE customer_id IS NULL AND direction = 'inbound')::int AS communications_without_customer, (SELECT COUNT(*) FROM tasks WHERE status = 'pending' AND due_at IS NOT NULL AND due_at < now())::int AS overdue_tasks, (SELECT COUNT(*) FROM tasks WHERE status = 'pending' AND (route = 'rever' OR action_code = 'REVIEW_MANUALLY'))::int AS review_tasks, (SELECT COUNT(*) FROM integration_outbox WHERE status IN ('failed','blocked') AND COALESCE(last_error,'') NOT ILIKE '%limpo manualmente%' AND COALESCE(last_error,'') NOT ILIKE '%resolvido manualmente%')::int AS blocked_outbox, ( (SELECT COUNT(*) FROM tasks WHERE status = 'pending' AND NOT (action_code LIKE 'FOLLOW_UP_%' AND due_at IS NOT NULL AND due_at > now())) + (SELECT COUNT(*) FROM integration_outbox WHERE status IN ('failed','blocked') AND COALESCE(last_error,'') NOT ILIKE '%limpo manualmente%' AND COALESCE(last_error,'') NOT ILIKE '%resolvido manualmente%') + (SELECT COUNT(*) FROM communications WHERE status = 'needs_review') )::int AS work_queue_total """)).mappings().first() or {} recent_outbox = conn.execute(text(""" SELECT id::text, target_system, action_type, status, last_error, created_at, updated_at FROM integration_outbox WHERE status IN ('pending','failed','blocked') AND NOT (status = 'failed' AND (COALESCE(last_error,'') ILIKE '%limpo manualmente%' OR COALESCE(last_error,'') ILIKE '%resolvido manualmente%')) ORDER BY created_at DESC LIMIT :limit """), {"limit": int(limit)}).mappings().all() recent_documents = conn.execute(text(""" SELECT cd.id::text, cd.opportunity_id::text, cd.document_kind, cd.status, cd.document_number, cd.external_id, cd.total_amount, cd.currency, c.name AS customer_name, cd.created_at FROM commercial_documents cd LEFT JOIN customers c ON c.id = cd.customer_id WHERE (cd.document_kind = 'quotation' AND cd.status NOT IN ('failed','cancelled','converted')) OR (cd.document_kind = 'invoice' AND cd.status IN ('issued','created')) OR (cd.status IN ('failed','blocked')) ORDER BY cd.created_at DESC LIMIT :limit """), {"limit": int(limit)}).mappings().all() problem_products = conn.execute(text(""" SELECT sku, name, default_unit_price FROM products WHERE active = TRUE AND COALESCE(jasmin_sales_item,'') = '' ORDER BY name LIMIT :limit """), {"limit": int(limit)}).mappings().all() incomplete_customers = conn.execute(text(""" SELECT id::text, name, tax_id, street_name, postal_zone, city_name, updated_at FROM customers WHERE COALESCE(tax_id,'') = '' OR COALESCE(street_name,'') = '' OR COALESCE(postal_zone,'') = '' OR COALESCE(city_name,'') = '' ORDER BY updated_at DESC LIMIT :limit """), {"limit": int(limit)}).mappings().all() recent_communications = conn.execute(text(""" SELECT c.id::text, c.sender_name, c.sender_email, c.subject, c.classification, c.confidence, c.status, c.created_at, c.customer_id::text, c.opportunity_id::text, cu.name AS customer_name, o.title AS opportunity_title FROM communications c LEFT JOIN customers cu ON cu.id = c.customer_id LEFT JOIN opportunities o ON o.id = c.opportunity_id WHERE c.status IN ('new','classified','needs_review') ORDER BY c.created_at DESC LIMIT :limit """), {"limit": int(limit)}).mappings().all() work_seed_rows = conn.execute(text(""" SELECT * FROM ( SELECT 'task' AS source, t.id::text AS id, t.created_at, t.due_at, CASE WHEN COALESCE(t.metadata->>'opportunity_linking_status','') = 'ambiguous' THEN 'alta' ELSE COALESCE(t.priority, CASE WHEN t.due_at < now() THEN 'alta' ELSE 'normal' END) END AS priority, CASE WHEN COALESCE(t.metadata->>'opportunity_linking_status','') = 'ambiguous' THEN 'rever' ELSE COALESCE(t.route, 'rever') END AS queue, CASE WHEN COALESCE(t.metadata->>'opportunity_linking_status','') = 'ambiguous' THEN 'ASSOCIATE_OPPORTUNITY' ELSE t.action_code END AS action_code, CASE WHEN COALESCE(t.metadata->>'opportunity_linking_status','') = 'ambiguous' THEN 'Confirmar associação da oportunidade' ELSE COALESCE(t.action, t.action_code, 'Tarefa') END AS title, CASE WHEN COALESCE(t.metadata->>'opportunity_linking_status','') = 'ambiguous' THEN 'O sistema encontrou mais de uma oportunidade aberta para este contacto. Confirme a oportunidade e o cliente fiscal antes de emitir documentos.' ELSE COALESCE(t.note, '') END AS detail, t.status, t.source_system, t.conversation_id, t.contact_id, COALESCE(m.clean_body, m.raw_body, re.payload->>'content', '') AS request_text, t.opportunity_id::text, COALESCE(o.title, '') AS opportunity_title, COALESCE(NULLIF(re.payload->'sender'->>'name',''), NULLIF(re.payload->'sender'->>'email',''), NULLIF(t.contact_id,''), '') AS customer_name, NULLIF(re.payload->'sender'->>'name','') AS sender_name, NULLIF(re.payload->'sender'->>'email','') AS sender_email, COALESCE(cu_opp.name, cu_task.name, '') AS fiscal_customer_name, COALESCE(cu_opp.email, cu_task.email, '') AS fiscal_customer_email, COALESCE(cu_opp.tax_id, cu_task.tax_id, '') AS fiscal_customer_tax_id, COALESCE(cu_opp.street_name, cu_task.street_name, '') AS fiscal_customer_street_name, COALESCE(cu_opp.postal_zone, cu_task.postal_zone, '') AS fiscal_customer_postal_zone, COALESCE(cu_opp.city_name, cu_task.city_name, '') AS fiscal_customer_city_name, COALESCE(NULLIF(re.payload->'sender'->>'name',''), NULLIF(re.payload->'sender'->>'email',''), NULLIF(t.contact_id,''), '') AS contact_display_name, COALESCE(NULLIF(re.payload->'sender'->>'email',''), '') AS customer_email, COALESCE(t.metadata->>'no_opportunity_reason', '') AS no_opportunity_reason, '/tasks/' || t.id::text AS href, 'Abrir' AS action_label, COALESCE(t.metadata->>'opportunity_linking_status','') AS opportunity_linking_status, COALESCE(t.metadata, '{}'::jsonb) AS item_metadata, COALESCE(o.metadata, '{}'::jsonb) AS opportunity_metadata, COALESCE(o.stage, '') AS opportunity_stage, COALESCE(o.lifecycle_state, '') AS opportunity_lifecycle_state, o.next_follow_up_at AS opportunity_next_follow_up_at, COALESCE(o.value_amount, 0) AS opportunity_value_amount, COALESCE(o.currency, 'EUR') AS opportunity_currency FROM ( SELECT * FROM tasks WHERE status = 'pending' AND NOT (action_code LIKE 'FOLLOW_UP_%' AND due_at IS NOT NULL AND due_at > now()) ORDER BY created_at DESC LIMIT :candidate_limit ) t LEFT JOIN opportunities o ON o.id = t.opportunity_id LEFT JOIN customers cu_opp ON cu_opp.id = o.local_customer_id LEFT JOIN customers cu_task ON cu_task.id::text = t.customer_id LEFT JOIN messages m ON m.id = t.message_id LEFT JOIN raw_events re ON re.id = t.raw_event_id WHERE TRUE UNION ALL SELECT 'outbox' AS source, io.id::text AS id, io.created_at, NULL::timestamptz AS due_at, CASE WHEN io.status = 'failed' THEN 'alta' ELSE 'normal' END AS priority, 'sistema' AS queue, upper(io.target_system || '_' || io.action_type) AS action_code, io.target_system || '.' || io.action_type AS title, COALESCE(io.last_error, io.idempotency_key, '') AS detail, io.status, io.target_system AS source_system, NULL::text AS conversation_id, NULL::text AS contact_id, ''::text AS request_text, NULLIF(io.payload->>'opportunity_id','') AS opportunity_id, COALESCE(o.title, '') AS opportunity_title, COALESCE(cu.name, '') AS customer_name, ''::text AS sender_name, ''::text AS sender_email, COALESCE(cu.name, '') AS fiscal_customer_name, COALESCE(cu.email, '') AS fiscal_customer_email, COALESCE(cu.tax_id, '') AS fiscal_customer_tax_id, COALESCE(cu.street_name, '') AS fiscal_customer_street_name, COALESCE(cu.postal_zone, '') AS fiscal_customer_postal_zone, COALESCE(cu.city_name, '') AS fiscal_customer_city_name, ''::text AS contact_display_name, ''::text AS customer_email, ''::text AS no_opportunity_reason, '/outbox/' || io.id::text AS href, CASE WHEN io.status = 'failed' THEN 'Reprocessar' ELSE 'Abrir' END AS action_label, ''::text AS opportunity_linking_status, COALESCE(io.payload, '{}'::jsonb) AS item_metadata, COALESCE(o.metadata, '{}'::jsonb) AS opportunity_metadata, COALESCE(o.stage, '') AS opportunity_stage, COALESCE(o.lifecycle_state, '') AS opportunity_lifecycle_state, o.next_follow_up_at AS opportunity_next_follow_up_at, COALESCE(o.value_amount, 0) AS opportunity_value_amount, COALESCE(o.currency, 'EUR') AS opportunity_currency FROM ( SELECT * FROM integration_outbox WHERE status IN ('failed','blocked') AND NOT (status = 'failed' AND (COALESCE(last_error,'') ILIKE '%limpo manualmente%' OR COALESCE(last_error,'') ILIKE '%resolvido manualmente%')) ORDER BY created_at DESC LIMIT :candidate_limit ) io LEFT JOIN opportunities o ON o.id::text = NULLIF(io.payload->>'opportunity_id','') LEFT JOIN customers cu ON cu.id = o.local_customer_id WHERE TRUE UNION ALL SELECT 'communication' AS source, c.id::text AS id, c.created_at, NULL::timestamptz AS due_at, CASE WHEN c.status = 'needs_review' THEN 'normal' ELSE 'normal' END AS priority, CASE WHEN c.classification IN ('comprovativo_pagamento','pedido_fatura','aceitacao_orcamento','dados_fiscais') THEN 'financeiro' WHEN c.classification IN ('pedido_tracking') THEN 'operacoes' WHEN c.classification IN ('reclamacao') THEN 'suporte' WHEN c.classification IN ('pedido_remocao_lista') THEN 'marketing' ELSE 'vendas' END AS queue, upper(COALESCE(c.classification, 'REVIEW_MANUALLY')) AS action_code, COALESCE(c.classification, 'Mensagem por classificar') AS title, COALESCE(c.subject, c.sender_email, '') AS detail, c.status, c.source_system, c.conversation_id, c.contact_id, COALESCE(c.body, '') AS request_text, c.opportunity_id::text, COALESCE(o.title, '') AS opportunity_title, COALESCE(cu.name, c.sender_name, c.sender_email, '') AS customer_name, COALESCE(c.sender_name, '') AS sender_name, COALESCE(c.sender_email, '') AS sender_email, COALESCE(cu.name, '') AS fiscal_customer_name, COALESCE(cu.email, '') AS fiscal_customer_email, COALESCE(cu.tax_id, '') AS fiscal_customer_tax_id, COALESCE(cu.street_name, '') AS fiscal_customer_street_name, COALESCE(cu.postal_zone, '') AS fiscal_customer_postal_zone, COALESCE(cu.city_name, '') AS fiscal_customer_city_name, COALESCE(c.sender_name, c.sender_email, c.contact_id, '') AS contact_display_name, COALESCE(c.sender_email, '') AS customer_email, ''::text AS no_opportunity_reason, '/communications/' || c.id::text AS href, CASE WHEN c.customer_id IS NULL THEN 'Associar cliente' ELSE 'Abrir' END AS action_label, ''::text AS opportunity_linking_status, COALESCE(c.metadata, '{}'::jsonb) || CASE WHEN c.task_id IS NOT NULL THEN jsonb_build_object('task_id', c.task_id::text) ELSE '{}'::jsonb END AS item_metadata, COALESCE(o.metadata, '{}'::jsonb) AS opportunity_metadata, COALESCE(o.stage, '') AS opportunity_stage, COALESCE(o.lifecycle_state, '') AS opportunity_lifecycle_state, o.next_follow_up_at AS opportunity_next_follow_up_at, COALESCE(o.value_amount, 0) AS opportunity_value_amount, COALESCE(o.currency, 'EUR') AS opportunity_currency FROM ( SELECT * FROM communications WHERE status IN ('new','classified','needs_review') ORDER BY created_at DESC LIMIT :candidate_limit ) c LEFT JOIN customers cu ON cu.id = c.customer_id LEFT JOIN opportunities o ON o.id = c.opportunity_id WHERE TRUE UNION ALL SELECT 'reconciliation' AS source, ri.id::text AS id, ri.created_at, NULL::timestamptz AS due_at, COALESCE(ri.priority, 'normal') AS priority, 'rever' AS queue, upper(COALESCE(ri.suggested_action, 'RECONCILE_DOCUMENTS')) AS action_code, COALESCE(ri.title, 'Rever reconciliação') AS title, COALESCE(ri.description, ri.resolution_note, '') AS detail, ri.status, ri.source_system, NULL::text AS conversation_id, NULL::text AS contact_id, COALESCE(ri.description, '') AS request_text, ri.opportunity_id::text, COALESCE(o.title, '') AS opportunity_title, COALESCE(cu.name, ri.customer_name, ri.customer_email, '') AS customer_name, COALESCE(ri.customer_name, '') AS sender_name, COALESCE(ri.customer_email, '') AS sender_email, COALESCE(cu.name, '') AS fiscal_customer_name, COALESCE(cu.email, '') AS fiscal_customer_email, COALESCE(cu.tax_id, '') AS fiscal_customer_tax_id, COALESCE(cu.street_name, '') AS fiscal_customer_street_name, COALESCE(cu.postal_zone, '') AS fiscal_customer_postal_zone, COALESCE(cu.city_name, '') AS fiscal_customer_city_name, COALESCE(cu.name, ri.customer_name, ri.customer_email, '') AS contact_display_name, COALESCE(ri.customer_email, '') AS customer_email, ''::text AS no_opportunity_reason, '/reconciliation' AS href, 'Rever' AS action_label, ''::text AS opportunity_linking_status, COALESCE(ri.payload, '{}'::jsonb) AS item_metadata, COALESCE(o.metadata, '{}'::jsonb) AS opportunity_metadata, COALESCE(o.stage, '') AS opportunity_stage, COALESCE(o.lifecycle_state, '') AS opportunity_lifecycle_state, o.next_follow_up_at AS opportunity_next_follow_up_at, COALESCE(o.value_amount, ri.amount, 0) AS opportunity_value_amount, COALESCE(o.currency, ri.currency, 'EUR') AS opportunity_currency FROM ( SELECT * FROM reconciliation_items WHERE status IN ('open','needs_review','conflict') AND """ + active_document_link_exclusion_sql("reconciliation_items") + """ ORDER BY created_at DESC LIMIT :candidate_limit ) ri LEFT JOIN opportunities o ON o.id = ri.opportunity_id LEFT JOIN customers cu ON cu.id = COALESCE(o.local_customer_id, ri.customer_id) WHERE TRUE ) items ORDER BY CASE lower(priority) WHEN 'alta' THEN 1 WHEN 'high' THEN 1 WHEN 'urgente' THEN 0 WHEN 'normal' THEN 2 ELSE 3 END, created_at DESC """), {"candidate_limit": candidate_limit}).mappings().all() seed_opportunity_ids = opportunity_ids_from_work_seeds(work_seed_rows) evidence_rows = [] if seed_opportunity_ids: evidence_rows = conn.execute(text(""" SELECT * FROM ( SELECT 'communication' AS source, c.id::text AS id, c.created_at, c.status, upper(COALESCE(c.classification, 'REVIEW_MANUALLY')) AS action_code, c.opportunity_id::text, c.source_system, c.conversation_id, COALESCE(c.metadata, '{}'::jsonb) || CASE WHEN c.task_id IS NOT NULL THEN jsonb_build_object('task_id', c.task_id::text) ELSE '{}'::jsonb END AS item_metadata FROM communications c WHERE c.opportunity_id = ANY(CAST(:opportunity_ids AS UUID[])) AND c.status NOT IN ('new','classified','needs_review') ORDER BY c.created_at DESC LIMIT :candidate_limit ) communication_evidence UNION ALL SELECT * FROM ( SELECT 'reconciliation' AS source, ri.id::text AS id, ri.created_at, ri.status, upper(COALESCE(ri.suggested_action, 'RECONCILE_DOCUMENTS')) AS action_code, ri.opportunity_id::text, ri.source_system, NULL::text AS conversation_id, COALESCE(ri.payload, '{}'::jsonb) AS item_metadata FROM reconciliation_items ri WHERE ri.opportunity_id = ANY(CAST(:opportunity_ids AS UUID[])) AND ri.status IN ('linked','resolved') ORDER BY ri.created_at DESC LIMIT :candidate_limit ) reconciliation_evidence """), { "opportunity_ids": seed_opportunity_ids, "candidate_limit": candidate_limit, }).mappings().all() conversation_ids = sorted({ str(row.get("conversation_id") or "").strip() for row in work_seed_rows if str(row.get("conversation_id") or "").strip() }) conversation_timeline = {} if conversation_ids: timeline_rows = conn.execute(text(""" SELECT conversation_id, max(created_at) FILTER (WHERE direction = 'inbound') AS latest_public_inbound, max(created_at) FILTER (WHERE direction = 'outbound') AS latest_public_outbound FROM messages WHERE source_system = 'chatwoot' AND conversation_id = ANY(CAST(:conversation_ids AS TEXT[])) GROUP BY conversation_id """), {"conversation_ids": conversation_ids}).mappings().all() conversation_timeline = {str(row["conversation_id"]): dict(row) for row in timeline_rows} seed_rows = [dict(r) for r in work_seed_rows] for row in seed_rows: timeline = conversation_timeline.get(str(row.get("conversation_id") or ""), {}) row.update({key: timeline.get(key) for key in ("latest_public_inbound", "latest_public_outbound")}) normalized_rows = _normalise_work_item_intent(seed_rows) opportunity_ids = opportunity_ids_from_work_seeds(normalized_rows) decisions = get_opportunity_next_actions(opportunity_ids) if opportunity_ids else {} projection = canonicalize_operations(normalized_rows, decisions, evidence_rows=[dict(row) for row in evidence_rows]) canonical_items = _attach_operation_urls(list(projection["items"])) partition = partition_canonical_items(canonical_items, display_limit=display_limit) actionable_items = partition["actionable_items"] all_waiting_items = partition["waiting_items"] cleaned_work_items = partition["visible_actionable_items"] waiting_items = partition["visible_waiting_items"] cleaned_counts = {k: _int(v) for k, v in dict(counts).items()} # v4.9.0: the visible Operations total should match the queue the # operator can actually act on, not raw pending tasks that include mailbox # noise awaiting cleanup. The cleanup script still fixes the data source. cleaned_counts["work_queue_total"] = partition["work_queue_total"] cleaned_counts["waiting_total"] = partition["waiting_total"] cleaned_counts["backlog_total"] = partition["backlog_total"] cleaned_counts["not_current_total"] = partition["not_current_total"] cleaned_counts["raw_source_count"] = projection["raw_source_count"] cleaned_counts["canonical_total"] = len(canonical_items) diagnostics = { "candidate_limit": candidate_limit, "raw_source_count": projection["raw_source_count"], "clean_source_count": projection["clean_source_count"], "seed_source_count": projection["seed_source_count"], "evidence_source_count": projection["evidence_source_count"], "canonical_count": len(canonical_items), "visible_count": len(cleaned_work_items), "waiting_total": len(all_waiting_items), "backlog_total": partition["backlog_total"], "not_current_total": partition["not_current_total"], } return { "counts": cleaned_counts, "recent_outbox": [dict(r) for r in recent_outbox], "recent_documents": [dict(r) for r in recent_documents], "problem_products": [dict(r) for r in problem_products], "incomplete_customers": [dict(r) for r in incomplete_customers], "recent_communications": [dict(r) for r in recent_communications], "work_items": cleaned_work_items, "waiting_items": waiting_items, "backlog_items": partition["backlog_items"][:display_limit], "not_current_items": partition["not_current_items"][:display_limit], **diagnostics, "projection_metrics": diagnostics, } def get_system_health_summary() -> Dict[str, Any]: """Return human and API friendly health details.""" db_ok = True db_error: Optional[str] = None try: with engine.begin() as conn: conn.execute(text("SELECT 1")) except Exception as exc: db_ok = False db_error = str(exc) outbox_counts: Dict[str, Dict[str, int]] = {} document_counts: Dict[str, Dict[str, int]] = {} operational_metrics: Dict[str, int] = {} if db_ok: with engine.begin() as conn: rows = conn.execute(text(""" SELECT target_system, status, COUNT(*)::int AS total FROM integration_outbox GROUP BY target_system, status ORDER BY target_system, status """)).mappings().all() for row in rows: outbox_counts.setdefault(row["target_system"] or "unknown", {})[row["status"] or "unknown"] = row["total"] rows = conn.execute(text(""" SELECT document_kind, status, COUNT(*)::int AS total FROM commercial_documents GROUP BY document_kind, status ORDER BY document_kind, status """)).mappings().all() for row in rows: document_counts.setdefault(row["document_kind"] or "unknown", {})[row["status"] or "unknown"] = row["total"] row = conn.execute(text(""" SELECT (SELECT COUNT(*) FROM tasks WHERE status = 'pending')::int AS tasks_pending, (SELECT COUNT(*) FROM tasks WHERE status = 'done' AND done_at >= now() - interval '24 hours')::int AS tasks_done_24h, (SELECT COUNT(*) FROM task_events WHERE event_type = 'task_auto_completed' AND created_at >= now() - interval '24 hours')::int AS tasks_auto_completed_24h, (SELECT COUNT(*) FROM tasks WHERE status = 'pending' AND COALESCE(metadata->>'opportunity_linking_status','') = 'ambiguous')::int AS ambiguous_opportunity_tasks, (SELECT COUNT(*) FROM integration_outbox WHERE status = 'processing')::int AS outbox_processing, (SELECT COUNT(*) FROM integration_outbox WHERE status = 'processing' AND locked_at < now() - interval '30 minutes')::int AS outbox_processing_stale, (SELECT COUNT(*) FROM integration_outbox WHERE status = 'stale')::int AS outbox_stale, (SELECT COUNT(*) FROM integration_outbox WHERE status IN ('failed','blocked','stale'))::int AS outbox_blocked_or_failed, (SELECT COUNT(*) FROM business_events WHERE event_type = 'operator_action' AND created_at >= now() - interval '24 hours')::int AS operator_actions_24h, (SELECT COUNT(*) FROM opportunities WHERE status = 'open' AND local_customer_id IS NULL)::int AS open_opportunities_without_fiscal_customer, (SELECT COUNT(*) FROM opportunities o JOIN customers c ON c.id = o.local_customer_id WHERE o.status = 'open' AND (COALESCE(c.tax_id,'') = '' OR COALESCE(c.email,'') = '' OR COALESCE(c.street_name,'') = '' OR COALESCE(c.postal_zone,'') = '' OR COALESCE(c.city_name,'') = ''))::int AS active_incomplete_fiscal_customers, (SELECT COUNT(*) FROM products WHERE active = TRUE AND COALESCE(jasmin_sales_item,'') = '')::int AS products_missing_external_code, (SELECT COUNT(*) FROM raw_events WHERE source_system = 'chatwoot')::int AS chatwoot_events_total, (SELECT COUNT(*) FROM raw_events WHERE source_system = 'chatwoot' AND event_type = 'message_created' AND processed = false AND ignored = false AND processing_error IS NULL AND COALESCE(payload #>> '{message_type}', '') = 'incoming')::int AS chatwoot_incoming_pending, (SELECT COUNT(*) FROM raw_events WHERE source_system = 'chatwoot' AND event_type = 'message_created' AND COALESCE(payload #>> '{message_type}', '') = 'incoming' AND created_at >= now() - interval '24 hours')::int AS chatwoot_incoming_24h, (SELECT EXTRACT(EPOCH FROM (now() - max(created_at)))::int FROM raw_events WHERE source_system = 'chatwoot')::int AS seconds_since_last_chatwoot_webhook, (SELECT COUNT(*) FROM tasks WHERE status = 'pending' AND route = 'vendas')::int AS tasks_pending_vendas, (SELECT COUNT(*) FROM tasks WHERE status = 'pending' AND route = 'financeiro')::int AS tasks_pending_financeiro, (SELECT COUNT(*) FROM tasks WHERE status = 'pending' AND route IN ('operacoes','logistica'))::int AS tasks_pending_operacoes, (SELECT COUNT(*) FROM tasks WHERE status = 'pending' AND route IN ('rever','revisao'))::int AS tasks_pending_rever """)).mappings().first() or {} operational_metrics = {key: _int(value) for key, value in dict(row).items()} configured_auto_codes = os.getenv("CHATWOOT_AUTO_COMPLETE_ACTION_CODES", "").strip() return { "status": "ok" if db_ok else "degraded", "database": {"ok": db_ok, "error": db_error}, "settings": { "app_name": settings.app_name, "env": settings.env, "jasmin_enabled": bool(settings.jasmin_enabled), "packlink_enabled": bool(settings.packlink_enabled), "jasmin_company_key": settings.jasmin_company_key, "jasmin_quotation_serie": settings.jasmin_quotation_serie, "packlink_default_service_id": settings.packlink_default_service_id, "chatwoot_auto_complete_on_outgoing": os.getenv("CHATWOOT_AUTO_COMPLETE_ON_OUTGOING", "true"), "chatwoot_auto_complete_action_codes": configured_auto_codes or "default_safe_codes", "outbox_stale_processing_minutes": os.getenv("OUTBOX_STALE_PROCESSING_MINUTES", "30"), "outbox_stale_recovery_mode": os.getenv("OUTBOX_STALE_RECOVERY_MODE", "manual_only"), }, "timers": { "jasmin": _systemd_state("clientflow-outbox-jasmin.timer"), "packlink": _systemd_state("clientflow-outbox-packlink.timer"), }, "outbox": outbox_counts, "documents": document_counts, "operational_metrics": operational_metrics, } def list_unified_opportunity_timeline(opportunity_id: str, limit: int = 60) -> List[Dict[str, Any]]: """Return a combined opportunity timeline from events, documents, outbox, shipments and items.""" params = {"opportunity_id": opportunity_id, "limit": int(limit)} with engine.begin() as conn: rows = conn.execute(text(""" SELECT * FROM ( SELECT created_at, COALESCE(source, 'timeline') AS source, title, COALESCE(description, '') AS detail, payload, NULL::text AS status, related_id AS external_id FROM timeline_events WHERE opportunity_id = CAST(:opportunity_id AS UUID) UNION ALL SELECT created_at, 'event' AS source, event_type AS title, COALESCE(note, '') AS detail, payload, NULL::text AS status, NULL::text AS external_id FROM opportunity_events WHERE opportunity_id = CAST(:opportunity_id AS UUID) UNION ALL SELECT created_at, 'document' AS source, CASE WHEN document_kind = 'quotation' THEN 'Orçamento Jasmin' ELSE 'Fatura Jasmin' END AS title, COALESCE(document_number, external_id, '') AS detail, payload, status, external_id FROM commercial_documents WHERE opportunity_id = CAST(:opportunity_id AS UUID) UNION ALL SELECT created_at, 'outbox' AS source, target_system || '.' || action_type AS title, COALESCE(last_error, idempotency_key, '') AS detail, payload, status, id::text AS external_id FROM integration_outbox WHERE payload->>'opportunity_id' = :opportunity_id UNION ALL SELECT created_at, 'shipment' AS source, 'Envio Packlink' AS title, COALESCE(carrier || ' · ' || service_name, external_reference, '') AS detail, payload, status, external_reference AS external_id FROM shipments WHERE opportunity_id = CAST(:opportunity_id AS UUID) UNION ALL SELECT created_at, 'communication' AS source, COALESCE(classification, 'Comunicação recebida') AS title, COALESCE(subject, sender_email, '') AS detail, metadata AS payload, status, id::text AS external_id FROM communications WHERE opportunity_id = CAST(:opportunity_id AS UUID) UNION ALL SELECT created_at, 'product' AS source, 'Produto adicionado' AS title, COALESCE(product_name, sku, '') AS detail, jsonb_build_object('sku', sku, 'jasmin_sales_item', jasmin_sales_item, 'quantity', quantity, 'unit_price', unit_price) AS payload, NULL::text AS status, id::text AS external_id FROM opportunity_items WHERE opportunity_id = CAST(:opportunity_id AS UUID) ) x ORDER BY created_at DESC LIMIT :limit """), params).mappings().all() return [dict(r) for r in rows]