"""Revenue-weighted same-item reconciliation for PChome growth automation."""

from __future__ import annotations

import hashlib
import json
import uuid
from datetime import datetime, timezone
from typing import Any, Callable

from sqlalchemy import bindparam, inspect, text

from services.pchome_growth_same_item_receipt_evidence import (
    build_target_candidate_evidence,
    select_next_machine_action,
)


CONTRACT_VERSION = "pchome_same_item_reconciliation_v1"
WORK_ITEM_ID = "GROWTH-P0-001-B"
ALLOWED_TARGET_ACTIONS = {
    "map_external_product",
    "review_external_candidate",
    "validate_external_source",
}


def _float(value: Any, default: float = 0.0) -> float:
    try:
        return float(value)
    except (TypeError, ValueError):
        return default


def _canonical_hash(payload: Any) -> str:
    encoded = json.dumps(
        payload,
        ensure_ascii=False,
        sort_keys=True,
        separators=(",", ":"),
        default=str,
    ).encode("utf-8")
    return hashlib.sha256(encoded).hexdigest()


def new_reconciliation_identity() -> dict[str, str]:
    run_id = uuid.uuid4().hex
    return {
        "trace_id": uuid.uuid4().hex,
        "run_id": run_id,
        "work_item_id": WORK_ITEM_ID,
    }


def build_revenue_weighted_targets(
    payload: dict[str, Any],
    limit: int,
) -> list[dict[str, Any]]:
    """Rank unresolved products by revenue at risk before generic priority."""
    unresolved: list[dict[str, Any]] = []
    for item in list((payload or {}).get("opportunities") or []):
        action = item.get("recommended_action") or {}
        if item.get("external_price"):
            continue
        if str(action.get("code") or "") not in ALLOWED_TARGET_ACTIONS:
            continue
        product_id = str(item.get("pchome_product_id") or "").strip()
        product_name = str(item.get("product_name") or "").strip()
        if not product_id or not product_name:
            continue
        sales_7d = max(0.0, _float(item.get("sales_7d")))
        sales_prev_7d = max(0.0, _float(item.get("sales_prev_7d")))
        unresolved.append({
            "product_id": product_id,
            "name": product_name,
            "price": item.get("pchome_price"),
            "sales_7d": sales_7d,
            "sales_prev_7d": sales_prev_7d,
            "revenue_at_risk": max(sales_7d, sales_prev_7d),
            "priority_score": _float(item.get("priority_score")),
            "target_state": (
                "candidate_validation"
                if item.get("review_candidate") or item.get("market_offers")
                else "unmatched"
            ),
            "existing_review_candidate": item.get("review_candidate") or None,
        })

    if not unresolved:
        return []

    total_revenue_at_risk = sum(item["revenue_at_risk"] for item in unresolved)
    max_revenue_at_risk = max(item["revenue_at_risk"] for item in unresolved) or 1.0
    for item in unresolved:
        revenue_index = item["revenue_at_risk"] / max_revenue_at_risk * 100
        validation_boost = 5.0 if item["target_state"] == "candidate_validation" else 0.0
        item["revenue_share_pct"] = round(
            item["revenue_at_risk"] / max(total_revenue_at_risk, 1.0) * 100,
            3,
        )
        item["execution_priority_score"] = round(
            revenue_index * 0.75 + item["priority_score"] * 0.20 + validation_boost,
            3,
        )

    unresolved.sort(
        key=lambda item: (
            item["execution_priority_score"],
            item["revenue_at_risk"],
            item["priority_score"],
            item["product_id"],
        ),
        reverse=True,
    )
    bounded_limit = max(1, min(int(limit or 12), 20))
    selected = unresolved[:bounded_limit]
    for rank, item in enumerate(selected, start=1):
        item["execution_rank"] = rank
    return selected


def _candidate_source_id(candidate: dict[str, Any]) -> str:
    return str(
        candidate.get("product_id")
        or candidate.get("goodsCode")
        or candidate.get("id")
        or ""
    ).strip()


def _candidate_name(candidate: dict[str, Any]) -> str:
    return str(candidate.get("name") or candidate.get("title") or "").strip()


def _verification_rank(candidate: dict[str, Any]) -> tuple[int, float, str]:
    compare_type = str(candidate.get("auto_compare_type") or "")
    type_rank = 2 if compare_type == "total_price" else 1
    return (
        type_rank,
        _float(candidate.get("target_match_score")),
        _candidate_source_id(candidate),
    )


