import json
import os
from typing import Any, Dict, List, Optional

from sqlalchemy import text

from app.action_catalog import get_action_config
from app.db import engine
from app.schemas import ActionResult


def _dump(value: Any) -> Dict[str, Any]:
    if value is None:
        return {}
    if hasattr(value, "model_dump"):
        return value.model_dump()
    if isinstance(value, dict):
        return value
    return dict(value)


def _json(value: Any) -> str:
    return json.dumps(value or {}, ensure_ascii=False)


def _uuid_or_none(value: Optional[str]) -> Optional[str]:
    value = str(value or "").strip()
    return value or None


def _initial_task_status(action_code: str, route: str, action_required: bool) -> str:
    """Define se uma decisão de triagem gera trabalho humano.

    v4.6: REVIEW_MANUALLY e REMOVE_FROM_LIST têm de aparecer como trabalho
    pendente.  Só spam e NO_ACTION são realmente ignorados/skipped.
    """
    code = str(action_code or "").strip().upper()
    route = str(route or "").strip().lower()

    if code == "IGNORE_BOUNCE":
        return "ignored"
    if code in {"IGNORE_SPAM", "NO_ACTION"} or route == "spam":
        return "skipped"
    if code in {"REVIEW_MANUALLY", "REMOVE_FROM_LIST"}:
        return "pending"
    return "pending" if action_required else "skipped"


def _priority_for_action(action_code: str, route: str) -> str:
    """Initial task priority for Operations.

    v4.9.0: generic review/marketing tasks must not compete with real
    document/payment/fulfilment work. Financial/document tasks remain high.
    """
    code = str(action_code or "").strip().upper()
    route = str(route or "").strip().lower()
    if code in {"SEND_PROFORMA", "SEND_INVOICE", "CONFIRM_PAYMENT", "PREPARE_ORDER", "VALIDATE_PHYSICAL_ORDER", "CREATE_SHIPMENT"}:
        return "alta"
    if code.startswith("FOLLOW_UP_") or code in {"CONFIRM_DELIVERY", "RECOVER_OPPORTUNITY", "REVIEW_NURTURE"}:
        return "normal"
    if code in {"REVIEW_MANUALLY", "REMOVE_FROM_LIST", "MARK_NO_INTEREST", "IGNORE_SPAM", "NO_ACTION", "IGNORE_BOUNCE"}:
        return "baixa"
    if route == "financeiro":
        return "alta"
    if route in {"marketing", "rever"}:
        return "baixa"
    return "normal"




OBSOLETE_AFTER_PAYMENT_ACTION_CODES = {
    "CONFIRM_PAYMENT",
    "FOLLOW_UP_PAYMENT",
    "FOLLOW_UP_PROFORMA",
    "FOLLOW_UP_QUOTE",
}


def mark_obsolete_payment_followup_tasks(
    opportunity_id: str,
    *,
    actor: str = "system",
    reason: str = "Pagamento confirmado; tarefa de confirmação/follow-up de pagamento obsoleta.",
) -> int:
    """Ignore pending payment/follow-up tasks that became obsolete.

    This is intentionally narrow: it only acts after an opportunity has confirmed
    payment and only on pending payment-confirmation/follow-up tasks.  It avoids
    the UI contradiction where the stage says "Pagamento confirmado" but the next
    action still asks the operator to follow up payment.
    """
    opportunity_id = str(opportunity_id or "").strip()
    if not opportunity_id:
        return 0
    with engine.begin() as conn:
        rows = conn.execute(text("""
            UPDATE tasks
            SET status = 'ignored',
                done_at = COALESCE(done_at, now()),
                done_by = COALESCE(NULLIF(done_by, ''), :actor),
                metadata = COALESCE(metadata, '{}'::jsonb) || CAST(:metadata AS JSONB),
                updated_at = now()
            WHERE opportunity_id = CAST(:opportunity_id AS UUID)
              AND status = 'pending'
              AND (upper(COALESCE(action_code, '')) IN ('CONFIRM_PAYMENT','FOLLOW_UP_PAYMENT','FOLLOW_UP_PROFORMA','FOLLOW_UP_QUOTE') OR (upper(COALESCE(action_code, '')) = 'CONFIRM_DELIVERY' AND COALESCE(metadata->>'follow_up_family','') = 'payment'))
            RETURNING id::text
        """), {
            "opportunity_id": opportunity_id,            "actor": actor or "system",
            "metadata": _json({
                "auto_ignored_reason": reason,
                "auto_ignored_after": "payment_confirmed",
            }),
        }).mappings().all()
        ids = [str(row.get("id")) for row in rows if row.get("id")]

    for task_id in ids:
        try:
            _record_task_timeline_event(task_id, event_type="task_skipped")
        except Exception:
            pass
    return len(ids)


def _timeline_payload_preview(metadata: Optional[Dict[str, Any]]) -> Dict[str, Any]:
    metadata = metadata or {}
    preview = dict(metadata)
    # Evita payloads demasiado grandes em eventos de timeline.
    if "message_preview" in preview:
        preview["message_preview"] = str(preview.get("message_preview") or "")[:500]
    return preview


def _record_task_timeline_event(task_id: str, event_type: str = "task_created") -> None:
    """Regista timeline quando a task já está associada a uma oportunidade.

    Se ainda não houver oportunidade, não força nada. O histórico principal
    continua em task_events/opportunity_events.
    """
    try:
        from app.communication_service import create_timeline_event

        task = get_task_detail(task_id)
        if not task or not task.get("opportunity_id"):
            return

        title_prefix = {
            "task_created": "Task criada",
            "task_done": "Task concluída",
            "task_auto_completed": "Task concluída automaticamente",
            "task_skipped": "Task ignorada",
        }.get(str(event_type or ""), "Task")

        create_timeline_event(
            opportunity_id=task.get("opportunity_id"),
            event_type=event_type,
            title=f"{title_prefix}: {task.get('action') or task.get('action_code')}",
            description=(task.get("note") or "")[:500],
            source=task.get("source_system") or "clientflow",
            related_type="task",
            related_id=task_id,
            payload={
                "action_code": task.get("action_code"),
                "route": task.get("route"),
                "status": task.get("status"),
                "conversation_id": task.get("conversation_id"),
                "contact_id": task.get("contact_id"),
                "source_system": task.get("source_system"),
            },
            created_by="system",
        )
    except Exception as exc:
        print(f"ClientFlow timeline event failed for task {task_id}: {exc}", flush=True)


