"""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]