def verify_same_item_candidates(
    targets: list[dict[str, Any]],
    candidates: list[dict[str, Any]],
    *,
    identity: dict[str, str],
    score_func: Callable[..., Any] | None = None,
    unit_price_func: Callable[..., dict[str, Any]] | None = None,
    min_exact_score: float = 0.76,
) -> dict[str, Any]:
    """Recompute identity and unit evidence without trusting crawler labels."""
    if score_func is None or unit_price_func is None:
        from services.marketplace_product_matcher import (
            build_unit_price_comparison,
            score_marketplace_match,
        )

        score_func = score_func or score_marketplace_match
        unit_price_func = unit_price_func or build_unit_price_comparison

    target_map = {str(item.get("product_id") or ""): item for item in targets}
    verified: list[dict[str, Any]] = []
    blocked: list[dict[str, Any]] = []

    for candidate in list(candidates or []):
        target_id = str(candidate.get("target_pchome_product_id") or "").strip()
        target = target_map.get(target_id)
        source_id = _candidate_source_id(candidate)
        source_name = _candidate_name(candidate)
        if not target:
            blocked.append({
                "target_pchome_product_id": target_id,
                "source_product_id": source_id,
                "source_name": source_name,
                "source_price": _float(candidate.get("price")),
                "search_term": str(candidate.get("target_search_term") or ""),
                "reason_code": "target_not_in_bounded_batch",
            })
            continue
        if not source_id or not source_name:
            blocked.append({
                "target_pchome_product_id": target_id,
                "source_product_id": source_id,
                "source_name": source_name,
                "source_price": _float(candidate.get("price")),
                "search_term": str(candidate.get("target_search_term") or ""),
                "reason_code": "source_identity_missing",
            })
            continue

        source_price = _float(candidate.get("price"))
        target_price = _float(target.get("price"))
        if source_price <= 0 or target_price <= 0:
            blocked.append({
                "target_pchome_product_id": target_id,
                "source_product_id": source_id,
                "source_name": source_name,
                "source_price": source_price,
                "search_term": str(candidate.get("target_search_term") or ""),
                "reason_code": "positive_price_evidence_missing",
            })
            continue

        diagnostics = score_func(
            source_name,
            str(target.get("name") or ""),
            momo_price=source_price,
            competitor_price=target_price,
        )
        exact_ready = bool(
            not getattr(diagnostics, "hard_veto", True)
            and _float(getattr(diagnostics, "score", 0.0)) >= min_exact_score
            and getattr(diagnostics, "comparison_mode", "") == "exact_identity"
            and getattr(diagnostics, "match_type", "") == "exact"
            and getattr(diagnostics, "price_basis", "") == "total_price"
            and getattr(diagnostics, "alert_tier", "") == "price_alert_exact"
        )
        unit_comparison: dict[str, Any] = {}
        unit_ready = False
        if (
            getattr(diagnostics, "comparison_mode", "") == "unit_comparable"
            or getattr(diagnostics, "price_basis", "") == "unit_price"
        ):
            unit_comparison = unit_price_func(
                source_name,
                str(target.get("name") or ""),
                momo_price=source_price,
                competitor_price=target_price,
            )
            unit_ready = bool(
                unit_comparison.get("comparable")
                and unit_comparison.get("unit_label")
                and _float(unit_comparison.get("momo_total_quantity")) > 0
                and _float(unit_comparison.get("competitor_total_quantity")) > 0
            )

        if exact_ready:
            compare_type = "total_price"
            price_basis = "total_price"
            review_status = "independent_verifier_exact"
        elif unit_ready:
            compare_type = "unit_price"
            price_basis = "unit_price"
            review_status = "independent_verifier_unit"
        else:
            blocked.append({
                "target_pchome_product_id": target_id,
                "source_product_id": source_id,
                "source_name": source_name,
                "source_price": source_price,
                "search_term": str(candidate.get("target_search_term") or ""),
                "reason_code": "same_item_verifier_rejected",
                "match_score": _float(getattr(diagnostics, "score", 0.0)),
                "hard_veto": bool(getattr(diagnostics, "hard_veto", False)),
                "comparison_mode": str(
                    getattr(diagnostics, "comparison_mode", "") or ""
                ),
                "match_type": str(getattr(diagnostics, "match_type", "") or ""),
                "price_basis": str(getattr(diagnostics, "price_basis", "") or ""),
                "alert_tier": str(getattr(diagnostics, "alert_tier", "") or ""),
                "reasons": list(getattr(diagnostics, "reasons", ()) or ())[:8],
            })
            continue

        claimed_type = str(candidate.get("auto_compare_type") or "manual_review")
        fingerprint_payload = {
            "target_id": target_id,
            "source_id": source_id,
            "target_name": target.get("name"),
            "source_name": source_name,
            "target_price": target_price,
            "source_price": source_price,
            "compare_type": compare_type,
            "match_score": _float(getattr(diagnostics, "score", 0.0)),
        }
        enriched = dict(candidate)
        enriched.update({
            "target_pchome_product_id": target_id,
            "target_pchome_name": target.get("name"),
            "target_pchome_price": target_price,
            "target_match_score": round(_float(getattr(diagnostics, "score", 0.0)), 3),
            "target_match_reasons": list(getattr(diagnostics, "reasons", ()) or ()),
            "target_comparison_mode": getattr(diagnostics, "comparison_mode", ""),
            "target_match_type": getattr(diagnostics, "match_type", ""),
            "target_alert_tier": getattr(diagnostics, "alert_tier", ""),
            "target_hard_veto": bool(getattr(diagnostics, "hard_veto", False)),
            "target_price_basis": price_basis,
            "target_unit_price_comparison": unit_comparison,
            "target_review_status": review_status,
            "auto_compare_type": compare_type,
            "can_auto_compare": True,
            "same_item_candidate_fingerprint": _canonical_hash(fingerprint_payload)[:24],
            "same_item_reconciliation": {
                **identity,
                "contract_version": CONTRACT_VERSION,
                "execution_rank": target.get("execution_rank"),
                "execution_priority_score": target.get("execution_priority_score"),
                "revenue_share_pct": target.get("revenue_share_pct"),
                "target_state_before": target.get("target_state"),
                "candidate_claimed_type": claimed_type,
                "candidate_claim_drift": claimed_type != compare_type,
                "independent_verifier_passed": True,
                "candidate_fingerprint": _canonical_hash(fingerprint_payload)[:24],
            },
        })
        verified.append(enriched)

    selected_by_target: dict[str, dict[str, Any]] = {}
    for candidate in verified:
        target_id = str(candidate.get("target_pchome_product_id") or "")
        current = selected_by_target.get(target_id)
        if current is None or _verification_rank(candidate) > _verification_rank(current):
            selected_by_target[target_id] = candidate

    selected = list(selected_by_target.values())
    selected.sort(
        key=lambda item: int(
            (item.get("same_item_reconciliation") or {}).get("execution_rank") or 999
        )
    )

    review_by_target: dict[str, dict[str, Any]] = {}
    selected_ids = set(selected_by_target)
    for candidate in list(candidates or []):
        target_id = str(candidate.get("target_pchome_product_id") or "").strip()
        if not target_id or target_id in selected_ids or target_id not in target_map:
            continue
        current = review_by_target.get(target_id)
        if current is None or _float(candidate.get("target_match_score")) > _float(
            current.get("target_match_score")
        ):
            review_candidate = dict(candidate)
            review_candidate["same_item_reconciliation"] = {
                **identity,
                "contract_version": CONTRACT_VERSION,
                "independent_verifier_passed": False,
                "target_state_before": target_map[target_id].get("target_state"),
            }
            review_by_target[target_id] = review_candidate

    return {
        "contract_version": CONTRACT_VERSION,
        "candidate_count": len(candidates or []),
        "verified_candidate_count": len(verified),
        "selected_candidate_count": len(selected),
        "blocked_candidate_count": len(blocked),
        "claim_drift_count": sum(
            1
            for item in selected
            if (item.get("same_item_reconciliation") or {}).get("candidate_claim_drift")
        ),
        "verified_candidates": selected,
        "review_candidates": list(review_by_target.values()),
        "blocked_candidates": blocked,
    }