def create_task_from_action_result(
    *,
    action_result: ActionResult | Dict[str, Any],
    action_run_id: Optional[str] = None,
    message_id: Optional[str] = None,
    raw_event_id: Optional[str] = None,
    conversation_id: Optional[str] = None,
    contact_id: Optional[str] = None,
    customer_id: Optional[str] = None,
    source_system: str = "clientflow",
    source_event_id: Optional[str] = None,
    metadata: Optional[Dict[str, Any]] = None,
) -> Optional[str]:
    data = _dump(action_result)

    action_code = data.get("action_code") or "REVIEW_MANUALLY"
    if str(action_code or "").strip().upper() == "IGNORE_BOUNCE":
        return None
    config = get_action_config(action_code)

    route = data.get("route") or config["route"]
    action = data.get("action") or config["action"]
    note = data.get("note") or action

    action_required = bool(data.get("action_required", False))
    safe_to_post = bool(data.get("safe_to_post", False))

    status = _initial_task_status(action_code, route, action_required)
    priority = _priority_for_action(action_code, route)

    # Idempotência forte:
    # - raw_event_id é estável quando o mesmo webhook é reenviado;
    # - para mensagens repetidas sem id externo estável, usa fingerprint semântico;
    # - só depois usa ids internos do action_run/message, que mudam em cada análise.
    content_fingerprint = str((metadata or {}).get("content_fingerprint") or "").strip()
    external_source_event = source_event_id if (source_event_id and source_event_id != message_id) else None
    idempotency_source = (
        raw_event_id
        or external_source_event
        or content_fingerprint
        or message_id
        or action_run_id
        or "no_source"
    )

    idempotency_key = ":".join([
        "task",
        str(source_system or "unknown"),
        str(conversation_id or "no_conversation"),
        str(idempotency_source),
        str(action_code),
    ])

    sql = text("""
        INSERT INTO tasks (
            action_run_id,
            message_id,
            raw_event_id,
            conversation_id,
            contact_id,
            customer_id,
            action_code,
            route,
            action,
            note,
            action_required,
            safe_to_post,
            status,
            source_system,
            source_event_id,
            idempotency_key,
            metadata,
            priority
        )
        VALUES (
            CAST(:action_run_id AS UUID),
            CAST(:message_id AS UUID),
            CAST(:raw_event_id AS UUID),
            :conversation_id,
            :contact_id,
            :customer_id,
            :action_code,
            :route,
            :action,
            :note,
            :action_required,
            :safe_to_post,
            :status,
            :source_system,
            :source_event_id,
            :idempotency_key,
            CAST(:metadata AS JSONB),
            :priority
        )
        ON CONFLICT (idempotency_key) DO NOTHING
        RETURNING id::text
    """)

    with engine.begin() as conn:
        row = conn.execute(sql, {
            "action_run_id": _uuid_or_none(action_run_id),
            "message_id": _uuid_or_none(message_id),
            "raw_event_id": _uuid_or_none(raw_event_id),
            "conversation_id": conversation_id,
            "contact_id": contact_id,
            "customer_id": customer_id,
            "action_code": action_code,
            "route": route,
            "action": action,
            "note": note,
            "action_required": action_required,
            "safe_to_post": safe_to_post,
            "status": status,
            "source_system": source_system,
            "source_event_id": source_event_id,
            "idempotency_key": idempotency_key,
            "metadata": _json(metadata or {}),
            "priority": priority,
        }).fetchone()

    task_id = row[0] if row else None

    if task_id:
        try:
            from app.opportunity_service import upsert_opportunity_for_task

            upsert_opportunity_for_task(
                task_id,
                trigger="task_created",
                created_by="system",
            )
            _record_task_timeline_event(task_id, event_type="task_created")
        except Exception as exc:
            print(f"ClientFlow opportunity upsert failed for task {task_id}: {exc}", flush=True)

    return task_id


def list_tasks(
    *,
    status: Optional[str] = None,
    route: Optional[str] = None,
    q: Optional[str] = None,
    limit: int = 100,
) -> List[Dict[str, Any]]:
    # v4928.1.5.92: choose schema-supported opportunity customer for task list
    try:
        from app.infra.schema_compat import opportunity_customer_column

        customer_column = opportunity_customer_column()
    except Exception:
        customer_column = "local_customer_id"
    if customer_column not in {"local_customer_id", "fiscal_customer_id"}:
        customer_column = "local_customer_id"

    where = []
    params: Dict[str, Any] = {"limit": limit}

    if status:
        where.append("t.status = :status")
        params["status"] = status

    if route:
        where.append("t.route = :route")
        params["route"] = route

    if q:
        params["q"] = f"%{q.strip().lower()}%"
        where.append("""
            (
                lower(coalesce(t.conversation_id, '')) like :q
                or lower(coalesce(o.conversation_id, '')) like :q
                or lower(coalesce(t.contact_id, '')) like :q
                or lower(coalesce(o.contact_id, '')) like :q
                or lower(coalesce(t.customer_id, '')) like :q
                or lower(coalesce(t.action_code, '')) like :q
                or lower(coalesce(t.route, '')) like :q
                or lower(coalesce(t.action, '')) like :q
                or lower(coalesce(t.note, '')) like :q
                or lower(coalesce(m.clean_body, '')) like :q
                or lower(coalesce(m.raw_body, '')) like :q
                or lower(coalesce(re.payload->'sender'->>'name', '')) like :q
                or lower(coalesce(re.payload->'sender'->>'email', '')) like :q
                or lower(coalesce(re.payload->'sender'->>'phone_number', '')) like :q
                or lower(coalesce(re.payload->'conversation'->'meta'->'sender'->>'name', '')) like :q
                or lower(coalesce(re.payload->'conversation'->'meta'->'sender'->>'email', '')) like :q
                or lower(coalesce(re.payload->'conversation'->'contact_inbox'->>'source_id', '')) like :q
                or lower(coalesce(re.payload->>'content', '')) like :q
            )
        """)

    where_sql = ""
    if where:
        where_sql = "WHERE " + " AND ".join(where)

    sql = text(f"""
        SELECT
            t.id::text,
            t.action_run_id::text,
            t.message_id::text,
            t.raw_event_id::text,
            t.opportunity_id::text,
            COALESCE(NULLIF(t.conversation_id, ''), NULLIF(o.conversation_id, '')) AS conversation_id,
            COALESCE(NULLIF(t.contact_id, ''), NULLIF(o.contact_id, '')) AS contact_id,
            t.customer_id,
            t.action_code,
            t.route,
            t.action,
            t.note,
            t.action_required,
            t.safe_to_post,
            t.status,
            t.priority,
            t.source_system,
            t.source_event_id,
            t.due_at,
            t.created_at,
            t.updated_at,
            t.done_at,
            t.done_by,
            t.metadata,
            o.title AS opportunity_title,
            o.customer_name AS opportunity_customer_name,
            o.customer_email AS opportunity_customer_email,
            o.conversation_id AS opportunity_conversation_id,
            o.contact_id AS opportunity_contact_id,
            cu.name AS linked_customer_name,
            cu.email AS linked_customer_email,

            COALESCE(
                NULLIF(cu.name, ''),
                NULLIF(o.customer_name, ''),
                NULLIF(re.payload->'sender'->>'name', ''),
                NULLIF(re.payload->'conversation'->'meta'->'sender'->>'name', ''),
                NULLIF(o.customer_email, ''),
                NULLIF(cu.email, ''),
                NULLIF(t.customer_id, ''),
                NULLIF(t.contact_id, '')
            ) AS customer_name,

            COALESCE(
                NULLIF(cu.email, ''),
                NULLIF(o.customer_email, ''),
                NULLIF(re.payload->'sender'->>'email', ''),
                NULLIF(re.payload->'conversation'->'meta'->'sender'->>'email', ''),
                NULLIF(re.payload->'conversation'->'contact_inbox'->>'source_id', '')
            ) AS customer_email,

            COALESCE(
                NULLIF(re.payload->'sender'->>'phone_number', ''),
                NULLIF(re.payload->'conversation'->'meta'->'sender'->>'phone_number', '')
            ) AS customer_phone,

            COALESCE(
                NULLIF(re.payload->'conversation'->'additional_attributes'->>'mail_subject', ''),
                NULLIF(re.payload->'content_attributes'->'email'->>'subject', ''),
                NULLIF(re.payload->'conversation'->'messages'->0->'content_attributes'->'email'->>'subject', '')
            ) AS message_subject,

            COALESCE(
                NULLIF(m.clean_body, ''),
                NULLIF(m.raw_body, ''),
                NULLIF(re.payload->>'content', '')
            ) AS request_text

        FROM tasks t
        LEFT JOIN messages m ON m.id = t.message_id
        LEFT JOIN raw_events re ON re.id = t.raw_event_id
        LEFT JOIN opportunities o ON o.id = t.opportunity_id
        -- Legacy guard anchor: LEFT JOIN customers cu ON cu.id = o.local_customer_id
        -- v4928.1.5.63: choose fiscal_customer_id only if deployment schema supports it.
        LEFT JOIN customers cu ON cu.id = o.{customer_column}
        {where_sql}
        ORDER BY
            CASE t.status
                WHEN 'pending' THEN 1
                WHEN 'failed' THEN 2
                WHEN 'skipped' THEN 3
                WHEN 'done' THEN 4
                ELSE 5
            END,
            CASE COALESCE(t.priority, 'normal')
                WHEN 'alta' THEN 1
                WHEN 'normal' THEN 2
                WHEN 'baixa' THEN 3
                ELSE 4
            END,
            t.due_at NULLS LAST,
            t.created_at DESC
        LIMIT :limit
    """)

    with engine.begin() as conn:
        rows = conn.execute(sql, params).mappings().all()

    return [dict(row) for row in rows]


