138 lines
6.6 KiB
Python
138 lines
6.6 KiB
Python
#!/usr/bin/env python3
|
|
"""Classify legacy document links into Document Reconciliation v2.
|
|
|
|
Dry-run is the default. This script changes ClientFlow only and never calls
|
|
Jasmin. Ambiguous evidence is explicitly classified REVIEW_REQUIRED.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import csv
|
|
import json
|
|
import os
|
|
import sys
|
|
from pathlib import Path
|
|
from typing import Any, Dict
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(ROOT))
|
|
|
|
from app.document_reconciliation_backfill import build_backfill_result
|
|
|
|
|
|
def load_documents(
|
|
opportunity_id: str | None = None,
|
|
resume_from: str | None = None,
|
|
batch_size: int = 500,
|
|
) -> tuple[list[Dict[str, Any]], int, int]:
|
|
from sqlalchemy import text
|
|
|
|
from app.db import engine
|
|
|
|
where, params = ["d.opportunity_id IS NOT NULL"], {"limit": batch_size}
|
|
if opportunity_id:
|
|
where.append("d.opportunity_id=CAST(:opportunity_id AS UUID)"); params["opportunity_id"] = opportunity_id
|
|
if resume_from:
|
|
try:
|
|
resume_opportunity_id, resume_document_kind = resume_from.split("|", 1)
|
|
except ValueError as exc:
|
|
raise ValueError("--resume-from must be OPPORTUNITY_ID|DOCUMENT_KIND") from exc
|
|
where.append("(d.opportunity_id::text, d.document_kind) > (:resume_opportunity_id, :resume_document_kind)")
|
|
params.update(resume_opportunity_id=resume_opportunity_id, resume_document_kind=resume_document_kind)
|
|
with engine.begin() as conn:
|
|
rows = [dict(r) for r in conn.execute(text(f"""
|
|
WITH selected_groups AS (
|
|
SELECT d.opportunity_id, d.document_kind
|
|
FROM commercial_documents d WHERE {' AND '.join(where)}
|
|
GROUP BY d.opportunity_id, d.document_kind
|
|
ORDER BY d.opportunity_id, d.document_kind
|
|
LIMIT :limit
|
|
)
|
|
SELECT d.id::text, d.opportunity_id::text, d.document_kind, d.external_id,
|
|
d.document_number, d.status, d.role, d.is_primary, d.is_active,
|
|
d.total_amount, d.amount, d.payload
|
|
FROM selected_groups g
|
|
JOIN commercial_documents d ON d.opportunity_id=g.opportunity_id
|
|
AND d.document_kind=g.document_kind
|
|
ORDER BY d.opportunity_id, d.document_kind, d.created_at, d.id
|
|
"""), params).mappings().all()]
|
|
orphan_lines = int(conn.execute(text("""SELECT count(*) FROM commercial_document_lines dl
|
|
LEFT JOIN commercial_documents d ON d.id=COALESCE(dl.commercial_document_id,dl.document_id)
|
|
WHERE d.id IS NULL""")).scalar() or 0)
|
|
multi = int(conn.execute(text("""SELECT count(*) FROM (SELECT COALESCE(external_id,document_number), count(DISTINCT opportunity_id)
|
|
FROM commercial_documents WHERE opportunity_id IS NOT NULL AND COALESCE(external_id,document_number) IS NOT NULL
|
|
GROUP BY 1 HAVING count(DISTINCT opportunity_id)>1) q""")).scalar() or 0)
|
|
return rows, orphan_lines, multi
|
|
|
|
|
|
def scan(opportunity_id: str | None = None, resume_from: str | None = None, batch_size: int = 500) -> Dict[str, Any]:
|
|
rows, orphan_lines, multi = load_documents(opportunity_id, resume_from, batch_size)
|
|
result = build_backfill_result(
|
|
rows,
|
|
lines_without_document=orphan_lines,
|
|
documents_in_multiple_opportunities=multi,
|
|
)
|
|
groups = sorted({(str(row["opportunity_id"]), str(row["document_kind"])) for row in rows})
|
|
result["summary"]["batch_groups"] = len(groups)
|
|
result["summary"]["checkpoint"] = "|".join(groups[-1]) if groups else resume_from
|
|
return result
|
|
|
|
|
|
def actions_for_only_unambiguous(actions: list[Dict[str, Any]]) -> list[Dict[str, Any]]:
|
|
"""Keep ambiguous documents represented; never partially migrate a group.
|
|
|
|
Option B of the rollout contract is used: REVIEW_REQUIRED links are written
|
|
alongside unambiguous links so every selected group becomes v2-complete.
|
|
"""
|
|
return list(actions)
|
|
|
|
|
|
def apply_actions(actions: list[Dict[str, Any]]) -> int:
|
|
from app.db import engine
|
|
from app.document_reconciliation_service import set_document_relationship
|
|
groups: Dict[tuple[str, str], list[Dict[str, Any]]] = {}
|
|
for action in actions:
|
|
groups.setdefault((str(action["opportunity_id"]), str(action["document_kind"])), []).append(action)
|
|
# A group is the rollout authority boundary. Links, events and legacy dual
|
|
# writes commit together; any failure rolls the entire group back.
|
|
for group in groups.values():
|
|
with engine.begin() as conn:
|
|
for action in group:
|
|
set_document_relationship(action["opportunity_id"], action["id"], action["relationship"],
|
|
actor="document_reconciliation_v2_backfill", reason=action["decision_reason"], is_manual=False,
|
|
source="backfill_v2", idempotency_key=f"backfill-v2:{action['opportunity_id']}:{action['id']}",
|
|
event_type="BACKFILLED", metadata={"legacy_role": action.get("role"), "legacy_is_primary": action.get("is_primary")},
|
|
_conn=conn)
|
|
return len(actions)
|
|
|
|
|
|
def write_reports(result: Dict[str, Any], json_path: str | None, csv_path: str | None) -> None:
|
|
if json_path: Path(json_path).write_text(json.dumps(result, ensure_ascii=False, indent=2, default=str) + "\n")
|
|
if csv_path:
|
|
fields = ["opportunity_id", "id", "document_kind", "document_number", "status", "relationship", "decision_reason"]
|
|
with Path(csv_path).open("w", newline="", encoding="utf-8") as handle:
|
|
writer = csv.DictWriter(handle, fieldnames=fields, extrasaction="ignore"); writer.writeheader(); writer.writerows(result["actions"])
|
|
|
|
|
|
def main() -> int:
|
|
os.chdir(ROOT)
|
|
parser = argparse.ArgumentParser()
|
|
mode = parser.add_mutually_exclusive_group(); mode.add_argument("--dry-run", action="store_true"); mode.add_argument("--apply", action="store_true")
|
|
parser.add_argument("--only-unambiguous", action="store_true")
|
|
parser.add_argument("--opportunity-id"); parser.add_argument("--batch-size", type=int, default=500)
|
|
parser.add_argument("--resume-from"); parser.add_argument("--output-json"); parser.add_argument("--output-csv")
|
|
args = parser.parse_args()
|
|
result = scan(args.opportunity_id, args.resume_from, args.batch_size)
|
|
actions = result["actions"]
|
|
if args.only_unambiguous: actions = actions_for_only_unambiguous(actions)
|
|
applied = 0
|
|
if args.apply:
|
|
applied = apply_actions(actions)
|
|
result["summary"]["mode"] = "apply" if args.apply else "dry-run"; result["summary"]["applied"] = applied
|
|
write_reports(result, args.output_json, args.output_csv)
|
|
print(json.dumps(result["summary"], ensure_ascii=False, indent=2, default=str))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__": raise SystemExit(main())
|