clientflow_backend/scripts/audit_task_identity_mismatch.py

#!/usr/bin/env python3
"""Audit and optionally detach unsafe fiscal-customer associations.

Default mode is read-only.  The audit is intentionally conservative: it only
flags a task when an opportunity has a linked fiscal customer and that fiscal
customer clearly does not match the process/company implied by the task title.

It does NOT flag normal B2B situations where the process is a company and the
Chatwoot sender is a person from that company, and it does NOT detach
opportunities that already have no fiscal customer.

Use:
  python scripts/audit_task_identity_mismatch.py
  python scripts/audit_task_identity_mismatch.py --apply
"""
from __future__ import annotations

import argparse
import json
import re
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

from dotenv import load_dotenv
from sqlalchemy import text

PROJECT_ROOT = Path(__file__).resolve().parents[1]
load_dotenv(PROJECT_ROOT / ".env")

from app.db import engine  # noqa: E402

STOPWORDS = {
    "da", "de", "do", "dos", "das", "e", "lda", "ltda", "unipessoal",
    "sa", "s.a", "s.a.", "email", "mail", "geral", "info", "office",
    "frontoffice", "comercial", "vendas", "admin", "contacto", "contact",
}


def norm(value: Any) -> str:
    return " ".join(re.sub(r"[^0-9a-zA-ZÀ-ÿ]+", " ", str(value or "").casefold()).split())


def tokens(value: Any) -> set[str]:
    return {part for part in norm(value).split() if len(part) >= 3 and part not in STOPWORDS}


def overlaps(left: Any, right: Any) -> bool:
    a = tokens(left)
    b = tokens(right)
    if not a or not b:
        return False
    if a & b:
        return True
    return any(x in y or y in x for x in a for y in b)


def email_domain(value: Any) -> str:
    email = str(value or "").strip().casefold()
    if "@" not in email:
        return ""
    domain = email.rsplit("@", 1)[-1].strip()
    return domain if "." in domain else ""


def domains_match(*values: Any) -> bool:
    domains = [email_domain(v) for v in values if email_domain(v)]
    if len(domains) < 2:
        return False
    first = domains[0]
    return any(d == first for d in domains[1:])


def is_unsafe_fiscal_link(row: dict[str, Any], hint: str) -> bool:
    # Safe-by-default: without a fiscal link there is nothing to detach.
    if not row.get("linked_customer_id"):
        return False

    fiscal = str(row.get("linked_customer_name") or "").strip()
    if not fiscal:
        return False

    # Normal B2B: fiscal company name shares strong tokens with the process
    # even when the sender/opportunity customer is a person.
    if overlaps(hint, fiscal):
        return False

    # If the fiscal email domain matches the opportunity/sender domain, treat
    # it as ambiguous, not automatically unsafe.
    fiscal_email = row.get("linked_customer_email")
    if domains_match(fiscal_email, row.get("opportunity_customer_email")) or domains_match(fiscal_email, row.get("sender_email")):
        return False

    return True


def process_hint(title: Any) -> str:
    value = str(title or "").strip()
    if not value:
        return ""
    if "·" in value:
        tail = value.rsplit("·", 1)[-1].strip()
        if tail and not tail.upper().startswith(("ORC.", "S0")):
            return tail
    match = re.search(r"\bpara\s+(?:a|o|as|os|à|ao)?\s*(.+)$", value, re.IGNORECASE)
    if not match:
        return ""
    candidate = re.sub(r"\s+", " ", match.group(1)).strip(" .:-–—")
    candidate = re.split(r"\s+(?:de:|from:|enviada:|sent:)", candidate, maxsplit=1, flags=re.IGNORECASE)[0].strip()
    return candidate if 2 <= len(candidate) <= 120 else ""


def find_suspects(limit: int = 500) -> list[dict[str, Any]]:
    with engine.begin() as conn:
        rows = conn.execute(text("""
            SELECT
                t.id::text AS task_id,
                t.created_at,
                t.action_code,
                t.status,
                t.opportunity_id::text AS opportunity_id,
                COALESCE(NULLIF(t.conversation_id, ''), NULLIF(o.conversation_id, '')) AS conversation_id,
                COALESCE(NULLIF(o.title, ''), NULLIF(re.payload->'conversation'->'additional_attributes'->>'mail_subject', ''), NULLIF(re.payload->'content_attributes'->'email'->>'subject', '')) AS title,
                o.customer_name AS opportunity_customer_name,
                o.customer_email AS opportunity_customer_email,
                o.local_customer_id::text AS linked_customer_id,
                cu.name AS linked_customer_name,
                cu.tax_id AS linked_customer_tax_id,
                cu.email AS linked_customer_email,
                COALESCE(NULLIF(re.payload->'sender'->>'name', ''), NULLIF(re.payload->'conversation'->'meta'->'sender'->>'name', '')) AS sender_name,
                COALESCE(NULLIF(re.payload->'sender'->>'email', ''), NULLIF(re.payload->'conversation'->'meta'->'sender'->>'email', '')) AS sender_email
            FROM tasks t
            LEFT JOIN raw_events re ON re.id = t.raw_event_id
            LEFT JOIN opportunities o ON o.id = t.opportunity_id
            LEFT JOIN customers cu ON cu.id = o.local_customer_id
            WHERE t.status = 'pending'
              AND t.opportunity_id IS NOT NULL
              AND (cu.name IS NOT NULL OR COALESCE(o.customer_name, '') <> '')
            ORDER BY t.created_at DESC
            LIMIT :limit
        """), {"limit": limit}).mappings().all()

    suspects: list[dict[str, Any]] = []
    seen: set[str] = set()
    for row in rows:
        row_dict = dict(row)
        hint = process_hint(row_dict.get("title"))
        if not hint:
            continue
        fiscal_mismatch = is_unsafe_fiscal_link(row_dict, hint)
        if not fiscal_mismatch:
            continue
        key = str(row_dict.get("task_id"))
        if key in seen:
            continue
        seen.add(key)
        opportunity_customer = str(row_dict.get("opportunity_customer_name") or "").strip()
        suspects.append({
            **row_dict,
            "process_customer_hint": hint,
            "fiscal_mismatch": fiscal_mismatch,
            "opportunity_mismatch": bool(opportunity_customer and not overlaps(hint, opportunity_customer)),
        })
    return suspects