def get_task(task_id: str) -> Optional[Dict[str, Any]]:
    sql = text("""
        SELECT
            id::text,
            action_run_id::text,
            message_id::text,
            raw_event_id::text,
            opportunity_id::text,
            conversation_id,
            contact_id,
            customer_id,
            action_code,
            route,
            action,
            note,
            action_required,
            safe_to_post,
            status,
            priority,
            source_system,
            source_event_id,
            due_at,
            created_at,
            updated_at,
            done_at,
            done_by,
            metadata
        FROM tasks
        WHERE id = CAST(:task_id AS UUID)
    """)

    with engine.begin() as conn:
        row = conn.execute(sql, {"task_id": task_id}).mappings().first()

    return dict(row) if row else None



# Legacy static anchor: o.local_customer_id::text AS opportunity_fiscal_customer_id
def get_task_detail(task_id: str) -> Optional[Dict[str, Any]]:
    try:
        from app.infra.schema_compat import opportunity_customer_column
        customer_column = opportunity_customer_column()
    except Exception:
        customer_column = "local_customer_id"
    if customer_column not in {"local_customer_id", "fiscal_customer_id"}:
        customer_column = "local_customer_id"

    sql = text(f"""
        SELECT
            t.id::text,
            t.action_run_id::text,
            t.message_id::text,
            t.raw_event_id::text,
            t.opportunity_id::text,
            COALESCE(NULLIF(t.conversation_id, ''), NULLIF(o.conversation_id, '')) AS conversation_id,
            COALESCE(NULLIF(t.contact_id, ''), NULLIF(o.contact_id, '')) AS contact_id,
            t.customer_id,
            t.action_code,
            t.route,
            t.action,
            t.note,
            t.action_required,
            t.safe_to_post,
            t.status,
            t.priority,
            t.source_system,
            t.source_event_id,
            t.idempotency_key,
            t.due_at,
            t.created_at,
            t.updated_at,
            t.done_at,
            t.done_by,
            t.metadata,
            COALESCE(o.{customer_column}, dcu.customer_id)::text AS linked_customer_id,
            COALESCE(o.{customer_column}, dcu.customer_id)::text AS opportunity_fiscal_customer_id,
            o.local_customer_id::text AS opportunity_local_customer_id,
            o.title AS opportunity_title,
            o.stage AS opportunity_stage,
            o.status AS opportunity_status,
            o.customer_name AS opportunity_customer_name,
            o.customer_email AS opportunity_customer_email,
            o.conversation_id AS opportunity_conversation_id,
            o.contact_id AS opportunity_contact_id,
            o.product_interest AS opportunity_product_interest,
            o.value_amount AS opportunity_value_amount,
            o.currency AS opportunity_currency,
            COALESCE(cu.name, dcu.customer_name) AS linked_customer_name,
            COALESCE(cu.email, dcu.customer_email) AS linked_customer_email,
            COALESCE(cu.tax_id, dcu.customer_tax_id) AS linked_customer_tax_id,
            -- Compatibility anchors kept for fiscal readiness regression tests:
            -- cu.street_name AS linked_customer_street_name
            -- cu.postal_zone AS linked_customer_postal_zone
            -- cu.city_name AS linked_customer_city_name
            COALESCE(cu.street_name, dcu.customer_street_name) AS linked_customer_street_name,
            COALESCE(cu.postal_zone, dcu.customer_postal_zone) AS linked_customer_postal_zone,
            COALESCE(cu.city_name, dcu.customer_city_name) AS linked_customer_city_name,
            COALESCE(cu.phone, dcu.customer_phone) AS linked_customer_phone,

            m.raw_body,
            m.clean_body,
            m.previous_context,
            m.metadata AS message_metadata,

            ar.provider,
            ar.decision_source,
            ar.action_decision,
            ar.action_result,
            ar.usage,
            ar.needs_review,

            re.event_type,
            re.payload AS raw_payload,
            re.processing_error,

            COALESCE(
                NULLIF(cu.name, ''),
                NULLIF(o.customer_name, ''),
                NULLIF(re.payload->'sender'->>'name', ''),
                NULLIF(re.payload->'conversation'->'meta'->'sender'->>'name', ''),
                NULLIF(o.customer_email, ''),
                NULLIF(cu.email, ''),
                NULLIF(t.customer_id, ''),
                NULLIF(t.contact_id, '')
            ) AS customer_name,

            COALESCE(
                NULLIF(cu.email, ''),
                NULLIF(o.customer_email, ''),
                NULLIF(re.payload->'sender'->>'email', ''),
                NULLIF(re.payload->'conversation'->'meta'->'sender'->>'email', ''),
                NULLIF(re.payload->'conversation'->'contact_inbox'->>'source_id', '')
            ) AS customer_email,

            COALESCE(
                NULLIF(re.payload->'sender'->>'phone_number', ''),
                NULLIF(re.payload->'conversation'->'meta'->'sender'->>'phone_number', '')
            ) AS customer_phone,

            COALESCE(
                NULLIF(re.payload->'conversation'->'additional_attributes'->>'mail_subject', ''),
                NULLIF(re.payload->'content_attributes'->'email'->>'subject', ''),
                NULLIF(re.payload->'conversation'->'messages'->0->'content_attributes'->'email'->>'subject', '')
            ) AS message_subject

        FROM tasks t
        LEFT JOIN messages m ON m.id = t.message_id
        LEFT JOIN action_runs ar ON ar.id = t.action_run_id
        LEFT JOIN raw_events re ON re.id = t.raw_event_id
        LEFT JOIN opportunities o ON o.id = t.opportunity_id
        LEFT JOIN LATERAL (
            SELECT
                cd.customer_id,
                COALESCE(cdoc.name, NULLIF(cd.company, ''), NULLIF(cd.payload->>'customer_name', '')) AS customer_name,
                COALESCE(cdoc.email, NULLIF(cd.payload->>'customer_email', ''), NULLIF(cd.payload->>'email', '')) AS customer_email,
                COALESCE(cdoc.tax_id, NULLIF(cd.payload->>'customer_tax_id', ''), NULLIF(cd.payload->>'tax_id', ''), NULLIF(cd.payload->>'nif', '')) AS customer_tax_id,
                cdoc.street_name AS customer_street_name,
                cdoc.postal_zone AS customer_postal_zone,
                cdoc.city_name AS customer_city_name,
                cdoc.phone AS customer_phone
            FROM commercial_documents cd
            LEFT JOIN customers cdoc ON cdoc.id = cd.customer_id
            WHERE cd.opportunity_id = o.id
              AND COALESCE(cd.is_active, TRUE) = TRUE
              AND cd.document_kind IN ('invoice', 'quotation', 'proforma')
            ORDER BY
              CASE cd.document_kind WHEN 'invoice' THEN 1 WHEN 'quotation' THEN 2 WHEN 'proforma' THEN 3 ELSE 4 END,
              CASE COALESCE(cd.role, 'current') WHEN 'current' THEN 1 WHEN 'accepted' THEN 2 WHEN 'historical' THEN 3 WHEN 'history' THEN 3 ELSE 4 END,
              COALESCE(cd.is_primary, FALSE) DESC,
              COALESCE(cd.document_date, cd.created_at::date) DESC,
              cd.created_at DESC
            LIMIT 1
        ) dcu ON TRUE
        -- Legacy guard anchor: LEFT JOIN customers cu ON cu.id = o.local_customer_id
        -- v4928.1.5.64: choose schema-supported opportunity customer, then document customer fallback.
        LEFT JOIN customers cu ON cu.id = o.{customer_column}
        WHERE t.id = CAST(:task_id AS UUID)
    """)

    with engine.begin() as conn:
        row = conn.execute(sql, {"task_id": task_id}).mappings().first()

    return dict(row) if row else None