def readback_external_offer_candidates(
    engine,
    candidates: list[dict[str, Any]],
) -> dict[str, Any]:
    """Verify committed offers by exact identity on a fresh DB connection."""
    expected = list(candidates or [])
    if not expected:
        return {
            "success": True,
            "status": "no_write_verified",
            "expected_count": 0,
            "readback_pass_count": 0,
            "readback": [],
        }
    if not inspect(engine).has_table("external_offers"):
        return {
            "success": False,
            "status": "external_offers_schema_missing",
            "expected_count": len(expected),
            "readback_pass_count": 0,
            "readback": [],
        }

    query = text("""
        SELECT
            pchome_product_id,
            source_product_id,
            match_status,
            data_quality_status,
            raw_payload_json,
            observed_at
        FROM external_offers
        WHERE source_code = 'momo_reference'
          AND ingestion_method = 'targeted_momo_search'
          AND pchome_product_id = :pchome_product_id
          AND source_product_id = :source_product_id
        ORDER BY observed_at DESC, id DESC
        LIMIT 1
    """)
    readback: list[dict[str, Any]] = []
    try:
        with engine.connect() as conn:
            for candidate in expected:
                target_id = str(candidate.get("target_pchome_product_id") or "")
                source_id = _candidate_source_id(candidate)
                expected_identity = candidate.get("same_item_reconciliation") or {}
                expected_fingerprint = str(
                    candidate.get("same_item_candidate_fingerprint") or ""
                )
                row = conn.execute(query, {
                    "pchome_product_id": target_id,
                    "source_product_id": source_id,
                }).mappings().first()
                raw_payload: dict[str, Any] = {}
                if row:
                    raw_payload_value = row.get("raw_payload_json")
                    if isinstance(raw_payload_value, dict):
                        raw_payload = raw_payload_value
                    elif isinstance(raw_payload_value, str):
                        try:
                            parsed = json.loads(raw_payload_value)
                            raw_payload = parsed if isinstance(parsed, dict) else {}
                        except json.JSONDecodeError:
                            raw_payload = {}
                observed_identity = raw_payload.get("same_item_reconciliation") or {}
                observed_fingerprint = str(
                    raw_payload.get("same_item_candidate_fingerprint") or ""
                )
                identity_keys = ("trace_id", "run_id", "work_item_id")
                identity_passed = bool(
                    all(str(expected_identity.get(key) or "") for key in identity_keys)
                    and all(
                        str(observed_identity.get(key) or "")
                        == str(expected_identity.get(key) or "")
                        for key in identity_keys
                    )
                )
                passed = bool(
                    row
                    and str(row.get("pchome_product_id") or "") == target_id
                    and str(row.get("source_product_id") or "") == source_id
                    and row.get("match_status") == "verified"
                    and row.get("data_quality_status") == "verified"
                    and expected_fingerprint
                    and observed_fingerprint == expected_fingerprint
                    and identity_passed
                )
                readback.append({
                    "target_pchome_product_id": target_id,
                    "source_product_id": source_id,
                    "observed_source_product_id": (
                        str(row.get("source_product_id") or "") if row else ""
                    ),
                    "candidate_fingerprint": expected_fingerprint,
                    "observed_candidate_fingerprint": observed_fingerprint,
                    "identity_passed": identity_passed,
                    "passed": passed,
                })
    except Exception as exc:
        return {
            "success": False,
            "status": "external_offer_readback_failed",
            "expected_count": len(expected),
            "readback_pass_count": sum(1 for item in readback if item["passed"]),
            "readback": readback,
            "error_type": type(exc).__name__,
        }

    pass_count = sum(1 for item in readback if item["passed"])
    success = pass_count == len(expected)
    return {
        "success": success,
        "status": "verified" if success else "readback_mismatch",
        "expected_count": len(expected),
        "readback_pass_count": pass_count,
        "readback": readback,
    }


