754 lines
34 KiB
Python
754 lines
34 KiB
Python
"""Operator helpers to reimport Jasmin document details into opportunities.
|
|
|
|
Used by both CLI backfills and the opportunity detail UI. The function is
|
|
idempotent because the underlying reconciliation import upserts commercial
|
|
documents, document lines and opportunity items by Jasmin identifiers.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from decimal import Decimal, InvalidOperation
|
|
from typing import Any, Dict, Iterable, List, Optional
|
|
import json
|
|
|
|
from sqlalchemy import text
|
|
|
|
from app.db import engine
|
|
from app.reconciliation_service import (
|
|
_apply_jasmin_documents_to_opportunity, # noqa: PLC2701 - deliberate operator maintenance helper
|
|
_jasmin_document_lines_from_item, # noqa: PLC2701
|
|
_jasmin_document_totals, # noqa: PLC2701
|
|
)
|
|
|
|
|
|
def _as_text(value: Any) -> str:
|
|
return str(value or "").strip()
|
|
|
|
|
|
def _json(value: Any) -> str:
|
|
return json.dumps(value, ensure_ascii=False, default=str)
|
|
|
|
|
|
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 _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 _record_from_jasmin_item(item: Dict[str, Any]) -> Dict[str, Any]:
|
|
payload = item.get("payload") if isinstance(item.get("payload"), dict) else {}
|
|
record = payload.get("record") if isinstance(payload.get("record"), dict) else {}
|
|
return record
|
|
|
|
|
|
def _field_text(record: Dict[str, Any], *keys: str) -> str:
|
|
for key in keys:
|
|
value = record.get(key)
|
|
if value not in (None, ""):
|
|
return _as_text(value)
|
|
return ""
|
|
|
|
|
|
def _field_bool(record: Dict[str, Any], *keys: str) -> bool:
|
|
for key in keys:
|
|
value = record.get(key)
|
|
if isinstance(value, bool):
|
|
return value
|
|
if isinstance(value, str) and value.strip().lower() in {"true", "1", "yes", "sim"}:
|
|
return True
|
|
if isinstance(value, (int, float)) and value == 1:
|
|
return True
|
|
return False
|
|
|
|
|
|
def jasmin_document_lifecycle(item: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""Classify a Jasmin reconciliation item before linking/importing it.
|
|
|
|
The Jasmin API stores lifecycle information inside payload.record, not in
|
|
reconciliation_items.status (which is ClientFlow resolution status such as
|
|
linked/rejected). The UI must not import an old/closed quotation when a newer
|
|
open quotation exists for the same fiscal customer.
|
|
"""
|
|
record = _record_from_jasmin_item(item)
|
|
status_desc = _field_text(record, "documentStatusDescription", "statusDescription", "documentLineStatusDescription")
|
|
status_code = _field_text(record, "documentStatus", "status")
|
|
line_status = " ".join(
|
|
_field_text(line, "documentLineStatusDescription", "lineStatusDescription", "statusDescription")
|
|
for line in (record.get("documentLines") or [])
|
|
if isinstance(line, dict)
|
|
).strip()
|
|
text_blob = " ".join([status_desc, status_code, line_status]).casefold()
|
|
external_type = _as_text(item.get("external_type"))
|
|
is_invoice = external_type == "jasmin_invoice"
|
|
is_deleted = _field_bool(record, "isDeleted", "deleted")
|
|
is_draft = _field_bool(record, "isDraft")
|
|
is_completed = _field_bool(record, "statusWasCompleted", "wasCompleted", "completed")
|
|
# Quotations/proformas must be open because importing an old converted quote
|
|
# can replace the active proposal. Invoices are different: a valid issued
|
|
# invoice may be Open, Completed or Paid and still belongs as a downstream
|
|
# document of the same process. Only destructive/cancelled invoice states
|
|
# should block operator association.
|
|
closed_terms = {
|
|
"closed", "close", "fechado", "fechada", "completed", "complete", "concluido", "concluído",
|
|
"concluida", "concluída", "converted", "convertido", "convertida", "cancelled", "canceled",
|
|
"cancelado", "cancelada", "anulado", "anulada", "void", "deleted", "apagado", "apagada",
|
|
}
|
|
invoice_block_terms = {
|
|
"cancelled", "canceled", "cancelado", "cancelada", "anulado", "anulada",
|
|
"void", "deleted", "apagado", "apagada",
|
|
}
|
|
open_terms = {"open", "aberto", "aberta", "active", "ativo", "ativa"}
|
|
has_closed_term = any(term in text_blob for term in closed_terms)
|
|
has_invoice_block_term = any(term in text_blob for term in invoice_block_terms)
|
|
has_open_term = any(term in text_blob for term in open_terms)
|
|
# In Jasmin quotations observed in production, documentStatus=1 +
|
|
# documentStatusDescription=Open means valid/open. Keep code 1 as a weak
|
|
# positive only when no closed term exists.
|
|
code_open = status_code.strip() == "1"
|
|
invalid_reasons: List[str] = []
|
|
if is_deleted:
|
|
invalid_reasons.append("deleted")
|
|
if is_draft:
|
|
invalid_reasons.append("draft")
|
|
if is_invoice:
|
|
if has_invoice_block_term:
|
|
invalid_reasons.append("cancelled_or_void_invoice")
|
|
else:
|
|
if is_completed:
|
|
invalid_reasons.append("completed")
|
|
if has_closed_term:
|
|
invalid_reasons.append("closed_status")
|
|
is_open = (has_open_term or code_open) and not invalid_reasons
|
|
is_valid = (
|
|
(external_type in {"jasmin_quotation", "jasmin_proforma"} and is_open)
|
|
or (is_invoice and not invalid_reasons)
|
|
)
|
|
if not is_valid and not invalid_reasons:
|
|
invalid_reasons.append("not_open_or_supported_jasmin_document")
|
|
label = status_desc or ("Open" if is_open else "Unknown")
|
|
return {
|
|
"status_label": label,
|
|
"status_code": status_code,
|
|
"line_status_label": line_status,
|
|
"is_open": bool(is_open),
|
|
"is_valid_candidate": bool(is_valid),
|
|
"invalid_reason": ",".join(dict.fromkeys(invalid_reasons)),
|
|
}
|
|
|
|
|
|
def _jasmin_candidate_sort_tuple(item: Dict[str, Any]) -> tuple:
|
|
lifecycle = item.get("lifecycle") if isinstance(item.get("lifecycle"), dict) else jasmin_document_lifecycle(item)
|
|
valid_rank = 1 if lifecycle.get("is_valid_candidate") else 0
|
|
external_type = _as_text(item.get("external_type"))
|
|
type_rank = {"jasmin_quotation": 3, "jasmin_proforma": 2, "jasmin_invoice": 1}.get(external_type, 0)
|
|
match_score = int(item.get("match_score") or 0)
|
|
record = _record_from_jasmin_item(item)
|
|
series_number = 0
|
|
try:
|
|
series_number = int(record.get("seriesNumber") or 0)
|
|
except Exception:
|
|
series_number = 0
|
|
doc_date = _as_text(item.get("document_date") or record.get("documentDate") or record.get("postingDate") or item.get("updated_at"))
|
|
return (valid_rank, type_rank, doc_date, series_number, match_score, _as_text(item.get("updated_at")))
|
|
|
|
|
|
def _jasmin_item_recency_tuple(item: Dict[str, Any]) -> tuple:
|
|
"""Comparable recency key for Jasmin documents.
|
|
|
|
Jasmin quotations are naturally ordered by document date and series number.
|
|
We use this to prevent an old still-open quotation from being offered as a
|
|
replacement when the opportunity already has a newer Jasmin document.
|
|
"""
|
|
record = _record_from_jasmin_item(item)
|
|
series_number = 0
|
|
try:
|
|
series_number = int(record.get("seriesNumber") or item.get("series_number") or 0)
|
|
except Exception:
|
|
series_number = 0
|
|
doc_date = _as_text(
|
|
item.get("document_date")
|
|
or record.get("documentDate")
|
|
or record.get("postingDate")
|
|
or item.get("created_at")
|
|
or item.get("updated_at")
|
|
)[:10]
|
|
return (doc_date, series_number, _as_text(item.get("document_number") or item.get("external_id")))
|
|
|
|
|
|
def _commercial_doc_as_jasmin_item(doc: Dict[str, Any]) -> Dict[str, Any]:
|
|
payload = doc.get("payload") if isinstance(doc.get("payload"), dict) else {}
|
|
record = payload.get("record") if isinstance(payload.get("record"), dict) else payload
|
|
return {
|
|
"external_type": f"jasmin_{doc.get('document_kind') or 'document'}",
|
|
"external_id": doc.get("external_id"),
|
|
"document_number": doc.get("document_number"),
|
|
"document_date": doc.get("document_date"),
|
|
"created_at": doc.get("created_at"),
|
|
"updated_at": doc.get("updated_at"),
|
|
"payload": {"record": record} if isinstance(record, dict) else {},
|
|
}
|
|
|
|
|
|
def load_jasmin_backfill_opportunity(opportunity_id: Optional[str] = None, document_number: Optional[str] = None) -> 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_jasmin_candidate_items_for_opportunity(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()
|
|
|
|
seen = set()
|
|
result: List[Dict[str, Any]] = []
|
|
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)
|
|
if external_type == "jasmin_proforma":
|
|
try:
|
|
return await client.get_quotation(external_id)
|
|
except Exception:
|
|
return None
|
|
return None
|
|
|
|
|
|
async def enrich_jasmin_item_with_detail(item: Dict[str, Any], *, fetch_detail: bool = True) -> Dict[str, Any]:
|
|
if not fetch_detail:
|
|
return item
|
|
existing_lines = _jasmin_document_lines_from_item(item)
|
|
if existing_lines:
|
|
return item
|
|
try:
|
|
detail = await _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"),
|
|
}
|
|
|
|
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_jasmin_items(items: Iterable[Dict[str, Any]]) -> List[Dict[str, Any]]:
|
|
result = []
|
|
for item in items:
|
|
lifecycle = jasmin_document_lifecycle(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": _jasmin_document_totals(item),
|
|
"lines": len(_jasmin_document_lines_from_item(item)),
|
|
"payload_keys": list((item.get("payload") or {}).keys()) if isinstance(item.get("payload"), dict) else [],
|
|
"lifecycle": lifecycle,
|
|
})
|
|
return result
|
|
|
|
|
|
def jasmin_backfill_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],
|
|
}
|
|
|
|
|
|
async def backfill_jasmin_opportunity_details_async(
|
|
*,
|
|
opportunity_id: Optional[str] = None,
|
|
document_number: Optional[str] = None,
|
|
fetch_detail: bool = True,
|
|
actor: str = "operator_ui_reimport",
|
|
dry_run: bool = False,
|
|
) -> Dict[str, Any]:
|
|
opportunity = load_jasmin_backfill_opportunity(opportunity_id=opportunity_id, document_number=document_number)
|
|
if not opportunity:
|
|
return {"ok": False, "error": "opportunity_not_found"}
|
|
|
|
items = load_jasmin_candidate_items_for_opportunity(opportunity)
|
|
enriched_items = [await enrich_jasmin_item_with_detail(item, fetch_detail=fetch_detail) for item in items]
|
|
result: Dict[str, Any] = {
|
|
"ok": bool(enriched_items),
|
|
"opportunity_id": opportunity.get("id"),
|
|
"title": opportunity.get("title"),
|
|
"candidate_items": summary_for_jasmin_items(enriched_items),
|
|
"dry_run": dry_run,
|
|
}
|
|
if not enriched_items:
|
|
result["error"] = "no_jasmin_reconciliation_items_found"
|
|
return result
|
|
if dry_run:
|
|
return result
|
|
|
|
with engine.begin() as conn:
|
|
import_result = _apply_jasmin_documents_to_opportunity(
|
|
conn,
|
|
enriched_items,
|
|
str(opportunity["id"]),
|
|
actor=actor,
|
|
)
|
|
result["import_result"] = import_result
|
|
result["summary"] = jasmin_backfill_post_import_summary(str(opportunity["id"]))
|
|
return result
|
|
|
|
|
|
def _email_domain(email: Any) -> str:
|
|
text_value = _as_text(email).lower().strip().strip(';')
|
|
if "@" not in text_value:
|
|
return ""
|
|
return text_value.rsplit("@", 1)[-1].strip()
|
|
|
|
|
|
def find_jasmin_document_candidates_for_opportunity(opportunity_id: str, *, limit: int = 5) -> List[Dict[str, Any]]:
|
|
"""Find Jasmin documents that probably belong to an opportunity but are not imported yet.
|
|
|
|
This is deliberately conservative and prioritizes fiscal identity (NIF/customer_id)
|
|
over weak name/domain matching. It is used in the opportunity detail UI before
|
|
offering to create a new Jasmin quotation, so operators can associate an existing
|
|
quotation instead of duplicating it.
|
|
"""
|
|
with engine.begin() as conn:
|
|
opp = conn.execute(text("""
|
|
SELECT
|
|
o.id::text,
|
|
o.customer_name,
|
|
o.customer_email,
|
|
o.local_customer_id::text,
|
|
c.name AS fiscal_customer_name,
|
|
c.tax_id AS fiscal_customer_tax_id,
|
|
c.email AS fiscal_customer_email
|
|
FROM opportunities o
|
|
LEFT JOIN customers c ON c.id = o.local_customer_id
|
|
WHERE o.id = CAST(:id AS UUID)
|
|
LIMIT 1
|
|
"""), {"id": opportunity_id}).mappings().first()
|
|
if not opp:
|
|
return []
|
|
|
|
fiscal_tax_id = _as_text(opp.get("fiscal_customer_tax_id"))
|
|
fiscal_customer_id = _as_text(opp.get("local_customer_id"))
|
|
fiscal_name = _as_text(opp.get("fiscal_customer_name"))
|
|
fiscal_email = _as_text(opp.get("fiscal_customer_email") or opp.get("customer_email")).lower()
|
|
domain = _email_domain(fiscal_email)
|
|
|
|
current_docs = conn.execute(text("""
|
|
SELECT
|
|
id::text, document_number, external_id, document_kind, document_date,
|
|
total_amount, amount, payload, created_at, updated_at
|
|
FROM commercial_documents
|
|
WHERE opportunity_id = CAST(:opportunity_id AS UUID)
|
|
AND system = 'jasmin'
|
|
AND document_kind IN ('quotation', 'proforma')
|
|
ORDER BY document_date DESC NULLS LAST, created_at DESC
|
|
"""), {"opportunity_id": opportunity_id}).mappings().all()
|
|
|
|
rows = conn.execute(text("""
|
|
SELECT
|
|
ri.id::text,
|
|
ri.source_system,
|
|
ri.external_type,
|
|
ri.external_id,
|
|
ri.title,
|
|
ri.status,
|
|
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.created_at,
|
|
ri.updated_at,
|
|
CASE
|
|
WHEN CAST(:fiscal_customer_id AS TEXT) <> '' AND ri.customer_id::text = CAST(:fiscal_customer_id AS TEXT) THEN 100
|
|
WHEN CAST(:fiscal_tax_id AS TEXT) <> '' AND ri.customer_tax_id = CAST(:fiscal_tax_id AS TEXT) THEN 98
|
|
WHEN CAST(:fiscal_tax_id AS TEXT) <> '' AND ri.payload::text ILIKE '%' || CAST(:fiscal_tax_id AS TEXT) || '%' THEN 96
|
|
WHEN CAST(:fiscal_email AS TEXT) <> '' AND LOWER(COALESCE(ri.customer_email, '')) = CAST(:fiscal_email AS TEXT) THEN 90
|
|
WHEN CAST(:domain AS TEXT) <> '' AND ri.payload::text ILIKE '%' || CAST(:domain AS TEXT) || '%' THEN 78
|
|
WHEN CAST(:fiscal_name AS TEXT) <> '' AND LOWER(COALESCE(ri.customer_name, '')) = LOWER(CAST(:fiscal_name AS TEXT)) THEN 75
|
|
ELSE 0
|
|
END AS match_score,
|
|
CASE
|
|
WHEN CAST(:fiscal_customer_id AS TEXT) <> '' AND ri.customer_id::text = CAST(:fiscal_customer_id AS TEXT) THEN 'customer_id'
|
|
WHEN CAST(:fiscal_tax_id AS TEXT) <> '' AND ri.customer_tax_id = CAST(:fiscal_tax_id AS TEXT) THEN 'nif'
|
|
WHEN CAST(:fiscal_tax_id AS TEXT) <> '' AND ri.payload::text ILIKE '%' || CAST(:fiscal_tax_id AS TEXT) || '%' THEN 'payload_nif'
|
|
WHEN CAST(:fiscal_email AS TEXT) <> '' AND LOWER(COALESCE(ri.customer_email, '')) = CAST(:fiscal_email AS TEXT) THEN 'email'
|
|
WHEN CAST(:domain AS TEXT) <> '' AND ri.payload::text ILIKE '%' || CAST(:domain AS TEXT) || '%' THEN 'domain_payload'
|
|
WHEN CAST(:fiscal_name AS TEXT) <> '' AND LOWER(COALESCE(ri.customer_name, '')) = LOWER(CAST(:fiscal_name AS TEXT)) THEN 'exact_name'
|
|
ELSE 'none'
|
|
END AS match_reason
|
|
FROM reconciliation_items ri
|
|
WHERE ri.source_system = 'jasmin'
|
|
AND ri.external_type IN ('jasmin_quotation', 'jasmin_proforma', 'jasmin_invoice')
|
|
AND (ri.opportunity_id IS NULL OR ri.opportunity_id = CAST(:opportunity_id AS UUID))
|
|
AND NOT EXISTS (
|
|
SELECT 1 FROM commercial_documents cd
|
|
WHERE cd.opportunity_id = CAST(:opportunity_id AS UUID)
|
|
AND (
|
|
(ri.document_number IS NOT NULL AND cd.document_number = ri.document_number)
|
|
OR (ri.external_id IS NOT NULL AND cd.external_id = ri.external_id)
|
|
)
|
|
)
|
|
AND (
|
|
(CAST(:fiscal_customer_id AS TEXT) <> '' AND ri.customer_id::text = CAST(:fiscal_customer_id AS TEXT))
|
|
OR (CAST(:fiscal_tax_id AS TEXT) <> '' AND ri.customer_tax_id = CAST(:fiscal_tax_id AS TEXT))
|
|
OR (CAST(:fiscal_tax_id AS TEXT) <> '' AND ri.payload::text ILIKE '%' || CAST(:fiscal_tax_id AS TEXT) || '%')
|
|
OR (CAST(:fiscal_email AS TEXT) <> '' AND LOWER(COALESCE(ri.customer_email, '')) = CAST(:fiscal_email AS TEXT))
|
|
OR (CAST(:domain AS TEXT) <> '' AND ri.payload::text ILIKE '%' || CAST(:domain AS TEXT) || '%')
|
|
OR (CAST(:fiscal_name AS TEXT) <> '' AND LOWER(COALESCE(ri.customer_name, '')) = LOWER(CAST(:fiscal_name AS TEXT)))
|
|
)
|
|
ORDER BY match_score DESC, ri.document_date DESC NULLS LAST, ri.updated_at DESC
|
|
LIMIT :limit
|
|
"""), {
|
|
"opportunity_id": opportunity_id,
|
|
"fiscal_customer_id": fiscal_customer_id,
|
|
"fiscal_tax_id": fiscal_tax_id,
|
|
"fiscal_email": fiscal_email,
|
|
"domain": domain,
|
|
"fiscal_name": fiscal_name,
|
|
"limit": int(limit),
|
|
}).mappings().all()
|
|
|
|
current_doc_items = [_commercial_doc_as_jasmin_item(dict(row)) for row in current_docs]
|
|
current_latest_key = max((_jasmin_item_recency_tuple(item) for item in current_doc_items), default=None)
|
|
current_doc_numbers = {
|
|
_as_text(row.get("document_number"))
|
|
for row in current_docs
|
|
if _as_text(row.get("document_number"))
|
|
}
|
|
current_external_ids = {
|
|
_as_text(row.get("external_id"))
|
|
for row in current_docs
|
|
if _as_text(row.get("external_id"))
|
|
}
|
|
|
|
candidates: List[Dict[str, Any]] = []
|
|
ignored: List[Dict[str, Any]] = []
|
|
hidden_old_count = 0
|
|
for row in rows:
|
|
item = dict(row)
|
|
item_doc_number = _as_text(item.get("document_number"))
|
|
item_external_id = _as_text(item.get("external_id"))
|
|
if item_doc_number in current_doc_numbers or item_external_id in current_external_ids:
|
|
continue
|
|
|
|
totals = _jasmin_document_totals(item)
|
|
lines = _jasmin_document_lines_from_item(item)
|
|
lifecycle = jasmin_document_lifecycle(item)
|
|
|
|
# If the opportunity already has a Jasmin quotation/proforma, do not show
|
|
# older/equal still-open quotations as actionable candidates. They caused
|
|
# operators to import ORC.137 after ORC.158 already existed. Downstream
|
|
# invoices are not replacement proposals; keep them visible so operators
|
|
# can associate the issued invoice to the same process.
|
|
recency_key = _jasmin_item_recency_tuple(item)
|
|
if _as_text(item.get("external_type")) in {"jasmin_quotation", "jasmin_proforma"} and current_latest_key and recency_key <= current_latest_key:
|
|
hidden_old_count += 1
|
|
continue
|
|
|
|
item["totals"] = totals
|
|
item["line_count"] = len(lines)
|
|
item["lifecycle"] = lifecycle
|
|
item["jasmin_status_label"] = lifecycle.get("status_label")
|
|
item["jasmin_status_code"] = lifecycle.get("status_code")
|
|
item["is_valid_candidate"] = lifecycle.get("is_valid_candidate")
|
|
item["invalid_reason"] = lifecycle.get("invalid_reason")
|
|
if lifecycle.get("is_valid_candidate"):
|
|
candidates.append(item)
|
|
else:
|
|
ignored.append(item)
|
|
candidates.sort(key=_jasmin_candidate_sort_tuple, reverse=True)
|
|
ignored.sort(key=_jasmin_candidate_sort_tuple, reverse=True)
|
|
|
|
result = (candidates + ignored)[: int(limit)]
|
|
if hidden_old_count and result:
|
|
result[0]["hidden_older_candidates_count"] = hidden_old_count
|
|
return result
|
|
|
|
|
|
async def replace_jasmin_document_for_opportunity_async(
|
|
*,
|
|
opportunity_id: str,
|
|
item_id: str,
|
|
actor: str = "operator_replace_jasmin_document",
|
|
dry_run: bool = False,
|
|
) -> Dict[str, Any]:
|
|
"""Remove current imported Jasmin quotation/proforma details and import a valid candidate.
|
|
|
|
This is intended for operator repair when an old/closed quotation was linked
|
|
by mistake. It only removes ClientFlow imported Jasmin artifacts from this
|
|
opportunity; it does not delete anything in Jasmin.
|
|
"""
|
|
with engine.begin() as conn:
|
|
current_docs = conn.execute(text("""
|
|
SELECT id::text, document_number, external_id, document_kind, status, total_amount
|
|
FROM commercial_documents
|
|
WHERE opportunity_id = CAST(:opportunity_id AS UUID)
|
|
AND system = 'jasmin'
|
|
AND document_kind IN ('quotation', 'proforma')
|
|
ORDER BY created_at DESC
|
|
"""), {"opportunity_id": opportunity_id}).mappings().all()
|
|
current_items = conn.execute(text("""
|
|
SELECT id::text, product_name, total_price, metadata
|
|
FROM opportunity_items
|
|
WHERE opportunity_id = CAST(:opportunity_id AS UUID)
|
|
AND (metadata->>'source_system' = 'jasmin' OR status = 'JASMIN_IMPORTED')
|
|
ORDER BY created_at
|
|
"""), {"opportunity_id": opportunity_id}).mappings().all()
|
|
|
|
if dry_run:
|
|
return {
|
|
"ok": True,
|
|
"dry_run": True,
|
|
"would_remove_documents": [dict(r) for r in current_docs],
|
|
"would_remove_items": [dict(r) for r in current_items],
|
|
"candidate_item_id": item_id,
|
|
}
|
|
|
|
# Validate candidate before deleting anything.
|
|
with engine.begin() as conn:
|
|
candidate = conn.execute(text("""
|
|
SELECT id::text, source_system, external_type, payload, document_number, opportunity_id::text
|
|
FROM reconciliation_items
|
|
WHERE id = CAST(:item_id AS UUID)
|
|
LIMIT 1
|
|
"""), {"item_id": item_id}).mappings().first()
|
|
if not candidate:
|
|
return {"ok": False, "error": "jasmin_item_not_found"}
|
|
lifecycle = jasmin_document_lifecycle(dict(candidate))
|
|
if candidate.get("source_system") != "jasmin" or not lifecycle.get("is_valid_candidate"):
|
|
return {"ok": False, "error": "jasmin_candidate_not_open_or_valid", "lifecycle": lifecycle}
|
|
|
|
with engine.begin() as conn:
|
|
conn.execute(text("""
|
|
DELETE FROM commercial_documents
|
|
WHERE opportunity_id = CAST(:opportunity_id AS UUID)
|
|
AND system = 'jasmin'
|
|
AND document_kind IN ('quotation', 'proforma')
|
|
"""), {"opportunity_id": opportunity_id})
|
|
conn.execute(text("""
|
|
DELETE FROM opportunity_items
|
|
WHERE opportunity_id = CAST(:opportunity_id AS UUID)
|
|
AND (metadata->>'source_system' = 'jasmin' OR status = 'JASMIN_IMPORTED')
|
|
"""), {"opportunity_id": opportunity_id})
|
|
conn.execute(text("""
|
|
UPDATE reconciliation_items
|
|
SET opportunity_id = NULL,
|
|
status = CASE WHEN status = 'linked' THEN 'open' ELSE status END,
|
|
updated_at = now(),
|
|
resolution_note = COALESCE(resolution_note, '') || '\nDesassociado por substituição de orçamento Jasmin em ClientFlow.'
|
|
WHERE source_system = 'jasmin'
|
|
AND opportunity_id = CAST(:opportunity_id AS UUID)
|
|
AND id <> CAST(:item_id AS UUID)
|
|
"""), {"opportunity_id": opportunity_id, "item_id": item_id})
|
|
|
|
result = await link_and_import_jasmin_candidate_async(opportunity_id=opportunity_id, item_id=item_id, actor=actor)
|
|
result["removed_documents"] = len(current_docs)
|
|
result["removed_items"] = len(current_items)
|
|
return result
|
|
|
|
|
|
async def link_and_import_jasmin_candidate_async(
|
|
*,
|
|
opportunity_id: str,
|
|
item_id: str,
|
|
actor: str = "operator_ui_link_existing_jasmin",
|
|
) -> Dict[str, Any]:
|
|
"""Link an existing Jasmin reconciliation item to an opportunity and import its details."""
|
|
with engine.begin() as conn:
|
|
item = conn.execute(text("""
|
|
SELECT id::text, source_system, external_type, opportunity_id::text, document_number, payload
|
|
FROM reconciliation_items
|
|
WHERE id = CAST(:item_id AS UUID)
|
|
LIMIT 1
|
|
"""), {"item_id": item_id}).mappings().first()
|
|
if not item:
|
|
return {"ok": False, "error": "jasmin_item_not_found"}
|
|
if item.get("source_system") != "jasmin":
|
|
return {"ok": False, "error": "item_is_not_jasmin"}
|
|
lifecycle = jasmin_document_lifecycle(item)
|
|
if not lifecycle.get("is_valid_candidate"):
|
|
return {
|
|
"ok": False,
|
|
"error": "jasmin_candidate_not_open_or_valid",
|
|
"status": lifecycle.get("status_label"),
|
|
"reason": lifecycle.get("invalid_reason"),
|
|
}
|
|
current_opportunity_id = _as_text(item.get("opportunity_id"))
|
|
if current_opportunity_id and current_opportunity_id != opportunity_id:
|
|
return {"ok": False, "error": "jasmin_item_already_linked_to_other_opportunity", "linked_opportunity_id": current_opportunity_id}
|
|
conn.execute(text("""
|
|
UPDATE reconciliation_items
|
|
SET opportunity_id = CAST(:opportunity_id AS UUID),
|
|
status = 'linked',
|
|
resolved_at = COALESCE(resolved_at, now()),
|
|
updated_at = now(),
|
|
resolution_note = COALESCE(resolution_note, 'Associado a oportunidade pela ficha comercial'),
|
|
payload = COALESCE(payload, '{}'::jsonb) || jsonb_build_object(
|
|
'manual_opportunity_link', jsonb_build_object(
|
|
'actor', CAST(:actor AS TEXT),
|
|
'opportunity_id', CAST(:opportunity_id AS TEXT),
|
|
'linked_at', now()
|
|
)
|
|
)
|
|
WHERE id = CAST(:item_id AS UUID)
|
|
"""), {"item_id": item_id, "opportunity_id": opportunity_id, "actor": actor})
|
|
conn.execute(text("""
|
|
INSERT INTO opportunity_events (id, opportunity_id, event_type, note, payload, created_by)
|
|
VALUES (gen_random_uuid(), CAST(:opportunity_id AS UUID), 'jasmin_existing_document_linked', :note, CAST(:payload AS JSONB), :actor)
|
|
"""), {
|
|
"opportunity_id": opportunity_id,
|
|
"note": "Documento Jasmin existente associado à oportunidade.",
|
|
"payload": _json({"reconciliation_item_id": item_id, "document_number": item.get("document_number")}),
|
|
"actor": actor,
|
|
})
|
|
|
|
result = await backfill_jasmin_opportunity_details_async(
|
|
opportunity_id=opportunity_id,
|
|
fetch_detail=True,
|
|
actor=actor,
|
|
dry_run=False,
|
|
)
|
|
result["linked_item_id"] = item_id
|
|
return result
|