def list_customer_task_history(
    *,
    contact_id: Optional[str] = None,
    conversation_id: Optional[str] = None,
    exclude_task_id: Optional[str] = None,
    limit: int = 10,
) -> List[Dict[str, Any]]:
    where = []
    params: Dict[str, Any] = {"limit": limit}

    if contact_id:
        where.append("contact_id = :contact_id")
        params["contact_id"] = contact_id

    if not contact_id and conversation_id:
        where.append("conversation_id = :conversation_id")
        params["conversation_id"] = conversation_id

    if exclude_task_id:
        where.append("id <> CAST(:exclude_task_id AS UUID)")
        params["exclude_task_id"] = exclude_task_id

    if not where:
        return []

    sql = text(f"""
        SELECT
            id::text,
            created_at,
            action_code,
            route,
            action,
            note,
            status,
            conversation_id,
            contact_id,
            source_system,
            done_at,
            done_by
        FROM tasks
        WHERE {" AND ".join(where)}
        ORDER BY created_at DESC
        LIMIT :limit
    """)

    with engine.begin() as conn:
        rows = conn.execute(sql, params).mappings().all()

    return [dict(row) for row in rows]



def get_external_mapping(
    *,
    local_system: str,
    local_entity_type: str,
    local_entity_id: str,
    external_system: str,
    external_entity_type: str,
) -> Optional[Dict[str, Any]]:
    sql = text("""
        SELECT
            id::text,
            local_system,
            local_entity_type,
            local_entity_id,
            external_system,
            external_entity_type,
            external_entity_id,
            external_url,
            match_key,
            match_value,
            confidence,
            metadata,
            created_at,
            updated_at
        FROM external_mappings
        WHERE local_system = :local_system
          AND local_entity_type = :local_entity_type
          AND local_entity_id = :local_entity_id
          AND external_system = :external_system
          AND external_entity_type = :external_entity_type
        ORDER BY updated_at DESC
        LIMIT 1
    """)

    with engine.begin() as conn:
        row = conn.execute(sql, {
            "local_system": local_system,
            "local_entity_type": local_entity_type,
            "local_entity_id": str(local_entity_id),
            "external_system": external_system,
            "external_entity_type": external_entity_type,
        }).mappings().first()

    return dict(row) if row else None


def get_customer_profile(contact_id: str) -> Optional[Dict[str, Any]]:
    sql = text("""
        WITH base AS (
            SELECT
                t.contact_id,
                t.conversation_id,
                t.created_at,
                t.status,
                COALESCE(
                    NULLIF(re.payload->'sender'->>'name', ''),
                    NULLIF(re.payload->'conversation'->'meta'->'sender'->>'name', ''),
                    NULLIF(t.customer_id, ''),
                    NULLIF(t.contact_id, '')
                ) AS customer_name,
                COALESCE(
                    NULLIF(re.payload->'sender'->>'email', ''),
                    NULLIF(re.payload->'conversation'->'meta'->'sender'->>'email', ''),
                    NULLIF(re.payload->'conversation'->'contact_inbox'->>'source_id', '')
                ) AS customer_email,
                COALESCE(
                    NULLIF(re.payload->'sender'->>'phone_number', ''),
                    NULLIF(re.payload->'conversation'->'meta'->'sender'->>'phone_number', '')
                ) AS customer_phone
            FROM tasks t
            LEFT JOIN raw_events re ON re.id = t.raw_event_id
            WHERE t.contact_id = :contact_id
        )
        SELECT
            :contact_id AS contact_id,
            COALESCE((array_agg(customer_name ORDER BY created_at DESC) FILTER (WHERE customer_name IS NOT NULL AND customer_name <> ''))[1], :contact_id) AS customer_name,
            COALESCE((array_agg(customer_email ORDER BY created_at DESC) FILTER (WHERE customer_email IS NOT NULL AND customer_email <> ''))[1], '') AS customer_email,
            COALESCE((array_agg(customer_phone ORDER BY created_at DESC) FILTER (WHERE customer_phone IS NOT NULL AND customer_phone <> ''))[1], '') AS customer_phone,
            COALESCE((array_agg(conversation_id ORDER BY created_at DESC) FILTER (WHERE conversation_id IS NOT NULL AND conversation_id <> ''))[1], '') AS latest_conversation_id,
            COUNT(*) AS task_count,
            COUNT(*) FILTER (WHERE status = 'pending') AS pending_count,
            COUNT(*) FILTER (WHERE status = 'done') AS done_count,
            COUNT(*) FILTER (WHERE status = 'skipped') AS skipped_count,
            MIN(created_at) AS first_seen_at,
            MAX(created_at) AS last_seen_at
        FROM base
    """)

    with engine.begin() as conn:
        row = conn.execute(sql, {"contact_id": str(contact_id)}).mappings().first()

    if not row:
        return None

    result = dict(row)
    if not result.get("task_count"):
        return None

    return result


def list_customer_tasks(contact_id: str, *, limit: int = 50) -> List[Dict[str, Any]]:
    sql = text("""
        SELECT
            id::text,
            created_at,
            updated_at,
            action_code,
            route,
            action,
            note,
            status,
            conversation_id,
            contact_id,
            source_system,
            done_at,
            done_by
        FROM tasks
        WHERE contact_id = :contact_id
        ORDER BY created_at DESC
        LIMIT :limit
    """)

    with engine.begin() as conn:
        rows = conn.execute(sql, {
            "contact_id": str(contact_id),
            "limit": limit,
        }).mappings().all()

    return [dict(row) for row in rows]


def list_customer_messages(contact_id: str, *, limit: int = 20) -> List[Dict[str, Any]]:
    sql = text("""
        SELECT
            id::text,
            created_at,
            conversation_id,
            contact_id,
            source_system,
            direction,
            raw_body,
            clean_body,
            previous_context,
            metadata
        FROM messages
        WHERE contact_id = :contact_id
        ORDER BY created_at DESC
        LIMIT :limit
    """)

    with engine.begin() as conn:
        rows = conn.execute(sql, {
            "contact_id": str(contact_id),
            "limit": limit,
        }).mappings().all()

    return [dict(row) for row in rows]


def complete_task_with_note(
    task_id: str,
    *,
    done_by: str = "admin",
    done_note: str = "",
) -> Optional[Dict[str, Any]]:
    sql = text("""
        UPDATE tasks
        SET
            status = 'done',
            done_at = now(),
            done_by = CAST(:done_by AS TEXT),
            updated_at = now(),
            metadata = COALESCE(metadata, '{}'::jsonb)
                || jsonb_build_object(
                    'done_note', CAST(:done_note AS TEXT),
                    'done_by', CAST(:done_by AS TEXT),
                    'done_at', now()
                )
        WHERE id = CAST(:task_id AS UUID)
        RETURNING
            id::text,
            status,
            done_at,
            done_by,
            metadata
    """)

    with engine.begin() as conn:
        row = conn.execute(sql, {
            "task_id": task_id,
            "done_by": done_by,
            "done_note": done_note or "",
        }).mappings().first()

    if row:
        try:
            from app.opportunity_service import advance_opportunity_after_task_done

            advance_opportunity_after_task_done(
                task_id,
                event_type="task_done_with_note",
                payload={"done_note": done_note or ""},
                created_by=done_by,
            )
        except Exception as exc:
            print(f"ClientFlow opportunity advance failed for task {task_id}: {exc}", flush=True)

        if str(done_by or "").lower() not in {"chatwoot_outgoing", "system"}:
            from app.operator_audit_service import record_operator_action_best_effort

            record_operator_action_best_effort(
                action="task_completed_with_note",
                entity_type="task",
                entity_id=task_id,
                task_id=task_id,
                actor=done_by,
                payload={"done_note": done_note or ""},
            )

    return dict(row) if row else None