def build_coverage_snapshot(payload: dict[str, Any]) -> dict[str, Any]:
    opportunities = list((payload or {}).get("opportunities") or [])
    total_revenue = sum(max(0.0, _float(item.get("sales_7d"))) for item in opportunities)
    ready_revenue = sum(
        max(0.0, _float(item.get("sales_7d")))
        for item in opportunities
        if item.get("external_price")
    )
    validation_revenue = sum(
        max(0.0, _float(item.get("sales_7d")))
        for item in opportunities
        if not item.get("external_price")
        and (item.get("review_candidate") or item.get("market_offers"))
    )
    unmatched_revenue = max(0.0, total_revenue - ready_revenue - validation_revenue)
    stats = (payload or {}).get("stats") or {}
    contract = stats.get("comparison_metric_contract") or {}
    partition = contract.get("state_partition") or {}
    return {
        "scope": contract.get("scope") or f"top_{len(opportunities)}_revenue_products",
        "comparison_ready_count": int(
            partition.get("comparison_ready") or stats.get("mapped_count") or 0
        ),
        "candidate_validation_count": int(
            partition.get("candidate_validation") or stats.get("candidate_validation_count") or 0
        ),
        "unmatched_count": int(partition.get("unmatched") or stats.get("unmatched_count") or 0),
        "total_count": int(partition.get("total") or len(opportunities)),
        "count_coverage_rate": round(_float(contract.get("rate") or stats.get("mapping_rate")), 3),
        "comparison_ready_sales_7d": round(ready_revenue, 2),
        "candidate_validation_sales_7d": round(validation_revenue, 2),
        "unmatched_sales_7d": round(unmatched_revenue, 2),
        "total_sales_7d": round(total_revenue, 2),
        "revenue_coverage_rate": round(ready_revenue / max(total_revenue, 1.0) * 100, 3),
    }


