#!/usr/bin/env python3 """Resolve stale Odoo reconciliation candidates already linked to opportunities. Dry-run is the default. The script deliberately does not alter opportunity stage, value, documents or operation_links. On --apply it only resolves stale reconciliation_items and then asks the central workflow engine to materialize a single supported human task (for example CREATE_SHIPMENT or PREPARE_ORDER). """ from __future__ import annotations import argparse import json import sys from pathlib import Path from typing import Any, Dict, List from sqlalchemy import text def _project_root() -> Path: script = Path(__file__).resolve() for candidate in [script.parent.parent, Path.cwd().resolve(), *script.parents]: if (candidate / "app" / "db.py").is_file(): return candidate return script.parent.parent PROJECT_ROOT = _project_root() if str(PROJECT_ROOT) not in sys.path: sys.path.insert(0, str(PROJECT_ROOT)) def _rows(*, limit: int, sales: List[str]) -> List[Dict[str, Any]]: from app.db import engine from app.reconciliation_service import ensure_reconciliation_schema ensure_reconciliation_schema() sale_filter = "" params: Dict[str, Any] = {"limit": max(1, min(int(limit), 5000))} if sales: sale_filter = "AND UPPER(COALESCE(ri.document_number, '')) = ANY(CAST(:sales AS TEXT[]))" params["sales"] = [str(value).strip().upper() for value in sales if str(value).strip()] with engine.begin() as conn: rows = conn.execute(text(f""" WITH candidates AS ( SELECT ri.id::text AS item_id, ri.external_id, ri.document_number, ri.customer_name, ri.amount, ri.status AS item_status, ri.suggested_action, ri.opportunity_id::text AS current_item_opportunity_id FROM reconciliation_items ri WHERE ri.source_system = 'odoo' AND ri.external_type = 'odoo_sale_order' AND ri.status IN ('open', 'needs_review', 'conflict') {sale_filter} ORDER BY ri.document_date NULLS LAST, ri.created_at LIMIT :limit ) SELECT c.*, COUNT(DISTINCT o.id)::int AS exact_link_count, MIN(o.id::text) AS linked_opportunity_id, MIN(o.title) AS opportunity_title, MIN(o.stage) AS opportunity_stage, MIN(o.status) AS opportunity_status, COUNT(t.id) FILTER (WHERE t.status = 'pending')::int AS pending_tasks FROM candidates c LEFT JOIN operation_links ol ON ol.system = 'odoo' AND ol.external_type = 'sale_order' AND ( (NULLIF(c.external_id, '') IS NOT NULL AND ol.external_id = c.external_id) OR (NULLIF(c.document_number, '') IS NOT NULL AND UPPER(COALESCE(ol.external_name, '')) = UPPER(c.document_number)) ) LEFT JOIN opportunities o ON o.id = ol.opportunity_id AND o.status = 'open' LEFT JOIN tasks t ON t.opportunity_id = o.id AND t.status = 'pending' GROUP BY c.item_id, c.external_id, c.document_number, c.customer_name, c.amount, c.item_status, c.suggested_action, c.current_item_opportunity_id ORDER BY c.document_number """), params).mappings().all() return [dict(row) for row in rows] def _resolve_item(item: Dict[str, Any]) -> bool: from app.db import engine from app.reconciliation_service import ensure_reconciliation_schema ensure_reconciliation_schema() with engine.begin() as conn: result = conn.execute(text(""" UPDATE reconciliation_items ri SET opportunity_id = CAST(:opportunity_id AS UUID), status = 'linked', resolution_note = 'Resolvido: venda Odoo já estava ligada à oportunidade; fase preservada', resolved_at = now(), updated_at = now(), payload = COALESCE(ri.payload, '{}'::jsonb) || CAST(:payload AS JSONB) WHERE ri.id = CAST(:item_id AS UUID) AND ri.status IN ('open', 'needs_review', 'conflict') """), { "item_id": item["item_id"], "opportunity_id": item["linked_opportunity_id"], "payload": json.dumps({ "resolved_as_existing_operation_link": True, "resolved_by": "migration_v4928_1_5_129", "preserve_opportunity_stage": True, "previous_suggested_action": item.get("suggested_action"), "sale_order": item.get("document_number") or item.get("external_id"), }, ensure_ascii=False), }) if not result.rowcount: return False try: with conn.begin_nested(): conn.execute(text(""" INSERT INTO reconciliation_decisions ( decision_type, status, item_ids, opportunity_id, note, actor, payload ) VALUES ( 'resolve_existing_operation_link', 'linked', CAST(:item_ids AS TEXT[]), CAST(:opportunity_id AS UUID), :note, 'migration_v4928_1_5_129', CAST(:payload AS JSONB) ) """), { "item_ids": [item["item_id"]], "opportunity_id": item["linked_opportunity_id"], "note": "Venda Odoo já ligada; candidato obsoleto resolvido sem regressão de fase.", "payload": json.dumps({ "sale_order": item.get("document_number") or item.get("external_id"), "preserve_stage": True, }, ensure_ascii=False), }) except Exception: # The core update remains valid if an older installation has not # yet created the optional decisions table. pass return True def _materialize(opportunity_id: str) -> Dict[str, Any]: from app.opportunity_action_task_materializer import ensure_pending_task_for_next_action from app.opportunity_next_action_service import get_opportunity_next_action next_action = get_opportunity_next_action(opportunity_id) result = ensure_pending_task_for_next_action( opportunity_id, next_action, source="reconciliation_cleanup_v129", actor="migration_v4928_1_5_129", ) return {"next_action": next_action, "materialization": result} def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("--apply", action="store_true", help="Apply the reviewed corrections.") parser.add_argument("--limit", type=int, default=500) parser.add_argument("--sale", action="append", default=[], help="Restrict to a sale number, e.g. S00325.") args = parser.parse_args() rows = _rows(limit=args.limit, sales=args.sale) safe = [row for row in rows if int(row.get("exact_link_count") or 0) == 1 and row.get("linked_opportunity_id")] conflicts = [row for row in rows if int(row.get("exact_link_count") or 0) > 1] unmatched = [row for row in rows if int(row.get("exact_link_count") or 0) == 0] print(f"Open Odoo reconciliation candidates: {len(rows)}") print(f"Safe existing links: {len(safe)}") print(f"Conflicting links: {len(conflicts)}") print(f"Without exact link: {len(unmatched)}") for row in rows: status = "SAFE" if row in safe else ("CONFLICT" if row in conflicts else "MANUAL") print( f"{status:8} | {str(row.get('document_number') or row.get('external_id') or ''):10} | " f"links={int(row.get('exact_link_count') or 0)} | " f"stage={row.get('opportunity_stage') or '-':18} | " f"{row.get('customer_name') or '-'}" ) if not args.apply: print("Dry-run only. Use --apply after reviewing the safe candidates.") return 0 resolved = 0 task_results: List[Dict[str, Any]] = [] for row in safe: if _resolve_item(row): resolved += 1 task_result = _materialize(str(row["linked_opportunity_id"])) task_results.append({ "sale": row.get("document_number") or row.get("external_id"), "opportunity_id": row.get("linked_opportunity_id"), **task_result, }) print(f"Candidates resolved: {resolved}") for result in task_results: next_action = result.get("next_action") or {} materialization = result.get("materialization") or {} print( f"{result.get('sale')} | next={next_action.get('action_code')} | " f"task={materialization.get('action_code') or '-'} | " f"created={bool(materialization.get('created'))} | " f"reason={materialization.get('reason') or '-'}" ) return 0 if __name__ == "__main__": raise SystemExit(main())