def get_admin_dashboard_metrics() -> Dict[str, Any]:
    sql = text("""
        WITH task_stats AS (
            SELECT
                count(*) FILTER (WHERE status = 'pending') AS pending_total,
                count(*) FILTER (WHERE status = 'done') AS done_total,
                count(*) FILTER (WHERE status = 'skipped') AS skipped_total,
                count(*) FILTER (WHERE status = 'failed') AS failed_total,

                count(*) FILTER (WHERE created_at >= date_trunc('day', now())) AS created_today,
                count(*) FILTER (WHERE done_at >= date_trunc('day', now())) AS done_today,

                count(*) FILTER (WHERE status = 'pending' AND route = 'vendas') AS pending_vendas,
                count(*) FILTER (WHERE status = 'pending' AND route = 'financeiro') AS pending_financeiro,
                count(*) FILTER (WHERE status = 'pending' AND route = 'operacoes') AS pending_operacoes,
                count(*) FILTER (WHERE status = 'pending' AND route = 'suporte') AS pending_suporte,
                count(*) FILTER (WHERE status = 'pending' AND route = 'rever') AS pending_rever,

                count(*) FILTER (
                    WHERE status = 'pending'
                      AND (
                        (due_at IS NOT NULL AND due_at < now())
                        OR (
                          due_at IS NULL AND (
                            (route = 'suporte' AND created_at < now() - interval '2 hours')
                            OR (route = 'vendas' AND created_at < now() - interval '4 hours')
                            OR (route = 'financeiro' AND created_at < now() - interval '8 hours')
                            OR (route = 'operacoes' AND created_at < now() - interval '24 hours')
                            OR (route = 'rever' AND created_at < now() - interval '24 hours')
                          )
                        )
                      )
                ) AS overdue_total,
                count(*) FILTER (WHERE status = 'pending' AND action_code LIKE 'FOLLOW_UP_%') AS pending_followups,
                count(*) FILTER (WHERE status = 'pending' AND action_code LIKE 'FOLLOW_UP_%' AND due_at <= now()) AS due_followups
            FROM tasks
        ),
        raw_stats AS (
            SELECT
                count(*) FILTER (WHERE created_at >= now() - interval '24 hours') AS webhooks_24h,
                count(*) FILTER (WHERE ignored = true AND created_at >= now() - interval '24 hours') AS ignored_24h,
                count(*) FILTER (WHERE processing_error IS NOT NULL AND processing_error <> '' AND created_at >= now() - interval '24 hours') AS webhook_errors_24h,
                count(*) FILTER (
                    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'
                ) AS chatwoot_incoming_pending
            FROM raw_events
        ),
        outbox_stats AS (
            SELECT
                count(*) FILTER (WHERE status = 'pending') AS outbox_pending,
                count(*) FILTER (WHERE status = 'failed') AS outbox_failed,
                count(*) FILTER (WHERE status = 'sent') AS outbox_sent
            FROM integration_outbox
        )
        SELECT *
        FROM task_stats, raw_stats, outbox_stats
    """)

    with engine.begin() as conn:
        row = conn.execute(sql).mappings().first()

    return dict(row or {})


def list_admin_recent_tasks(*, limit: int = 8) -> List[Dict[str, Any]]:
    sql = text("""
        SELECT
            id::text,
            created_at,
            action_code,
            route,
            action,
            note,
            status,
            conversation_id,
            contact_id,
            source_system
        FROM tasks
        ORDER BY created_at DESC
        LIMIT :limit
    """)

    with engine.begin() as conn:
        rows = conn.execute(sql, {"limit": limit}).mappings().all()

    return [dict(row) for row in rows]


def list_admin_recent_raw_events(*, limit: int = 8) -> List[Dict[str, Any]]:
    sql = text("""
        SELECT
            created_at,
            source_system,
            event_type,
            source_event_id,
            conversation_id,
            processed,
            ignored,
            processing_error
        FROM raw_events
        ORDER BY created_at DESC
        LIMIT :limit
    """)

    with engine.begin() as conn:
        rows = conn.execute(sql, {"limit": limit}).mappings().all()

    return [dict(row) for row in rows]


def list_customer_opportunities(contact_id: str, *, limit: int = 20) -> List[Dict[str, Any]]:
    """Lista oportunidades internas ClientFlow associadas ao contacto.

    Substitui a antiga consulta de oportunidades mapeadas em CRM externo.
    """
    sql = text("""
        SELECT
            o.id::text,
            o.title,
            o.stage,
            o.status,
            o.product_interest,
            o.value_amount,
            o.currency,
            o.conversation_id,
            o.customer_name,
            o.customer_email,
            o.last_action_code,
            o.last_task_id::text,
            o.created_at,
            o.updated_at,
            count(t.id) AS task_count,
            count(t.id) FILTER (WHERE t.status = 'pending') AS pending_task_count
        FROM opportunities o
        LEFT JOIN tasks t ON t.opportunity_id = o.id
        WHERE o.contact_id = :contact_id
        GROUP BY o.id
        ORDER BY o.updated_at DESC
        LIMIT :limit
    """)

    with engine.begin() as conn:
        rows = conn.execute(sql, {
            "contact_id": str(contact_id),
            "limit": limit,
        }).mappings().all()

    return [dict(row) for row in rows]


# Compatibilidade para código antigo/imports anteriores.
list_customer_opportunity_mappings = list_customer_opportunities

def get_system_health_metrics() -> Dict[str, Any]:
    sql = text("""
        WITH task_stats AS (
            SELECT
                count(*) AS tasks_total,
                count(*) FILTER (WHERE status = 'pending') AS tasks_pending,
                count(*) FILTER (WHERE status = 'done') AS tasks_done,
                count(*) FILTER (WHERE status = 'failed') AS tasks_failed,
                count(*) FILTER (WHERE created_at >= now() - interval '24 hours') AS tasks_24h
            FROM tasks
        ),
        raw_stats AS (
            SELECT
                count(*) FILTER (WHERE created_at >= now() - interval '24 hours') AS webhooks_24h,
                count(*) FILTER (WHERE ignored = true AND created_at >= now() - interval '24 hours') AS webhooks_ignored_24h,
                count(*) FILTER (
                    WHERE processing_error IS NOT NULL
                      AND processing_error <> ''
                      AND created_at >= now() - interval '24 hours'
                ) AS webhooks_errors_24h
            FROM raw_events
        ),
        outbox_stats AS (
            SELECT
                count(*) FILTER (WHERE status = 'pending') AS outbox_pending,
                count(*) FILTER (WHERE status = 'failed') AS outbox_failed,
                count(*) FILTER (WHERE status = 'sent') AS outbox_sent
            FROM integration_outbox
            WHERE target_system <> ('t' || 'wenty')
        ),
        opportunity_stats AS (
            SELECT
                count(*) AS opportunities_total,
                count(*) FILTER (WHERE status = 'open') AS opportunities_open,
                count(*) FILTER (WHERE stage = 'WAITING_PAYMENT') AS opportunities_waiting_payment,
                count(*) FILTER (WHERE stage IN ('ORDER_PREPARATION', 'SHIPPED')) AS opportunities_operations
            FROM opportunities
        )
        SELECT *
        FROM task_stats, raw_stats, outbox_stats, opportunity_stats
    """)

    with engine.begin() as conn:
        row = conn.execute(sql).mappings().first()

    return dict(row or {})

