#!/usr/bin/env python3
"""
ClientFlow deep system audit (read-only)

Purpose:
  Validate end-to-end operational correctness, not only technical health.
  This script is intentionally READ-ONLY: it performs no UPDATE/INSERT/DELETE.

Usage from project root:
  cd /mnt/ssd/home/plx/clientflow_backend
  source .venv/bin/activate
  export PYTHONPATH=/mnt/ssd/home/plx/clientflow_backend
  set -a; source .env; set +a
  python scripts/audit_deep_system.py --window-hours 72 --sample-limit 25

Outputs:
  audit_reports/deep_audit_YYYYMMDD_HHMMSS.txt
  audit_reports/deep_audit_YYYYMMDD_HHMMSS.json
"""
from __future__ import annotations

import argparse
import json
import os
import re
import sys
from collections import Counter, defaultdict
from dataclasses import dataclass, field, asdict
from datetime import date, datetime, timezone
from decimal import Decimal
from uuid import UUID
from pathlib import Path
from typing import Any, Iterable

try:
    from dotenv import load_dotenv
except Exception:  # pragma: no cover
    load_dotenv = None

PROJECT_ROOT = Path(__file__).resolve().parents[1] if len(Path(__file__).resolve().parents) > 1 else Path.cwd()
if load_dotenv:
    load_dotenv(PROJECT_ROOT / ".env")

try:
    from sqlalchemy import text
    from sqlalchemy.exc import OperationalError, ProgrammingError
    from app.db import engine
except Exception as exc:  # pragma: no cover
    print(f"FATAL: não foi possível importar app.db/SQLAlchemy: {exc}")
    print("Confirma: cd projeto, source .venv/bin/activate, export PYTHONPATH=<root>, source .env")
    sys.exit(2)


SEVERITY_ORDER = {"FAIL": 3, "WARN": 2, "OK": 1, "INFO": 0}
EXECUTABLE_EXTENSIONS = [".exe", ".msi", ".bat", ".cmd", ".ps1", ".scr", ".vbs", ".js", ".jar"]
REMOTE_ACCESS_KEYWORDS = ["screenconnect", "connectwise", "anydesk", "teamviewer", "rustdesk", "dwservice", "remoteutilities"]
DANGEROUS_FINANCIAL_ACTIONS = {"CONFIRM_PAYMENT", "SEND_INVOICE", "SEND_PROFORMA", "SEND_QUOTE"}
FINANCIAL_ACTIONS = {"CONFIRM_PAYMENT", "SEND_INVOICE", "SEND_PROFORMA"}
MARKETING_ACTIONS = {"REMOVE_FROM_LIST"}
REVIEW_ACTIONS = {"REVIEW_MANUALLY"}


@dataclass
class Finding:
    severity: str
    area: str
    title: str
    detail: str = ""
    count: int | None = None
    samples: list[dict[str, Any]] = field(default_factory=list)


class Audit:
    def __init__(self, *, sample_limit: int):
        self.sample_limit = sample_limit
        self.findings: list[Finding] = []
        self.schema: dict[str, set[str]] = {}
        self.table_cache: set[str] = set()
        self.started_at = datetime.now(timezone.utc).isoformat()

    def add(self, severity: str, area: str, title: str, detail: str = "", count: int | None = None, samples: list[dict[str, Any]] | None = None):
        if severity not in SEVERITY_ORDER:
            raise ValueError(f"invalid severity {severity}")
        self.findings.append(Finding(severity, area, title, detail, count, samples or []))

    def summary(self) -> dict[str, int]:
        c = Counter(f.severity for f in self.findings)
        return {k: c.get(k, 0) for k in ["OK", "WARN", "FAIL", "INFO"]}

    def exit_code(self) -> int:
        s = self.summary()
        if s["FAIL"]:
            return 2
        if s["WARN"]:
            return 1
        return 0


def json_safe(value: Any) -> Any:
    if isinstance(value, (datetime, date)):
        return value.isoformat()
    if isinstance(value, Decimal):
        # Keep integer-like decimals as int; otherwise float for compact reports.
        return int(value) if value == value.to_integral_value() else float(value)
    if isinstance(value, UUID):
        return str(value)
    if isinstance(value, (list, tuple, set)):
        return [json_safe(v) for v in value]
    if isinstance(value, dict):
        return {str(k): json_safe(v) for k, v in value.items()}
    return value