def build_coverage_post_verifier(
    *,
    before_payload: dict[str, Any],
    after_payload: dict[str, Any],
    verified_candidates: list[dict[str, Any]],
    sync_result: dict[str, Any],
) -> dict[str, Any]:
    before = build_coverage_snapshot(before_payload)
    after = build_coverage_snapshot(after_payload)
    after_by_id = {
        str(item.get("pchome_product_id") or ""): item
        for item in list((after_payload or {}).get("opportunities") or [])
    }
    independent_readback = (sync_result or {}).get("independent_readback") or {}
    independent_required = bool(
        (sync_result or {}).get("independent_readback_required")
    )
    direct_by_pair = {
        (
            str(item.get("target_pchome_product_id") or ""),
            str(item.get("source_product_id") or ""),
        ): item
        for item in independent_readback.get("readback") or []
    }
    readback: list[dict[str, Any]] = []
    for candidate in verified_candidates:
        target_id = str(candidate.get("target_pchome_product_id") or "")
        source_id = _candidate_source_id(candidate)
        direct = direct_by_pair.get((target_id, source_id))
        if direct is not None:
            observed_source_id = str(direct.get("observed_source_product_id") or "")
            passed = bool(direct.get("passed"))
            readback_source = "external_offers_exact_identity"
            expected_fingerprint = direct.get("candidate_fingerprint")
            observed_fingerprint = direct.get("observed_candidate_fingerprint")
            identity_passed = bool(direct.get("identity_passed"))
        else:
            external_price = (after_by_id.get(target_id) or {}).get("external_price") or {}
            observed_source_id = str(external_price.get("momo_sku") or "")
            passed = bool(
                not independent_required
                and external_price
                and observed_source_id == source_id
            )
            readback_source = (
                "missing_required_external_offer_readback"
                if independent_required
                else "top50_payload_compatibility_fallback"
            )
            expected_fingerprint = candidate.get("same_item_candidate_fingerprint")
            observed_fingerprint = None
            identity_passed = None
        readback.append({
            "target_pchome_product_id": target_id,
            "source_product_id": source_id,
            "observed_source_product_id": observed_source_id,
            "readback_source": readback_source,
            "candidate_fingerprint": expected_fingerprint,
            "observed_candidate_fingerprint": observed_fingerprint,
            "identity_passed": identity_passed,
            "passed": passed,
        })

    selected_count = len(verified_candidates)
    written_count = int((sync_result or {}).get("written_count") or 0)
    readback_pass_count = sum(1 for item in readback if item["passed"])
    no_write_verified = selected_count == 0 and written_count == 0
    apply_verified = bool(
        selected_count > 0
        and written_count == selected_count
        and readback_pass_count == selected_count
        and (
            not independent_required
            or bool(independent_readback.get("success"))
        )
    )
    no_regression = (
        after["comparison_ready_count"] >= before["comparison_ready_count"]
        and after["count_coverage_rate"] >= before["count_coverage_rate"]
        and after["revenue_coverage_rate"] >= before["revenue_coverage_rate"]
    )
    all_checks_passed = bool((no_write_verified or apply_verified) and no_regression)
    return {
        "contract_version": CONTRACT_VERSION,
        "status": (
            "verified"
            if apply_verified and no_regression
            else "no_write_verified"
            if no_write_verified and no_regression
            else "failed"
        ),
        "selected_candidate_count": selected_count,
        "written_count": written_count,
        "readback_pass_count": readback_pass_count,
        "independent_readback_required": independent_required,
        "independent_readback_status": independent_readback.get("status"),
        "count_coverage_delta": round(
            after["count_coverage_rate"] - before["count_coverage_rate"], 3
        ),
        "comparison_ready_delta": (
            after["comparison_ready_count"] - before["comparison_ready_count"]
        ),
        "revenue_coverage_delta": round(
            after["revenue_coverage_rate"] - before["revenue_coverage_rate"], 3
        ),
        "no_regression": no_regression,
        "all_checks_passed": all_checks_passed,
        "before": before,
        "after": after,
        "readback": readback,
    }