def create_task_event(
    *,
    task_id: str,
    event_type: str,
    payload: Optional[Dict[str, Any]] = None,
    created_by: str = "operator",
) -> None:
    sql = text("""
        INSERT INTO task_events (
            task_id,
            event_type,
            payload,
            created_by
        )
        VALUES (
            CAST(:task_id AS UUID),
            CAST(:event_type AS TEXT),
            CAST(:payload AS JSONB),
            CAST(:created_by AS TEXT)
        )
    """)

    with engine.begin() as conn:
        conn.execute(sql, {
            "task_id": task_id,
            "event_type": event_type,
            "payload": _json(payload or {}),
            "created_by": created_by,
        })


def create_business_event(
    *,
    event_type: str,
    task: Dict[str, Any],
    payload: Optional[Dict[str, Any]] = None,
    created_by: str = "operator",
) -> Optional[str]:
    idempotency_key = f"business_event:{task['id']}:{event_type}"

    sql = text("""
        INSERT INTO business_events (
            event_type,
            task_id,
            action_run_id,
            message_id,
            raw_event_id,
            customer_id,
            conversation_id,
            contact_id,
            payload,
            idempotency_key,
            created_by
        )
        VALUES (
            CAST(:event_type AS TEXT),
            CAST(:task_id AS UUID),
            CAST(:action_run_id AS UUID),
            CAST(:message_id AS UUID),
            CAST(:raw_event_id AS UUID),
            :customer_id,
            :conversation_id,
            :contact_id,
            CAST(:payload AS JSONB),
            CAST(:idempotency_key AS TEXT),
            CAST(:created_by AS TEXT)
        )
        ON CONFLICT (idempotency_key) DO NOTHING
        RETURNING id::text
    """)

    with engine.begin() as conn:
        row = conn.execute(sql, {
            "event_type": event_type,
            "task_id": task["id"],
            "action_run_id": _uuid_or_none(task.get("action_run_id")),
            "message_id": _uuid_or_none(task.get("message_id")),
            "raw_event_id": _uuid_or_none(task.get("raw_event_id")),
            "customer_id": task.get("customer_id"),
            "conversation_id": task.get("conversation_id"),
            "contact_id": task.get("contact_id"),
            "payload": _json(payload or {}),
            "idempotency_key": idempotency_key,
            "created_by": created_by,
        }).fetchone()

    return row[0] if row else None


def create_next_task_after_business_event(
    *,
    event_type: str,
    task: Dict[str, Any],
    created_by: str = "system",
) -> Optional[str]:
    """Não cria action_codes operacionais legados.

    A continuação após pagamento/preparação passa por `business_events`,
    `operation_links` e pela página de operações da oportunidade.
    """
    return None


def complete_task(
    *,
    task_id: str,
    done_by: str = "operator",
    payload: Optional[Dict[str, Any]] = None,
) -> Dict[str, Any]:
    task = get_task(task_id)

    if task is None:
        return {"ok": False, "error": "task_not_found"}

    if task["status"] == "done":
        return {"ok": True, "status": "already_done", "task": task}

    with engine.begin() as conn:
        conn.execute(text("""
            UPDATE tasks
            SET
                status = 'done',
                done_at = now(),
                done_by = CAST(:done_by AS TEXT),
                updated_at = now()
            WHERE id = CAST(:task_id AS UUID)
        """), {
            "task_id": task_id,
            "done_by": done_by,
        })

    create_task_event(
        task_id=task_id,
        event_type="task_done",
        payload=payload or {},
        created_by=done_by,
    )

    if str(done_by or "") == "chatwoot_outgoing":
        create_task_event(
            task_id=task_id,
            event_type="task_auto_completed",
            payload=payload or {},
            created_by=done_by,
        )
    elif str(done_by or "").lower() != "system":
        from app.operator_audit_service import record_operator_action_best_effort

        record_operator_action_best_effort(
            action="task_completed",
            entity_type="task",
            entity_id=task_id,
            task_id=task_id,
            actor=done_by,
            before={"status": task.get("status")},
            after={"status": "done"},
            payload=payload or {},
        )

    config = get_action_config(task["action_code"])
    event_type = config.get("business_event_on_done")

    business_event_id = None
    next_task_id = None

    if event_type:
        business_event_id = create_business_event(
            event_type=event_type,
            task=task,
            payload={
                "task_action_code": task["action_code"],
                "task_action": task["action"],
                "task_note": task["note"],
                **(payload or {}),
            },
            created_by=done_by,
        )

        try:
            from app.integration_outbox_service import create_outbox_for_business_event

            create_outbox_for_business_event(
                business_event_id=business_event_id,
                event_type=event_type,
                task=task,
                payload={
                    "task_action_code": task["action_code"],
                    "task_action": task["action"],
                    "task_note": task["note"],
                    **(payload or {}),
                },
            )
        except Exception as exc:
            print(f"ClientFlow outbox creation failed: {exc}")

        next_task_id = create_next_task_after_business_event(
            event_type=event_type,
            task=task,
            created_by="system",
        )

    try:
        from app.opportunity_service import advance_opportunity_after_task_done

        advance_opportunity_after_task_done(
            task_id,
            event_type=event_type or "task_done",
            payload=payload or {},
            created_by=done_by,
        )
        timeline_event_type = "task_auto_completed" if str(done_by) == "chatwoot_outgoing" else "task_done"
        _record_task_timeline_event(task_id, event_type=timeline_event_type)
    except Exception as exc:
        print(f"ClientFlow opportunity advance failed for task {task_id}: {exc}", flush=True)

    return {
        "ok": True,
        "status": "done",
        "task_id": task_id,
        "business_event_id": business_event_id,
        "next_task_id": next_task_id,
    }


def skip_task(
    *,
    task_id: str,
    skipped_by: str = "operator",
    reason: str = "",
) -> Dict[str, Any]:
    task = get_task(task_id)

    if task is None:
        return {"ok": False, "error": "task_not_found"}

    with engine.begin() as conn:
        conn.execute(text("""
            UPDATE tasks
            SET
                status = 'skipped',
                updated_at = now()
            WHERE id = CAST(:task_id AS UUID)
        """), {"task_id": task_id})

    create_task_event(
        task_id=task_id,
        event_type="task_skipped",
        payload={"reason": reason},
        created_by=skipped_by,
    )

    from app.operator_audit_service import record_operator_action_best_effort

    auto_archive_result = None
    if str(task.get("action_code") or "").upper() == "IGNORE_SPAM" and task.get("opportunity_id"):
        try:
            from app.opportunity_service import archive_spam_opportunity_if_safe

            auto_archive_result = archive_spam_opportunity_if_safe(
                str(task.get("opportunity_id")),
                reason=reason or "spam_task_skipped",
                actor=skipped_by,
                source_task_id=task_id,
            )
        except Exception as exc:
            auto_archive_result = {"ok": False, "reason": f"archive_exception:{exc}"}

    record_operator_action_best_effort(
        action="task_skipped",
        entity_type="task",
        entity_id=task_id,
        task_id=task_id,
        actor=skipped_by,
        before={"status": task.get("status")},
        after={"status": "skipped"},
        payload={"reason": reason},
    )

    return {"ok": True, "status": "skipped", "task_id": task_id, "auto_archive_result": auto_archive_result}


