#!/usr/bin/env python3 """Audit ClientFlow Work Center actions and reconciliation coherence. READ-ONLY production helper for ClientFlow v4928.1.5.130+. The auditor compares four views of the same process: 1. the action actually shown in /operations; 2. the pending task stored in the database; 3. the central opportunity next-action engine; 4. mandatory blockers, especially reconciliation and reconstructed processes. It also detects work that should be in the Work Center but is missing, and association candidates that are stale, conflicting or attached to the wrong opportunity. No INSERT/UPDATE/DELETE is executed. There is deliberately no --apply option. Examples: python scripts/audit_work_center_coherence.py --limit 500 python scripts/audit_work_center_coherence.py --opportunity-id python scripts/audit_work_center_coherence.py --fail-on high python scripts/audit_work_center_coherence.py --self-test """ from __future__ import annotations import argparse import csv import json import sys from collections import Counter, defaultdict from dataclasses import asdict, dataclass, field from datetime import date, datetime, timezone from pathlib import Path from typing import Any, Dict, Iterable, List, Mapping, Optional, Sequence, Tuple def _detect_project_root() -> Path: script_path = Path(__file__).resolve() candidates = [script_path.parent.parent, Path.cwd().resolve(), *script_path.parents] seen: set[str] = set() for candidate in candidates: key = str(candidate) if key in seen: continue seen.add(key) if (candidate / "app" / "db.py").is_file(): return candidate return script_path.parent.parent PROJECT_ROOT = _detect_project_root() if str(PROJECT_ROOT) not in sys.path: sys.path.insert(0, str(PROJECT_ROOT)) from sqlalchemy import text # noqa: E402 OPEN_RECONCILIATION_STATUSES = {"open", "needs_review", "conflict"} TERMINAL_STAGES = {"WON", "LOST", "NO_INTEREST", "CLOSED"} ASSOCIATION_ACTIONS = { "ASSOCIATE_OPPORTUNITY", "REVIEW_ASSOCIATION", "LINK_DOCUMENT", "RECONCILE_DOCUMENTS", } RECONSTRUCTED_REVIEW_ACTIONS = { "REVIEW_RECONSTRUCTED_PROCESS", "REVIEW_MANUALLY", } SENSITIVE_ACTIONS = { "SEND_PROFORMA", "SEND_INVOICE", "CONFIRM_PAYMENT", "CONFIRM_PAYMENT_AND_PREPARE_SHIPMENT", "PREPARE_ORDER", "VALIDATE_PHYSICAL_ORDER", "CREATE_SHIPMENT", "SHIP_ORDER", "CLOSE_OPPORTUNITY", } HUMAN_ACTIONS = { "VALIDATE_FISCAL_CUSTOMER", "RECONCILE_DOCUMENTS", "CREATE_QUOTE", "SEND_QUOTE", "SEND_INVOICE", "CONFIRM_PAYMENT", "FOLLOW_UP_PAYMENT", "FOLLOW_UP_QUOTE", "FOLLOW_UP_CUSTOMER_REVIEW", "FOLLOW_UP", "PREPARE_ORDER", "VALIDATE_PHYSICAL_ORDER", "CREATE_SHIPMENT", "SHIP_ORDER", "CLOSE_OPPORTUNITY", "REVIEW", } NON_TASK_ACTIONS = {"NO_ACTION", "WAIT_PRODUCTION", "NOT_FOUND"} ACTION_ALIASES = { "SHIP_ORDER": "CREATE_SHIPMENT", "RECONCILE_DOCUMENTS": "ASSOCIATE_OPPORTUNITY", "REVIEW_ASSOCIATION": "ASSOCIATE_OPPORTUNITY", "LINK_DOCUMENT": "ASSOCIATE_OPPORTUNITY", "REVIEW": "REVIEW_MANUALLY", } SEVERITY_ORDER = {"critical": 0, "high": 1, "medium": 2, "low": 3, "info": 4} @dataclass class Finding: severity: str code: str title: str recommendation: str source: str = "" work_item_id: str = "" task_id: str = "" opportunity_id: str = "" opportunity_title: str = "" customer_name: str = "" displayed_action: str = "" stored_action: str = "" expected_action: str = "" association_item_id: str = "" external_ref: str = "" evidence: Dict[str, Any] = field(default_factory=dict) proposed_correction: Dict[str, Any] = field(default_factory=dict) def _s(value: Any) -> str: return str(value or "").strip() def _upper(value: Any) -> str: return _s(value).upper() def _lower(value: Any) -> str: return _s(value).lower() def _as_dict(value: Any) -> Dict[str, Any]: if isinstance(value, dict): return value if isinstance(value, str) and value.strip(): try: parsed = json.loads(value) return parsed if isinstance(parsed, dict) else {} except Exception: return {} return {} def _json_default(value: Any) -> str: if isinstance(value, (datetime, date)): return value.isoformat() return str(value) def _canonical_action(value: Any) -> str: code = _upper(value) return ACTION_ALIASES.get(code, code) def _actions_match(left: Any, right: Any) -> bool: a, b = _canonical_action(left), _canonical_action(right) if not a or not b: return False if a == b: return True # REVIEW_RECONSTRUCTED_PROCESS is a stricter review than generic REVIEW. if {a, b} <= {"REVIEW_MANUALLY", "REVIEW_RECONSTRUCTED_PROCESS"}: return True # Specific follow-up variants are compatible with a generic follow-up. if a == "FOLLOW_UP" and b.startswith("FOLLOW_UP_"): return True if b == "FOLLOW_UP" and a.startswith("FOLLOW_UP_"): return True return False def _external_refs(item: Mapping[str, Any]) -> set[str]: refs = { _s(item.get("external_id")), _s(item.get("document_number")), } payload = _as_dict(item.get("payload")) record = payload.get("record") if isinstance(payload.get("record"), dict) else payload for key in ("id", "external_id", "name", "number", "document_number", "order_name"): refs.add(_s(record.get(key))) return {ref.casefold() for ref in refs if ref} def _link_system(source_system: Any) -> str: source = _lower(source_system) if source.startswith("odoo"): return "odoo" if source.startswith("jasmin"): return "jasmin" if source.startswith("packlink"): return "packlink" return source def _record_mode(metadata: Any) -> str: return _lower(_as_dict(metadata).get("clientflow_record_mode")) def _is_reconstructed(metadata: Any) -> bool: return _record_mode(metadata) in { "reconstructed_invoice_review", "historical_reconstructed", "legacy_review", } def _review_type(task: Mapping[str, Any]) -> str: metadata = _as_dict(task.get("metadata")) return _lower(metadata.get("review_type") or metadata.get("task_type")) def _has_completed_reconstructed_review(tasks: Sequence[Mapping[str, Any]], opportunity_metadata: Any) -> bool: metadata = _as_dict(opportunity_metadata) if metadata.get("reconstructed_process_validated") is True: return True if metadata.get("historical_evidence_review_required") is False: return True for task in tasks: if _lower(task.get("status")) not in {"done", "completed", "concluida", "concluída"}: continue code = _upper(task.get("action_code")) review_type = _review_type(task) if code == "REVIEW_RECONSTRUCTED_PROCESS": return True if code == "REVIEW_MANUALLY" and review_type in { "reconstructed_process", "historical_evidence_review", "historical_evidence_review_required", }: return True return False def _is_manual_association_action(item: Mapping[str, Any]) -> bool: code = _upper(item.get("action_code")) return code in ASSOCIATION_ACTIONS or _lower(item.get("opportunity_linking_status")) in { "ambiguous", "review_required", "needs_review", } def _severity_at_or_above(value: str, threshold: str) -> bool: return SEVERITY_ORDER.get(value, 99) <= SEVERITY_ORDER.get(threshold, -1) class WorkCenterAuditor: def __init__(self, *, limit: int, opportunity_limit: int, opportunity_ids: Sequence[str]) -> None: self.limit = min(max(int(limit), 1), 5000) self.opportunity_limit = min(max(int(opportunity_limit), 1), 5000) self.opportunity_ids = [_s(value) for value in opportunity_ids if _s(value)] self.work_items: List[Dict[str, Any]] = [] self.opportunities: Dict[str, Dict[str, Any]] = {} self.tasks_by_opportunity: Dict[str, List[Dict[str, Any]]] = defaultdict(list) self.pending_tasks_by_opportunity: Dict[str, List[Dict[str, Any]]] = defaultdict(list) self.reconciliation_items: List[Dict[str, Any]] = [] self.reconciliation_by_opportunity: Dict[str, List[Dict[str, Any]]] = defaultdict(list) self.operation_links: List[Dict[str, Any]] = [] self.documents_by_opportunity: Dict[str, List[Dict[str, Any]]] = defaultdict(list) self.next_actions: Dict[str, Dict[str, Any]] = {} self.findings: List[Finding] = [] self.summary: Dict[str, Any] = {} def add(self, finding: Finding) -> None: self.findings.append(finding) def load(self) -> None: try: from app.db import engine from app.operations_service import get_operations_summary from app.opportunity_next_action_service import get_opportunity_next_action except ModuleNotFoundError as exc: raise SystemExit( "Não foi possível importar o backend ClientFlow. " f"Raiz detetada: {PROJECT_ROOT}. Instale o script em /scripts/." ) from exc # This is the exact normalized queue consumed by /operations. operations = get_operations_summary(limit=self.limit) self.work_items = [dict(item) for item in operations.get("work_items", [])] self.summary["operations_counts"] = operations.get("counts", {}) params: Dict[str, Any] = {"opportunity_limit": self.opportunity_limit} opportunity_filter = "" if self.opportunity_ids: opportunity_filter = "AND o.id::text = ANY(:opportunity_ids)" params["opportunity_ids"] = self.opportunity_ids with engine.connect() as conn: tx = conn.begin() try: conn.execute(text("SET TRANSACTION READ ONLY")) required_tables = { "opportunities", "tasks", "reconciliation_items", "operation_links", "commercial_documents", "customers", } existing = set(conn.execute(text(""" SELECT table_name FROM information_schema.tables WHERE table_schema = current_schema() AND table_name = ANY(:tables) """), {"tables": sorted(required_tables)}).scalars().all()) missing = sorted(required_tables - existing) if missing: raise RuntimeError("Tabelas necessárias em falta: " + ", ".join(missing)) rows = conn.execute(text(f""" SELECT o.id::text, o.title, o.customer_name, o.customer_email, o.stage, o.status, o.value_amount, o.currency, o.local_customer_id::text, o.lifecycle_state, o.next_follow_up_at, o.follow_up_attempts, o.last_customer_activity_at, o.last_operator_activity_at, o.last_commercial_activity_at, o.metadata, c.name AS fiscal_customer_name, c.email AS fiscal_customer_email, c.tax_id AS fiscal_customer_tax_id FROM opportunities o LEFT JOIN customers c ON c.id = o.local_customer_id WHERE COALESCE(o.status, 'open') = 'open' AND COALESCE(o.stage, '') <> ALL(:terminal_stages) {opportunity_filter} ORDER BY o.updated_at DESC, o.created_at DESC LIMIT :opportunity_limit """), {**params, "terminal_stages": sorted(TERMINAL_STAGES)}).mappings().all() self.opportunities = {str(row["id"]): dict(row) for row in rows} relevant_ids = set(self.opportunities) relevant_ids.update( _s(item.get("opportunity_id")) for item in self.work_items if _s(item.get("opportunity_id")) ) if relevant_ids: ids = sorted(relevant_ids) task_rows = conn.execute(text(""" SELECT t.id::text, t.opportunity_id::text, t.action_code, t.action, t.note, t.route, t.priority, t.status, t.due_at, t.source_system, t.metadata, t.created_at, t.done_at FROM tasks t WHERE t.opportunity_id::text = ANY(:ids) ORDER BY t.created_at DESC """), {"ids": ids}).mappings().all() for row in task_rows: task = dict(row) oid = _s(task.get("opportunity_id")) self.tasks_by_opportunity[oid].append(task) if _lower(task.get("status")) == "pending": self.pending_tasks_by_opportunity[oid].append(task) recon_rows = conn.execute(text(""" SELECT ri.id::text, ri.opportunity_id::text, ri.customer_id::text, ri.source_system, ri.external_type, ri.external_id, ri.document_number, ri.customer_name, ri.customer_email, ri.customer_tax_id, ri.amount, ri.currency, ri.suggested_action, ri.confidence, ri.status, ri.payload, ri.resolution_note, ri.created_at, ri.updated_at FROM reconciliation_items ri WHERE ri.status IN ('open','needs_review','conflict') OR ri.opportunity_id::text = ANY(:ids) ORDER BY ri.updated_at DESC, ri.created_at DESC """), {"ids": ids}).mappings().all() self.reconciliation_items = [dict(row) for row in recon_rows] for item in self.reconciliation_items: oid = _s(item.get("opportunity_id")) if oid and _lower(item.get("status")) in OPEN_RECONCILIATION_STATUSES: self.reconciliation_by_opportunity[oid].append(item) self.operation_links = [dict(row) for row in conn.execute(text(""" SELECT ol.id::text, ol.opportunity_id::text, ol.system, ol.external_type, ol.external_id, ol.external_name, ol.status, ol.payload, ol.updated_at FROM operation_links ol WHERE ol.opportunity_id::text = ANY(:ids) OR COALESCE(ol.status, '') NOT IN ('ignored','deleted') """), {"ids": ids}).mappings().all()] doc_rows = conn.execute(text(""" SELECT cd.id::text, cd.opportunity_id::text, cd.system, cd.document_kind, cd.external_id, cd.document_number, cd.status, cd.total_amount, cd.currency, cd.role, cd.is_active, cd.is_primary, cd.document_date FROM commercial_documents cd WHERE cd.opportunity_id::text = ANY(:ids) ORDER BY cd.created_at DESC """), {"ids": ids}).mappings().all() for row in doc_rows: doc = dict(row) self.documents_by_opportunity[_s(doc.get("opportunity_id"))].append(doc) finally: tx.rollback() # Central decisions are read-only. Do this after the snapshot load so # failures are isolated and visible per opportunity. for opportunity_id in sorted(self.opportunities): try: self.next_actions[opportunity_id] = dict(get_opportunity_next_action(opportunity_id)) except Exception as exc: self.next_actions[opportunity_id] = { "action_code": "AUDIT_ERROR", "label": "Erro ao calcular próxima ação", "can_execute": False, "reason_if_blocked": str(exc), } def _matching_links(self, item: Mapping[str, Any]) -> List[Dict[str, Any]]: system = _link_system(item.get("source_system")) refs = _external_refs(item) matches: List[Dict[str, Any]] = [] for link in self.operation_links: if _lower(link.get("system")) != system: continue values = { _s(link.get("external_id")).casefold(), _s(link.get("external_name")).casefold(), } if refs.intersection({value for value in values if value}): matches.append(link) return matches def _effective_expected_action(self, opportunity_id: str) -> Tuple[str, Dict[str, Any]]: opportunity = self.opportunities.get(opportunity_id, {}) central = self.next_actions.get(opportunity_id, {}) open_reconciliation = self.reconciliation_by_opportunity.get(opportunity_id, []) tasks = self.tasks_by_opportunity.get(opportunity_id, []) if open_reconciliation: return "ASSOCIATE_OPPORTUNITY", { "reason": "open_reconciliation_candidate", "candidate_ids": [_s(item.get("id")) for item in open_reconciliation], "central_action": _upper(central.get("action_code")), } central_code = _upper(central.get("action_code")) if _is_reconstructed(opportunity.get("metadata")) and not _has_completed_reconstructed_review(tasks, opportunity.get("metadata")): if central_code in SENSITIVE_ACTIONS or central_code in HUMAN_ACTIONS: return "REVIEW_RECONSTRUCTED_PROCESS", { "reason": "reconstructed_process_not_validated", "central_action": central_code, "record_mode": _record_mode(opportunity.get("metadata")), } return central_code, { "reason": "central_decision", "central_action": central_code, "can_execute": central.get("can_execute"), "reason_if_blocked": central.get("reason_if_blocked"), } def _common(self, *, opportunity_id: str = "", item: Mapping[str, Any] | None = None) -> Dict[str, Any]: opportunity = self.opportunities.get(opportunity_id, {}) item = item or {} return { "source": _s(item.get("source")), "work_item_id": _s(item.get("id")), "task_id": _s(item.get("id")) if _s(item.get("source")) == "task" else "", "opportunity_id": opportunity_id, "opportunity_title": _s(opportunity.get("title") or item.get("opportunity_title")), "customer_name": _s( opportunity.get("fiscal_customer_name") or opportunity.get("customer_name") or item.get("fiscal_customer_name") or item.get("customer_name") ), } def _audit_work_item(self, item: Mapping[str, Any]) -> None: source = _lower(item.get("source")) opportunity_id = _s(item.get("opportunity_id")) displayed = _upper(item.get("action_code")) stored = _upper(item.get("original_action_code") or item.get("action_code")) common = self._common(opportunity_id=opportunity_id, item=item) if source != "task": # Outbox and communication items are audited only for impossible # association states; their action vocabulary is not opportunity flow. if opportunity_id and self.reconciliation_by_opportunity.get(opportunity_id) and displayed not in ASSOCIATION_ACTIONS: self.add(Finding( severity="medium", code="NON_TASK_WORK_BYPASSES_ASSOCIATION", title="Item do Centro de Trabalho ignora uma associação por validar", recommendation="Resolver a associação antes de executar a comunicação ou reprocessamento ligado à oportunidade.", displayed_action=displayed, expected_action="ASSOCIATE_OPPORTUNITY", evidence={"source": source}, proposed_correction={"action": "block_until_association_resolved", "automatic": False}, **common, )) return if not opportunity_id: # Support/no-opportunity tasks can be valid. Only flag sensitive work. if displayed in SENSITIVE_ACTIONS: self.add(Finding( severity="high", code="SENSITIVE_TASK_WITHOUT_OPPORTUNITY", title="Ação sensível sem oportunidade associada", recommendation="Associar ou criar a oportunidade antes de executar ação financeira, fiscal ou logística.", displayed_action=displayed, stored_action=stored, expected_action="ASSOCIATE_OPPORTUNITY", proposed_correction={"action": "require_opportunity_link", "automatic": False}, **common, )) return opportunity = self.opportunities.get(opportunity_id) if not opportunity: self.add(Finding( severity="critical", code="WORK_ITEM_OPPORTUNITY_NOT_FOUND", title="Task do Centro de Trabalho aponta para oportunidade inexistente ou fora do conjunto ativo", recommendation="Rever a ligação da task antes de a executar.", displayed_action=displayed, stored_action=stored, expected_action="REVIEW_MANUALLY", proposed_correction={"action": "repair_task_opportunity_link", "automatic": False}, **common, )) return expected, expected_context = self._effective_expected_action(opportunity_id) if not _actions_match(displayed, expected): severity = "high" if expected in {"ASSOCIATE_OPPORTUNITY", "REVIEW_RECONSTRUCTED_PROCESS"} else "medium" if displayed in SENSITIVE_ACTIONS and expected in {"ASSOCIATE_OPPORTUNITY", "REVIEW_RECONSTRUCTED_PROCESS"}: severity = "critical" self.add(Finding( severity=severity, code="WORK_CENTER_ACTION_MISMATCH", title="A ação mostrada no Centro de Trabalho não corresponde à primeira ação segura", recommendation="Mostrar e executar primeiro o bloqueio/decisão efetiva; só depois materializar a ação operacional seguinte.", displayed_action=displayed, stored_action=stored, expected_action=expected, evidence={ "expected_context": expected_context, "stage": opportunity.get("stage"), "task_title": item.get("title"), "task_detail": item.get("detail"), }, proposed_correction={ "action": "recompute_or_reclassify_task", "replacement_action": expected, "automatic": False, }, **common, )) central = self.next_actions.get(opportunity_id, {}) if central.get("can_execute") is False and _actions_match(displayed, central.get("action_code")): self.add(Finding( severity="high", code="BLOCKED_ACTION_VISIBLE_AS_EXECUTABLE", title="Ação bloqueada pelo motor central aparece como executável", recommendation="Apresentar o motivo do bloqueio e impedir conclusão até a condição ser resolvida.", displayed_action=displayed, stored_action=stored, expected_action=_upper(central.get("action_code")), evidence={"reason_if_blocked": central.get("reason_if_blocked")}, proposed_correction={"action": "surface_central_blocker", "automatic": False}, **common, )) pending = self.pending_tasks_by_opportunity.get(opportunity_id, []) if len(pending) > 1: codes = [_upper(task.get("action_code")) for task in pending] self.add(Finding( severity="high", code="MULTIPLE_PENDING_TASKS_FOR_OPPORTUNITY", title="A oportunidade tem várias tasks pendentes concorrentes", recommendation="Manter apenas a primeira ação humana segura e resolver/ignorar as restantes após revisão.", displayed_action=displayed, stored_action=stored, expected_action=expected, evidence={"pending_task_ids": [_s(task.get("id")) for task in pending], "pending_codes": codes}, proposed_correction={"action": "review_competing_tasks", "automatic": False}, **common, )) if expected in NON_TASK_ACTIONS: self.add(Finding( severity="high", code="TASK_EXISTS_FOR_NON_ACTION_STATE", title="Existe task humana para um estado que deveria ser apenas espera/sem ação", recommendation="Resolver a task obsoleta e deixar o sistema aguardar nova evidência.", displayed_action=displayed, stored_action=stored, expected_action=expected, evidence={"stage": opportunity.get("stage"), "expected_context": expected_context}, proposed_correction={"action": "ignore_obsolete_task", "automatic": False}, **common, )) def _audit_association_item(self, item: Mapping[str, Any]) -> None: if _lower(item.get("status")) not in OPEN_RECONCILIATION_STATUSES: return matches = self._matching_links(item) linked_opportunities = sorted({_s(link.get("opportunity_id")) for link in matches if _s(link.get("opportunity_id"))}) assigned_opportunity = _s(item.get("opportunity_id")) common = self._common(opportunity_id=assigned_opportunity) external_ref = _s(item.get("document_number") or item.get("external_id")) base = { **common, "association_item_id": _s(item.get("id")), "external_ref": external_ref, } if len(linked_opportunities) > 1: self.add(Finding( severity="critical", code="ASSOCIATION_EXTERNAL_REF_MULTIPLE_OPPORTUNITIES", title="A mesma evidência externa está ligada a várias oportunidades", recommendation="Não aplicar o candidato. Rever e manter uma única ligação correta.", expected_action="ASSOCIATE_OPPORTUNITY", evidence={"matching_opportunity_ids": linked_opportunities, "matching_links": matches}, proposed_correction={"action": "manual_resolve_duplicate_external_link", "automatic": False}, **base, )) elif len(linked_opportunities) == 1: exact = linked_opportunities[0] if assigned_opportunity and assigned_opportunity != exact: self.add(Finding( severity="critical", code="ASSOCIATION_POINTS_TO_WRONG_OPPORTUNITY", title="O candidato aponta para uma oportunidade diferente da ligação externa existente", recommendation="Corrigir a associação antes de qualquer ação comercial, fiscal ou logística.", expected_action="ASSOCIATE_OPPORTUNITY", evidence={"candidate_opportunity_id": assigned_opportunity, "existing_link_opportunity_id": exact}, proposed_correction={"action": "review_wrong_association", "automatic": False}, **base, )) else: opportunity = self.opportunities.get(exact, {}) linked_common = self._common(opportunity_id=exact) linked_common["source"] = "reconciliation" self.add(Finding( severity="high", code="OPEN_ASSOCIATION_ALREADY_LINKED", title="Candidato de reconciliação continua aberto apesar de a evidência já estar ligada", recommendation="Resolver como já ligado, preservando a fase, documentos e tasks atuais.", expected_action="RESOLVE_AS_EXISTING_LINK", evidence={"matching_link": matches[0]}, proposed_correction={"action": "resolve_as_existing_link", "guard": "exactly_one_link", "automatic": True}, association_item_id=_s(item.get("id")), external_ref=external_ref, **linked_common, )) if assigned_opportunity: opportunity = self.opportunities.get(assigned_opportunity, {}) fiscal_tax_id = _s(opportunity.get("fiscal_customer_tax_id")) candidate_tax_id = _s(item.get("customer_tax_id")) if fiscal_tax_id and candidate_tax_id and fiscal_tax_id != candidate_tax_id: self.add(Finding( severity="critical", code="ASSOCIATION_NIF_MISMATCH", title="O NIF do candidato não corresponde ao cliente fiscal da oportunidade", recommendation="Bloquear a associação e validar o cliente fiscal correto.", expected_action="ASSOCIATE_OPPORTUNITY", evidence={"candidate_tax_id": candidate_tax_id, "opportunity_tax_id": fiscal_tax_id}, proposed_correction={"action": "block_nif_mismatch", "automatic": False}, **base, )) pending = self.pending_tasks_by_opportunity.get(assigned_opportunity, []) if pending and not any(_upper(task.get("action_code")) in ASSOCIATION_ACTIONS for task in pending): self.add(Finding( severity="high", code="OPEN_ASSOCIATION_WITH_DOWNSTREAM_TASK", title="Existe associação por validar mas a task pendente é uma ação posterior", recommendation="Bloquear/reclassificar a task posterior e mostrar primeiro “Validar associação”.", stored_action=", ".join(_upper(task.get("action_code")) for task in pending), expected_action="ASSOCIATE_OPPORTUNITY", evidence={"pending_task_ids": [_s(task.get("id")) for task in pending]}, proposed_correction={"action": "replace_downstream_task_with_association_review", "automatic": False}, **base, )) def _audit_missing_work(self, opportunity_id: str, opportunity: Mapping[str, Any]) -> None: expected, context = self._effective_expected_action(opportunity_id) pending = self.pending_tasks_by_opportunity.get(opportunity_id, []) visible = [ item for item in self.work_items if _s(item.get("opportunity_id")) == opportunity_id ] common = self._common(opportunity_id=opportunity_id) if expected in NON_TASK_ACTIONS: return if expected not in HUMAN_ACTIONS and expected not in { "ASSOCIATE_OPPORTUNITY", "REVIEW_RECONSTRUCTED_PROCESS", }: return if not pending and not visible: severity = "high" if expected in { "ASSOCIATE_OPPORTUNITY", "REVIEW_RECONSTRUCTED_PROCESS", "SEND_INVOICE", "CONFIRM_PAYMENT", "PREPARE_ORDER", "VALIDATE_PHYSICAL_ORDER", "CREATE_SHIPMENT", } else "medium" self.add(Finding( severity=severity, code="REQUIRED_HUMAN_ACTION_MISSING_FROM_WORK_CENTER", title="A oportunidade exige ação humana mas não tem task no Centro de Trabalho", recommendation="Materializar uma única task para a primeira ação segura, respeitando associações e bloqueios.", expected_action=expected, evidence={"stage": opportunity.get("stage"), "context": context}, proposed_correction={"action": "materialize_effective_next_action", "action_code": expected, "automatic": False}, **common, )) if pending and not visible: self.add(Finding( severity="medium", code="PENDING_TASK_NOT_VISIBLE_IN_WORK_CENTER", title="Existe task pendente mas não aparece na fila normalizada do Centro de Trabalho", recommendation="Verificar filtros de ruído, rota, data e classificação da task.", stored_action=", ".join(_upper(task.get("action_code")) for task in pending), expected_action=expected, evidence={"pending_task_ids": [_s(task.get("id")) for task in pending], "context": context}, proposed_correction={"action": "review_work_center_visibility", "automatic": False}, **common, )) if visible and not pending and all(_lower(item.get("source")) != "task" for item in visible): self.add(Finding( severity="low", code="WORK_CENTER_ACTION_WITHOUT_TASK", title="A oportunidade aparece no Centro de Trabalho apenas por comunicação/outbox", recommendation="Confirmar se deve existir uma task humana persistida ou se o item técnico é suficiente.", displayed_action=", ".join(_upper(item.get("action_code")) for item in visible), expected_action=expected, evidence={"visible_sources": [_s(item.get("source")) for item in visible]}, proposed_correction={"action": "review_materialization_policy", "automatic": False}, **common, )) def _audit_stale_association_tasks(self) -> None: for opportunity_id, tasks in self.pending_tasks_by_opportunity.items(): association_tasks = [task for task in tasks if _upper(task.get("action_code")) in ASSOCIATION_ACTIONS] if not association_tasks: continue if self.reconciliation_by_opportunity.get(opportunity_id): continue common = self._common(opportunity_id=opportunity_id) self.add(Finding( severity="medium", code="ASSOCIATION_TASK_WITHOUT_OPEN_CANDIDATE", title="Existe task de associação mas não há candidato de reconciliação aberto", recommendation="Rever se a task ficou obsoleta após ligação/resolução do candidato.", stored_action=", ".join(_upper(task.get("action_code")) for task in association_tasks), expected_action=_upper(self.next_actions.get(opportunity_id, {}).get("action_code")), evidence={"task_ids": [_s(task.get("id")) for task in association_tasks]}, proposed_correction={"action": "review_stale_association_task", "automatic": False}, **common, )) def audit(self) -> None: for item in self.work_items: self._audit_work_item(item) for item in self.reconciliation_items: self._audit_association_item(item) for opportunity_id, opportunity in self.opportunities.items(): self._audit_missing_work(opportunity_id, opportunity) self._audit_stale_association_tasks() # Avoid exact duplicate findings produced by multiple queue views. unique: Dict[Tuple[str, str, str, str, str], Finding] = {} for finding in self.findings: key = ( finding.code, finding.opportunity_id, finding.work_item_id, finding.association_item_id, finding.expected_action, ) unique.setdefault(key, finding) self.findings = sorted( unique.values(), key=lambda finding: ( SEVERITY_ORDER.get(finding.severity, 99), finding.customer_name.casefold(), finding.opportunity_id, finding.code, ), ) visible_opportunities = { _s(item.get("opportunity_id")) for item in self.work_items if _s(item.get("opportunity_id")) } self.summary.update({ "generated_at": datetime.now(timezone.utc).isoformat(), "project_root": str(PROJECT_ROOT), "work_items_audited": len(self.work_items), "visible_opportunities": len(visible_opportunities), "open_opportunities_audited": len(self.opportunities), "open_reconciliation_items": sum( 1 for item in self.reconciliation_items if _lower(item.get("status")) in OPEN_RECONCILIATION_STATUSES ), "findings": len(self.findings), "severity": dict(Counter(finding.severity for finding in self.findings)), "codes": dict(Counter(finding.code for finding in self.findings)), }) def report(self, output_dir: Path) -> Dict[str, str]: output_dir.mkdir(parents=True, exist_ok=True) json_path = output_dir / "work_center_audit.json" markdown_path = output_dir / "WORK_CENTER_AUDIT.md" csv_path = output_dir / "work_center_findings.csv" correction_path = output_dir / "correction_plan.json" sql_path = output_dir / "diagnostic_queries.sql" payload = { "summary": self.summary, "findings": [asdict(finding) for finding in self.findings], "work_items": self.work_items, "next_actions": self.next_actions, } json_path.write_text(json.dumps(payload, ensure_ascii=False, indent=2, default=_json_default), encoding="utf-8") fieldnames = list(Finding.__dataclass_fields__.keys()) with csv_path.open("w", encoding="utf-8", newline="") as handle: writer = csv.DictWriter(handle, fieldnames=fieldnames) writer.writeheader() for finding in self.findings: row = asdict(finding) row["evidence"] = json.dumps(row["evidence"], ensure_ascii=False, default=_json_default) row["proposed_correction"] = json.dumps(row["proposed_correction"], ensure_ascii=False, default=_json_default) writer.writerow(row) correction_plan = [ { "severity": finding.severity, "code": finding.code, "opportunity_id": finding.opportunity_id, "task_id": finding.task_id, "association_item_id": finding.association_item_id, "external_ref": finding.external_ref, "expected_action": finding.expected_action, "correction": finding.proposed_correction, "requires_review": not bool(finding.proposed_correction.get("automatic")), } for finding in self.findings if finding.proposed_correction ] correction_path.write_text(json.dumps(correction_plan, ensure_ascii=False, indent=2, default=_json_default), encoding="utf-8") lines = [ "# Auditoria do Centro de Trabalho e associações", "", f"Gerada em: `{self.summary.get('generated_at')}`", f"Itens visíveis auditados: **{self.summary.get('work_items_audited', 0)}**", f"Oportunidades abertas auditadas: **{self.summary.get('open_opportunities_audited', 0)}**", f"Candidatos de reconciliação abertos: **{self.summary.get('open_reconciliation_items', 0)}**", f"Achados: **{self.summary.get('findings', 0)}**", "", "## Resumo por severidade", "", ] severities = self.summary.get("severity", {}) for severity in ("critical", "high", "medium", "low", "info"): lines.append(f"- {severity}: **{severities.get(severity, 0)}**") lines.extend(["", "## Achados", ""]) if not self.findings: lines.append("Nenhuma incoerência encontrada.") for index, finding in enumerate(self.findings, 1): lines.extend([ f"### {index}. [{finding.severity.upper()}] {finding.title}", "", f"- Código: `{finding.code}`", f"- Cliente: {finding.customer_name or '—'}", f"- Oportunidade: `{finding.opportunity_id or '—'}` {finding.opportunity_title}", f"- Task/item: `{finding.task_id or finding.work_item_id or '—'}`", f"- Reconciliação: `{finding.association_item_id or '—'}` · `{finding.external_ref or '—'}`", f"- Ação mostrada: `{finding.displayed_action or '—'}`", f"- Ação persistida: `{finding.stored_action or '—'}`", f"- Primeira ação segura: `{finding.expected_action or '—'}`", f"- Recomendação: {finding.recommendation}", f"- Evidência: `{json.dumps(finding.evidence, ensure_ascii=False, default=_json_default)}`", f"- Correção proposta: `{json.dumps(finding.proposed_correction, ensure_ascii=False, default=_json_default)}`", "", ]) lines.extend([ "## Garantia de segurança", "", "O auditor não possui modo de aplicação. As consultas próprias são executadas numa transação PostgreSQL `READ ONLY`; os serviços centrais chamados são funções de leitura. Nenhum item, oportunidade, documento, ligação ou task é alterado.", "", ]) markdown_path.write_text("\n".join(lines), encoding="utf-8") sql_path.write_text("""-- Diagnóstico complementar (READ ONLY) -- Tasks pendentes por oportunidade SELECT o.customer_name, o.stage, t.id, t.action_code, t.action, t.route, t.due_at FROM tasks t JOIN opportunities o ON o.id = t.opportunity_id WHERE t.status = 'pending' ORDER BY o.customer_name, t.created_at; -- Candidatos de reconciliação abertos e respetiva oportunidade SELECT ri.id, ri.source_system, ri.external_type, ri.document_number, ri.customer_name, ri.customer_tax_id, ri.opportunity_id, o.title, o.stage, ri.suggested_action, ri.status FROM reconciliation_items ri LEFT JOIN opportunities o ON o.id = ri.opportunity_id WHERE ri.status IN ('open','needs_review','conflict') ORDER BY ri.updated_at DESC; -- Referências externas ligadas a mais de uma oportunidade SELECT system, external_type, COALESCE(external_id, external_name) AS external_ref, COUNT(DISTINCT opportunity_id) AS opportunities, array_agg(DISTINCT opportunity_id) AS opportunity_ids FROM operation_links WHERE COALESCE(status, '') NOT IN ('ignored','deleted') GROUP BY system, external_type, COALESCE(external_id, external_name) HAVING COUNT(DISTINCT opportunity_id) > 1; -- Oportunidades com múltiplas tasks pendentes SELECT o.id, o.customer_name, o.stage, COUNT(*) AS pending_tasks, string_agg(t.action_code, ', ' ORDER BY t.created_at) AS actions FROM opportunities o JOIN tasks t ON t.opportunity_id = o.id AND t.status = 'pending' WHERE o.status = 'open' GROUP BY o.id, o.customer_name, o.stage HAVING COUNT(*) > 1 ORDER BY COUNT(*) DESC; """, encoding="utf-8") return { "json": str(json_path), "markdown": str(markdown_path), "csv": str(csv_path), "correction_plan": str(correction_path), "diagnostic_sql": str(sql_path), } def _self_test() -> None: assert _canonical_action("SHIP_ORDER") == "CREATE_SHIPMENT" assert _actions_match("SHIP_ORDER", "CREATE_SHIPMENT") assert _actions_match("RECONCILE_DOCUMENTS", "ASSOCIATE_OPPORTUNITY") assert _actions_match("FOLLOW_UP", "FOLLOW_UP_PAYMENT") assert not _actions_match("SEND_INVOICE", "CONFIRM_PAYMENT") assert _is_reconstructed({"clientflow_record_mode": "historical_reconstructed"}) assert not _is_reconstructed({"clientflow_record_mode": "normal"}) assert _external_refs({"external_id": "325", "document_number": "S00325"}) == {"325", "s00325"} assert _has_completed_reconstructed_review([ {"status": "done", "action_code": "REVIEW_MANUALLY", "metadata": {"review_type": "reconstructed_process"}} ], {}) assert not _has_completed_reconstructed_review([ {"status": "pending", "action_code": "REVIEW_MANUALLY", "metadata": {"review_type": "reconstructed_process"}} ], {}) print("Self-test OK") def _parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--limit", type=int, default=500, help="Máximo de itens normalizados do Centro de Trabalho.") parser.add_argument("--opportunity-limit", type=int, default=1000, help="Máximo de oportunidades abertas a auditar para cobertura.") parser.add_argument("--opportunity-id", action="append", default=[], help="Limitar a uma oportunidade; repetível.") parser.add_argument("--output-dir", default="", help="Diretório do relatório; por defeito audit_reports/work_center_.") parser.add_argument("--fail-on", choices=["critical", "high", "medium", "low", "info", "never"], default="critical", help="Código 2 quando existe achado nesta severidade ou superior.") parser.add_argument("--self-test", action="store_true") return parser def main() -> int: args = _parser().parse_args() if args.self_test: _self_test() return 0 auditor = WorkCenterAuditor( limit=args.limit, opportunity_limit=args.opportunity_limit, opportunity_ids=args.opportunity_id, ) auditor.load() auditor.audit() timestamp = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S") output_dir = Path(args.output_dir) if args.output_dir else Path("audit_reports") / f"work_center_{timestamp}" paths = auditor.report(output_dir) severity = auditor.summary.get("severity", {}) print("Centro de Trabalho auditado em modo READ ONLY") print(f"Itens visíveis auditados: {auditor.summary.get('work_items_audited', 0)}") print(f"Oportunidades abertas auditadas: {auditor.summary.get('open_opportunities_audited', 0)}") print(f"Candidatos de reconciliação abertos: {auditor.summary.get('open_reconciliation_items', 0)}") print(f"Achados: {auditor.summary.get('findings', 0)}") for level in ("critical", "high", "medium", "low", "info"): print(f" {level:8}: {severity.get(level, 0)}") for finding in auditor.findings[:80]: subject = finding.customer_name or finding.opportunity_title or finding.external_ref or finding.work_item_id print( f"{finding.severity.upper():8} | {finding.code:48} | " f"{finding.displayed_action or '-':28} -> {finding.expected_action or '-':28} | {subject}" ) if len(auditor.findings) > 80: print(f"... {len(auditor.findings) - 80} achado(s) adicionais no relatório.") print("Relatórios:") for key, value in paths.items(): print(f" {key}: {value}") if args.fail_on != "never" and any(_severity_at_or_above(f.severity, args.fail_on) for f in auditor.findings): return 2 return 0 if __name__ == "__main__": raise SystemExit(main())