#!/usr/bin/env python3 """Auditoria read-only da reconciliação ClientFlow. Objetivos: - detetar candidatos que já estão ligados a uma oportunidade; - comparar NIF, email, nome fiscal, documentos, valores, fase e próxima ação; - detetar risco de regressão ao ligar um candidato antigo/incompleto; - opcionalmente confirmar a evidência com pedidos GET/search_read a Odoo, Jasmin e Packlink; - produzir um plano de correções sem alterar a base de dados. O script abre a transação PostgreSQL como READ ONLY. Não cria, atualiza, resolve nem liga qualquer item de reconciliação. """ from __future__ import annotations import argparse import asyncio import csv import json import re import sys import unicodedata from collections import defaultdict from dataclasses import asdict, dataclass, field from datetime import date, datetime, timezone from decimal import Decimal, InvalidOperation from pathlib import Path from typing import Any, Dict, Iterable, List, Mapping, Optional, Sequence, Tuple from sqlalchemy import text def _detect_project_root() -> Path: """Locate the ClientFlow backend root without depending on cwd/PYTHONPATH.""" script_path = Path(__file__).resolve() candidates = [script_path.parent.parent, Path.cwd().resolve(), *script_path.parents] seen = 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)) OPEN_STATUSES = ("open", "needs_review", "conflict") STAGE_RANK = { "REVIEW": 5, "NEW_LEAD": 10, "INFO_REQUESTED": 20, "INFO_SENT": 30, "QUOTE_REQUESTED": 40, "QUOTE_SENT": 50, "PROFORMA_REQUESTED": 55, "PROFORMA_SENT": 60, "INVOICE_REQUESTED": 62, "INVOICE_SENT": 65, "WAITING_PAYMENT": 70, "PAYMENT_CONFIRMED": 80, "ODOO_ORDER_CREATED": 85, "IN_PRODUCTION": 88, "ORDER_PREPARATION": 90, "READY_TO_SHIP": 92, "INVOICED": 94, "SHIPMENT_CREATED": 98, "SHIPPED": 100, "TRACKING_SENT": 102, "DELIVERED": 108, "WON": 110, "NO_INTEREST": 118, "LOST": 120, "ARCHIVED": 130, } SUGGESTED_STAGE_BY_TYPE = { "jasmin_quotation": "QUOTE_SENT", "jasmin_proforma": "WAITING_PAYMENT", "jasmin_invoice": "INVOICE_SENT", "odoo_sale_order": "ODOO_ORDER_CREATED", "packlink_shipment": "SHIPMENT_CREATED", "payment_proof": "WAITING_PAYMENT", "payment_receipt": "WAITING_PAYMENT", "manual_request": "QUOTE_REQUESTED", } HUMAN_ACTIONS_BY_STAGE = { "QUOTE_SENT": {"FOLLOW_UP_QUOTE", "SEND_PROFORMA", "REVIEW_MANUALLY"}, "PROFORMA_SENT": {"FOLLOW_UP_PAYMENT", "CONFIRM_PAYMENT", "REVIEW_MANUALLY"}, "INVOICE_SENT": {"FOLLOW_UP_PAYMENT", "CONFIRM_PAYMENT", "REVIEW_MANUALLY"}, "WAITING_PAYMENT": {"FOLLOW_UP_PAYMENT", "CONFIRM_PAYMENT", "REVIEW_MANUALLY"}, "PAYMENT_CONFIRMED": {"PREPARE_ORDER", "REVIEW_MANUALLY"}, "ODOO_ORDER_CREATED": {"SEND_INVOICE", "PREPARE_ORDER", "REVIEW_MANUALLY"}, "IN_PRODUCTION": {"PREPARE_ORDER", "REVIEW_MANUALLY"}, "ORDER_PREPARATION": {"PREPARE_ORDER", "VALIDATE_PHYSICAL_ORDER", "CREATE_SHIPMENT", "SHIP_ORDER", "REVIEW_MANUALLY"}, "READY_TO_SHIP": {"CREATE_SHIPMENT", "SHIP_ORDER", "REVIEW_MANUALLY"}, "SHIPMENT_CREATED": {"SEND_TRACKING", "SHIP_ORDER", "REVIEW_MANUALLY"}, } SEVERITY_ORDER = {"critical": 0, "high": 1, "medium": 2, "low": 3, "info": 4} def _clean(value: Any) -> str: return str(value or "").strip() def _jsonable(value: Any) -> Any: if isinstance(value, (datetime, date)): return value.isoformat() if isinstance(value, Decimal): return str(value) if isinstance(value, Mapping): return {str(k): _jsonable(v) for k, v in value.items()} if isinstance(value, (list, tuple, set)): return [_jsonable(v) for v in value] return value def _normalize_tax_id(value: Any) -> str: raw = re.sub(r"[^A-Z0-9]", "", _clean(value).upper()) if raw.startswith("PT"): raw = raw[2:] return raw def _normalize_email(value: Any) -> str: return _clean(value).lower() def _normalize_name(value: Any) -> str: raw = unicodedata.normalize("NFKD", _clean(value).lower()) raw = "".join(ch for ch in raw if not unicodedata.combining(ch)) raw = re.sub(r"\b(lda|ltd|unipessoal|sa|s a|sociedade anonima|limitada)\b", " ", raw) raw = re.sub(r"[^a-z0-9]+", " ", raw) return " ".join(raw.split()) def _money(value: Any) -> Optional[Decimal]: if value is None or value == "": return None raw = _clean(value).replace("€", "").replace(" ", "").replace(",", ".") try: return Decimal(raw).quantize(Decimal("0.01")) except (InvalidOperation, ValueError): return None def _stage_rank(stage: Any) -> int: return STAGE_RANK.get(_clean(stage).upper(), 0) def _external_refs(item: Mapping[str, Any]) -> List[str]: values = [item.get("external_id"), item.get("document_number")] payload = item.get("payload") if isinstance(item.get("payload"), dict) else {} record = payload.get("record") if isinstance(payload.get("record"), dict) else {} values.extend([ record.get("id"), record.get("name"), record.get("documentNumber"), record.get("naturalKey"), record.get("reference"), ]) refs: List[str] = [] for value in values: value = _clean(value) if value and value.lower() not in {ref.lower() for ref in refs}: refs.append(value) return refs def _system_for_link(source_system: Any) -> str: value = _clean(source_system).lower() if value in {"odoo", "jasmin", "packlink"}: return value return value def _expected_link_types(external_type: Any) -> set[str]: return { "odoo_sale_order": {"sale_order", "odoo_sale_order"}, "packlink_shipment": {"shipment", "packlink_shipment"}, "jasmin_quotation": {"quotation", "jasmin_quotation"}, "jasmin_proforma": {"proforma", "jasmin_proforma"}, "jasmin_invoice": {"invoice", "jasmin_invoice"}, }.get(_clean(external_type), set()) def _document_kind(external_type: Any) -> str: return { "jasmin_quotation": "quotation", "jasmin_proforma": "proforma", "jasmin_invoice": "invoice", }.get(_clean(external_type), "") def _close_amount(left: Any, right: Any, *, absolute: Decimal = Decimal("0.02"), relative: Decimal = Decimal("0.02")) -> bool: a, b = _money(left), _money(right) if a is None or b is None: return False diff = abs(a - b) if diff <= absolute: return True base = max(abs(a), abs(b), Decimal("1")) return (diff / base) <= relative def _looks_like_tax_difference(net: Any, gross: Any) -> bool: a, b = _money(net), _money(gross) if a is None or b is None or a <= 0 or b <= 0: return False ratio = max(a, b) / min(a, b) return Decimal("1.20") <= ratio <= Decimal("1.27") def _picking_state_from_payload(item: Mapping[str, Any]) -> Tuple[str, str]: payload = item.get("payload") if isinstance(item.get("payload"), dict) else {} record = payload.get("record") if isinstance(payload.get("record"), dict) else payload fulfilment = record.get("fulfilment") if isinstance(record.get("fulfilment"), dict) else {} pickings = fulfilment.get("outgoing_pickings") if isinstance(fulfilment.get("outgoing_pickings"), list) else [] if not pickings and isinstance(record.get("pickings"), list): pickings = record.get("pickings") if not pickings: return "", "" ordered = sorted(pickings, key=lambda p: _clean(p.get("date_done") or p.get("scheduled_date")), reverse=True) state = _clean(ordered[0].get("state")).lower() scheduled = _clean(ordered[0].get("scheduled_date") or ordered[0].get("date_done")) return state, scheduled @dataclass class Finding: severity: str code: str title: str recommendation: str item_id: str = "" source_system: str = "" external_type: str = "" external_ref: str = "" customer_name: str = "" opportunity_id: str = "" opportunity_title: str = "" evidence: Dict[str, Any] = field(default_factory=dict) proposed_correction: Dict[str, Any] = field(default_factory=dict) class Auditor: def __init__(self, *, days: int, limit: int, source: str = "", item_id: str = "") -> None: self.days = max(int(days), 1) self.limit = min(max(int(limit), 1), 5000) self.source = _clean(source).lower() self.item_id = _clean(item_id) self.items: List[Dict[str, Any]] = [] self.opportunities: Dict[str, Dict[str, Any]] = {} self.operation_links: List[Dict[str, Any]] = [] self.documents: List[Dict[str, Any]] = [] self.tasks: List[Dict[str, Any]] = [] self.findings: List[Finding] = [] self.live_results: Dict[str, Any] = {} def load(self) -> None: try: from app.db import engine 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/ ou execute-o " "a partir da raiz do backend." ) from exc source_sql = "" params: Dict[str, Any] = {"days": self.days, "limit": self.limit} if self.source: source_sql += " AND lower(ri.source_system) = :source" params["source"] = self.source if self.item_id: source_sql += " AND ri.id = CAST(:item_id AS UUID)" params["item_id"] = self.item_id with engine.connect() as conn: transaction = conn.begin() try: conn.execute(text("SET TRANSACTION READ ONLY")) existing_tables = set(conn.execute(text(""" SELECT table_name FROM information_schema.tables WHERE table_schema = current_schema() AND table_name IN ( 'reconciliation_items','opportunities','customers', 'operation_links','commercial_documents','tasks' ) """)).scalars().all()) required = {"reconciliation_items", "opportunities", "operation_links", "commercial_documents", "tasks"} missing = sorted(required - existing_tables) if missing: raise RuntimeError("Tabelas necessárias em falta: " + ", ".join(missing)) self.items = [dict(row) for row in conn.execute(text(f""" SELECT ri.id::text, ri.source_system, ri.external_type, ri.external_id, ri.title, ri.description, ri.status, ri.priority, ri.suggested_action, ri.confidence, ri.opportunity_id::text, ri.customer_id::text, ri.customer_name, ri.customer_email, ri.customer_tax_id, ri.document_number, ri.document_date, ri.amount, ri.currency, ri.payload, ri.resolution_note, ri.created_at, ri.updated_at, o.title AS linked_opportunity_title, o.stage AS linked_opportunity_stage, o.status AS linked_opportunity_status FROM reconciliation_items ri LEFT JOIN opportunities o ON o.id = ri.opportunity_id WHERE ri.status IN ('open','needs_review','conflict') AND (ri.document_date IS NULL OR ri.document_date >= CURRENT_DATE - :days) {source_sql} ORDER BY ri.updated_at DESC, ri.created_at DESC LIMIT :limit """), params).mappings().all()] opportunity_rows = conn.execute(text(""" SELECT o.id::text, o.title, o.stage, o.status, o.customer_name, o.customer_email, o.value_amount, o.local_customer_id::text, o.metadata, o.updated_at, c.name AS fiscal_name, c.email AS fiscal_email, c.tax_id AS fiscal_tax_id FROM opportunities o LEFT JOIN customers c ON c.id = o.local_customer_id WHERE o.status IN ('open','won','lost','no_interest','archived') """)).mappings().all() self.opportunities = {str(row["id"]): dict(row) for row in opportunity_rows} 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.last_synced_at, ol.updated_at, o.title AS opportunity_title, o.stage AS opportunity_stage, o.status AS opportunity_status, o.customer_name AS opportunity_customer_name FROM operation_links ol JOIN opportunities o ON o.id = ol.opportunity_id """)).mappings().all()] self.documents = [dict(row) for row in conn.execute(text(""" SELECT cd.id::text, cd.opportunity_id::text, cd.customer_id::text, cd.system, cd.document_kind, cd.external_id, cd.document_number, cd.status, cd.amount, cd.tax_amount, cd.total_amount, cd.currency, cd.role, cd.is_primary, cd.is_active, cd.document_date, cd.payload, cd.updated_at, o.title AS opportunity_title, o.stage AS opportunity_stage, o.status AS opportunity_status FROM commercial_documents cd LEFT JOIN opportunities o ON o.id = cd.opportunity_id """)).mappings().all()] self.tasks = [dict(row) for row in conn.execute(text(""" SELECT t.id::text, t.opportunity_id::text, t.action_code, t.action, t.route, t.status, t.priority, t.due_at, t.source_system, t.metadata, t.created_at FROM tasks t WHERE t.status = 'pending' """)).mappings().all()] finally: transaction.rollback() def add(self, finding: Finding) -> None: self.findings.append(finding) def _matching_links(self, item: Mapping[str, Any]) -> List[Dict[str, Any]]: system = _system_for_link(item.get("source_system")) refs = {ref.lower() for ref in _external_refs(item)} expected_types = _expected_link_types(item.get("external_type")) matches = [] for link in self.operation_links: if _clean(link.get("system")).lower() != system: continue values = {_clean(link.get("external_id")).lower(), _clean(link.get("external_name")).lower()} if refs & {value for value in values if value}: matches.append(link) continue if expected_types and _clean(link.get("external_type")) in expected_types: # Same type alone is not a match; keep strict external identity. continue return matches def _matching_documents(self, item: Mapping[str, Any]) -> List[Dict[str, Any]]: if _clean(item.get("source_system")).lower() != "jasmin": return [] refs = {ref.lower() for ref in _external_refs(item)} kind = _document_kind(item.get("external_type")) matches = [] for doc in self.documents: if _clean(doc.get("system")).lower() != "jasmin": continue if kind and _clean(doc.get("document_kind")) != kind: continue values = {_clean(doc.get("external_id")).lower(), _clean(doc.get("document_number")).lower()} if refs & {value for value in values if value}: matches.append(doc) return matches def _identity_candidates(self, item: Mapping[str, Any]) -> List[Dict[str, Any]]: tax_id = _normalize_tax_id(item.get("customer_tax_id")) email = _normalize_email(item.get("customer_email")) name = _normalize_name(item.get("customer_name")) amount = _money(item.get("amount")) scored: List[Tuple[int, Dict[str, Any], List[str]]] = [] for opp in self.opportunities.values(): if _clean(opp.get("status")).lower() != "open": continue score = 0 reasons: List[str] = [] opp_tax = _normalize_tax_id(opp.get("fiscal_tax_id")) opp_email_values = {_normalize_email(opp.get("customer_email")), _normalize_email(opp.get("fiscal_email"))} opp_names = {_normalize_name(opp.get("customer_name")), _normalize_name(opp.get("fiscal_name"))} if tax_id and opp_tax and tax_id == opp_tax: score += 120 reasons.append("NIF exato") if email and email in {x for x in opp_email_values if x}: score += 80 reasons.append("email exato") if name and name in {x for x in opp_names if x}: score += 60 reasons.append("nome fiscal exato") elif name and any(name in x or x in name for x in opp_names if len(x) >= 5): score += 35 reasons.append("nome fiscal aproximado") if amount is not None and _close_amount(amount, opp.get("value_amount"), relative=Decimal("0.05")): score += 10 reasons.append("valor próximo") if score >= 35: copy = dict(opp) copy["match_score"] = score copy["match_reasons"] = reasons scored.append((score, copy, reasons)) scored.sort(key=lambda row: (-row[0], -_stage_rank(row[1].get("stage")), _clean(row[1].get("title")))) return [row[1] for row in scored[:5]] def _documents_for_opportunity(self, opportunity_id: str) -> List[Dict[str, Any]]: return [doc for doc in self.documents if _clean(doc.get("opportunity_id")) == opportunity_id and bool(doc.get("is_active", True))] def _tasks_for_opportunity(self, opportunity_id: str) -> List[Dict[str, Any]]: return [task for task in self.tasks if _clean(task.get("opportunity_id")) == opportunity_id] def _audit_item(self, item: Dict[str, Any]) -> None: item_id = _clean(item.get("id")) refs = _external_refs(item) external_ref = refs[0] if refs else "" common = { "item_id": item_id, "source_system": _clean(item.get("source_system")), "external_type": _clean(item.get("external_type")), "external_ref": external_ref, "customer_name": _clean(item.get("customer_name")), } links = self._matching_links(item) docs = self._matching_documents(item) exact_opp_ids = { _clean(link.get("opportunity_id")) for link in links if _clean(link.get("opportunity_id")) } | { _clean(doc.get("opportunity_id")) for doc in docs if _clean(doc.get("opportunity_id")) } if _clean(item.get("opportunity_id")): exact_opp_ids.add(_clean(item.get("opportunity_id"))) if len(exact_opp_ids) > 1: self.add(Finding( severity="critical", code="EXTERNAL_ID_LINKED_TO_MULTIPLE_OPPORTUNITIES", title="A mesma evidência externa aparece ligada a várias oportunidades", recommendation="Bloquear a ligação e rever manualmente as associações antes de qualquer correção de fase ou documento.", evidence={"opportunity_ids": sorted(exact_opp_ids), "links": links, "documents": docs}, proposed_correction={"action": "block_and_review", "automatic": False}, **common, )) return exact_opp_id = next(iter(exact_opp_ids), "") exact_opp = self.opportunities.get(exact_opp_id) if exact_opp_id else None if exact_opp: common.update({ "opportunity_id": exact_opp_id, "opportunity_title": _clean(exact_opp.get("title")), }) if _clean(item.get("status")) in OPEN_STATUSES: self.add(Finding( severity="high", code="OPEN_ITEM_ALREADY_LINKED", title="Candidato aberto apesar de a evidência já estar ligada", recommendation="Resolver o item como já ligado, sem reaplicar a fase sugerida nem recriar tasks.", evidence={ "item_status": item.get("status"), "existing_links": links, "existing_documents": docs, "current_stage": exact_opp.get("stage"), }, proposed_correction={ "action": "resolve_as_existing_link", "automatic_candidate": True, "guard": "confirm exactly one existing opportunity", }, **common, )) suggested_stage = SUGGESTED_STAGE_BY_TYPE.get(_clean(item.get("external_type")), "") current_stage = _clean(exact_opp.get("stage")).upper() if suggested_stage and _stage_rank(current_stage) > _stage_rank(suggested_stage): self.add(Finding( severity="high", code="STAGE_REGRESSION_RISK", title="A reconciliação sugere uma fase anterior à fase atual", recommendation="Ao resolver o item, preservar a fase atual e usar a evidência apenas como contexto/documento.", evidence={"current_stage": current_stage, "suggested_stage": suggested_stage, "suggested_action": item.get("suggested_action")}, proposed_correction={"action": "preserve_current_stage", "automatic_candidate": True}, **common, )) opp_docs = self._documents_for_opportunity(exact_opp_id) has_invoice = any(_clean(doc.get("document_kind")) == "invoice" and bool(doc.get("is_active", True)) for doc in opp_docs) has_quotation = any(_clean(doc.get("document_kind")) == "quotation" and bool(doc.get("is_active", True)) for doc in opp_docs) suggested_action = _clean(item.get("suggested_action")).upper() if suggested_action == "SEND_INVOICE" and has_invoice: self.add(Finding( severity="high", code="OBSOLETE_SEND_INVOICE", title="A reconciliação pede envio de fatura, mas a oportunidade já tem fatura", recommendation="Marcar o candidato como resolvido/obsoleto e recalcular a próxima ação a partir da fase operacional atual.", evidence={"documents": opp_docs, "current_stage": current_stage}, proposed_correction={"action": "resolve_obsolete_action", "replacement_action": self._recommended_action(exact_opp, opp_docs)}, **common, )) if suggested_action in {"SEND_QUOTE", "SEND_PROFORMA"} and has_quotation and _stage_rank(current_stage) >= _stage_rank("QUOTE_SENT"): self.add(Finding( severity="medium", code="OBSOLETE_DOCUMENT_ACTION", title="Ação documental sugerida já foi ultrapassada", recommendation="Preservar os documentos atuais e recalcular a próxima ação sem repetir o envio histórico.", evidence={"suggested_action": suggested_action, "current_stage": current_stage}, proposed_correction={"action": "recompute_next_action"}, **common, )) item_amount = _money(item.get("amount")) opp_amount = _money(exact_opp.get("value_amount")) doc_amounts = [ _money(doc.get("total_amount")) or _money(doc.get("amount")) for doc in opp_docs if (_money(doc.get("total_amount")) or _money(doc.get("amount"))) is not None ] if item_amount is not None: comparable = [value for value in [opp_amount, *doc_amounts] if value is not None] if comparable and not any(_close_amount(item_amount, value, relative=Decimal("0.03")) for value in comparable): severity = "medium" if any(_looks_like_tax_difference(item_amount, value) for value in comparable) else "high" self.add(Finding( severity=severity, code="AMOUNT_MISMATCH", title="Valor da reconciliação diverge do valor da oportunidade/documentos", recommendation="Validar se a diferença corresponde a IVA, transporte, desconto ou linhas em falta antes de aplicar a associação.", evidence={ "reconciliation_amount": item_amount, "opportunity_amount": opp_amount, "document_amounts": comparable, "possible_tax_difference": any(_looks_like_tax_difference(item_amount, value) for value in comparable), }, proposed_correction={"action": "review_amount_and_lines", "automatic": False}, **common, )) self._audit_missing_work_task(exact_opp, opp_docs, common) else: candidates = self._identity_candidates(item) if candidates: best = candidates[0] common.update({"opportunity_id": _clean(best.get("id")), "opportunity_title": _clean(best.get("title"))}) severity = "medium" if int(best.get("match_score") or 0) >= 80 else "low" self.add(Finding( severity=severity, code="EXISTING_OPPORTUNITY_CANDIDATE", title="Existe uma oportunidade aberta potencialmente compatível", recommendation="Comparar NIF, cliente fiscal, documentos e compra antes de criar uma nova oportunidade.", evidence={"candidates": candidates}, proposed_correction={"action": "review_link_to_existing", "automatic": False}, **common, )) else: self.add(Finding( severity="info", code="NO_EXISTING_OPPORTUNITY_FOUND", title="Não foi encontrada oportunidade aberta compatível", recommendation="Confirmar identidade e decidir entre criar nova oportunidade, histórico ou ignorar.", evidence={"identity": {"tax_id": item.get("customer_tax_id"), "email": item.get("customer_email"), "name": item.get("customer_name")}}, proposed_correction={"action": "manual_triage"}, **common, )) picking_state, picking_date = _picking_state_from_payload(item) if picking_state in {"assigned", "confirmed", "waiting"} and _clean(item.get("suggested_action")).upper() == "SEND_INVOICE": self.add(Finding( severity="medium", code="ODOO_PHYSICAL_STATE_ACTION_MISMATCH", title="O estado físico Odoo está avançado, mas a ação sugerida é apenas enviar fatura", recommendation="Verificar primeiro se já existe fatura/pagamento e, se existir, orientar a próxima ação para expedição.", evidence={"picking_state": picking_state, "scheduled_date": picking_date, "suggested_action": item.get("suggested_action")}, proposed_correction={"action": "recompute_from_full_operation_context"}, **common, )) if picking_date: try: parsed = datetime.fromisoformat(picking_date.replace("Z", "+00:00")).date() except Exception: parsed = None if parsed and parsed > datetime.now(timezone.utc).date(): self.add(Finding( severity="low", code="FUTURE_OPERATION_DATE_LABEL", title="Data futura apresentada como evento já ocorrido", recommendation="Apresentar como data prevista/programada e não como entrega concluída.", evidence={"scheduled_date": picking_date, "picking_state": picking_state}, proposed_correction={"action": "fix_ui_label"}, **common, )) def _recommended_action(self, opportunity: Mapping[str, Any], docs: Sequence[Mapping[str, Any]]) -> str: stage = _clean(opportunity.get("stage")).upper() has_invoice = any(_clean(doc.get("document_kind")) == "invoice" and bool(doc.get("is_active", True)) for doc in docs) if stage == "ORDER_PREPARATION" and picking_state == "assigned": return "VALIDATE_PHYSICAL_ORDER" if stage == "READY_TO_SHIP": return "CREATE_SHIPMENT" if stage in {"ORDER_PREPARATION", "IN_PRODUCTION"}: return "PREPARE_ORDER" if stage in {"ODOO_ORDER_CREATED", "PAYMENT_CONFIRMED"}: return "PREPARE_ORDER" if has_invoice else "SEND_INVOICE" if stage in {"INVOICE_SENT", "WAITING_PAYMENT"}: return "FOLLOW_UP_PAYMENT" if stage == "QUOTE_SENT": return "FOLLOW_UP_QUOTE" return "REVIEW_MANUALLY" def _audit_missing_work_task(self, opportunity: Mapping[str, Any], docs: Sequence[Mapping[str, Any]], common: Dict[str, Any]) -> None: opportunity_id = _clean(opportunity.get("id")) stage = _clean(opportunity.get("stage")).upper() expected = HUMAN_ACTIONS_BY_STAGE.get(stage) if not expected: return pending = self._tasks_for_opportunity(opportunity_id) pending_codes = {_clean(task.get("action_code")).upper() for task in pending} recommended = self._recommended_action(opportunity, docs) if not pending and recommended in expected: severity = "high" if stage in {"READY_TO_SHIP", "SHIPMENT_CREATED"} else "medium" self.add(Finding( severity=severity, code="MISSING_HUMAN_TASK", title="A oportunidade tem próxima ação humana, mas não tem task pendente", recommendation="Materializar uma única task operacional coerente com a fase, sem regressar a oportunidade.", evidence={"stage": stage, "recommended_action": recommended, "pending_tasks": []}, proposed_correction={"action": "materialize_single_task", "action_code": recommended, "automatic": False}, **common, )) elif pending and recommended not in pending_codes and stage in {"READY_TO_SHIP", "SHIPMENT_CREATED"}: self.add(Finding( severity="medium", code="PENDING_TASK_NOT_ALIGNED_WITH_STAGE", title="A task pendente não corresponde à fase operacional mais avançada", recommendation="Rever a task antes de executar; evitar ações financeiras/comerciais já ultrapassadas.", evidence={"stage": stage, "recommended_action": recommended, "pending_tasks": pending}, proposed_correction={"action": "review_pending_task", "replacement_action": recommended}, **common, )) def audit(self) -> None: for item in self.items: self._audit_item(item) self.findings.sort(key=lambda f: (SEVERITY_ORDER.get(f.severity, 9), f.customer_name, f.external_ref, f.code)) async def _probe_jasmin(self, items: Sequence[Mapping[str, Any]], max_live: int) -> Dict[str, Any]: from app.jasmin_client import JasminClient client = JasminClient() results: List[Dict[str, Any]] = [] selected = [item for item in items if _clean(item.get("source_system")).lower() == "jasmin"][:max_live] for item in selected: external_id = _clean(item.get("external_id")) external_type = _clean(item.get("external_type")) row = {"item_id": item.get("id"), "external_id": external_id, "external_type": external_type, "ok": False} if not external_id: row["error"] = "external_id em falta; GET ignorado" row["skipped"] = True results.append(row) continue try: if external_type == "jasmin_quotation": payload = await client.get_quotation(external_id) elif external_type == "jasmin_invoice": payload = await client.get_invoice(external_id) else: row["error"] = "tipo sem GET direto implementado" row["skipped"] = True results.append(row) continue row.update({"ok": True, "payload": payload}) except Exception as exc: row["error"] = str(exc) results.append(row) return {"system": "jasmin", "count": len(results), "results": results} def _probe_odoo(self, items: Sequence[Mapping[str, Any]], max_live: int) -> Dict[str, Any]: from app.odoo_client import OdooClient client = OdooClient() results: List[Dict[str, Any]] = [] fields = ["id", "name", "state", "partner_id", "amount_total", "amount_untaxed", "currency_id", "date_order", "invoice_status", "write_date"] selected = [item for item in items if _clean(item.get("source_system")).lower() == "odoo"][:max_live] for item in selected: external_id = _clean(item.get("external_id")) order_name = _clean(item.get("document_number")) row = {"item_id": item.get("id"), "external_id": external_id, "document_number": order_name, "ok": False} try: records: List[Dict[str, Any]] = [] if external_id.isdigit(): records = client.search_read("sale.order", [["id", "=", int(external_id)]], fields, limit=2) if not records and order_name: records = client.search_read("sale.order", [["name", "=", order_name]], fields, limit=2) if not records: row["error"] = "venda não encontrada no Odoo" else: sale = records[0] pickings = client.search_read( "stock.picking", [["origin", "ilike", _clean(sale.get("name"))]], ["id", "name", "state", "origin", "picking_type_id", "scheduled_date", "date_done"], limit=100, order="id desc", ) row.update({"ok": True, "sale": sale, "pickings": pickings}) except Exception as exc: row["error"] = str(exc) results.append(row) return {"system": "odoo", "count": len(results), "results": results} async def _probe_packlink(self, items: Sequence[Mapping[str, Any]], max_live: int) -> Dict[str, Any]: from app.packlink_client import PacklinkClient client = PacklinkClient() results: List[Dict[str, Any]] = [] selected = [item for item in items if _clean(item.get("source_system")).lower() == "packlink"][:max_live] for item in selected: ref = _clean(item.get("external_id") or item.get("document_number")) row = {"item_id": item.get("id"), "reference": ref, "ok": False} if not ref: row["error"] = "referência Packlink em falta" row["skipped"] = True results.append(row) continue try: payload = await client.get_shipment(ref) row.update({"ok": True, "payload": payload}) except Exception as exc: row["error"] = str(exc) results.append(row) return {"system": "packlink", "count": len(results), "results": results} def probe_live(self, systems: Sequence[str], max_live: int) -> None: chosen = {system.strip().lower() for system in systems if system.strip()} if "all" in chosen: chosen = {"odoo", "jasmin", "packlink"} if "odoo" in chosen: try: self.live_results["odoo"] = self._probe_odoo(self.items, max_live) except Exception as exc: self.live_results["odoo"] = {"system": "odoo", "error": str(exc), "results": []} if "jasmin" in chosen: try: self.live_results["jasmin"] = asyncio.run(self._probe_jasmin(self.items, max_live)) except Exception as exc: self.live_results["jasmin"] = {"system": "jasmin", "error": str(exc), "results": []} if "packlink" in chosen: try: self.live_results["packlink"] = asyncio.run(self._probe_packlink(self.items, max_live)) except Exception as exc: self.live_results["packlink"] = {"system": "packlink", "error": str(exc), "results": []} self._audit_live_results() def _audit_live_results(self) -> None: item_by_id = {_clean(item.get("id")): item for item in self.items} for system, payload in self.live_results.items(): if payload.get("error"): self.add(Finding( severity="high", code=f"LIVE_{system.upper()}_UNAVAILABLE", title=f"Integração {system} indisponível para auditoria live", recommendation="Corrigir configuração/credenciais antes de confiar nos candidatos dessa origem.", source_system=system, evidence={"error": payload.get("error")}, proposed_correction={"action": "fix_integration_configuration", "automatic": False}, )) for result in payload.get("results") or []: if result.get("skipped"): continue item = item_by_id.get(_clean(result.get("item_id"))) if not item: continue common = { "item_id": _clean(item.get("id")), "source_system": _clean(item.get("source_system")), "external_type": _clean(item.get("external_type")), "external_ref": (_external_refs(item) or [""])[0], "customer_name": _clean(item.get("customer_name")), } if not result.get("ok"): severity = "high" if "401" in _clean(result.get("error")) else "medium" self.add(Finding( severity=severity, code=f"LIVE_{system.upper()}_GET_FAILED", title=f"Falha na confirmação live em {system}", recommendation="Corrigir credenciais/identificador ou tratar o candidato como não confirmado; não ligar automaticamente.", evidence=result, proposed_correction={"action": "fix_integration_or_review", "automatic": False}, **common, )) elif system == "odoo": sale = result.get("sale") or {} live_amount = _money(sale.get("amount_total")) staged_amount = _money(item.get("amount")) if live_amount is not None and staged_amount is not None and not _close_amount(live_amount, staged_amount): self.add(Finding( severity="high", code="LIVE_ODOO_AMOUNT_CHANGED", title="O valor Odoo mudou desde a criação do candidato", recommendation="Atualizar o staging e rever linhas antes de ligar o processo.", evidence={"staged_amount": staged_amount, "live_amount": live_amount, "sale": sale}, proposed_correction={"action": "refresh_candidate_from_odoo"}, **common, )) self.findings.sort(key=lambda f: (SEVERITY_ORDER.get(f.severity, 9), f.customer_name, f.external_ref, f.code)) def summary(self) -> Dict[str, Any]: by_severity: Dict[str, int] = defaultdict(int) by_code: Dict[str, int] = defaultdict(int) for finding in self.findings: by_severity[finding.severity] += 1 by_code[finding.code] += 1 return { "generated_at": datetime.now(timezone.utc).isoformat(), "days": self.days, "items_audited": len(self.items), "findings": len(self.findings), "by_severity": dict(sorted(by_severity.items(), key=lambda x: SEVERITY_ORDER.get(x[0], 9))), "by_code": dict(sorted(by_code.items())), "live_systems": sorted(self.live_results), "read_only": True, } def write_reports(self, output_dir: Path) -> Dict[str, str]: output_dir.mkdir(parents=True, exist_ok=True) summary = self.summary() findings_data = [_jsonable(asdict(finding)) for finding in self.findings] items_data = [_jsonable(item) for item in self.items] full = { "summary": summary, "findings": findings_data, "items": items_data, "live_results": _jsonable(self.live_results), } json_path = output_dir / "reconciliation_audit.json" json_path.write_text(json.dumps(full, ensure_ascii=False, indent=2), encoding="utf-8") plan_path = output_dir / "correction_plan.json" plan = { "generated_at": summary["generated_at"], "read_only": True, "instructions": "Rever cada proposta antes de criar qualquer script --apply.", "corrections": [ { "item_id": finding.item_id, "code": finding.code, "severity": finding.severity, "opportunity_id": finding.opportunity_id, "external_ref": finding.external_ref, "proposal": _jsonable(finding.proposed_correction), "guard_evidence": _jsonable(finding.evidence), } for finding in self.findings if finding.proposed_correction ], } plan_path.write_text(json.dumps(plan, ensure_ascii=False, indent=2), encoding="utf-8") csv_path = output_dir / "reconciliation_findings.csv" columns = [ "severity", "code", "customer_name", "source_system", "external_type", "external_ref", "item_id", "opportunity_id", "opportunity_title", "title", "recommendation", "proposed_correction", "evidence", ] with csv_path.open("w", encoding="utf-8", newline="") as handle: writer = csv.DictWriter(handle, fieldnames=columns) writer.writeheader() for finding in self.findings: row = asdict(finding) row["proposed_correction"] = json.dumps(_jsonable(row["proposed_correction"]), ensure_ascii=False) row["evidence"] = json.dumps(_jsonable(row["evidence"]), ensure_ascii=False) writer.writerow({key: row.get(key, "") for key in columns}) md_path = output_dir / "RECONCILIATION_AUDIT.md" lines = [ "# Auditoria de coerência da reconciliação", "", f"Gerada em: `{summary['generated_at']}`", f"Janela: `{self.days}` dia(s)", f"Itens auditados: **{summary['items_audited']}**", f"Achados: **{summary['findings']}**", "", "## Resumo por severidade", "", ] for severity in ["critical", "high", "medium", "low", "info"]: lines.append(f"- {severity}: **{summary['by_severity'].get(severity, 0)}**") lines.extend(["", "## Achados", ""]) if not self.findings: lines.append("Nenhuma incoerência encontrada dentro da janela auditada.") 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"- Origem: `{finding.source_system}` / `{finding.external_type}` / `{finding.external_ref or '—'}`", f"- Item: `{finding.item_id or '—'}`", f"- Oportunidade: `{finding.opportunity_id or '—'}` {finding.opportunity_title or ''}", f"- Recomendação: {finding.recommendation}", f"- Correção proposta: `{json.dumps(_jsonable(finding.proposed_correction), ensure_ascii=False)}`", "", ]) lines.extend([ "## Garantia de segurança", "", "Este relatório foi produzido numa transação PostgreSQL `READ ONLY`. O script não alterou itens, oportunidades, documentos, ligações ou tasks.", ]) md_path.write_text("\n".join(lines), encoding="utf-8") sql_path = output_dir / "diagnostic_queries.sql" sql_path.write_text("""-- Consultas apenas de diagnóstico; não contêm UPDATE/DELETE/INSERT.\n\nSELECT status, source_system, external_type, COUNT(*)\nFROM reconciliation_items\nWHERE status IN ('open','needs_review','conflict')\nGROUP BY status, source_system, external_type\nORDER BY status, source_system, external_type;\n\nSELECT ri.id, ri.source_system, ri.external_type, ri.external_id, ri.document_number,\n ri.customer_name, ri.opportunity_id, o.title, o.stage\nFROM reconciliation_items ri\nLEFT JOIN opportunities o ON o.id = ri.opportunity_id\nWHERE ri.status IN ('open','needs_review','conflict')\nORDER BY ri.updated_at DESC;\n""", encoding="utf-8") return { "json": str(json_path), "markdown": str(md_path), "csv": str(csv_path), "correction_plan": str(plan_path), "diagnostic_sql": str(sql_path), } def _self_test() -> None: assert _normalize_tax_id("PT 510 177 441") == "510177441" assert _normalize_email(" Test@Example.COM ") == "test@example.com" assert _normalize_name("NOLTIA SYSTEM, LDA") == "noltia system" assert _close_amount("269,37", Decimal("269.37")) assert _looks_like_tax_difference("219.00", "269.37") assert _stage_rank("READY_TO_SHIP") > _stage_rank("ODOO_ORDER_CREATED") print("Self-test OK") def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description="Auditoria read-only da reconciliação ClientFlow") parser.add_argument("--days", type=int, default=3, help="Janela operacional em dias (default: 3)") parser.add_argument("--limit", type=int, default=500, help="Máximo de itens a auditar") parser.add_argument("--source", choices=["odoo", "jasmin", "packlink", "manual"], default=None, help="Filtrar origem") parser.add_argument("--item-id", default="", help="Auditar apenas um reconciliation_item UUID") parser.add_argument("--live", default="", help="GET live separados por vírgula: odoo,jasmin,packlink,all") parser.add_argument("--max-live", type=int, default=30, help="Máximo de pedidos live por sistema") parser.add_argument("--output-dir", default="", help="Diretório dos relatórios") parser.add_argument("--no-files", action="store_true", help="Não escrever relatórios em disco") parser.add_argument("--self-test", action="store_true", help="Executar testes internos e terminar") return parser.parse_args() def main() -> int: args = parse_args() if args.self_test: _self_test() return 0 auditor = Auditor(days=args.days, limit=args.limit, source=args.source, item_id=args.item_id) auditor.load() auditor.audit() if args.live: auditor.probe_live([value for value in args.live.split(",") if value.strip()], max(1, int(args.max_live))) summary = auditor.summary() print("Reconciliação auditada em modo READ ONLY") print(f"Itens auditados: {summary['items_audited']}") print(f"Achados: {summary['findings']}") for severity in ["critical", "high", "medium", "low", "info"]: print(f" {severity:8}: {summary['by_severity'].get(severity, 0)}") for finding in auditor.findings[:50]: print( f"{finding.severity.upper():8} | {finding.code:42} | " f"{finding.external_ref or '—':18} | {finding.customer_name or '—'}" ) if len(auditor.findings) > 50: print(f"... mais {len(auditor.findings) - 50} achado(s) no relatório") if not args.no_files: if args.output_dir: output_dir = Path(args.output_dir) else: stamp = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S") output_dir = Path("audit_reports") / f"reconciliation_{stamp}" paths = auditor.write_reports(output_dir) print("Relatórios:") for name, path in paths.items(): print(f" {name}: {path}") return 2 if summary["by_severity"].get("critical", 0) else 0 if __name__ == "__main__": raise SystemExit(main())