def reschedule_task_due_at(
    *,
    task_id: str,
    delay_days: int = 2,
    rescheduled_by: str = "operator",
    reason: str = "",
) -> Dict[str, Any]:
    """Adia uma tarefa pendente usando due_at.

    Usado sobretudo por follow-ups semi-automáticos: o sistema mantém o
    controlo da data, mas o operador decide se adia ou fecha.
    """
    task = get_task(task_id)
    if task is None:
        return {"ok": False, "error": "task_not_found"}

    delay_days = max(0, min(int(delay_days or 0), 90))
    reason = str(reason or "").strip()

    with engine.begin() as conn:
        row = conn.execute(text("""
            UPDATE tasks
            SET
                due_at = now() + make_interval(days => :delay_days),
                updated_at = now(),
                metadata = COALESCE(metadata, '{}'::jsonb)
                    || jsonb_build_object(
                        'rescheduled_at', now(),
                        'rescheduled_by', CAST(:rescheduled_by AS TEXT),
                        'reschedule_reason', CAST(:reason AS TEXT),
                        'reschedule_delay_days', :delay_days
                    )
            WHERE id = CAST(:task_id AS UUID)
            RETURNING id::text, due_at, status
        """), {
            "task_id": task_id,
            "delay_days": delay_days,
            "rescheduled_by": rescheduled_by,
            "reason": reason,
        }).mappings().first()

    if not row:
        return {"ok": False, "error": "task_not_found"}

    create_task_event(
        task_id=task_id,
        event_type="task_rescheduled",
        payload={"delay_days": delay_days, "reason": reason},
        created_by=rescheduled_by,
    )

    try:
        from app.operator_audit_service import record_operator_action_best_effort

        record_operator_action_best_effort(
            action="task_rescheduled",
            entity_type="task",
            entity_id=task_id,
            task_id=task_id,
            actor=rescheduled_by,
            before={"due_at": task.get("due_at")},
            after={"due_at": row.get("due_at")},
            payload={"delay_days": delay_days, "reason": reason},
        )
    except Exception:
        pass

    return {"ok": True, "status": "rescheduled", "task_id": task_id, "due_at": row.get("due_at")}


def auto_complete_task_from_outgoing_message(
    *,
    conversation_id: Optional[str],
    contact_id: Optional[str] = None,
    outgoing_message_id: Optional[str] = None,
    outgoing_content: str = "",
    sender_name: Optional[str] = None,
    sender_type: Optional[str] = None,
    is_private: bool = False,
) -> Dict[str, Any]:
    """
    Quando um operador responde no Chatwoot, fecha automaticamente a tarefa pendente
    da mesma conversa, mas só para action_codes onde responder ao cliente é a ação.
    """

    enabled = os.getenv("CHATWOOT_AUTO_COMPLETE_ON_OUTGOING", "true").lower() in {
        "1", "true", "yes", "sim"
    }

    if not enabled:
        return {"ok": True, "status": "disabled"}

    if is_private:
        return {"ok": True, "status": "ignored_private_note"}

    if not conversation_id:
        return {"ok": False, "status": "missing_conversation_id"}

    content = str(outgoing_content or "").strip()
    if not content:
        return {"ok": True, "status": "ignored_empty_outgoing"}

    # Evitar que notas internas/automáticas do próprio ClientFlow fechem tarefas.
    lowered = content.lower()
    if "clientflow" in lowered and ("ação recomendada" in lowered or "acao recomendada" in lowered):
        return {"ok": True, "status": "ignored_clientflow_note"}

    # Guardrail v4.6.7 / v4.6.8:
    # completar automaticamente só ações em que a resposta no Chatwoot é a
    # própria execução da task. Há ações que nunca devem ser fechadas por uma
    # mensagem outgoing, mesmo que alguém as coloque por engano no ENV.
    default_codes = [
        "SEND_INFO",
        "SEND_QUOTE",
        "SEND_PROFORMA",
        "SEND_INVOICE",
        "SUPPORT",
        "FOLLOW_UP_QUOTE",
        "FOLLOW_UP_PROFORMA",
        "FOLLOW_UP_PAYMENT",
        "FOLLOW_UP_CUSTOMER_REVIEW",
        "FOLLOW_UP_GENERIC",
        "CONFIRM_DELIVERY",
    ]
    configured = os.getenv("CHATWOOT_AUTO_COMPLETE_ACTION_CODES", "").strip()
    never_auto_complete_codes = {
        "CONFIRM_PAYMENT",
        "CONFIRM_PAYMENT_AND_PREPARE_SHIPMENT",
        "MARK_NO_INTEREST",
        "REVIEW_MANUALLY",
        "PREPARE_ORDER",
        "CREATE_SHIPMENT",
        "REMOVE_FROM_LIST",
    }
    requested_codes = [
        c.strip().upper()
        for c in (configured.split(",") if configured else default_codes)
        if c.strip()
    ]
    blocked_by_policy = [c for c in requested_codes if c in never_auto_complete_codes]
    codes = [c for c in requested_codes if c not in never_auto_complete_codes]

    if blocked_by_policy:
        print(
            "ClientFlow auto-complete policy ignored action_codes="
            + ",".join(sorted(set(blocked_by_policy))),
            flush=True,
        )

    if not codes:
        return {
            "ok": True,
            "status": "no_action_codes_configured",
            "blocked_by_policy": sorted(set(blocked_by_policy)),
        }

    placeholders = ", ".join([f":code_{i}" for i in range(len(codes))])
    params: Dict[str, Any] = {
        "conversation_id": str(conversation_id),
        **{f"code_{i}": code for i, code in enumerate(codes)},
    }

    sql = text(f"""
        SELECT
            id::text,
            action_code,
            route,
            action,
            note,
            created_at,
            metadata
        FROM tasks
        WHERE status = 'pending'
          AND conversation_id = :conversation_id
          AND action_code IN ({placeholders})
        ORDER BY
            CASE route
                WHEN 'financeiro' THEN 1
                WHEN 'vendas' THEN 2
                WHEN 'suporte' THEN 3
                ELSE 4
            END,
            created_at DESC
        LIMIT 1
    """)

    with engine.begin() as conn:
        task = conn.execute(sql, params).mappings().first()

    if not task:
        return {
            "ok": True,
            "status": "no_matching_pending_task",
            "conversation_id": conversation_id,
        }

    metadata = task.get("metadata") or {}
    if isinstance(metadata, str):
        try:
            metadata = json.loads(metadata)
        except Exception:
            metadata = {}
    if str(metadata.get("opportunity_linking_status") or "") == "ambiguous":
        return {
            "ok": True,
            "status": "blocked_ambiguous_opportunity",
            "reason": "operator_must_confirm_opportunity_before_auto_complete",
            "task_id": task["id"],
            "action_code": task["action_code"],
            "route": task["route"],
            "conversation_id": conversation_id,
        }

    payload = {
        "auto_completed": True,
        "reason": "operator_outgoing_message_in_chatwoot",
        "outgoing_message_id": outgoing_message_id,
        "outgoing_content_preview": content[:500],
        "sender_name": sender_name,
        "sender_type": sender_type,
        "conversation_id": conversation_id,
        "contact_id": contact_id,
    }

    result = complete_task(
        task_id=task["id"],
        done_by="chatwoot_outgoing",
        payload=payload,
    )

    return {
        "ok": result.get("ok", False),
        "status": "auto_completed" if result.get("ok") else "failed",
        "task_id": task["id"],
        "action_code": task["action_code"],
        "route": task["route"],
        "conversation_id": conversation_id,
        "result": result,
    }



def get_latest_task_preparation(task_id: str) -> Optional[Dict[str, Any]]:
    sql = text("""
        select
          id::text,
          task_id::text,
          conversation_id,
          contact_id,
          prep_type,
          status,
          extracted_data,
          missing_fields,
          suggested_reply,
          confidence,
          model,
          provider,
          total_tokens,
          cost,
          created_at,
          updated_at
        from task_preparations
        where task_id = :task_id
        order by created_at desc
        limit 1
    """)

    with engine.begin() as conn:
        row = conn.execute(sql, {"task_id": task_id}).mappings().first()

    return dict(row) if row else None