def ensure_evidence_receipt_table(engine) -> dict[str, Any]:
    """Idempotently apply the existing additive Migration 045 schema contract."""
    table_name = "external_offer_evidence_receipts"
    if inspect(engine).has_table(table_name):
        return {
            "success": True,
            "status": "already_ready",
            "table": table_name,
            "created": False,
        }

    dialect = engine.dialect.name
    if dialect == "postgresql":
        create_sql = """
            CREATE TABLE IF NOT EXISTS external_offer_evidence_receipts (
                receipt_id TEXT PRIMARY KEY,
                pchome_product_id TEXT NOT NULL,
                automation_decision TEXT NOT NULL,
                payload_hash TEXT NOT NULL,
                evidence_delta_json JSONB NOT NULL DEFAULT '{}'::jsonb,
                created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
                applied_at TIMESTAMPTZ,
                apply_status TEXT NOT NULL DEFAULT 'previewed'
            )
        """
    else:
        create_sql = """
            CREATE TABLE IF NOT EXISTS external_offer_evidence_receipts (
                receipt_id TEXT PRIMARY KEY,
                pchome_product_id TEXT NOT NULL,
                automation_decision TEXT NOT NULL,
                payload_hash TEXT NOT NULL,
                evidence_delta_json TEXT NOT NULL DEFAULT '{}',
                created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
                applied_at TEXT,
                apply_status TEXT NOT NULL DEFAULT 'previewed'
            )
        """
    index_sql = (
        "CREATE INDEX IF NOT EXISTS idx_external_offer_evidence_receipts_pchome_product_id "
        "ON external_offer_evidence_receipts (pchome_product_id)",
        "CREATE INDEX IF NOT EXISTS idx_external_offer_evidence_receipts_payload_hash "
        "ON external_offer_evidence_receipts (payload_hash)",
        "CREATE INDEX IF NOT EXISTS idx_external_offer_evidence_receipts_apply_status "
        "ON external_offer_evidence_receipts (apply_status)",
    )
    try:
        with engine.begin() as conn:
            conn.execute(text(create_sql))
            for statement in index_sql:
                conn.execute(text(statement))
        ready = inspect(engine).has_table(table_name)
        return {
            "success": ready,
            "status": "created_and_verified" if ready else "verification_failed",
            "table": table_name,
            "created": ready,
        }
    except Exception as exc:
        return {
            "success": False,
            "status": "apply_failed",
            "table": table_name,
            "created": False,
            "error_type": type(exc).__name__,
        }


def _target_receipt_decision(
    target: dict[str, Any],
    run_receipt: dict[str, Any],
) -> tuple[str, str]:
    target_id = str(target.get("product_id") or "")
    verification = run_receipt.get("candidate_verification") or {}
    selected_ids = {
        str(item.get("target_pchome_product_id") or "")
        for item in verification.get("verified_candidates") or []
    }
    blocked_ids = {
        str(item.get("target_pchome_product_id") or "")
        for item in verification.get("blocked_candidates") or []
    }
    passed_ids = {
        str(item.get("target_pchome_product_id") or "")
        for item in (run_receipt.get("post_verifier") or {}).get("readback") or []
        if item.get("passed")
    }
    if target_id in passed_ids:
        return "promoted_verified", "verified"
    if target_id in selected_ids:
        return "apply_not_verified", "failed"
    if target_id in blocked_ids:
        return "candidate_rejected", "no_write"
    return "no_candidate", "no_write"