def rows_to_dicts(rows: Iterable[Any], limit: int | None = None) -> list[dict[str, Any]]:
    out: list[dict[str, Any]] = []
    for i, r in enumerate(rows):
        if limit is not None and i >= limit:
            break
        out.append({k: json_safe(v) for k, v in dict(r).items()})
    return out


def q(conn, sql: str, params: dict[str, Any] | None = None):
    return conn.execute(text(sql), params or {}).mappings().all()


def s(conn, sql: str, params: dict[str, Any] | None = None):
    return conn.execute(text(sql), params or {}).scalar()


def table_exists(conn, table: str) -> bool:
    return bool(s(conn, """
        SELECT EXISTS (
            SELECT 1 FROM information_schema.tables
            WHERE table_schema = 'public' AND table_name = :table
        )
    """, {"table": table}))


def columns(conn, table: str) -> set[str]:
    return {r["column_name"] for r in q(conn, """
        SELECT column_name FROM information_schema.columns
        WHERE table_schema = 'public' AND table_name = :table
    """, {"table": table})}


def require(audit: Audit, conn, tables: list[str]):
    missing = [t for t in tables if not table_exists(conn, t)]
    if missing:
        audit.add("FAIL", "schema", "Tabelas essenciais em falta", ", ".join(missing), len(missing))
        return False
    return True


def safe_query(audit: Audit, conn, area: str, title: str, sql: str, params: dict[str, Any] | None = None):
    try:
        return q(conn, sql, params)
    except OperationalError as exc:
        msg = str(exc)
        if "DeadlockDetected" in msg or "deadlock detected" in msg or "LockNotAvailable" in msg:
            audit.add("WARN", area, f"{title}: lock/deadlock temporário", msg[:800])
            return []
        audit.add("FAIL", area, f"{title}: erro DB", msg[:1200])
        return []
    except ProgrammingError as exc:
        audit.add("FAIL", area, f"{title}: erro SQL/schema", str(exc)[:1200])
        return []


def inspect_schema(audit: Audit, conn):
    tables = [r["table_name"] for r in q(conn, """
        SELECT table_name FROM information_schema.tables
        WHERE table_schema = 'public'
        ORDER BY table_name
    """)]
    audit.table_cache = set(tables)
    for t in tables:
        audit.schema[t] = columns(conn, t)
    audit.add("INFO", "schema", "Tabelas encontradas", count=len(tables), samples=[{"tables": tables[:80]}])


def check_base_health(audit: Audit, conn, window_hours: int):
    ok_tables = require(audit, conn, ["tasks", "raw_events", "opportunities"])
    if ok_tables:
        audit.add("OK", "base", "Tabelas essenciais existem")
    for table in ["tasks", "raw_events", "opportunities", "customers", "integration_outbox"]:
        if table in audit.table_cache:
            try:
                count = s(conn, f"SELECT COUNT(*) FROM {table}")
                audit.add("INFO", "base", f"Contagem {table}", count=int(count))
            except Exception as exc:
                audit.add("WARN", "base", f"Não foi possível contar {table}", str(exc)[:500])


