"""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, ) 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. """ 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_items = 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.value_amount, 0) AS opportunity_value_amount, COALESCE(o.currency, 'EUR') AS opportunity_currency FROM tasks 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 t.status = 'pending' AND NOT (t.action_code LIKE 'FOLLOW_UP_%' AND t.due_at IS NOT NULL AND t.due_at > now()) 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.value_amount, 0) AS opportunity_value_amount, COALESCE(o.currency, 'EUR') AS opportunity_currency FROM integration_outbox 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 io.status IN ('pending','failed','blocked') AND NOT (io.status = 'failed' AND (COALESCE(io.last_error,'') ILIKE '%limpo manualmente%' OR COALESCE(io.last_error,'') ILIKE '%resolvido manualmente%')) 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) AS item_metadata, COALESCE(o.metadata, '{}'::jsonb) AS opportunity_metadata, COALESCE(o.stage, '') AS opportunity_stage, COALESCE(o.value_amount, 0) AS opportunity_value_amount, COALESCE(o.currency, 'EUR') AS opportunity_currency 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') ) 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 LIMIT :limit """), {"limit": int(limit)}).mappings().all() cleaned_work_items = _attach_operation_urls(_normalise_work_item_intent([dict(r) for r in work_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"] = len(cleaned_work_items) 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, } 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]