#!/usr/bin/env python3 """Audit and repair stale Odoo reconciliation candidates. Problem addressed ----------------- A reconciliation item may remain ``open`` even though an exact Odoo sale-order link already exists in ``operation_links``. The current synchronizer can ignore links whose opportunity is closed because ``_existing_odoo_sale_links`` filters with ``o.status = 'open'``. Default behaviour is read-only. Use ``--apply`` to: 1. remove that exact source-code filter, creating a timestamped backup; and 2. mark unambiguous stale reconciliation items as ``linked``. Ambiguous cases with more than one linked opportunity are never changed. Documents supplied through ``--exclude`` are also never changed. Run from the ClientFlow repository root, for example: PYTHONPATH="$PWD" .venv/bin/python scripts/audit_fix_odoo_reconciliation_links.py PYTHONPATH="$PWD" .venv/bin/python scripts/audit_fix_odoo_reconciliation_links.py --apply Exit codes: 0: audit/apply completed 2: configuration/source validation error 3: database operation error """ from __future__ import annotations import argparse import json import shutil import sys from collections import Counter from datetime import datetime, timezone from pathlib import Path from typing import Any, Iterable from sqlalchemy import text from app.db import engine SOURCE_FILE = Path("app/external_reconciliation_sync.py") BUGGY_FILTER = " AND o.status = 'open'\n" AUDIT_SQL = text( r""" WITH candidate_links AS ( SELECT ri.id AS reconciliation_item_id, ri.document_number, ri.external_id AS reconciliation_external_id, ri.customer_name, ri.customer_tax_id, ri.status AS reconciliation_status, ri.opportunity_id AS current_opportunity_id, ri.resolution_note, ri.resolved_at, ol.opportunity_id, ol.external_id AS link_external_id, ol.external_name, o.title AS opportunity_title, o.status AS opportunity_status, o.stage AS opportunity_stage, o.closed_at FROM reconciliation_items ri JOIN operation_links ol ON ol.system = 'odoo' AND ol.external_type = 'sale_order' AND ( ( NULLIF(BTRIM(COALESCE(ri.external_id, '')), '') IS NOT NULL AND ol.external_id = ri.external_id ) OR ( NULLIF(BTRIM(COALESCE(ri.document_number, '')), '') IS NOT NULL AND UPPER(BTRIM(COALESCE(ol.external_name, ''))) = UPPER(BTRIM(ri.document_number)) ) ) JOIN opportunities o ON o.id = ol.opportunity_id WHERE ri.source_system = 'odoo' AND ri.external_type = 'odoo_sale_order' AND ri.status IN ('open', 'needs_review', 'conflict') ), grouped AS ( SELECT reconciliation_item_id, document_number, reconciliation_external_id, customer_name, customer_tax_id, reconciliation_status, current_opportunity_id, resolution_note, resolved_at, COUNT(DISTINCT opportunity_id) AS opportunity_count, MIN(opportunity_id::text) AS single_opportunity_id FROM candidate_links GROUP BY reconciliation_item_id, document_number, reconciliation_external_id, customer_name, customer_tax_id, reconciliation_status, current_opportunity_id, resolution_note, resolved_at ) SELECT g.reconciliation_item_id::text, g.document_number, g.reconciliation_external_id, g.customer_name, g.customer_tax_id, g.reconciliation_status, g.current_opportunity_id::text, g.opportunity_count, CASE WHEN g.opportunity_count = 1 THEN g.single_opportunity_id ELSE NULL END AS opportunity_id, CASE WHEN g.opportunity_count = 1 THEN o.title ELSE NULL END AS opportunity_title, CASE WHEN g.opportunity_count = 1 THEN o.status ELSE NULL END AS opportunity_status, CASE WHEN g.opportunity_count = 1 THEN o.stage ELSE NULL END AS opportunity_stage, CASE WHEN g.opportunity_count = 1 THEN o.closed_at ELSE NULL END AS closed_at, g.resolution_note, g.resolved_at FROM grouped g LEFT JOIN opportunities o ON g.opportunity_count = 1 AND o.id = CAST(g.single_opportunity_id AS UUID) ORDER BY CASE WHEN g.opportunity_count = 1 THEN 0 ELSE 1 END, g.document_number """ ) UPDATE_ONE_SQL = text( r""" UPDATE reconciliation_items SET opportunity_id = CAST(:opportunity_id AS UUID), status = 'linked', resolution_note = CASE WHEN COALESCE(BTRIM(resolution_note), '') = '' THEN 'Resolvido por auditoria: ligação Odoo exata já existente em operation_links.' WHEN POSITION('Resolvido por auditoria: ligação Odoo exata' IN resolution_note) > 0 THEN resolution_note ELSE resolution_note || E'\nResolvido por auditoria: ligação Odoo exata já existente em operation_links.' END, resolved_at = COALESCE(resolved_at, now()), updated_at = now(), payload = COALESCE(payload, '{}'::jsonb) || jsonb_build_object( 'resolved_as_existing_operation_link', TRUE, 'resolved_by', 'audit_fix_odoo_reconciliation_links', 'resolved_at_audit', now() ) WHERE id = CAST(:reconciliation_item_id AS UUID) AND status IN ('open', 'needs_review', 'conflict') RETURNING id::text AS reconciliation_item_id, document_number, opportunity_id::text, status, resolved_at """ ) def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser( description="Audita e corrige candidatos Odoo já ligados a oportunidades." ) parser.add_argument( "--apply", action="store_true", help="Aplica a correção no código e na base de dados. Sem esta opção é dry-run.", ) parser.add_argument( "--data-only", action="store_true", help="Com --apply, corrige apenas a base de dados, sem alterar o código.", ) parser.add_argument( "--code-only", action="store_true", help="Com --apply, corrige apenas o código, sem alterar a base de dados.", ) parser.add_argument( "--exclude", action="append", default=[], metavar="DOCUMENTO", help="Não altera este documento. Pode repetir, por exemplo --exclude S00330.", ) parser.add_argument( "--json", action="store_true", help="Emite o relatório de auditoria em JSON.", ) parser.add_argument( "--source-file", default=str(SOURCE_FILE), help="Caminho do ficheiro external_reconciliation_sync.py.", ) return parser.parse_args() def audit_code(path: Path) -> dict[str, Any]: if not path.exists(): return { "path": str(path), "exists": False, "buggy_filter_count": None, "needs_fix": None, } source = path.read_text(encoding="utf-8") count = source.count(BUGGY_FILTER) return { "path": str(path), "exists": True, "buggy_filter_count": count, "needs_fix": count > 0, } def apply_code_fix(path: Path) -> dict[str, Any]: if not path.exists(): raise RuntimeError(f"Ficheiro não encontrado: {path}") source = path.read_text(encoding="utf-8") count = source.count(BUGGY_FILTER) if count == 0: return {"changed": False, "reason": "Filtro já não existe."} if count != 1: raise RuntimeError( f"Esperado exatamente 1 filtro {BUGGY_FILTER!r}; encontrados {count}." ) timestamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") backup = path.with_suffix(path.suffix + f".bak.{timestamp}") shutil.copy2(path, backup) updated = source.replace(BUGGY_FILTER, "", 1) path.write_text(updated, encoding="utf-8") return {"changed": True, "backup": str(backup)} def audit_database() -> list[dict[str, Any]]: with engine.connect() as conn: rows = conn.execute(AUDIT_SQL).mappings().all() return [dict(row) for row in rows] def apply_database_fix(excluded: Iterable[str]) -> list[dict[str, Any]]: excluded_set = {str(value).strip() for value in excluded if str(value).strip()} audit_rows = audit_database() eligible = [ row for row in audit_rows if int(row.get("opportunity_count") or 0) == 1 and row.get("opportunity_id") and row.get("document_number") not in excluded_set ] changed: list[dict[str, Any]] = [] with engine.begin() as conn: # Prevent two operators/jobs from applying the same repair concurrently. conn.execute(text("SELECT pg_advisory_xact_lock(hashtext(:lock_name))"), { "lock_name": "clientflow.audit_fix_odoo_reconciliation_links", }) for row in eligible: updated = conn.execute(UPDATE_ONE_SQL, { "reconciliation_item_id": row["reconciliation_item_id"], "opportunity_id": row["opportunity_id"], }).mappings().first() if updated: changed.append(dict(updated)) return changed def serialize(value: Any) -> Any: if isinstance(value, datetime): return value.isoformat() return value def print_audit(rows: list[dict[str, Any]], excluded: set[str]) -> None: counts = Counter() for row in rows: if int(row["opportunity_count"] or 0) == 1: counts[f"single:{row.get('opportunity_status') or 'unknown'}"] += 1 else: counts["ambiguous"] += 1 if row.get("document_number") in excluded: counts["excluded"] += 1 print("\nAUDITORIA — candidatos Odoo com ligação existente") print("=" * 72) print(f"Total encontrado: {len(rows)}") print(f"Ligações únicas / oportunidade fechada: {counts['single:closed']}") print(f"Ligações únicas / oportunidade aberta: {counts['single:open']}") other_single = sum( count for key, count in counts.items() if key.startswith("single:") and key not in {"single:closed", "single:open"} ) print(f"Ligações únicas / outros estados: {other_single}") print(f"Ambíguos (mais de uma oportunidade): {counts['ambiguous']}") print(f"Excluídos por opção: {counts['excluded']}") if not rows: print("Nenhuma inconsistência encontrada.") return print("\nDetalhe:") for row in rows: document = row.get("document_number") or "(sem número)" count = int(row.get("opportunity_count") or 0) excluded_marker = " [EXCLUÍDO]" if document in excluded else "" if count == 1: print( f"- {document}{excluded_marker}: {row.get('reconciliation_status')} -> " f"{row.get('opportunity_id')} | " f"{row.get('opportunity_status')}/{row.get('opportunity_stage')} | " f"{row.get('opportunity_title')}" ) else: print(f"- {document}{excluded_marker}: AMBÍGUO ({count} oportunidades)") def main() -> int: args = parse_args() if args.data_only and args.code_only: print("ERRO: --data-only e --code-only não podem ser usados em conjunto.", file=sys.stderr) return 2 source_path = Path(args.source_file) excluded = {value.strip() for value in args.exclude if value.strip()} code_report = audit_code(source_path) try: rows_before = audit_database() except Exception as exc: print(f"ERRO ao auditar a base de dados: {exc}", file=sys.stderr) return 3 if args.json: report = { "mode": "apply" if args.apply else "dry-run", "code": code_report, "excluded": sorted(excluded), "database": [{k: serialize(v) for k, v in row.items()} for row in rows_before], } print(json.dumps(report, ensure_ascii=False, indent=2)) else: print("AUDITORIA DO CÓDIGO") print("=" * 72) print(f"Ficheiro: {code_report['path']}") print(f"Existe: {code_report['exists']}") print(f"Filtro incorreto encontrado: {code_report['buggy_filter_count']}") print_audit(rows_before, excluded) if not args.apply: if not args.json: print("\nDRY-RUN: nenhuma alteração aplicada.") print("Use --apply para corrigir código e dados.") return 0 code_result: dict[str, Any] | None = None changed_rows: list[dict[str, Any]] = [] try: if not args.data_only: code_result = apply_code_fix(source_path) if not args.code_only: changed_rows = apply_database_fix(excluded) except Exception as exc: print(f"ERRO durante a aplicação: {exc}", file=sys.stderr) return 3 try: rows_after = audit_database() except Exception as exc: print(f"ERRO na auditoria posterior: {exc}", file=sys.stderr) return 3 if args.json: result = { "code_result": code_result, "database_rows_changed": [ {k: serialize(v) for k, v in row.items()} for row in changed_rows ], "remaining_inconsistencies": [ {k: serialize(v) for k, v in row.items()} for row in rows_after ], } print(json.dumps(result, ensure_ascii=False, indent=2)) else: print("\nAPLICAÇÃO") print("=" * 72) if code_result is not None: print(f"Código: {code_result}") print(f"Itens corrigidos na base de dados: {len(changed_rows)}") for row in changed_rows: print( f"- {row['document_number']} -> {row['opportunity_id']} " f"({row['status']})" ) print_audit(rows_after, excluded) return 0 if __name__ == "__main__": raise SystemExit(main())