def check_raw_events(audit: Audit, conn, window_hours: int):
    if "raw_events" not in audit.table_cache:
        return
    cols = audit.schema["raw_events"]
    needed = {"processed", "ignored", "created_at", "payload"}
    if not needed.issubset(cols):
        audit.add("WARN", "raw_events", "raw_events sem colunas esperadas", detail=str(sorted(needed - cols)))
        return

    pending = safe_query(audit, conn, "raw_events", "incoming pendentes", """
        SELECT id::text AS raw_event_id, created_at, source_system, source_event_id,
               payload #>> '{message_type}' AS message_type,
               payload #>> '{conversation,id}' AS conversation_id,
               payload #>> '{sender,email}' AS sender_email,
               LEFT(COALESCE(payload #>> '{content}', payload::text), 500) AS content_excerpt
        FROM raw_events
        WHERE COALESCE(processed, false) = false
          AND COALESCE(ignored, false) = false
          AND created_at >= now() - (:window_hours || ' hours')::interval
        ORDER BY created_at DESC
        LIMIT :limit
    """, {"window_hours": window_hours, "limit": audit.sample_limit})
    if pending:
        audit.add("FAIL", "raw_events", "raw_events pendentes por processar", count=len(pending), samples=rows_to_dicts(pending))
    else:
        audit.add("OK", "raw_events", "Sem raw_events pendentes recentes")

    errors = safe_query(audit, conn, "raw_events", "erros recentes", """
        SELECT id::text AS raw_event_id, created_at, source_system, source_event_id,
               payload #>> '{message_type}' AS message_type,
               LEFT(COALESCE(processing_error, ''), 700) AS processing_error
        FROM raw_events
        WHERE processing_error IS NOT NULL
          AND created_at >= now() - (:window_hours || ' hours')::interval
        ORDER BY created_at DESC
        LIMIT :limit
    """, {"window_hours": window_hours, "limit": audit.sample_limit})
    real_errors = [r for r in errors if "processing_exception" in (r["processing_error"] or "")]
    if real_errors:
        audit.add("FAIL", "raw_events", "processing_exception em raw_events recentes", count=len(real_errors), samples=rows_to_dicts(real_errors))
    elif errors:
        audit.add("INFO", "raw_events", "Apenas erros/ignored benignos recentes", count=len(errors), samples=rows_to_dicts(errors))
    else:
        audit.add("OK", "raw_events", "Sem erros recentes em raw_events")

    dupes = safe_query(audit, conn, "raw_events", "source_event_id duplicado", """
        SELECT source_system, source_event_id, COUNT(*) AS total,
               MIN(created_at) AS first_seen, MAX(created_at) AS last_seen
        FROM raw_events
        WHERE source_event_id IS NOT NULL
          AND created_at >= now() - (:window_hours || ' hours')::interval
        GROUP BY source_system, source_event_id
        HAVING COUNT(*) > 1
        ORDER BY total DESC, last_seen DESC
        LIMIT :limit
    """, {"window_hours": window_hours, "limit": audit.sample_limit})
    if dupes:
        audit.add("WARN", "raw_events", "source_event_id duplicados recentes", count=len(dupes), samples=rows_to_dicts(dupes))
    else:
        audit.add("OK", "raw_events", "Sem duplicados recentes por source_event_id")


def check_security_content(audit: Audit, conn, window_hours: int):
    if "raw_events" not in audit.table_cache:
        return
    risky_sql = """
        SELECT
            r.id::text AS raw_event_id,
            r.created_at,
            r.source_event_id,
            r.payload #>> '{message_type}' AS message_type,
            r.payload #>> '{conversation,id}' AS conversation_id,
            r.payload #>> '{sender,name}' AS sender_name,
            r.payload #>> '{sender,email}' AS sender_email,
            LEFT(COALESCE(r.payload #>> '{content}', r.payload::text), 1200) AS content_excerpt,
            t.id::text AS task_id,
            t.route,
            t.action_code,
            t.status,
            t.priority,
            t.metadata AS task_metadata
        FROM raw_events r
        LEFT JOIN tasks t ON t.raw_event_id = r.id
        WHERE r.created_at >= now() - (:window_hours || ' hours')::interval
          AND (
                r.payload::text ILIKE '%%.exe%%'
             OR r.payload::text ILIKE '%%.msi%%'
             OR r.payload::text ILIKE '%%.bat%%'
             OR r.payload::text ILIKE '%%.cmd%%'
             OR r.payload::text ILIKE '%%.ps1%%'
             OR r.payload::text ILIKE '%%.scr%%'
             OR r.payload::text ILIKE '%%screenconnect%%'
             OR r.payload::text ILIKE '%%connectwise%%'
             OR r.payload::text ILIKE '%%anydesk%%'
             OR r.payload::text ILIKE '%%teamviewer%%'
          )
        ORDER BY r.created_at DESC
        LIMIT :limit
    """
    risky = safe_query(audit, conn, "security", "conteúdo com indicadores de risco", risky_sql, {"window_hours": window_hours * 14, "limit": audit.sample_limit})
    bad_financial = [r for r in risky if r.get("status") == "pending" and r.get("action_code") in DANGEROUS_FINANCIAL_ACTIONS and not (r.get("task_metadata") or {}).get("security_risk")]
    if bad_financial:
        audit.add("FAIL", "security", "Indicador de malware em task perigosa não bloqueada", count=len(bad_financial), samples=rows_to_dicts(bad_financial))
    elif risky:
        audit.add("WARN", "security", "Conteúdo com indicadores de risco encontrado", "Validar que está em REVIEW_MANUALLY/security_risk", len(risky), rows_to_dicts(risky))
    else:
        audit.add("OK", "security", "Sem indicadores de links executáveis/acesso remoto em eventos recentes")

    if "tasks" in audit.table_cache and "metadata" in audit.schema["tasks"]:
        security_tasks = safe_query(audit, conn, "security", "tasks security_risk pendentes", """
            SELECT id::text AS task_id, created_at, route, action_code, priority, status,
                   conversation_id, note, metadata
            FROM tasks
            WHERE status = 'pending'
              AND COALESCE((metadata ->> 'security_risk')::boolean, false) = true
            ORDER BY created_at DESC
            LIMIT :limit
        """, {"limit": audit.sample_limit})
        if security_tasks:
            audit.add("WARN", "security", "Tasks de segurança pendentes", count=len(security_tasks), samples=rows_to_dicts(security_tasks))
        else:
            audit.add("OK", "security", "Sem tasks security_risk pendentes")


