"""Communication/inbox-classification helpers for ClientFlow v4.5. A communication is the durable record of an inbound/outbound email, Chatwoot message or other customer message. Tasks and outbox items are the actions that come from it; this module deliberately keeps the original communication separate from human work and automation work. """ from __future__ import annotations import json from typing import Any, Dict, List, Optional from sqlalchemy import text from app.db import engine COMMUNICATION_STATUSES = { "new", "classified", "needs_review", "linked", "task_created", "done", "ignored", } ACTION_CLASSIFICATION_MAP: dict[str, tuple[str, str, str]] = { "pedido_orcamento": ("Comercial", "Criar/validar oportunidade", "cf-chip-blue"), "pedido_informacao": ("Comercial", "Responder pedido de informação", "cf-chip-blue"), "aceitacao_orcamento": ("Financeiro", "Converter em fatura/pró-forma", "cf-chip-green"), "comprovativo_pagamento": ("Financeiro", "Confirmar pagamento", "cf-chip-green"), "pedido_fatura": ("Financeiro", "Emitir/enviar fatura", "cf-chip-green"), "dados_fiscais": ("Financeiro", "Atualizar dados fiscais", "cf-chip-green"), "pedido_tracking": ("Operações", "Verificar envio/tracking", "cf-chip-purple"), "reclamacao": ("Suporte", "Responder reclamação", "cf-chip-orange"), "pedido_remocao_lista": ("Marketing", "Remover contacto da lista", "cf-chip-gray"), } def ensure_communication_schema() -> None: """Create/upgrade v4.5 communications and timeline tables. The statements are additive to be safe on field deployments. """ with engine.begin() as conn: conn.execute(text("CREATE EXTENSION IF NOT EXISTS pgcrypto")) conn.execute(text(""" CREATE TABLE IF NOT EXISTS communications ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), source_system TEXT NOT NULL DEFAULT 'email', source_message_id TEXT, thread_id TEXT, conversation_id TEXT, contact_id TEXT, direction TEXT NOT NULL DEFAULT 'inbound', sender_name TEXT, sender_email TEXT, recipient TEXT, subject TEXT, body TEXT, classification TEXT, confidence NUMERIC(4,3), status TEXT NOT NULL DEFAULT 'new', customer_id UUID, opportunity_id UUID, task_id UUID, metadata JSONB NOT NULL DEFAULT '{}'::jsonb, created_at TIMESTAMPTZ NOT NULL DEFAULT now(), updated_at TIMESTAMPTZ NOT NULL DEFAULT now() ) """)) for statement in [ "ALTER TABLE communications ADD COLUMN IF NOT EXISTS source_system TEXT NOT NULL DEFAULT 'email'", "ALTER TABLE communications ADD COLUMN IF NOT EXISTS source_message_id TEXT", "ALTER TABLE communications ADD COLUMN IF NOT EXISTS thread_id TEXT", "ALTER TABLE communications ADD COLUMN IF NOT EXISTS conversation_id TEXT", "ALTER TABLE communications ADD COLUMN IF NOT EXISTS contact_id TEXT", "ALTER TABLE communications ADD COLUMN IF NOT EXISTS direction TEXT NOT NULL DEFAULT 'inbound'", "ALTER TABLE communications ADD COLUMN IF NOT EXISTS sender_name TEXT", "ALTER TABLE communications ADD COLUMN IF NOT EXISTS sender_email TEXT", "ALTER TABLE communications ADD COLUMN IF NOT EXISTS recipient TEXT", "ALTER TABLE communications ADD COLUMN IF NOT EXISTS subject TEXT", "ALTER TABLE communications ADD COLUMN IF NOT EXISTS body TEXT", "ALTER TABLE communications ADD COLUMN IF NOT EXISTS classification TEXT", "ALTER TABLE communications ADD COLUMN IF NOT EXISTS confidence NUMERIC(4,3)", "ALTER TABLE communications ADD COLUMN IF NOT EXISTS status TEXT NOT NULL DEFAULT 'new'", "ALTER TABLE communications ADD COLUMN IF NOT EXISTS customer_id UUID", "ALTER TABLE communications ADD COLUMN IF NOT EXISTS opportunity_id UUID", "ALTER TABLE communications ADD COLUMN IF NOT EXISTS task_id UUID", "ALTER TABLE communications ADD COLUMN IF NOT EXISTS metadata JSONB NOT NULL DEFAULT '{}'::jsonb", "ALTER TABLE communications ADD COLUMN IF NOT EXISTS created_at TIMESTAMPTZ NOT NULL DEFAULT now()", "ALTER TABLE communications ADD COLUMN IF NOT EXISTS updated_at TIMESTAMPTZ NOT NULL DEFAULT now()", ]: conn.execute(text(statement)) conn.execute(text(""" CREATE UNIQUE INDEX IF NOT EXISTS ux_communications_source_message ON communications(source_system, source_message_id) WHERE source_message_id IS NOT NULL """)) conn.execute(text("CREATE INDEX IF NOT EXISTS idx_communications_created ON communications(created_at DESC)")) conn.execute(text("CREATE INDEX IF NOT EXISTS idx_communications_status ON communications(status)")) conn.execute(text("CREATE INDEX IF NOT EXISTS idx_communications_classification ON communications(classification)")) conn.execute(text("CREATE INDEX IF NOT EXISTS idx_communications_sender_email ON communications(sender_email)")) conn.execute(text("CREATE INDEX IF NOT EXISTS idx_communications_customer ON communications(customer_id)")) conn.execute(text("CREATE INDEX IF NOT EXISTS idx_communications_opportunity ON communications(opportunity_id)")) # v4.5 task context columns. Existing code still uses action/route; these # columns let the UI connect a task back to the communication/document/etc. for statement in [ "ALTER TABLE tasks ADD COLUMN IF NOT EXISTS communication_id UUID", "ALTER TABLE tasks ADD COLUMN IF NOT EXISTS document_id UUID", "ALTER TABLE tasks ADD COLUMN IF NOT EXISTS shipment_id UUID", "ALTER TABLE tasks ADD COLUMN IF NOT EXISTS outbox_id UUID", "ALTER TABLE tasks ADD COLUMN IF NOT EXISTS priority TEXT NOT NULL DEFAULT 'normal'", "ALTER TABLE tasks ADD COLUMN IF NOT EXISTS assigned_to TEXT", ]: conn.execute(text(statement)) conn.execute(text("CREATE INDEX IF NOT EXISTS idx_tasks_communication ON tasks(communication_id)")) conn.execute(text("CREATE INDEX IF NOT EXISTS idx_tasks_document ON tasks(document_id)")) conn.execute(text("CREATE INDEX IF NOT EXISTS idx_tasks_shipment ON tasks(shipment_id)")) conn.execute(text("CREATE INDEX IF NOT EXISTS idx_tasks_outbox ON tasks(outbox_id)")) conn.execute(text("CREATE INDEX IF NOT EXISTS idx_tasks_priority ON tasks(priority)")) conn.execute(text("CREATE INDEX IF NOT EXISTS idx_tasks_due_status ON tasks(status, due_at)")) conn.execute(text(""" CREATE TABLE IF NOT EXISTS timeline_events ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), opportunity_id UUID, customer_id UUID, event_type TEXT NOT NULL, title TEXT NOT NULL, description TEXT, source TEXT NOT NULL DEFAULT 'clientflow', related_type TEXT, related_id TEXT, payload JSONB NOT NULL DEFAULT '{}'::jsonb, created_by TEXT NOT NULL DEFAULT 'system', created_at TIMESTAMPTZ NOT NULL DEFAULT now() ) """)) conn.execute(text("CREATE INDEX IF NOT EXISTS idx_timeline_opportunity ON timeline_events(opportunity_id, created_at DESC)")) conn.execute(text("CREATE INDEX IF NOT EXISTS idx_timeline_customer ON timeline_events(customer_id, created_at DESC)")) conn.execute(text("CREATE INDEX IF NOT EXISTS idx_timeline_related ON timeline_events(related_type, related_id)")) def _row_dict(row: Any) -> Dict[str, Any]: return dict(row) if row is not None else {} def normalize_status(status: Optional[str]) -> Optional[str]: if not status: return None status = status.strip().lower() return status if status in COMMUNICATION_STATUSES else None def list_communications( *, status: Optional[str] = None, classification: Optional[str] = None, q: Optional[str] = None, opportunity_id: Optional[str] = None, limit: int = 50, ) -> List[Dict[str, Any]]: """List classified inbox items with optional filters.""" filters = [] params: Dict[str, Any] = {"limit": int(limit)} if normalize_status(status): filters.append("c.status = :status") params["status"] = normalize_status(status) if classification: filters.append("c.classification = :classification") params["classification"] = classification if opportunity_id: filters.append("c.opportunity_id = CAST(:opportunity_id AS UUID)") params["opportunity_id"] = opportunity_id if q: filters.append("(c.sender_email ILIKE :q OR c.sender_name ILIKE :q OR c.subject ILIKE :q OR c.body ILIKE :q OR c.classification ILIKE :q)") params["q"] = f"%{q}%" where = "WHERE " + " AND ".join(filters) if filters else "" with engine.begin() as conn: rows = conn.execute(text(f""" SELECT c.id::text, c.source_system, c.source_message_id, c.thread_id, c.conversation_id, c.contact_id, c.direction, c.sender_name, c.sender_email, c.recipient, c.subject, c.body, c.classification, c.confidence, c.status, c.customer_id::text, c.opportunity_id::text, c.task_id::text, c.metadata, c.created_at, c.updated_at, cu.name AS customer_name, o.title AS opportunity_title FROM communications c LEFT JOIN customers cu ON cu.id = c.customer_id LEFT JOIN opportunities o ON o.id = c.opportunity_id {where} ORDER BY c.created_at DESC LIMIT :limit """), params).mappings().all() return [dict(row) for row in rows] def _list_message_backed_chatwoot_items_for_opportunity(opportunity_id: str, limit: int = 20) -> List[Dict[str, Any]]: """Return inbound Chatwoot messages visible from raw_events/messages. Some historical Chatwoot ingestions created `messages`/`raw_events` and tasks/opportunities but did not create rows in the later `communications` inbox table. The opportunity detail must still show those messages; hiding them makes the page say a conversation is linked while "no messages" exist. """ if not str(opportunity_id or "").strip(): return [] with engine.begin() as conn: rows = conn.execute(text(""" WITH opp AS ( SELECT id, conversation_id, contact_id FROM opportunities WHERE id = CAST(:opportunity_id AS UUID) LIMIT 1 ), ranked AS ( SELECT DISTINCT ON (m.id) m.id::text AS id, m.source_system, COALESCE(re.source_event_id, m.source_event_id, m.id::text) AS source_message_id, m.conversation_id, m.contact_id, m.direction, COALESCE( re.payload #>> '{sender,name}', re.payload #>> '{sender,available_name}', re.payload #>> '{sender,display_name}', re.payload #>> '{contact,name}', re.payload #>> '{conversation,contact,name}', re.payload #>> '{message,sender,name}' ) AS sender_name, COALESCE( re.payload #>> '{sender,email}', re.payload #>> '{contact,email}', re.payload #>> '{conversation,contact,email}', re.payload #>> '{message,sender,email}' ) AS sender_email, NULL::text AS recipient, COALESCE( re.payload #>> '{message,content_attributes,email,subject}', re.payload #>> '{content_attributes,email,subject}', re.payload #>> '{message,content_attributes,subject}', re.payload #>> '{content_attributes,subject}', re.payload #>> '{conversation,additional_attributes,mail_subject}', 'Chatwoot #' || COALESCE(re.source_event_id, m.source_event_id, m.id::text) ) AS subject, COALESCE(m.clean_body, m.raw_body, re.payload #>> '{content}', re.payload #>> '{message,content}') AS body, COALESCE(ar.action_result ->> 'action_code', ar.action_decision ->> 'action_code') AS classification, COALESCE(t.status, 'indexed') AS status, NULL::text AS customer_id, COALESCE(t.opportunity_id::text, opp.id::text) AS opportunity_id, t.id::text AS task_id, jsonb_build_object( 'source_kind', 'message_raw_event', 'raw_event_id', re.id::text, 'task_id', t.id::text, 'linked_by', CASE WHEN t.opportunity_id = opp.id THEN 'task_opportunity' WHEN NULLIF(opp.conversation_id, '') IS NOT NULL AND m.conversation_id = opp.conversation_id THEN 'conversation_id' WHEN NULLIF(opp.conversation_id, '') IS NOT NULL AND re.conversation_id = opp.conversation_id THEN 'raw_event_conversation_id' ELSE 'unknown' END ) AS metadata, m.created_at, m.created_at AS updated_at FROM opp JOIN messages m ON m.source_system = 'chatwoot' LEFT JOIN raw_events re ON re.id = m.raw_event_id OR re.message_id = m.id LEFT JOIN action_runs ar ON ar.message_id = m.id OR ar.raw_event_id = re.id LEFT JOIN tasks t ON t.message_id = m.id OR t.raw_event_id = re.id OR t.action_run_id = ar.id WHERE t.opportunity_id = opp.id OR ( NULLIF(opp.conversation_id, '') IS NOT NULL AND (m.conversation_id = opp.conversation_id OR re.conversation_id = opp.conversation_id) ) ORDER BY m.id, COALESCE(t.created_at, m.created_at) DESC ) SELECT * FROM ranked ORDER BY created_at DESC LIMIT :limit """), {"opportunity_id": opportunity_id, "limit": max(1, int(limit or 20))}).mappings().all() return [dict(row) for row in rows] def list_communications_for_opportunity(opportunity_id: str, limit: int = 20) -> List[Dict[str, Any]]: canonical = list_communications(opportunity_id=opportunity_id, limit=limit) fallback = _list_message_backed_chatwoot_items_for_opportunity(opportunity_id, limit=limit) seen: set[str] = set() combined: List[Dict[str, Any]] = [] for item in canonical + fallback: key = f"{item.get('source_system')}:{item.get('source_message_id') or item.get('id')}" if key in seen: continue seen.add(key) combined.append(item) combined.sort(key=lambda row: str(row.get("created_at") or ""), reverse=True) return combined[: max(1, int(limit or 20))] def get_communication(communication_id: str) -> Optional[Dict[str, Any]]: with engine.begin() as conn: row = conn.execute(text(""" SELECT c.id::text, c.source_system, c.source_message_id, c.thread_id, c.conversation_id, c.contact_id, c.direction, c.sender_name, c.sender_email, c.recipient, c.subject, c.body, c.classification, c.confidence, c.status, c.customer_id::text, c.opportunity_id::text, c.task_id::text, c.metadata, c.created_at, c.updated_at, cu.name AS customer_name, o.title AS opportunity_title FROM communications c LEFT JOIN customers cu ON cu.id = c.customer_id LEFT JOIN opportunities o ON o.id = c.opportunity_id WHERE c.id = CAST(:id AS UUID) """), {"id": communication_id}).mappings().first() return dict(row) if row else None def get_communications_summary() -> Dict[str, int]: with engine.begin() as conn: row = conn.execute(text(""" SELECT COUNT(*)::int AS total, COUNT(*) FILTER (WHERE status IN ('new','classified','needs_review'))::int AS open, COUNT(*) FILTER (WHERE status = 'needs_review')::int AS needs_review, COUNT(*) FILTER (WHERE customer_id IS NULL AND direction = 'inbound')::int AS without_customer, COUNT(*) FILTER (WHERE opportunity_id IS NULL AND direction = 'inbound')::int AS without_opportunity, COUNT(*) FILTER (WHERE task_id IS NOT NULL)::int AS with_task, COUNT(*) FILTER (WHERE created_at >= now() - interval '24 hours')::int AS last_24h FROM communications """)).mappings().first() return {k: int(v or 0) for k, v in dict(row or {}).items()} def set_communication_status(communication_id: str, status: str) -> None: status = normalize_status(status) if not status: raise ValueError("Estado de comunicação inválido.") with engine.begin() as conn: conn.execute(text(""" UPDATE communications SET status = :status, updated_at = now() WHERE id = CAST(:id AS UUID) """), {"id": communication_id, "status": status}) def link_communication_to_customer(communication_id: str, customer_id: Optional[str]) -> None: with engine.begin() as conn: conn.execute(text(""" UPDATE communications SET customer_id = CASE WHEN :customer_id = '' THEN NULL ELSE CAST(:customer_id AS UUID) END, status = CASE WHEN status IN ('new','classified','needs_review') THEN 'linked' ELSE status END, updated_at = now() WHERE id = CAST(:id AS UUID) """), {"id": communication_id, "customer_id": customer_id or ""}) def link_communication_to_opportunity(communication_id: str, opportunity_id: Optional[str]) -> None: with engine.begin() as conn: conn.execute(text(""" UPDATE communications SET opportunity_id = CASE WHEN :opportunity_id = '' THEN NULL ELSE CAST(:opportunity_id AS UUID) END, status = CASE WHEN status IN ('new','classified','needs_review') THEN 'linked' ELSE status END, updated_at = now() WHERE id = CAST(:id AS UUID) """), {"id": communication_id, "opportunity_id": opportunity_id or ""}) def record_outbound_communication( *, source_system: str = "chatwoot", source_message_id: Optional[str] = None, conversation_id: Optional[str] = None, contact_id: Optional[str] = None, body: str, customer_id: Optional[str] = None, opportunity_id: Optional[str] = None, task_id: Optional[str] = None, metadata: Optional[Dict[str, Any]] = None, ) -> Optional[str]: """Persist an outbound customer message in ClientFlow. This makes the opportunity timeline auditable even when Chatwoot remains the transport layer. """ ensure_communication_schema() with engine.begin() as conn: row = conn.execute(text(""" INSERT INTO communications ( source_system, source_message_id, conversation_id, contact_id, direction, body, status, customer_id, opportunity_id, task_id, metadata ) VALUES ( :source_system, :source_message_id, :conversation_id, :contact_id, 'outbound', :body, 'done', CASE WHEN :customer_id = '' THEN NULL ELSE CAST(:customer_id AS UUID) END, CASE WHEN :opportunity_id = '' THEN NULL ELSE CAST(:opportunity_id AS UUID) END, CASE WHEN :task_id = '' THEN NULL ELSE CAST(:task_id AS UUID) END, CAST(:metadata AS JSONB) ) ON CONFLICT (source_system, source_message_id) WHERE source_message_id IS NOT NULL DO UPDATE SET body = EXCLUDED.body, status = 'done', customer_id = COALESCE(EXCLUDED.customer_id, communications.customer_id), opportunity_id = COALESCE(EXCLUDED.opportunity_id, communications.opportunity_id), task_id = COALESCE(EXCLUDED.task_id, communications.task_id), metadata = COALESCE(communications.metadata, '{}'::jsonb) || EXCLUDED.metadata, updated_at = now() RETURNING id::text """), { "source_system": source_system or "chatwoot", "source_message_id": source_message_id, "conversation_id": conversation_id or "", "contact_id": contact_id or "", "body": body or "", "customer_id": customer_id or "", "opportunity_id": opportunity_id or "", "task_id": task_id or "", "metadata": json.dumps(metadata or {}, ensure_ascii=False, default=str), }).first() return str(row[0]) if row else None def create_timeline_event( *, opportunity_id: Optional[str] = None, customer_id: Optional[str] = None, event_type: str, title: str, description: str = "", source: str = "clientflow", related_type: Optional[str] = None, related_id: Optional[str] = None, payload: Optional[Dict[str, Any]] = None, created_by: str = "system", ) -> Optional[str]: if not opportunity_id and not customer_id: return None with engine.begin() as conn: row = conn.execute(text(""" INSERT INTO timeline_events ( opportunity_id, customer_id, event_type, title, description, source, related_type, related_id, payload, created_by ) VALUES ( CASE WHEN :opportunity_id = '' THEN NULL ELSE CAST(:opportunity_id AS UUID) END, CASE WHEN :customer_id = '' THEN NULL ELSE CAST(:customer_id AS UUID) END, :event_type, :title, :description, :source, :related_type, :related_id, CAST(:payload AS JSONB), :created_by ) RETURNING id::text """), { "opportunity_id": opportunity_id or "", "customer_id": customer_id or "", "event_type": event_type, "title": title, "description": description, "source": source, "related_type": related_type, "related_id": related_id, "payload": json.dumps(payload or {}, ensure_ascii=False), "created_by": created_by, }).first() return str(row[0]) if row else None def classification_action(classification: Optional[str]) -> Dict[str, str]: key = (classification or "").strip().lower() queue, action, chip = ACTION_CLASSIFICATION_MAP.get(key, ("Revisão", "Rever classificação", "cf-chip-orange")) return {"queue": queue, "action": action, "chip": chip}