def print_suspects(suspects: list[dict[str, Any]]) -> None:
    print(f"Suspeitas de cliente fiscal inseguro: {len(suspects)}")
    for row in suspects[:100]:
        print("=" * 100)
        print(f"task_id: {row['task_id']}")
        print(f"opportunity_id: {row.get('opportunity_id') or '—'}")
        print(f"action_code/status: {row['action_code']} / {row['status']}")
        print(f"conversation: {row.get('conversation_id')}")
        print(f"processo: {row.get('process_customer_hint')}")
        print(f"cliente_fiscal: {row.get('linked_customer_name') or '—'} · NIF {row.get('linked_customer_tax_id') or '—'} · {row.get('linked_customer_email') or '—'}")
        print(f"oportunidade.customer: {row.get('opportunity_customer_name') or '—'} · {row.get('opportunity_customer_email') or '—'}")
        print(f"contacto_chatwoot: {row.get('sender_name') or '—'} · {row.get('sender_email') or '—'}")
        print(f"titulo: {row.get('title')}")


def apply_detach(suspects: list[dict[str, Any]], *, actor: str) -> int:
    now = datetime.now(timezone.utc).isoformat()
    by_opportunity: dict[str, list[dict[str, Any]]] = {}
    for row in suspects:
        opportunity_id = str(row.get("opportunity_id") or "").strip()
        if opportunity_id:
            by_opportunity.setdefault(opportunity_id, []).append(row)

    if not by_opportunity:
        return 0

    with engine.begin() as conn:
        for opportunity_id, rows in by_opportunity.items():
            sample = rows[0]
            audit_payload = {
                "detached_at": now,
                "detached_by": actor,
                "reason": "task_identity_mismatch_process_hint",
                "process_customer_hint": sample.get("process_customer_hint"),
                "previous_local_customer_id": sample.get("linked_customer_id"),
                "previous_linked_customer_name": sample.get("linked_customer_name"),
                "previous_linked_customer_email": sample.get("linked_customer_email"),
                "previous_opportunity_customer_name": sample.get("opportunity_customer_name"),
                "previous_opportunity_customer_email": sample.get("opportunity_customer_email"),
                "task_ids": [r.get("task_id") for r in rows],
            }
            conn.execute(text("""
                UPDATE opportunities
                SET
                    local_customer_id = NULL,
                    customer_name = COALESCE(NULLIF(:process_customer_hint, ''), customer_name),
                    customer_email = COALESCE(NULLIF(:sender_email, ''), customer_email),
                    metadata = COALESCE(metadata, '{}'::jsonb)
                        || jsonb_build_object('identity_detached', CAST(:audit_payload AS JSONB)),
                    updated_at = now()
                WHERE id = CAST(:opportunity_id AS UUID)
            """), {
                "opportunity_id": opportunity_id,
                "process_customer_hint": sample.get("process_customer_hint") or None,
                "sender_email": sample.get("sender_email") or None,
                "audit_payload": json.dumps(audit_payload, ensure_ascii=False),
            })
            conn.execute(text("""
                UPDATE tasks
                SET
                    metadata = COALESCE(metadata, '{}'::jsonb)
                        || jsonb_build_object(
                            'identity_repair', CAST(:audit_payload AS JSONB),
                            'requires_fiscal_customer_review', true
                        ),
                    updated_at = now()
                WHERE opportunity_id = CAST(:opportunity_id AS UUID)
                  AND status = 'pending'
            """), {
                "opportunity_id": opportunity_id,
                "audit_payload": json.dumps(audit_payload, ensure_ascii=False),
            })
    return len(by_opportunity)


def main() -> int:
    parser = argparse.ArgumentParser(description="Audit/repair task identity mismatches")
    parser.add_argument("--apply", action="store_true", help="Detach only clearly unsafe fiscal customers from affected opportunities")
    parser.add_argument("--limit", type=int, default=500)
    parser.add_argument("--actor", default="admin")
    args = parser.parse_args()

    suspects = find_suspects(limit=args.limit)
    print_suspects(suspects)
    if not args.apply:
        if suspects:
            print("\nDry-run: nenhuma alteração aplicada. Usa --apply para desassociar clientes fiscais inseguros.")
        return 1 if suspects else 0

    count = apply_detach(suspects, actor=args.actor)
    print(f"\nAlterações aplicadas: {count} oportunidade(s) desassociada(s) de cliente fiscal claramente inseguro.")
    remaining = find_suspects(limit=args.limit)
    if remaining:
        print("\nAinda há suspeitas após apply:")
        print_suspects(remaining)
    return 1 if remaining else 0


if __name__ == "__main__":
    raise SystemExit(main())