def check_tasks(audit: Audit, conn, window_hours: int):
    if "tasks" not in audit.table_cache:
        return
    cols = audit.schema["tasks"]
    pending_summary = safe_query(audit, conn, "tasks", "backlog por route/action", """
        SELECT route, action_code, priority, status, COUNT(*) AS total,
               MIN(created_at) AS oldest, MAX(created_at) AS newest
        FROM tasks
        WHERE status = 'pending'
        GROUP BY route, action_code, priority, status
        ORDER BY total DESC, oldest ASC
    """)
    audit.add("INFO", "tasks", "Backlog pendente por rota/action", count=sum(int(r["total"]) for r in pending_summary), samples=rows_to_dicts(pending_summary, audit.sample_limit))

    old = safe_query(audit, conn, "tasks", "tasks antigas", """
        SELECT t.id::text AS task_id, t.created_at, t.route, t.action_code, t.action, t.note,
               t.priority, t.conversation_id, t.opportunity_id::text AS opportunity_id,
               o.title AS opportunity_title, o.customer_name AS opportunity_customer_name,
               o.customer_email AS opportunity_customer_email
        FROM tasks t
        LEFT JOIN opportunities o ON o.id = t.opportunity_id
        WHERE t.status = 'pending'
          AND t.created_at < now() - interval '14 days'
        ORDER BY t.created_at ASC
        LIMIT :limit
    """, {"limit": audit.sample_limit})
    if old:
        audit.add("WARN", "tasks", "Tasks pendentes antigas > 14 dias", count=len(old), samples=rows_to_dicts(old))
    else:
        audit.add("OK", "tasks", "Sem tasks pendentes com mais de 14 dias")

    review_recent = safe_query(audit, conn, "tasks", "REVIEW_MANUALLY recentes", """
        SELECT t.id::text AS task_id, t.created_at, t.route, t.action_code, t.priority,
               t.conversation_id, t.note, t.metadata,
               o.title AS opportunity_title, o.customer_name AS opportunity_customer_name
        FROM tasks t
        LEFT JOIN opportunities o ON o.id = t.opportunity_id
        WHERE t.status = 'pending'
          AND t.action_code = 'REVIEW_MANUALLY'
          AND t.created_at >= now() - (:window_hours || ' hours')::interval
        ORDER BY t.created_at DESC
        LIMIT :limit
    """, {"window_hours": window_hours, "limit": audit.sample_limit})
    if review_recent:
        audit.add("WARN", "tasks", "REVIEW_MANUALLY pendentes recentes", count=len(review_recent), samples=rows_to_dicts(review_recent))
    else:
        audit.add("OK", "tasks", "Sem REVIEW_MANUALLY recente pendente")

    dupes = safe_query(audit, conn, "tasks", "tasks duplicadas por conversa/action", """
        SELECT conversation_id, action_code, status, COUNT(*) AS total,
               MIN(created_at) AS first_created, MAX(created_at) AS last_created,
               array_agg(id::text ORDER BY created_at DESC) AS task_ids
        FROM tasks
        WHERE status = 'pending'
          AND conversation_id IS NOT NULL
        GROUP BY conversation_id, action_code, status
        HAVING COUNT(*) > 1
        ORDER BY total DESC, last_created DESC
        LIMIT :limit
    """, {"limit": audit.sample_limit})
    if dupes:
        audit.add("WARN", "tasks", "Possíveis tasks duplicadas pendentes por conversa/action", count=len(dupes), samples=rows_to_dicts(dupes))
    else:
        audit.add("OK", "tasks", "Sem duplicados pendentes simples por conversa/action")