def persist_reconciliation_receipts(
    engine,
    *,
    targets: list[dict[str, Any]],
    run_receipt: dict[str, Any],
) -> dict[str, Any]:
    """Persist one bounded, replayable receipt for each targeted product."""
    identity = run_receipt.get("identity") or {}
    run_id = str(identity.get("run_id") or uuid.uuid4().hex)
    generated_at = str(run_receipt.get("generated_at") or datetime.now(timezone.utc).isoformat())
    receipt_targets = list(targets or []) or [{
        "product_id": "__run__",
        "target_state": "not_needed",
        "execution_rank": 0,
        "revenue_share_pct": 0,
    }]
    rows: list[dict[str, Any]] = []
    for target in receipt_targets:
        target_id = str(target.get("product_id") or "__run__")
        decision, apply_status = _target_receipt_decision(target, run_receipt)
        candidate_evidence = build_target_candidate_evidence(target_id, run_receipt)
        next_machine_action = select_next_machine_action(
            decision,
            candidate_evidence,
        )
        evidence = {
            "contract_version": CONTRACT_VERSION,
            "identity": identity,
            "generated_at": generated_at,
            "target": {
                "pchome_product_id": target_id,
                "name": target.get("name"),
                "price": target.get("price"),
                "sales_7d": target.get("sales_7d"),
                "revenue_at_risk": target.get("revenue_at_risk"),
                "target_state_before": target.get("target_state"),
                "execution_rank": target.get("execution_rank"),
                "execution_priority_score": target.get("execution_priority_score"),
                "revenue_share_pct": target.get("revenue_share_pct"),
            },
            "source_receipt": run_receipt.get("source_receipt") or {},
            "source_of_truth_diff": run_receipt.get("coverage_diff") or {},
            "ai_decision": decision,
            "candidate_evidence": candidate_evidence,
            "next_machine_action": next_machine_action,
            "risk_policy": run_receipt.get("risk_policy") or {},
            "check_mode": run_receipt.get("check_mode") or {},
            "execution": run_receipt.get("execution") or {},
            "post_verifier": run_receipt.get("post_verifier") or {},
            "terminal": run_receipt.get("terminal") or {},
        }
        payload_hash = _canonical_hash(evidence)
        rows.append({
            "receipt_id": f"sameitem-{_canonical_hash([run_id, target_id])[:24]}",
            "pchome_product_id": target_id,
            "automation_decision": decision,
            "payload_hash": payload_hash,
            "evidence_delta_json": json.dumps(evidence, ensure_ascii=False, default=str),
            "applied_at": generated_at if apply_status == "verified" else None,
            "apply_status": apply_status,
        })

    if engine.dialect.name == "postgresql":
        insert_sql = """
            INSERT INTO external_offer_evidence_receipts (
                receipt_id, pchome_product_id, automation_decision, payload_hash,
                evidence_delta_json, applied_at, apply_status
            ) VALUES (
                :receipt_id, :pchome_product_id, :automation_decision, :payload_hash,
                CAST(:evidence_delta_json AS JSONB), CAST(:applied_at AS TIMESTAMPTZ), :apply_status
            )
            ON CONFLICT (receipt_id) DO UPDATE SET
                automation_decision = EXCLUDED.automation_decision,
                payload_hash = EXCLUDED.payload_hash,
                evidence_delta_json = EXCLUDED.evidence_delta_json,
                applied_at = EXCLUDED.applied_at,
                apply_status = EXCLUDED.apply_status
        """
    else:
        insert_sql = """
            INSERT INTO external_offer_evidence_receipts (
                receipt_id, pchome_product_id, automation_decision, payload_hash,
                evidence_delta_json, applied_at, apply_status
            ) VALUES (
                :receipt_id, :pchome_product_id, :automation_decision, :payload_hash,
                :evidence_delta_json, :applied_at, :apply_status
            )
            ON CONFLICT (receipt_id) DO UPDATE SET
                automation_decision = excluded.automation_decision,
                payload_hash = excluded.payload_hash,
                evidence_delta_json = excluded.evidence_delta_json,
                applied_at = excluded.applied_at,
                apply_status = excluded.apply_status
        """
    try:
        with engine.begin() as conn:
            conn.execute(text(insert_sql), rows)
            ids = [row["receipt_id"] for row in rows]
            readback_stmt = text(
                "SELECT receipt_id, apply_status FROM external_offer_evidence_receipts "
                "WHERE receipt_id IN :receipt_ids"
            ).bindparams(bindparam("receipt_ids", expanding=True))
            readback = conn.execute(readback_stmt, {"receipt_ids": ids}).mappings().all()
        readback_ids = {str(row.get("receipt_id") or "") for row in readback}
        return {
            "success": len(readback_ids) == len(rows),
            "status": "persisted_and_verified" if len(readback_ids) == len(rows) else "readback_mismatch",
            "written_count": len(rows),
            "readback_count": len(readback_ids),
            "receipt_ids": [row["receipt_id"] for row in rows],
        }
    except Exception as exc:
        return {
            "success": False,
            "status": "persistence_failed",
            "written_count": 0,
            "readback_count": 0,
            "receipt_ids": [],
            "error_type": type(exc).__name__,
        }


