ewoooc/services/pchome_growth_same_item_receipt_evidence.py

"""Public-safe evidence shaping for same-item reconciliation receipts."""

from __future__ import annotations

from typing import Any


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


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


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


def safe_candidate_evidence(candidate: dict[str, Any]) -> dict[str, Any]:
    reconciliation = candidate.get("same_item_reconciliation") or {}
    return {
        "target_pchome_product_id": str(
            candidate.get("target_pchome_product_id") or ""
        ),
        "source_product_id": _candidate_source_id(candidate),
        "source_name": _candidate_name(candidate),
        "source_price": _float(
            candidate.get("price"),
            _float(candidate.get("source_price")),
        ),
        "search_term": str(
            candidate.get("target_search_term")
            or candidate.get("search_term")
            or ""
        ),
        "match_score": _float(
            candidate.get("target_match_score"),
            _float(candidate.get("match_score")),
        ),
        "comparison_mode": str(
            candidate.get("target_comparison_mode")
            or candidate.get("comparison_mode")
            or ""
        ),
        "match_type": str(
            candidate.get("target_match_type")
            or candidate.get("match_type")
            or ""
        ),
        "price_basis": str(
            candidate.get("target_price_basis")
            or candidate.get("price_basis")
            or ""
        ),
        "alert_tier": str(
            candidate.get("target_alert_tier")
            or candidate.get("alert_tier")
            or ""
        ),
        "hard_veto": bool(
            candidate.get("target_hard_veto")
            if "target_hard_veto" in candidate
            else candidate.get("hard_veto")
        ),
        "reason_code": str(candidate.get("reason_code") or ""),
        "reasons": list(
            candidate.get("target_match_reasons")
            or candidate.get("reasons")
            or []
        )[:8],
        "auto_compare_type": str(candidate.get("auto_compare_type") or ""),
        "candidate_fingerprint": str(
            candidate.get("same_item_candidate_fingerprint")
            or reconciliation.get("candidate_fingerprint")
            or ""
        ),
        "independent_verifier_passed": bool(
            reconciliation.get("independent_verifier_passed")
        ),
    }


def build_target_candidate_evidence(
    target_id: str,
    run_receipt: dict[str, Any],
) -> dict[str, Any]:
    verification = run_receipt.get("candidate_verification") or {}

    def matching(items: list[dict[str, Any]]) -> list[dict[str, Any]]:
        return [
            safe_candidate_evidence(item)
            for item in items
            if str(item.get("target_pchome_product_id") or "") == target_id
        ]

    verified = matching(list(verification.get("verified_candidates") or []))
    review = matching(list(verification.get("review_candidates") or []))
    blocked = matching(list(verification.get("blocked_candidates") or []))
    reason_codes = sorted({
        item.get("reason_code")
        for item in blocked
        if item.get("reason_code")
    })
    verifier_reasons = sorted({
        reason
        for item in [*verified, *review, *blocked]
        for reason in item.get("reasons") or []
        if reason
    })
    return {
        "verified_candidates": verified,
        "review_candidates": review,
        "blocked_candidates": blocked,
        "verified_candidate_count": len(verified),
        "review_candidate_count": len(review),
        "blocked_candidate_count": len(blocked),
        "reason_codes": reason_codes,
        "verifier_reasons": verifier_reasons,
    }


def select_next_machine_action(
    decision: str,
    candidate_evidence: dict[str, Any],
) -> str:
    if decision == "promoted_verified":
        return "continue_next_revenue_weighted_batch"
    if decision == "apply_not_verified":
        return "retry_exact_identity_readback_before_next_apply"

    reason_codes = set(candidate_evidence.get("reason_codes") or [])
    verifier_reasons = set(candidate_evidence.get("verifier_reasons") or [])
    if decision == "no_candidate":
        return "expand_search_terms_with_brand_pack_and_spec_anchors"
    if "source_identity_missing" in reason_codes:
        return "refresh_source_identity_then_retry"
    if "positive_price_evidence_missing" in reason_codes:
        return "refresh_positive_price_evidence_then_retry"
    if verifier_reasons & {
        "count_conflict",
        "bundle_offer_conflict",
        "multi_component_conflict",
        "component_count_conflict",
        "pack_quantity_difference",
    }:
        return "retry_search_with_exact_pack_quantity"
    if verifier_reasons & {
        "variant_descriptor_conflict",
        "variant_option_conflict",
        "variant_selection_review",
        "makeup_catalog_selection_gap",
    }:
        return "retry_search_with_exact_variant_evidence"
    if verifier_reasons & {"unit_comparable", "unit_price_review"}:
        return "replay_unit_basis_before_promotion"
    return "refresh_source_candidates_with_evidence_delta"


__all__ = (
    "build_target_candidate_evidence",
    "safe_candidate_evidence",
    "select_next_machine_action",
)