def check_identity(audit: Audit, conn, window_hours: int):
    if not require(audit, conn, ["tasks", "opportunities"]):
        return
    opp_cols = audit.schema["opportunities"]
    if "local_customer_id" not in opp_cols:
        audit.add("WARN", "identity", "opportunities sem local_customer_id")
        return

    # Use customers table if available; otherwise only detect metadata/security detached.
    has_customers = "customers" in audit.table_cache and {"id", "name"}.issubset(audit.schema.get("customers", set()))
    if has_customers:
        email_col = "email" if "email" in audit.schema["customers"] else "NULL::text"
        name_expr = "c.name"
        email_expr = f"c.{email_col}" if email_col != "NULL::text" else "NULL::text"
        suspected = safe_query(audit, conn, "identity", "cliente fiscal possivelmente inseguro", f"""
            WITH data AS (
                SELECT
                    t.id::text AS task_id,
                    t.action_code,
                    t.route,
                    t.status,
                    t.conversation_id,
                    o.id::text AS opportunity_id,
                    o.title,
                    o.customer_name,
                    o.customer_email,
                    o.local_customer_id::text AS local_customer_id,
                    {name_expr} AS fiscal_name,
                    {email_expr} AS fiscal_email,
                    lower(regexp_replace(coalesce(o.title, '') || ' ' || coalesce(o.customer_name, ''), '[^[:alnum:]À-ÿ]+', ' ', 'g')) AS process_text,
                    lower(coalesce({name_expr}, '')) AS fiscal_text,
                    lower(split_part(coalesce(o.customer_email, ''), '@', 2)) AS opp_domain,
                    lower(split_part(coalesce({email_expr}, ''), '@', 2)) AS fiscal_domain
                FROM tasks t
                JOIN opportunities o ON o.id = t.opportunity_id
                JOIN customers c ON c.id = o.local_customer_id
                WHERE t.status = 'pending'
                  AND o.local_customer_id IS NOT NULL
            )
            SELECT * FROM data
            WHERE
                -- suspicious: fiscal email domain does not match opportunity email domain
                (fiscal_domain IS NOT NULL AND fiscal_domain <> '' AND opp_domain IS NOT NULL AND opp_domain <> '' AND fiscal_domain <> opp_domain)
                AND NOT (
                    -- accept if process text contains a strong fiscal token >= 5 chars
                    EXISTS (
                        SELECT 1
                        FROM regexp_split_to_table(fiscal_text, '\\s+') tok
                        WHERE length(tok) >= 5
                          AND process_text LIKE '%%' || tok || '%%'
                    )
                )
            ORDER BY title
            LIMIT :limit
        """, {"limit": audit.sample_limit})
        if suspected:
            audit.add("FAIL", "identity", "Cliente fiscal inseguro em task pendente", count=len(suspected), samples=rows_to_dicts(suspected))
        else:
            audit.add("OK", "identity", "Sem suspeitas fortes de cliente fiscal inseguro")
    else:
        audit.add("INFO", "identity", "Tabela customers indisponível para comparação nominal detalhada")

    repair_marked = safe_query(audit, conn, "identity", "identity repair/detached ainda pendente", """
        SELECT t.id::text AS task_id, t.route, t.action_code, t.status, t.priority,
               t.conversation_id, o.id::text AS opportunity_id, o.title,
               o.customer_name, o.customer_email, o.local_customer_id::text AS local_customer_id,
               o.metadata AS opportunity_metadata, t.metadata AS task_metadata
        FROM tasks t
        LEFT JOIN opportunities o ON o.id = t.opportunity_id
        WHERE t.status = 'pending'
          AND (
                t.metadata::text ILIKE '%%identity_detached%%'
             OR t.metadata::text ILIKE '%%identity_repair%%'
             OR o.metadata::text ILIKE '%%identity_detached%%'
             OR o.metadata::text ILIKE '%%identity_repair%%'
             OR COALESCE((t.metadata ->> 'requires_fiscal_customer_review')::boolean, false) = true
          )
        ORDER BY t.created_at DESC
        LIMIT :limit
    """, {"limit": audit.sample_limit})
    if repair_marked:
        audit.add("WARN", "identity", "Tasks pendentes marcadas para revisão de identidade/fiscal", count=len(repair_marked), samples=rows_to_dicts(repair_marked))
    else:
        audit.add("OK", "identity", "Sem tasks pendentes marcadas por identity repair")


