313 lines
12 KiB
Python
Executable File
313 lines
12 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Backfill Jasmin document details into existing opportunities.
|
|
|
|
Use when an opportunity was created from a Jasmin reconciliation item before
|
|
v4926.6, so it still has no commercial_documents/opportunity_items/value.
|
|
|
|
Examples:
|
|
PYTHONPATH=. python scripts/backfill_jasmin_opportunity_details.py \
|
|
--opportunity-id a4e210f5-b870-48b1-882d-b55c71fbfd38
|
|
|
|
PYTHONPATH=. python scripts/backfill_jasmin_opportunity_details.py \
|
|
--document-number ORC.ORC2026.156
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import asyncio
|
|
import json
|
|
import sys
|
|
from decimal import Decimal, InvalidOperation
|
|
from typing import Any, Dict, Iterable, List, Optional
|
|
|
|
from sqlalchemy import text
|
|
|
|
from app.db import engine
|
|
from app.reconciliation_service import (
|
|
_apply_jasmin_documents_to_opportunity, # noqa: PLC2701 - deliberate operator backfill script
|
|
_jasmin_document_lines_from_item, # noqa: PLC2701
|
|
_jasmin_document_totals, # noqa: PLC2701
|
|
_payload_record, # noqa: PLC2701
|
|
)
|
|
|
|
|
|
def _as_text(value: Any) -> str:
|
|
return str(value or "").strip()
|
|
|
|
|
|
def _money_value(value: Any) -> Any:
|
|
if isinstance(value, dict):
|
|
for key in ("amount", "baseAmount", "reportingAmount", "value"):
|
|
if value.get(key) not in (None, ""):
|
|
return value.get(key)
|
|
return None
|
|
return value
|
|
|
|
|
|
def _decimal_or_none(value: Any) -> Optional[str]:
|
|
value = _money_value(value)
|
|
if value in (None, ""):
|
|
return None
|
|
try:
|
|
return str(Decimal(str(value).replace(",", ".")).quantize(Decimal("0.01")))
|
|
except (InvalidOperation, ValueError):
|
|
return None
|
|
|
|
|
|
def _json(value: Any) -> str:
|
|
return json.dumps(value, ensure_ascii=False, default=str)
|
|
|
|
|
|
def _ids_from_metadata(metadata: Any) -> List[str]:
|
|
if not isinstance(metadata, dict):
|
|
return []
|
|
ids: List[str] = []
|
|
for key in ("created_from_reconciliation_item_id", "reconciliation_item_id"):
|
|
value = metadata.get(key)
|
|
if value:
|
|
ids.append(str(value))
|
|
for key in ("item_ids", "reconciliation_item_ids"):
|
|
value = metadata.get(key)
|
|
if isinstance(value, list):
|
|
ids.extend(str(v) for v in value if v)
|
|
return list(dict.fromkeys(ids))
|
|
|
|
|
|
def _load_opportunity(opportunity_id: Optional[str], document_number: Optional[str]) -> Optional[Dict[str, Any]]:
|
|
with engine.begin() as conn:
|
|
if opportunity_id:
|
|
row = conn.execute(text("""
|
|
SELECT id::text, title, value_amount, product_interest, local_customer_id::text,
|
|
customer_name, customer_email, metadata
|
|
FROM opportunities
|
|
WHERE id = CAST(:id AS UUID)
|
|
LIMIT 1
|
|
"""), {"id": opportunity_id}).mappings().first()
|
|
return dict(row) if row else None
|
|
if document_number:
|
|
row = conn.execute(text("""
|
|
SELECT id::text, title, value_amount, product_interest, local_customer_id::text,
|
|
customer_name, customer_email, metadata
|
|
FROM opportunities
|
|
WHERE metadata->>'document_number' = :document_number
|
|
OR title ILIKE '%' || :document_number || '%'
|
|
ORDER BY updated_at DESC
|
|
LIMIT 1
|
|
"""), {"document_number": document_number}).mappings().first()
|
|
return dict(row) if row else None
|
|
return None
|
|
|
|
|
|
def _load_candidate_items(opportunity: Dict[str, Any]) -> List[Dict[str, Any]]:
|
|
metadata = opportunity.get("metadata") if isinstance(opportunity.get("metadata"), dict) else {}
|
|
ids = _ids_from_metadata(metadata)
|
|
external_id = _as_text(metadata.get("external_id"))
|
|
document_number = _as_text(metadata.get("document_number"))
|
|
opportunity_id = _as_text(opportunity.get("id"))
|
|
|
|
with engine.begin() as conn:
|
|
rows = conn.execute(text("""
|
|
SELECT id::text, source_system, external_type, external_id, title, description,
|
|
status, priority, suggested_action, confidence, opportunity_id::text,
|
|
customer_id::text, customer_name, customer_email, customer_tax_id,
|
|
document_number, document_date, amount, currency, payload,
|
|
resolution_note, created_at, updated_at, resolved_at
|
|
FROM reconciliation_items
|
|
WHERE source_system = 'jasmin'
|
|
AND (
|
|
opportunity_id = CAST(:opportunity_id AS UUID)
|
|
OR (CAST(:ids AS TEXT[]) IS NOT NULL AND id::text = ANY(CAST(:ids AS TEXT[])))
|
|
OR (CAST(:external_id AS TEXT) <> '' AND external_id = CAST(:external_id AS TEXT))
|
|
OR (CAST(:document_number AS TEXT) <> '' AND document_number = CAST(:document_number AS TEXT))
|
|
OR (CAST(:document_number AS TEXT) <> '' AND payload::text ILIKE '%' || CAST(:document_number AS TEXT) || '%')
|
|
)
|
|
ORDER BY updated_at DESC, created_at DESC
|
|
"""), {
|
|
"opportunity_id": opportunity_id,
|
|
"ids": ids or [],
|
|
"external_id": external_id,
|
|
"document_number": document_number,
|
|
}).mappings().all()
|
|
|
|
# De-duplicate while keeping recency order.
|
|
seen = set()
|
|
result = []
|
|
for row in rows:
|
|
item = dict(row)
|
|
item_id = item.get("id")
|
|
if item_id in seen:
|
|
continue
|
|
seen.add(item_id)
|
|
result.append(item)
|
|
return result
|
|
|
|
|
|
async def _fetch_jasmin_detail_async(item: Dict[str, Any]) -> Optional[Dict[str, Any]]:
|
|
external_type = _as_text(item.get("external_type"))
|
|
external_id = _as_text(item.get("external_id"))
|
|
if not external_id:
|
|
return None
|
|
from app.jasmin_client import JasminClient
|
|
|
|
client = JasminClient()
|
|
if external_type == "jasmin_quotation":
|
|
return await client.get_quotation(external_id)
|
|
if external_type == "jasmin_invoice":
|
|
return await client.get_invoice(external_id)
|
|
# Some tenants represent pro-forma as a quotation. Try quotation detail as a
|
|
# conservative fallback when the external id is present.
|
|
if external_type == "jasmin_proforma":
|
|
try:
|
|
return await client.get_quotation(external_id)
|
|
except Exception:
|
|
return None
|
|
return None
|
|
|
|
|
|
def _with_jasmin_detail(item: Dict[str, Any], *, fetch_detail: bool) -> Dict[str, Any]:
|
|
if not fetch_detail:
|
|
return item
|
|
existing_lines = _jasmin_document_lines_from_item(item)
|
|
if existing_lines:
|
|
return item
|
|
try:
|
|
detail = asyncio.run(_fetch_jasmin_detail_async(item))
|
|
except Exception as exc:
|
|
item = dict(item)
|
|
payload = item.get("payload") if isinstance(item.get("payload"), dict) else {}
|
|
item["payload"] = {
|
|
**payload,
|
|
"detail_fetch_error": f"{type(exc).__name__}: {exc}",
|
|
}
|
|
return item
|
|
if not isinstance(detail, dict):
|
|
return item
|
|
|
|
payload = item.get("payload") if isinstance(item.get("payload"), dict) else {}
|
|
enriched = dict(item)
|
|
enriched["payload"] = {
|
|
**payload,
|
|
"record": detail,
|
|
"detail_source": "jasmin_api",
|
|
"previous_record": payload.get("record"),
|
|
}
|
|
|
|
# Fill top-level fields if the detailed document exposes them only there.
|
|
record_number = detail.get("documentNumber") or detail.get("naturalKey") or detail.get("number")
|
|
if record_number and not enriched.get("document_number"):
|
|
enriched["document_number"] = record_number
|
|
total = (
|
|
detail.get("payableAmount")
|
|
or detail.get("totalAmount")
|
|
or detail.get("grossAmount")
|
|
or detail.get("amount")
|
|
)
|
|
if total and not enriched.get("amount"):
|
|
enriched["amount"] = _decimal_or_none(total) or total
|
|
return enriched
|
|
|
|
|
|
def _summary_for_items(items: Iterable[Dict[str, Any]]) -> List[Dict[str, Any]]:
|
|
result = []
|
|
for item in items:
|
|
totals = _jasmin_document_totals(item)
|
|
lines = _jasmin_document_lines_from_item(item)
|
|
result.append({
|
|
"id": item.get("id"),
|
|
"external_type": item.get("external_type"),
|
|
"external_id": item.get("external_id"),
|
|
"document_number": item.get("document_number"),
|
|
"amount": item.get("amount"),
|
|
"totals": totals,
|
|
"lines": len(lines),
|
|
"payload_keys": list((item.get("payload") or {}).keys()) if isinstance(item.get("payload"), dict) else [],
|
|
})
|
|
return result
|
|
|
|
|
|
def _post_import_summary(opportunity_id: str) -> Dict[str, Any]:
|
|
with engine.begin() as conn:
|
|
opportunity = conn.execute(text("""
|
|
SELECT id::text, title, value_amount, product_interest, metadata
|
|
FROM opportunities
|
|
WHERE id = CAST(:id AS UUID)
|
|
"""), {"id": opportunity_id}).mappings().first()
|
|
docs = conn.execute(text("""
|
|
SELECT id::text, document_kind, document_number, amount, total_amount, currency, document_date
|
|
FROM commercial_documents
|
|
WHERE opportunity_id = CAST(:id AS UUID)
|
|
ORDER BY created_at DESC
|
|
"""), {"id": opportunity_id}).mappings().all()
|
|
items = conn.execute(text("""
|
|
SELECT product_name, quantity, unit_price, total_price, jasmin_sales_item
|
|
FROM opportunity_items
|
|
WHERE opportunity_id = CAST(:id AS UUID)
|
|
ORDER BY created_at
|
|
"""), {"id": opportunity_id}).mappings().all()
|
|
lines = conn.execute(text("""
|
|
SELECT cdl.description, cdl.quantity, cdl.unit_price, cdl.total_amount, cdl.jasmin_sales_item
|
|
FROM commercial_document_lines cdl
|
|
JOIN commercial_documents cd ON cd.id = cdl.document_id
|
|
WHERE cd.opportunity_id = CAST(:id AS UUID)
|
|
ORDER BY cdl.line_index
|
|
"""), {"id": opportunity_id}).mappings().all()
|
|
return {
|
|
"opportunity": dict(opportunity or {}),
|
|
"documents": [dict(r) for r in docs],
|
|
"opportunity_items": [dict(r) for r in items],
|
|
"document_lines": [dict(r) for r in lines],
|
|
}
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--opportunity-id")
|
|
parser.add_argument("--document-number")
|
|
parser.add_argument("--no-fetch-jasmin-detail", action="store_true")
|
|
parser.add_argument("--dry-run", action="store_true")
|
|
parser.add_argument("--actor", default="operator_backfill")
|
|
args = parser.parse_args()
|
|
|
|
if not args.opportunity_id and not args.document_number:
|
|
parser.error("usa --opportunity-id ou --document-number")
|
|
|
|
opportunity = _load_opportunity(args.opportunity_id, args.document_number)
|
|
if not opportunity:
|
|
print(json.dumps({"ok": False, "error": "opportunity_not_found"}, ensure_ascii=False, indent=2))
|
|
return 2
|
|
|
|
items = _load_candidate_items(opportunity)
|
|
enriched_items = [
|
|
_with_jasmin_detail(item, fetch_detail=not args.no_fetch_jasmin_detail)
|
|
for item in items
|
|
]
|
|
print(json.dumps({
|
|
"opportunity_id": opportunity.get("id"),
|
|
"title": opportunity.get("title"),
|
|
"candidate_items": _summary_for_items(enriched_items),
|
|
"dry_run": args.dry_run,
|
|
}, ensure_ascii=False, indent=2, default=str))
|
|
|
|
if not enriched_items:
|
|
print(json.dumps({"ok": False, "error": "no_jasmin_reconciliation_items_found"}, ensure_ascii=False, indent=2))
|
|
return 3
|
|
|
|
if args.dry_run:
|
|
return 0
|
|
|
|
with engine.begin() as conn:
|
|
result = _apply_jasmin_documents_to_opportunity(
|
|
conn,
|
|
enriched_items,
|
|
str(opportunity["id"]),
|
|
actor=args.actor,
|
|
)
|
|
print(json.dumps({"ok": True, "import_result": result}, ensure_ascii=False, indent=2, default=str))
|
|
print(json.dumps(_post_import_summary(str(opportunity["id"])), ensure_ascii=False, indent=2, default=str))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|