def read_latest_reconciliation_receipts(engine, *, limit: int = 20) -> dict[str, Any]:
    bounded_limit = max(1, min(int(limit or 20), 100))
    if not inspect(engine).has_table("external_offer_evidence_receipts"):
        return {
            "contract_version": CONTRACT_VERSION,
            "status": "schema_missing",
            "receipt_count": 0,
            "receipts": [],
        }
    with engine.connect() as conn:
        rows = conn.execute(
            text(
                "SELECT receipt_id, pchome_product_id, automation_decision, payload_hash, "
                "evidence_delta_json, created_at, applied_at, apply_status "
                "FROM external_offer_evidence_receipts "
                "WHERE receipt_id LIKE 'sameitem-%' "
                "ORDER BY created_at DESC LIMIT :limit"
            ),
            {"limit": bounded_limit},
        ).mappings().all()
    receipts: list[dict[str, Any]] = []
    for row in rows:
        evidence = row.get("evidence_delta_json")
        if isinstance(evidence, str):
            try:
                evidence = json.loads(evidence)
            except json.JSONDecodeError:
                evidence = {}
        evidence = evidence if isinstance(evidence, dict) else {}
        candidate_evidence = evidence.get("candidate_evidence") or {}
        receipts.append({
            "receipt_id": row.get("receipt_id"),
            "pchome_product_id": row.get("pchome_product_id"),
            "automation_decision": row.get("automation_decision"),
            "apply_status": row.get("apply_status"),
            "payload_hash": row.get("payload_hash"),
            "created_at": str(row.get("created_at") or ""),
            "applied_at": str(row.get("applied_at") or ""),
            "trace_id": (evidence.get("identity") or {}).get("trace_id"),
            "run_id": (evidence.get("identity") or {}).get("run_id"),
            "work_item_id": (evidence.get("identity") or {}).get("work_item_id"),
            "post_verifier_status": (evidence.get("post_verifier") or {}).get("status"),
            "next_machine_action": evidence.get("next_machine_action"),
            "candidate_evidence_summary": {
                "verified_candidate_count": int(
                    candidate_evidence.get("verified_candidate_count") or 0
                ),
                "review_candidate_count": int(
                    candidate_evidence.get("review_candidate_count") or 0
                ),
                "blocked_candidate_count": int(
                    candidate_evidence.get("blocked_candidate_count") or 0
                ),
                "reason_codes": list(candidate_evidence.get("reason_codes") or []),
                "verifier_reasons": list(
                    candidate_evidence.get("verifier_reasons") or []
                )[:8],
            },
        })
    return {
        "contract_version": CONTRACT_VERSION,
        "status": "ready",
        "receipt_count": len(receipts),
        "latest_run_id": receipts[0].get("run_id") if receipts else None,
        "verified_count": sum(1 for item in receipts if item.get("apply_status") == "verified"),
        "no_write_count": sum(1 for item in receipts if item.get("apply_status") == "no_write"),
        "failed_count": sum(1 for item in receipts if item.get("apply_status") == "failed"),
        "receipts": receipts,
    }


__all__ = (
    "CONTRACT_VERSION",
    "WORK_ITEM_ID",
    "build_coverage_post_verifier",
    "build_coverage_snapshot",
    "build_revenue_weighted_targets",
    "ensure_evidence_receipt_table",
    "new_reconciliation_identity",
    "persist_reconciliation_receipts",
    "readback_external_offer_candidates",
    "read_latest_reconciliation_receipts",
    "verify_same_item_candidates",
)