def check_financial_safety(audit: Audit, conn, window_hours: int):
    if not require(audit, conn, ["tasks", "opportunities"]):
        return
    no_customer = safe_query(audit, conn, "financial", "financeiras sem cliente fiscal", """
        SELECT t.id::text AS task_id, t.created_at, t.route, t.action_code, t.priority,
               t.conversation_id, t.action, t.note,
               o.id::text AS opportunity_id, o.title, o.customer_name, o.customer_email,
               o.local_customer_id::text AS local_customer_id, t.metadata
        FROM tasks t
        LEFT JOIN opportunities o ON o.id = t.opportunity_id
        WHERE t.status = 'pending'
          AND t.action_code IN ('CONFIRM_PAYMENT', 'SEND_INVOICE', 'SEND_PROFORMA')
          AND (o.local_customer_id IS NULL OR t.opportunity_id IS NULL)
        ORDER BY t.created_at DESC
        LIMIT :limit
    """, {"limit": audit.sample_limit})
    if no_customer:
        audit.add("FAIL", "financial", "Tasks financeiras pendentes sem cliente fiscal confirmado", count=len(no_customer), samples=rows_to_dicts(no_customer))
    else:
        audit.add("OK", "financial", "Sem tasks financeiras pendentes sem cliente fiscal")

    security_fin = safe_query(audit, conn, "financial", "financeiras com security_risk", """
        SELECT id::text AS task_id, created_at, route, action_code, priority, status,
               conversation_id, note, metadata
        FROM tasks
        WHERE status = 'pending'
          AND action_code IN ('CONFIRM_PAYMENT', 'SEND_INVOICE', 'SEND_PROFORMA')
          AND COALESCE((metadata ->> 'security_risk')::boolean, false) = true
        ORDER BY created_at DESC
        LIMIT :limit
    """, {"limit": audit.sample_limit})
    if security_fin:
        audit.add("FAIL", "financial", "Task financeira com security_risk ainda não bloqueada", count=len(security_fin), samples=rows_to_dicts(security_fin))
    else:
        audit.add("OK", "financial", "Nenhuma task financeira pendente marcada como security_risk")


def check_llm_quality(audit: Audit, conn, window_hours: int):
    if "tasks" not in audit.table_cache or "metadata" not in audit.schema["tasks"]:
        return
    summary = safe_query(audit, conn, "llm", "qualidade LLM por decision_source/action", """
        SELECT
            COALESCE(metadata ->> 'decision_source', 'unknown') AS decision_source,
            action_code,
            route,
            COUNT(*) AS total,
            COUNT(*) FILTER (WHERE COALESCE((metadata ->> 'needs_review')::boolean, false) = true) AS needs_review,
            COUNT(*) FILTER (WHERE COALESCE((metadata ->> 'llm_confidence')::numeric, 0) = 0) AS zero_confidence,
            ROUND(AVG(NULLIF((metadata ->> 'llm_confidence')::numeric, 0)), 3) AS avg_nonzero_confidence
        FROM tasks
        WHERE created_at >= now() - (:window_hours || ' hours')::interval
        GROUP BY decision_source, action_code, route
        ORDER BY total DESC
        LIMIT :limit
    """, {"window_hours": window_hours, "limit": audit.sample_limit})
    audit.add("INFO", "llm", "Resumo de classificação LLM recente", count=sum(int(r["total"]) for r in summary), samples=rows_to_dicts(summary))

    invalid_json = safe_query(audit, conn, "llm", "LLM sem JSON/parse inválido", """
        SELECT id::text AS task_id, created_at, route, action_code, priority, conversation_id,
               note, metadata
        FROM tasks
        WHERE created_at >= now() - (:window_hours || ' hours')::interval
          AND (
                note ILIKE '%%sem JSON%%'
             OR note ILIKE '%%fallback-parse%%'
             OR note ILIKE '%%Resposta LLM inválida%%'
          )
        ORDER BY created_at DESC
        LIMIT :limit
    """, {"window_hours": window_hours, "limit": audit.sample_limit})
    if invalid_json:
        audit.add("WARN", "llm", "Falhas recentes de formato/parse LLM", count=len(invalid_json), samples=rows_to_dicts(invalid_json))
    else:
        audit.add("OK", "llm", "Sem falhas recentes de formato/parse LLM")

    high_conf_review = safe_query(audit, conn, "llm", "Alta confiança mas revisão/segurança", """
        SELECT id::text AS task_id, created_at, route, action_code, priority, conversation_id,
               note, metadata
        FROM tasks
        WHERE created_at >= now() - (:window_hours || ' hours')::interval
          AND COALESCE((metadata ->> 'llm_confidence')::numeric, 0) >= 0.90
          AND (action_code = 'REVIEW_MANUALLY' OR COALESCE((metadata ->> 'security_risk')::boolean, false) = true)
        ORDER BY created_at DESC
        LIMIT :limit
    """, {"window_hours": window_hours, "limit": audit.sample_limit})
    if high_conf_review:
        audit.add("WARN", "llm", "Alta confiança mas caiu em revisão/segurança", "Pode indicar regra de segurança ou erro da IA", len(high_conf_review), rows_to_dicts(high_conf_review))


