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", "CREATE_SHIPMENT"}: return "alta" if code.startswith("FOLLOW_UP_"): 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') 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]]: 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 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} 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", ] 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"), "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"), } 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 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 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, } 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]