"""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