def check_opportunities(audit: Audit, conn, window_hours: int):
    if "opportunities" not in audit.table_cache:
        return
    cols = audit.schema["opportunities"]
    audit.add("INFO", "opportunities", "Colunas opportunities", samples=[{"columns": sorted(cols)}])

    dupes = safe_query(audit, conn, "opportunities", "possíveis duplicados por email", """
        SELECT lower(customer_email) AS customer_email, COUNT(*) AS total,
               array_agg(id::text ORDER BY created_at DESC) AS opportunity_ids,
               array_agg(title ORDER BY created_at DESC) AS titles,
               MAX(created_at) AS newest
        FROM opportunities
        WHERE customer_email IS NOT NULL AND customer_email <> ''
        GROUP BY lower(customer_email)
        HAVING COUNT(*) > 1
        ORDER BY total DESC, newest DESC
        LIMIT :limit
    """, {"limit": audit.sample_limit})
    if dupes:
        audit.add("WARN", "opportunities", "Possíveis oportunidades duplicadas por email", count=len(dupes), samples=rows_to_dicts(dupes))
    else:
        audit.add("OK", "opportunities", "Sem duplicados simples por customer_email")

    no_task = safe_query(audit, conn, "opportunities", "oportunidades abertas sem task pendente", """
        SELECT o.id::text AS opportunity_id, o.created_at, o.title, o.customer_name, o.customer_email,
               o.local_customer_id::text AS local_customer_id, o.status, o.stage, o.metadata
        FROM opportunities o
        LEFT JOIN tasks t ON t.opportunity_id = o.id AND t.status = 'pending'
        WHERE t.id IS NULL
          AND o.created_at >= now() - (:window_hours || ' hours')::interval
          AND COALESCE(o.status, '') NOT IN ('closed', 'won', 'lost', 'DONE')
        ORDER BY o.created_at DESC
        LIMIT :limit
    """, {"window_hours": window_hours * 7, "limit": audit.sample_limit})
    if no_task:
        audit.add("WARN", "opportunities", "Oportunidades recentes sem task pendente", count=len(no_task), samples=rows_to_dicts(no_task))
    else:
        audit.add("OK", "opportunities", "Sem oportunidades recentes abertas sem task pendente")