ACTION_ROUTE_MAP = {
    "SEND_INFO": ("vendas", "Enviar informação ao cliente"),
    "SEND_QUOTE": ("vendas", "Enviar proposta/cotação"),
    "SEND_PROFORMA": ("financeiro", "Enviar orçamento para pagamento"),
    "SEND_INVOICE": ("financeiro", "Enviar fatura ao cliente"),
    "CONFIRM_PAYMENT": ("financeiro", "Confirmar pagamento"),
    "PREPARE_ORDER": ("operacoes", "Preparar encomenda / Odoo"),
    "VALIDATE_PHYSICAL_ORDER": ("logistica", "Validar encomenda física"),
    "CREATE_SHIPMENT": ("logistica", "Enviar encomenda"),
    "REVIEW_RECONSTRUCTED_PROCESS": ("rever", "Validar processo reconstruído"),
    "SUPPORT": ("suporte", "Responder ao pedido de suporte"),
    "REMOVE_FROM_LIST": ("marketing", "Remover contacto da lista"),
    "MARK_NO_INTEREST": ("vendas", "Marcar sem interesse"),
    "NO_ACTION": ("rever", "Sem ação necessária"),
    "REVIEW_MANUALLY": ("rever", "Rever manualmente"),
    "IGNORE_SPAM": ("spam", "Ignorar spam"),
    "FOLLOW_UP_QUOTE": ("vendas", "Fazer follow-up do orçamento"),
    "FOLLOW_UP_PROFORMA": ("financeiro", "Fazer follow-up do orçamento para pagamento"),
    "FOLLOW_UP_PAYMENT": ("financeiro", "Fazer follow-up de pagamento"),
    "FOLLOW_UP_CUSTOMER_REVIEW": ("vendas", "Fazer follow-up da informação enviada"),
    "FOLLOW_UP_GENERIC": ("vendas", "Fazer follow-up manual"),
    "CONFIRM_DELIVERY": ("vendas", "Confirmar receção da comunicação"),
    "RECOVER_OPPORTUNITY": ("vendas", "Recuperar oportunidade sem resposta"),
    "REVIEW_NURTURE": ("vendas", "Rever acompanhamento futuro"),
}


def reclassify_task(
    *,
    task_id: str,
    new_action_code: str,
    reason: str = "",
    reclassified_by: str = "operator",
    reopen: bool = True,
) -> Dict[str, Any]:
    import json as _json

    new_action_code = str(new_action_code or "").strip().upper()

    if new_action_code not in ACTION_ROUTE_MAP:
        raise ValueError(f"Unsupported action_code: {new_action_code}")

    new_route, new_action = ACTION_ROUTE_MAP[new_action_code]

    # Só spam/sem ação saem do trabalho operacional.
    # REVIEW_MANUALLY e REMOVE_FROM_LIST devem ficar pendentes para o operador.
    new_status = "skipped" if new_action_code in {"IGNORE_SPAM", "NO_ACTION"} else "pending"
    new_note = reason.strip() or f"Reclassificado manualmente para {new_action_code}."

    with engine.begin() as conn:
        old = conn.execute(
            text("""
                select
                  id::text,
                  action_code,
                  route,
                  action,
                  note,
                  status,
                  opportunity_id::text as opportunity_id
                from tasks
                where id = :task_id
                limit 1
            """),
            {"task_id": task_id},
        ).mappings().first()

        if not old:
            return {"ok": False, "status": "not_found", "task_id": task_id}

        metadata_patch = {
            "manual_reclassified": True,
            "manual_reclassified_at": "now",
            "manual_reclassified_by": reclassified_by,
            "manual_reclassified_reason": reason,
            "previous_action_code": old["action_code"],
            "previous_route": old["route"],
            "previous_status": old["status"],
        }

        event_payload = {
            "old": {
                "action_code": old["action_code"],
                "route": old["route"],
                "status": old["status"],
            },
            "new": {
                "action_code": new_action_code,
                "route": new_route,
                "status": new_status,
            },
            "reason": reason,
        }

        conn.execute(
            text("""
                update tasks
                set
                  action_code = :action_code,
                  route = :route,
                  action = :action,
                  note = :note,
                  status = :status,
                  done_at = null,
                  done_by = null,
                  metadata = coalesce(metadata, '{}'::jsonb)
                    || cast(:metadata_patch as jsonb)
                    || jsonb_build_object('manual_reclassified_at', now()),
                  updated_at = now()
                where id = :task_id
            """),
            {
                "task_id": task_id,
                "action_code": new_action_code,
                "route": new_route,
                "action": new_action,
                "note": new_note,
                "status": new_status,
                "metadata_patch": _json.dumps(metadata_patch, ensure_ascii=False),
            },
        )

        conn.execute(
            text("""
                insert into task_events (
                  task_id,
                  event_type,
                  created_by,
                  payload
                )
                values (
                  :task_id,
                  'task_reclassified',
                  :created_by,
                  cast(:payload as jsonb)
                )
            """),
            {
                "task_id": task_id,
                "created_by": reclassified_by,
                "payload": _json.dumps(event_payload, ensure_ascii=False),
            },
        )

    from app.operator_audit_service import record_operator_action_best_effort

    auto_archive_result = None
    if new_action_code == "IGNORE_SPAM" and old.get("opportunity_id"):
        try:
            from app.opportunity_service import archive_spam_opportunity_if_safe

            auto_archive_result = archive_spam_opportunity_if_safe(
                str(old.get("opportunity_id")),
                reason=reason or "task_reclassified_as_spam",
                actor=reclassified_by,
                source_task_id=task_id,
            )
        except Exception as exc:
            auto_archive_result = {"ok": False, "reason": f"archive_exception:{exc}"}

    record_operator_action_best_effort(
        action="task_reclassified",
        entity_type="task",
        entity_id=task_id,
        task_id=task_id,
        actor=reclassified_by,
        before={"action_code": old["action_code"], "route": old["route"], "status": old["status"]},
        after={"action_code": new_action_code, "route": new_route, "status": new_status},
        payload={"reason": reason},
    )

    return {
        "ok": True,
        "status": "reclassified",
        "task_id": task_id,
        "old_action_code": old["action_code"],
        "old_route": old["route"],
        "old_status": old["status"],
        "new_action_code": new_action_code,
        "new_route": new_route,
        "new_status": new_status,
        "auto_archive_result": auto_archive_result,
    }





def list_tasks_board_v3(
    *,
    status: str = "pending",
    route: str = "",
    q: str = "",
    limit: int = 200,
) -> List[Dict[str, Any]]:
    filters = []
    params: Dict[str, Any] = {"limit": int(limit)}

    status = str(status or "").strip()
    route = str(route or "").strip()
    q = str(q or "").strip()

    if status and status != "all":
        filters.append("t.status = :status")
        params["status"] = status

    if route and route != "all":
        filters.append("t.route = :route")
        params["route"] = route

    if q:
        filters.append("""
            (
              t.conversation_id ilike :q
              or coalesce(t.contact_id, '') ilike :q
              or coalesce(t.action_code, '') ilike :q
              or coalesce(t.action, '') ilike :q
              or coalesce(t.note, '') ilike :q
            )
        """)
        params["q"] = f"%{q}%"

    where_sql = " and ".join(filters) if filters else "true"

    sql = text(f"""
        select
          t.id::text,
          t.created_at,
          t.updated_at,
          t.conversation_id,
          t.contact_id,
          t.action_code,
          t.route,
          t.status,
          t.action,
          t.note,
          t.done_by,
          t.done_at,
          coalesce(t.metadata, '{{}}'::jsonb) as metadata
        from tasks t
        where {where_sql}
        order by
          case t.status
            when 'pending' then 1
            when 'skipped' then 2
            when 'done' then 3
            else 4
          end,
          case t.route
            when 'financeiro' then 1
            when 'vendas' then 2
            when 'suporte' then 3
            when 'rever' then 4
            when 'spam' then 5
            else 9
          end,
          t.created_at desc
        limit :limit
    """)

    with engine.begin() as conn:
        rows = conn.execute(sql, params).mappings().all()

    return [dict(r) for r in rows]
