#!/usr/bin/env python3 """ Auditoria read-only do ClientFlow Backend. Uso recomendado dentro do projeto: cd /mnt/ssd/home/plx/clientflow_backend source .venv/bin/activate export PYTHONPATH=/mnt/ssd/home/plx/clientflow_backend python scripts/audit_system_health.py O script NÃO altera dados. Valida: - import da app FastAPI - endpoint /health local - serviço systemd clientflow-api.service - ligação PostgreSQL - tabelas/colunas essenciais - raw_events Chatwoot pendentes/erros - tasks recentes e pendentes - action_codes inválidos - outbox/filas de integração, quando existirem - workers systemd de outbox, quando existirem Exit codes: 0 = OK 1 = WARN 2 = FAIL """ from __future__ import annotations import argparse import datetime as dt import importlib import json import os import platform import socket import subprocess import sys import time import traceback import urllib.error import urllib.request from dataclasses import dataclass, asdict from pathlib import Path from typing import Any, Iterable try: from sqlalchemy import text except Exception: # pragma: no cover text = None # type: ignore PROJECT_ROOT = Path(__file__).resolve().parents[1] DEFAULT_BASE_URL = "http://127.0.0.1:8020" DEFAULT_SERVICE = "clientflow-api.service" OUTBOX_SERVICES = [ "clientflow-outbox-jasmin.service", "clientflow-outbox-packlink.service", ] ALLOWED_ACTION_CODES = { "SEND_INFO", "SEND_QUOTE", "SEND_PROFORMA", "SEND_INVOICE", "CONFIRM_PAYMENT", "SUPPORT", "REMOVE_FROM_LIST", "IGNORE_SPAM", "IGNORE_BOUNCE", "REVIEW_MANUALLY", "NO_ACTION", "MARK_NO_INTEREST", "FOLLOW_UP_QUOTE", "FOLLOW_UP_PROFORMA", "FOLLOW_UP_PAYMENT", "FOLLOW_UP_CUSTOMER_REVIEW", "FOLLOW_UP_GENERIC", # Ações operacionais/backoffice válidas que não são usadas pelo classificador LLM "PREPARE_ORDER", "REVIEW_RECONCILIATION", } @dataclass class Check: name: str status: str # OK, WARN, FAIL, INFO summary: str details: Any = None class Audit: def __init__(self) -> None: self.checks: list[Check] = [] self.started_at = dt.datetime.now(dt.timezone.utc) def add(self, name: str, status: str, summary: str, details: Any = None) -> None: status = status.upper() self.checks.append(Check(name=name, status=status, summary=summary, details=details)) icon = {"OK": "✅", "WARN": "⚠️", "FAIL": "❌", "INFO": "ℹ️"}.get(status, "•") print(f"{icon} [{status}] {name}: {summary}") if details is not None and isinstance(details, (str, int, float)): print(f" {details}") def exit_code(self) -> int: if any(c.status == "FAIL" for c in self.checks): return 2 if any(c.status == "WARN" for c in self.checks): return 1 return 0 def counts(self) -> dict[str, int]: out: dict[str, int] = {"OK": 0, "WARN": 0, "FAIL": 0, "INFO": 0} for c in self.checks: out[c.status] = out.get(c.status, 0) + 1 return out def to_json(self) -> dict[str, Any]: finished = dt.datetime.now(dt.timezone.utc) return { "started_at": self.started_at.isoformat(), "finished_at": finished.isoformat(), "duration_seconds": round((finished - self.started_at).total_seconds(), 3), "host": socket.gethostname(), "project_root": str(PROJECT_ROOT), "python": sys.version.split()[0], "platform": platform.platform(), "exit_code": self.exit_code(), "counts": self.counts(), "checks": [asdict(c) for c in self.checks], } def run_cmd(cmd: list[str], timeout: int = 10) -> tuple[int, str, str]: try: p = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout) return p.returncode, p.stdout.strip(), p.stderr.strip() except FileNotFoundError as exc: return 127, "", str(exc) except subprocess.TimeoutExpired as exc: return 124, exc.stdout or "", f"timeout after {timeout}s" def http_get_json(url: str, timeout: int = 4) -> tuple[int | None, Any, str | None]: try: req = urllib.request.Request(url, headers={"User-Agent": "clientflow-audit/1.0"}) with urllib.request.urlopen(req, timeout=timeout) as resp: body = resp.read().decode("utf-8", errors="replace") try: payload = json.loads(body) except Exception: payload = body[:1000] return resp.status, payload, None except urllib.error.HTTPError as exc: try: body = exc.read().decode("utf-8", errors="replace")[:1000] except Exception: body = "" return exc.code, body, str(exc) except Exception as exc: return None, None, str(exc) def load_app_and_db(audit: Audit) -> tuple[Any | None, Any | None, Any | None]: if str(PROJECT_ROOT) not in sys.path: sys.path.insert(0, str(PROJECT_ROOT)) os.environ.setdefault("PYTHONPATH", str(PROJECT_ROOT)) try: app_main = importlib.import_module("app.main") app_obj = getattr(app_main, "app", None) audit.add("Import app.main", "OK", "app.main importado com sucesso", f"app={type(app_obj).__name__}") except Exception: audit.add("Import app.main", "FAIL", "Falhou import de app.main", traceback.format_exc()) return None, None, None try: db_mod = importlib.import_module("app.db") config_mod = importlib.import_module("app.config") engine = getattr(db_mod, "engine") settings = getattr(config_mod, "settings") masked_db = str(settings.database_url) if "@" in masked_db: prefix, suffix = masked_db.rsplit("@", 1) masked_db = prefix.split(":", 1)[0] + ":***@" + suffix audit.add("Config DB", "OK", "DATABASE_URL carregado", masked_db) return app_main, engine, settings except Exception: audit.add("Config DB", "FAIL", "Falhou import/config da base de dados", traceback.format_exc()) return app_main, None, None def check_systemd(audit: Audit, service: str, outbox_services: list[str]) -> None: rc, out, err = run_cmd(["systemctl", "is-active", service]) if rc == 0 and out.strip() == "active": audit.add("systemd API", "OK", f"{service} ativo") else: audit.add("systemd API", "FAIL", f"{service} não está ativo", {"rc": rc, "stdout": out, "stderr": err}) rc, out, err = run_cmd(["systemctl", "show", service, "--property=ActiveState,SubState,ExecMainPID,ExecMainStatus,RestartUSec,WorkingDirectory", "--no-pager"]) if rc == 0: audit.add("systemd API detalhes", "INFO", "Estado systemd lido", out) for svc in outbox_services: rc, out, err = run_cmd(["systemctl", "is-active", svc]) if rc == 0 and out.strip() == "active": audit.add(f"systemd worker {svc}", "OK", "ativo") elif out.strip() == "failed": audit.add(f"systemd worker {svc}", "WARN", "failed", {"stderr": err}) elif out.strip() == "inactive": audit.add(f"systemd worker {svc}", "INFO", "inactive") else: audit.add(f"systemd worker {svc}", "INFO", f"estado={out or 'desconhecido'}", {"rc": rc, "stderr": err}) def check_http(audit: Audit, base_url: str) -> None: url = base_url.rstrip("/") + "/health" # Uvicorn pode levar 2-3s após restart; dar duas tentativas curtas. last = None for _ in range(2): status, payload, err = http_get_json(url, timeout=4) last = (status, payload, err) if status == 200: break time.sleep(1.0) status, payload, err = last or (None, None, "not run") if status == 200: audit.add("HTTP /health", "OK", f"{url} respondeu 200", payload) else: audit.add("HTTP /health", "FAIL", f"{url} não respondeu OK", {"status": status, "error": err, "payload": payload}) def scalar(conn: Any, sql: str, params: dict[str, Any] | None = None) -> Any: return conn.execute(text(sql), params or {}).scalar() def rows(conn: Any, sql: str, params: dict[str, Any] | None = None) -> list[dict[str, Any]]: return [dict(r) for r in conn.execute(text(sql), params or {}).mappings().all()] def is_transient_db_lock_error(exc: BaseException) -> bool: text_value = " ".join(str(part) for part in [type(exc).__name__, exc, getattr(exc, "orig", "")]) return any(token in text_value for token in [ "DeadlockDetected", "deadlock detected", "LockNotAvailable", "could not obtain lock", "canceling statement due to lock timeout", ]) def table_exists(conn: Any, table: str) -> bool: return bool(scalar(conn, "SELECT to_regclass(:name) IS NOT NULL", {"name": f"public.{table}"})) def column_exists(conn: Any, table: str, column: str) -> bool: return bool(scalar(conn, """ SELECT EXISTS ( SELECT 1 FROM information_schema.columns WHERE table_schema = 'public' AND table_name = :table AND column_name = :column ) """, {"table": table, "column": column})) def count_table(conn: Any, table: str) -> int | None: if not table_exists(conn, table): return None return int(scalar(conn, f"SELECT COUNT(*) FROM {table}")) def safe_count_table(audit: Audit, conn: Any, table: str) -> int | None: try: return count_table(conn, table) except Exception as exc: if is_transient_db_lock_error(exc): audit.add( "DB contagem temporariamente bloqueada", "WARN", f"COUNT(*) em {table} encontrou lock/deadlock transitório", {"table": table, "error": str(exc)[:500]}, ) return None raise def check_db(audit: Audit, engine: Any, window_hours: int) -> None: if engine is None: audit.add("DB", "FAIL", "Sem engine DB") return try: with engine.begin() as conn: db_info = rows(conn, """ SELECT current_database() AS database, current_user AS user, version() AS version, now() AS now_utc """)[0] audit.add("DB conexão", "OK", "PostgreSQL acessível", db_info) except Exception: audit.add("DB conexão", "FAIL", "Falhou ligação/query PostgreSQL", traceback.format_exc()) return with engine.begin() as conn: essential = ["raw_events", "messages", "action_runs", "tasks", "business_events", "task_preparations"] missing = [t for t in essential if not table_exists(conn, t)] if missing: audit.add("DB tabelas essenciais", "FAIL", "Faltam tabelas essenciais", missing) else: audit.add("DB tabelas essenciais", "OK", "Tabelas essenciais existem") counts = {t: safe_count_table(audit, conn, t) for t in [ "raw_events", "messages", "action_runs", "tasks", "business_events", "task_preparations", "opportunities", "customers", "integration_outbox", "operation_links", ]} audit.add("DB contagens", "INFO", "Contagens por tabela", counts) if table_exists(conn, "raw_events"): pending = rows(conn, """ SELECT source_system, COALESCE(event_type, '-') AS event_type, COALESCE(payload #>> '{message_type}', '-') AS message_type, COUNT(*) AS total, MIN(created_at) AS oldest, MAX(created_at) AS newest FROM raw_events WHERE processed = false AND ignored = false AND processing_error IS NULL GROUP BY 1,2,3 ORDER BY total DESC, newest DESC LIMIT 20 """) incoming_pending = sum(int(r["total"]) for r in pending if r.get("source_system") == "chatwoot" and r.get("message_type") == "incoming") if incoming_pending > 0: audit.add("Chatwoot incoming pendente", "FAIL", f"{incoming_pending} incoming pendentes", pending) elif pending: audit.add("raw_events pendentes", "WARN", "Existem eventos pendentes não incoming", pending) else: audit.add("raw_events pendentes", "OK", "Sem raw_events pendentes por processar") errs = rows(conn, """ SELECT source_system, COALESCE(event_type, '-') AS event_type, COALESCE(payload #>> '{message_type}', '-') AS message_type, LEFT(COALESCE(processing_error, ''), 160) AS error, COUNT(*) AS total, MAX(created_at) AS newest FROM raw_events WHERE processing_error IS NOT NULL AND created_at >= now() - (:hours || ' hours')::interval GROUP BY 1,2,3,4 ORDER BY total DESC, newest DESC LIMIT 20 """, {"hours": window_hours}) fatal_errs = [ r for r in errs if str(r.get("error") or "").startswith("processing_exception:") ] benign_errs = [ r for r in errs if str(r.get("error") or "").startswith(("outgoing:", "ignored:")) ] if fatal_errs: audit.add("raw_events erros recentes", "FAIL", f"{sum(int(r.get('total', 0)) for r in fatal_errs)} erros reais de processamento nas últimas {window_hours}h", errs) elif errs: audit.add("raw_events erros recentes", "INFO", f"Só há erros benignos/ignorados nas últimas {window_hours}h", errs) else: audit.add("raw_events erros recentes", "OK", f"Sem processing_error nas últimas {window_hours}h") recent_chatwoot = rows(conn, """ SELECT COALESCE(payload #>> '{message_type}', '-') AS message_type, processed, ignored, CASE WHEN processing_error IS NULL THEN 'no_error' ELSE 'has_error' END AS error_state, COUNT(*) AS total, MIN(created_at) AS oldest, MAX(created_at) AS newest FROM raw_events WHERE source_system = 'chatwoot' AND event_type = 'message_created' AND created_at >= now() - (:hours || ' hours')::interval GROUP BY 1,2,3,4 ORDER BY newest DESC, total DESC LIMIT 30 """, {"hours": window_hours}) audit.add("Chatwoot raw_events recentes", "INFO", f"Resumo últimas {window_hours}h", recent_chatwoot) if table_exists(conn, "tasks"): action_status = rows(conn, """ SELECT action_code, status, COALESCE(priority, '-') AS priority, COUNT(*) AS total, MIN(created_at) AS oldest, MAX(created_at) AS newest FROM tasks WHERE created_at >= now() - (:hours || ' hours')::interval GROUP BY 1,2,3 ORDER BY total DESC, action_code LIMIT 30 """, {"hours": window_hours}) audit.add("tasks recentes", "INFO", f"Tasks criadas nas últimas {window_hours}h", action_status) pending_by_route = rows(conn, """ SELECT COALESCE(route, '-') AS route, action_code, status, COUNT(*) AS total, MIN(created_at) AS oldest, MAX(created_at) AS newest FROM tasks WHERE status = 'pending' GROUP BY 1,2,3 ORDER BY total DESC, oldest ASC LIMIT 30 """) audit.add("tasks pendentes", "INFO", "Backlog pendente por rota/action", pending_by_route) stale_pending = int(scalar(conn, """ SELECT COUNT(*) FROM tasks WHERE status = 'pending' AND created_at < now() - interval '14 days' """)) if stale_pending > 0: audit.add("tasks antigas", "WARN", f"{stale_pending} tasks pendentes com mais de 14 dias") else: audit.add("tasks antigas", "OK", "Sem tasks pendentes com mais de 14 dias") invalid_actions = rows(conn, """ SELECT action_code, COUNT(*) AS total, MAX(created_at) AS newest FROM tasks WHERE action_code <> ALL(:allowed) GROUP BY 1 ORDER BY total DESC """, {"allowed": list(ALLOWED_ACTION_CODES)}) if invalid_actions: audit.add("action_codes inválidos", "FAIL", "Há action_code fora da allow-list", invalid_actions) else: audit.add("action_codes inválidos", "OK", "Todos os action_code conhecidos estão na allow-list") review_recent = rows(conn, """ SELECT id::text AS task_id, created_at, conversation_id, action_code, status, LEFT(COALESCE(note, ''), 220) AS note, metadata FROM tasks WHERE action_code = 'REVIEW_MANUALLY' AND status = 'pending' AND created_at >= now() - (:hours || ' hours')::interval ORDER BY created_at DESC LIMIT 10 """, {"hours": window_hours}) if review_recent: audit.add("REVIEW_MANUALLY recente", "WARN", f"{len(review_recent)} amostras pendentes recentes", review_recent) else: audit.add("REVIEW_MANUALLY recente", "OK", f"Sem REVIEW_MANUALLY pendente nas últimas {window_hours}h") if table_exists(conn, "integration_outbox"): outbox_cols = {c for c in ["status", "target_system", "operation", "created_at", "updated_at", "attempts"] if column_exists(conn, "integration_outbox", c)} if {"status", "target_system"}.issubset(outbox_cols): outbox = rows(conn, """ SELECT target_system, status, COUNT(*) AS total, MIN(created_at) AS oldest, MAX(created_at) AS newest FROM integration_outbox GROUP BY 1,2 ORDER BY target_system, status """) failed = sum(int(r["total"]) for r in outbox if str(r.get("status", "")).lower() in {"failed", "error"}) if failed: audit.add("integration_outbox", "WARN", f"{failed} itens failed/error", outbox) else: audit.add("integration_outbox", "OK", "Sem itens failed/error", outbox) if table_exists(conn, "opportunities"): cols = ["state", "status", "stage"] group_col = next((c for c in cols if column_exists(conn, "opportunities", c)), None) if group_col: opps = rows(conn, f""" SELECT COALESCE({group_col}::text, '-') AS {group_col}, COUNT(*) AS total FROM opportunities GROUP BY 1 ORDER BY total DESC LIMIT 30 """) audit.add("opportunities", "INFO", f"Resumo por {group_col}", opps) else: audit.add("opportunities", "INFO", "Tabela existe; sem coluna state/status/stage reconhecida", {"total": count_table(conn, "opportunities")}) def check_env_settings(audit: Audit, settings: Any | None) -> None: if settings is None: return keys = { "env": getattr(settings, "env", None), "openrouter_model": getattr(settings, "openrouter_model", None), "openrouter_api_key_present": bool(getattr(settings, "openrouter_api_key", "")), "openai_api_key_present": bool(getattr(settings, "openai_api_key", "")), "chatwoot_write_enabled": getattr(settings, "chatwoot_write_enabled", None), "chatwoot_base_url_present": bool(getattr(settings, "chatwoot_base_url", "")), "chatwoot_api_token_present": bool(getattr(settings, "chatwoot_api_token", "")), "odoo_enabled": getattr(settings, "odoo_enabled", None), "jasmin_enabled": getattr(settings, "jasmin_enabled", None), "packlink_enabled": getattr(settings, "packlink_enabled", None), "external_company_lookup_enabled": getattr(settings, "external_company_lookup_enabled", None), "reply_llm_enabled": getattr(settings, "clientflow_reply_llm_enabled", None), "email_reply_agent_enabled": getattr(settings, "clientflow_email_reply_agent_enabled", None), } audit.add("Config funcional", "INFO", "Flags principais carregadas; segredos mascarados", keys) if not keys["openrouter_api_key_present"]: audit.add("OPENROUTER_API_KEY", "WARN", "OpenRouter API key ausente; triagem LLM pode cair em fallback/revisão") else: audit.add("OPENROUTER_API_KEY", "OK", "OpenRouter API key presente") def compile_check(audit: Audit) -> None: rc, out, err = run_cmd([sys.executable, "-m", "compileall", "-q", "app", "scripts"], timeout=60) if rc == 0: audit.add("compileall", "OK", "app/ e scripts/ compilam sem erro") else: audit.add("compileall", "FAIL", "Erro de compilação Python", {"stdout": out, "stderr": err}) def write_reports(audit: Audit, report_dir: Path) -> tuple[Path, Path]: report_dir.mkdir(parents=True, exist_ok=True) ts = dt.datetime.now().strftime("%Y%m%d_%H%M%S") json_path = report_dir / f"clientflow_audit_{ts}.json" txt_path = report_dir / f"clientflow_audit_{ts}.txt" data = audit.to_json() json_path.write_text(json.dumps(data, ensure_ascii=False, indent=2, default=str), encoding="utf-8") lines = [] lines.append("ClientFlow Audit Report") lines.append("=" * 80) lines.append(f"started_at: {data['started_at']}") lines.append(f"finished_at: {data['finished_at']}") lines.append(f"host: {data['host']}") lines.append(f"project_root: {data['project_root']}") lines.append(f"exit_code: {data['exit_code']}") lines.append(f"counts: {data['counts']}") lines.append("") for c in audit.checks: lines.append(f"[{c.status}] {c.name}: {c.summary}") if c.details is not None: lines.append(json.dumps(c.details, ensure_ascii=False, indent=2, default=str)) lines.append("-" * 80) txt_path.write_text("\n".join(lines), encoding="utf-8") return txt_path, json_path def main() -> int: parser = argparse.ArgumentParser(description="Auditoria read-only do ClientFlow Backend") parser.add_argument("--base-url", default=DEFAULT_BASE_URL, help="URL local da API ClientFlow") parser.add_argument("--service", default=DEFAULT_SERVICE, help="Nome do serviço systemd da API") parser.add_argument("--window-hours", type=int, default=24, help="Janela temporal para erros/eventos recentes") parser.add_argument("--report-dir", default=str(PROJECT_ROOT / "audit_reports"), help="Diretório para relatórios") parser.add_argument("--skip-systemd", action="store_true", help="Não verificar systemd") parser.add_argument("--skip-http", action="store_true", help="Não chamar /health") parser.add_argument("--skip-compile", action="store_true", help="Não correr compileall") args = parser.parse_args() print("ClientFlow system audit") print("=" * 80) print(f"project_root={PROJECT_ROOT}") print(f"base_url={args.base_url}") print(f"window_hours={args.window_hours}") print("") audit = Audit() audit.add("Ambiente", "INFO", "Contexto de execução", { "cwd": os.getcwd(), "project_root": str(PROJECT_ROOT), "python": sys.executable, "python_version": sys.version, "hostname": socket.gethostname(), }) app_main, engine, settings = load_app_and_db(audit) check_env_settings(audit, settings) if not args.skip_compile: compile_check(audit) if not args.skip_systemd: check_systemd(audit, args.service, OUTBOX_SERVICES) if not args.skip_http: check_http(audit, args.base_url) check_db(audit, engine, args.window_hours) txt_path, json_path = write_reports(audit, Path(args.report_dir)) print("") print("=" * 80) print(f"Relatório TXT: {txt_path}") print(f"Relatório JSON: {json_path}") print(f"Resumo: {audit.counts()} | exit_code={audit.exit_code()}") return audit.exit_code() if __name__ == "__main__": raise SystemExit(main())