def check_outbox(audit: Audit, conn, window_hours: int):
    if "integration_outbox" not in audit.table_cache:
        audit.add("INFO", "outbox", "Tabela integration_outbox não encontrada")
        return
    cols = audit.schema["integration_outbox"]
    status_col = "status" if "status" in cols else None
    if not status_col:
        audit.add("WARN", "outbox", "integration_outbox sem coluna status")
        return
    failed = safe_query(audit, conn, "outbox", "failed/error", """
        SELECT * FROM integration_outbox
        WHERE status IN ('failed', 'error')
        ORDER BY created_at DESC NULLS LAST
        LIMIT :limit
    """, {"limit": min(audit.sample_limit, 10)})
    if failed:
        audit.add("FAIL", "outbox", "integration_outbox com failed/error", count=len(failed), samples=rows_to_dicts(failed))
    else:
        audit.add("OK", "outbox", "Sem integration_outbox failed/error")

    pending_old = safe_query(audit, conn, "outbox", "pending antigo", """
        SELECT * FROM integration_outbox
        WHERE status IN ('pending', 'claimed')
          AND created_at < now() - interval '24 hours'
        ORDER BY created_at ASC NULLS LAST
        LIMIT :limit
    """, {"limit": min(audit.sample_limit, 10)})
    if pending_old:
        audit.add("WARN", "outbox", "integration_outbox pending/claimed antigo", count=len(pending_old), samples=rows_to_dicts(pending_old))
    else:
        audit.add("OK", "outbox", "Sem outbox pending/claimed antigo")


def write_reports(audit: Audit, out_dir: Path):
    out_dir.mkdir(parents=True, exist_ok=True)
    stamp = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S")
    txt_path = out_dir / f"deep_audit_{stamp}.txt"
    json_path = out_dir / f"deep_audit_{stamp}.json"

    lines: list[str] = []
    lines.append("ClientFlow Deep System Audit")
    lines.append("=" * 90)
    lines.append(f"started_at={audit.started_at}")
    lines.append(f"summary={audit.summary()} exit_code={audit.exit_code()}")
    lines.append("")
    for f in audit.findings:
        icon = {"OK": "✅", "WARN": "⚠️", "FAIL": "❌", "INFO": "ℹ️"}[f.severity]
        count = "" if f.count is None else f" count={f.count}"
        lines.append(f"{icon} [{f.severity}] {f.area}: {f.title}{count}")
        if f.detail:
            lines.append(f"    {f.detail}")
        for sample in f.samples[:5]:
            lines.append("    sample: " + json.dumps(json_safe(sample), ensure_ascii=False)[:2000])
        lines.append("")

    txt_path.write_text("\n".join(lines), encoding="utf-8")
    json_path.write_text(json.dumps({
        "started_at": audit.started_at,
        "summary": audit.summary(),
        "exit_code": audit.exit_code(),
        "findings": [asdict(f) for f in audit.findings],
    }, ensure_ascii=False, indent=2, default=str), encoding="utf-8")
    return txt_path, json_path


def main() -> int:
    parser = argparse.ArgumentParser(description="ClientFlow deep read-only system audit")
    parser.add_argument("--window-hours", type=int, default=72)
    parser.add_argument("--sample-limit", type=int, default=25)
    parser.add_argument("--out-dir", default=str(PROJECT_ROOT / "audit_reports"))
    args = parser.parse_args()

    audit = Audit(sample_limit=args.sample_limit)
    try:
        with engine.begin() as conn:
            inspect_schema(audit, conn)
            check_base_health(audit, conn, args.window_hours)
            check_raw_events(audit, conn, args.window_hours)
            check_security_content(audit, conn, args.window_hours)
            check_tasks(audit, conn, args.window_hours)
            check_identity(audit, conn, args.window_hours)
            check_financial_safety(audit, conn, args.window_hours)
            check_llm_quality(audit, conn, args.window_hours)
            check_opportunities(audit, conn, args.window_hours)
            check_outbox(audit, conn, args.window_hours)
    except Exception as exc:
        audit.add("FAIL", "runtime", "Auditoria abortou com exceção", str(exc)[:2000])

    txt_path, json_path = write_reports(audit, Path(args.out_dir))
    print("ClientFlow Deep System Audit")
    print("=" * 90)
    for k, v in audit.summary().items():
        print(f"{k}: {v}")
    print(f"Relatório TXT:  {txt_path}")
    print(f"Relatório JSON: {json_path}")
    print(f"exit_code={audit.exit_code()}")
    # Print FAIL/WARN concise list
    for f in audit.findings:
        if f.severity in {"FAIL", "WARN"}:
            print(f"{f.severity} | {f.area} | {f.title} | count={f.count}")
            if f.samples:
                print(json.dumps(f.samples[:2], ensure_ascii=False, indent=2, default=str)[:3000])
    return audit.exit_code()


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