"""Controlled PixelRAG marketplace adapter dry-run worker.

This worker turns marketplace adapter preflight receipts into deterministic
adapter dry-run receipts. It does not fetch marketplace pages, call models,
write databases, or promote candidate prices into production truth.
"""

from __future__ import annotations

import json
import os
import re
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Mapping

from services.pixelrag_crawler_integration_service import DEFAULT_ARTIFACT_MAX_AGE_HOURS
from services.pixelrag_marketplace_source_contract_adapter_preflight_service import (
    ADAPTER_PREFLIGHT_VERSION,
    DEFAULT_OUTPUT_ROOT as DEFAULT_ADAPTER_PREFLIGHT_RECEIPT_ROOT,
)
from services.pixelrag_source_contract_replay_worker_service import SOURCE_CONTRACT_VERSION


POLICY = "controlled_pixelrag_marketplace_source_contract_adapter_dry_run_v1"
ADAPTER_DRY_RUN_VERSION = "pixelrag_marketplace_adapter_dry_run_v1"
DEFAULT_LIMIT = 25
DEFAULT_OUTPUT_ROOT = os.getenv(
    "PIXELRAG_MARKETPLACE_ADAPTER_DRY_RUN_RECEIPT_ROOT",
    "/app/data/ai_automation/pixelrag_marketplace_adapter_dry_run_receipts"
    if Path("/app/data").exists()
    else "runtime_artifacts/pixelrag_marketplace_adapter_dry_run_receipts",
)

CANONICAL_OFFER_FIELDS = {
    "platform",
    "platform_product_id",
    "title",
    "url",
    "image_url",
    "price",
    "currency",
    "availability",
    "seller",
    "promotion_badges",
    "shipping_badges",
    "evidence_time",
    "source_method",
    "confidence",
}
REQUIRED_PROMOTION_GATES = {
    "public_source_boundary",
    "rate_limit_contract",
    "provenance_contract",
    "identity_matcher_replay",
    "promotion_gate",
    "embedding_signature_guard",
}


def _normalise_platforms(
    platform: str | tuple[str, ...] | list[str] | None,
) -> tuple[str, ...]:
    if isinstance(platform, str):
        value = platform.strip().lower()
        return (value,) if value else ()
    return tuple(
        str(item or "").strip().lower()
        for item in (platform or ())
        if str(item or "").strip()
    )


def _safe_segment(value: Any) -> str:
    text = str(value or "unknown").strip().lower()
    text = re.sub(r"[^a-z0-9._-]+", "-", text)
    return text.strip("-") or "unknown"


def _parse_iso_datetime(value: Any) -> datetime | None:
    if not value:
        return None
    try:
        return datetime.fromisoformat(str(value).replace("Z", "+00:00"))
    except ValueError:
        return None


def _receipt_candidates(
    root: Path,
    *,
    platforms: tuple[str, ...],
    limit: int,
) -> list[Path]:
    if not root.exists():
        return []
    candidates: list[Path] = []
    if platforms:
        for platform in platforms:
            candidates.extend(
                (root / platform).glob("*/marketplace_adapter_preflight_receipt.json")
            )
    else:
        candidates.extend(root.glob("*/*/marketplace_adapter_preflight_receipt.json"))
    return sorted(candidates, key=lambda path: path.stat().st_mtime, reverse=True)[:limit]


def _load_receipt(path: Path) -> tuple[dict[str, Any], list[str]]:
    try:
        return json.loads(path.read_text(encoding="utf-8")), []
    except (OSError, json.JSONDecodeError) as exc:
        return {}, [str(exc)[:300]]


def _field_source_path(field: str) -> str:
    return {
        "platform": "literal.platform",
        "platform_product_id": "source.product_id_or_url_identity",
        "title": "source.title_text",
        "url": "source.public_product_url",
        "image_url": "source.public_image_url",
        "price": "source.price_text_or_amount",
        "currency": "source.currency_text_or_locale_default",
        "availability": "source.availability_text",
        "seller": "source.seller_text",
        "promotion_badges": "source.promotion_badges[]",
        "shipping_badges": "source.shipping_badges[]",
        "evidence_time": "receipt.generated_at",
        "source_method": "source_contract.source_method",
        "confidence": "adapter.confidence_contract",
    }.get(field, f"source.{field}")


def _field_contract(field: str) -> dict[str, Any]:
    return {
        "field": field,
        "mapped": True,
        "required": field in CANONICAL_OFFER_FIELDS,
        "source_path": _field_source_path(field),
        "value_materialized": False,
        "write_target": None,
        "requires_evidence": field in {"price", "title", "url", "evidence_time"},
        "requires_identity_match": field in {"platform_product_id", "title", "url"},
        "requires_promotion_gate": field in {
            "price",
            "promotion_badges",
            "shipping_badges",
            "confidence",
        },
    }


def _preflight_payload(receipt: Mapping[str, Any]) -> Mapping[str, Any]:
    payload = receipt.get("adapter_preflight")
    return payload if isinstance(payload, Mapping) else {}


def _promotion_boundary(receipt: Mapping[str, Any]) -> Mapping[str, Any]:
    payload = receipt.get("promotion_boundary")
    return payload if isinstance(payload, Mapping) else {}


def _expected_fields(preflight: Mapping[str, Any]) -> list[str]:
    fields = [str(item) for item in list(preflight.get("expected_offer_fields") or [])]
    seen: set[str] = set()
    ordered: list[str] = []
    for field in fields + sorted(CANONICAL_OFFER_FIELDS):
        if field and field not in seen:
            ordered.append(field)
            seen.add(field)
    return ordered


def _adapter_dry_run_payload(
    receipt: Mapping[str, Any],
    *,
    receipt_path: Path,
) -> dict[str, Any]:
    preflight = _preflight_payload(receipt)
    fields = _expected_fields(preflight)
    source_count = int(preflight.get("source_count") or 0)
    output_count = max(0, min(source_count, 50))
    field_contracts = [_field_contract(field) for field in fields]
    return {
        "adapter_dry_run_version": ADAPTER_DRY_RUN_VERSION,
        "adapter_preflight_version": preflight.get("adapter_preflight_version"),
        "source_contract_version": preflight.get("source_contract_version"),
        "adapter_code": preflight.get("adapter_code") or receipt.get("adapter_code"),
        "contract_id": preflight.get("contract_id"),
        "barrier_type": preflight.get("barrier_type") or receipt.get("barrier_type"),
        "capture_runtime_unavailable": bool(preflight.get("capture_runtime_unavailable")),
        "input_preflight_receipt_path": str(receipt_path),
        "source_contract_replay_receipt_path": receipt.get(
            "source_contract_replay_receipt_path"
        ),
        "source_worker_receipt_path": receipt.get("source_worker_receipt_path"),
        "source_receipt_path": receipt.get("source_receipt_path"),
        "dry_run_execution_mode": "deterministic_mapping_only_no_network_no_db",
        "source_count": source_count,
        "dry_run_output_contract_count": output_count,
        "expected_offer_fields": fields,
        "field_contract_count": len(field_contracts),
        "field_contracts": field_contracts,
        "offer_candidate_contracts": [
            {
                "candidate_index": index,
                "candidate_id": (
                    f"{receipt.get('platform') or 'unknown'}:"
                    f"{receipt.get('manifest_id') or 'unknown'}:{index}"
                ),
                "field_count": len(field_contracts),
                "field_contracts_materialized": True,
                "values_materialized": False,
                "ready_for_identity_matcher_replay": True,
                "ready_for_price_table_write": False,
                "price_write_blocked_until_promotion_gate": True,
            }
            for index in range(output_count)
        ],
        "required_before_data_promotion": list(
            preflight.get("required_before_data_promotion") or []
        ),
        "allowed_next_step": "run_marketplace_adapter_identity_matcher_replay",
    }


def _adapter_dry_run_checks(
    receipt: Mapping[str, Any],
    *,
    stale: bool,
    errors: list[str],
    dry_run_payload: Mapping[str, Any],
) -> dict[str, bool]:
    preflight = _preflight_payload(receipt)
    boundary = _promotion_boundary(receipt)
    preflight_checks = receipt.get("adapter_preflight_checks")
    if not isinstance(preflight_checks, Mapping):
        preflight_checks = {}
    required_before = set(
        str(item) for item in list(preflight.get("required_before_data_promotion") or [])
    )
    expected_fields = set(str(item) for item in list(preflight.get("expected_offer_fields") or []))
    return {
        "receipt_parse_ok": not errors,
        "receipt_fresh": not stale,
        "adapter_preflight_ready": (
            receipt.get("worker_status") == "executed_adapter_preflight_ready"
        ),
        "adapter_preflight_version_supported": (
            preflight.get("adapter_preflight_version") == ADAPTER_PREFLIGHT_VERSION
        ),
        "source_contract_version_supported": (
            preflight.get("source_contract_version") == SOURCE_CONTRACT_VERSION
        ),
        "adapter_preflight_checks_all_passed": (
            int(receipt.get("adapter_preflight_check_pass_count") or 0)
            == int(receipt.get("adapter_preflight_check_count") or -1)
            and int(receipt.get("adapter_preflight_check_count") or 0) > 0
        ),
        "blocked_page_not_product_data": bool(
            preflight_checks.get("blocked_page_not_product_data")
        ),
        "adapter_code_present": bool(preflight.get("adapter_code") or receipt.get("adapter_code")),
        "source_count_positive": int(preflight.get("source_count") or 0) > 0,
        "canonical_offer_fields_present": CANONICAL_OFFER_FIELDS.issubset(expected_fields),
        "promotion_gates_required": REQUIRED_PROMOTION_GATES.issubset(required_before),
        "direct_price_write_disallowed": not bool(
            boundary.get("direct_price_write_allowed")
        ),
        "direct_ai_insights_write_disallowed": not bool(
            boundary.get("direct_ai_insights_write_allowed")
        ),
        "database_write_disallowed": not bool(receipt.get("writes_database"))
        and int(receipt.get("writes_database_count") or 0) == 0,
        "network_call_disallowed": not bool(receipt.get("network_call_performed")),
        "model_call_disallowed": not bool(receipt.get("model_call_performed")),
        "dry_run_schema_generated": bool(dry_run_payload.get("field_contracts")),
        "dry_run_output_contract_generated": int(
            dry_run_payload.get("dry_run_output_contract_count") or 0
        )
        > 0,
        "price_write_blocked_until_promotion_gate": all(
            bool(item.get("price_write_blocked_until_promotion_gate"))
            for item in list(dry_run_payload.get("offer_candidate_contracts") or [])
        ),
    }


def _worker_item(
    receipt_path: Path,
    *,
    now: datetime,
    max_age_hours: int,
    execute: bool,
) -> dict[str, Any]:
    receipt, errors = _load_receipt(receipt_path)
    generated_at = _parse_iso_datetime(receipt.get("generated_at"))
    if generated_at is None:
        try:
            generated_at = datetime.fromtimestamp(receipt_path.stat().st_mtime, tz=timezone.utc)
        except OSError:
            generated_at = None
    age_hours = ((now - generated_at).total_seconds() / 3600) if generated_at else None
    stale = age_hours is None or age_hours > max_age_hours
    dry_run_payload = _adapter_dry_run_payload(receipt, receipt_path=receipt_path)
    checks = _adapter_dry_run_checks(
        receipt,
        stale=stale,
        errors=errors,
        dry_run_payload=dry_run_payload,
    )
    check_count = len(checks)
    pass_count = sum(1 for passed in checks.values() if passed)
    ready = pass_count == check_count
    platform = str(receipt.get("platform") or receipt_path.parent.parent.name).strip().lower()
    manifest_id = str(receipt.get("manifest_id") or receipt_path.parent.name).strip()
    status = (
        "executed_marketplace_adapter_dry_run_ready"
        if execute and ready
        else (
            "dry_run_ready_for_marketplace_adapter_dry_run"
            if ready
            else "skipped_marketplace_adapter_dry_run_guard_failed"
        )
    )
    return {
        "worker_status": status,
        "platform": platform,
        "manifest_id": manifest_id,
        "source_type": "marketplace_adapter_preflight_receipt",
        "adapter_preflight_receipt_path": str(receipt_path),
        "source_contract_replay_receipt_path": receipt.get(
            "source_contract_replay_receipt_path"
        ),
        "source_worker_receipt_path": receipt.get("source_worker_receipt_path"),
        "source_receipt_path": receipt.get("source_receipt_path"),
        "adapter_code": dry_run_payload.get("adapter_code"),
        "adapter_dry_run_status": "ready" if ready else "blocked",
        "ready_for_execution": ready,
        "execute": bool(execute),
        "generated_at": generated_at.isoformat() if generated_at else None,
        "age_hours": round(age_hours, 3) if age_hours is not None else None,
        "stale": stale,
        "network_call_performed": False,
        "model_call_performed": False,
        "artifact_write_performed": False,
        "writes_database": False,
        "writes_database_count": 0,
        "primary_human_gate_count": 0,
        "adapter_dry_run_checks": checks,
        "adapter_dry_run_check_count": check_count,
        "adapter_dry_run_check_pass_count": pass_count,
        "adapter_dry_run": dry_run_payload,
        "promotion_boundary": {
            "direct_ai_insights_write_allowed": False,
            "direct_price_write_allowed": False,
            "adapter_dry_run_only": True,
            "requires_public_source_boundary": True,
            "requires_rate_limit_contract": True,
            "requires_provenance_contract": True,
            "requires_identity_matcher_replay": True,
            "requires_promotion_gate": True,
            "requires_embedding_signature_guard": True,
        },
        "next_machine_action": (
            "run_marketplace_adapter_identity_matcher_replay"
            if execute and ready
            else (
                "run_marketplace_source_contract_adapter_dry_run_execute"
                if ready
                else "repair_marketplace_source_contract_adapter_dry_run_inputs"
            )
        ),
    }


def _write_dry_run_receipt(*, output_root: Path, item: Mapping[str, Any]) -> str:
    target = (
        output_root
        / _safe_segment(item.get("platform"))
        / _safe_segment(item.get("manifest_id"))
        / "marketplace_adapter_dry_run_receipt.json"
    )
    target.parent.mkdir(parents=True, exist_ok=True)
    receipt = dict(item)
    receipt["artifact_write_performed"] = True
    receipt["receipt_path"] = str(target)
    receipt["generated_at"] = datetime.now(timezone.utc).isoformat()
    receipt["policy"] = POLICY
    target.write_text(
        json.dumps(receipt, ensure_ascii=False, indent=2, sort_keys=True),
        encoding="utf-8",
    )
    return str(target)


def run_pixelrag_marketplace_source_contract_adapter_dry_run(
    *,
    adapter_preflight_receipt_root: str | Path | None = None,
    output_root: str | Path | None = None,
    platform: str | tuple[str, ...] | list[str] | None = None,
    max_age_hours: int | None = None,
    limit: int | None = None,
    execute: bool = False,
    write_receipt: bool = False,
) -> dict[str, Any]:
    """Run or dry-run marketplace adapter deterministic mapping."""
    source_root = Path(adapter_preflight_receipt_root or DEFAULT_ADAPTER_PREFLIGHT_RECEIPT_ROOT)
    output = Path(output_root or DEFAULT_OUTPUT_ROOT)
    platforms = _normalise_platforms(platform)
    max_age = max(1, int(max_age_hours or DEFAULT_ARTIFACT_MAX_AGE_HOURS))
    item_limit = max(1, min(int(limit or DEFAULT_LIMIT), 250))
    now = datetime.now(timezone.utc)
    receipt_paths = _receipt_candidates(source_root, platforms=platforms, limit=item_limit)
    worker_items: list[dict[str, Any]] = []
    for receipt_path in receipt_paths:
        item = _worker_item(
            receipt_path,
            now=now,
            max_age_hours=max_age,
            execute=execute,
        )
        if execute and write_receipt and item.get("ready_for_execution"):
            item["receipt_path"] = _write_dry_run_receipt(output_root=output, item=item)
            item["artifact_write_performed"] = True
        worker_items.append(item)

    ready_count = sum(1 for item in worker_items if item.get("ready_for_execution"))
    blocked_count = len(worker_items) - ready_count
    dry_run_count = sum(
        1 for item in worker_items if str(item.get("worker_status") or "").startswith("dry_run_")
    )
    executed_count = sum(
        1 for item in worker_items if str(item.get("worker_status") or "").startswith("executed_")
    )
    receipt_written_count = sum(1 for item in worker_items if item.get("receipt_path"))
    guard_failed_count = sum(
        1 for item in worker_items if item.get("adapter_dry_run_status") != "ready"
    )
    capture_runtime_gap_count = sum(
        1
        for item in worker_items
        if (item.get("adapter_dry_run") or {}).get("capture_runtime_unavailable")
    )
    field_contract_count = sum(
        int((item.get("adapter_dry_run") or {}).get("field_contract_count") or 0)
        for item in worker_items
    )
    output_contract_count = sum(
        int((item.get("adapter_dry_run") or {}).get("dry_run_output_contract_count") or 0)
        for item in worker_items
    )

    if not worker_items:
        status = "warning"
    elif guard_failed_count:
        status = "warning"
    else:
        status = "ok"

    if not worker_items:
        next_action = "run_marketplace_source_contract_adapter_preflight"
    elif not execute and ready_count:
        next_action = "run_marketplace_source_contract_adapter_dry_run_execute"
    elif execute and receipt_written_count:
        next_action = "run_marketplace_adapter_identity_matcher_replay"
    elif guard_failed_count:
        next_action = "repair_marketplace_source_contract_adapter_dry_run_inputs"
    else:
        next_action = "refresh_marketplace_source_contract_adapter_dry_run_candidates"

    summary = {
        "candidate_count": len(worker_items),
        "ready_count": ready_count,
        "blocked_count": blocked_count,
        "dry_run_count": dry_run_count,
        "executed_count": executed_count,
        "receipt_written_count": receipt_written_count,
        "guard_failed_count": guard_failed_count,
        "capture_runtime_gap_count": capture_runtime_gap_count,
        "field_contract_count": field_contract_count,
        "dry_run_output_contract_count": output_contract_count,
        "platforms": sorted({str(item.get("platform") or "unknown") for item in worker_items}),
        "network_call_performed": False,
        "model_call_performed": False,
        "artifact_write_performed": bool(receipt_written_count),
        "writes_database_count": 0,
        "primary_human_gate_count": 0,
    }
    return {
        "success": status != "critical",
        "policy": POLICY,
        "status": status,
        "generated_at": now.isoformat(),
        "adapter_dry_run_version": ADAPTER_DRY_RUN_VERSION,
        "adapter_preflight_receipt_root": str(source_root),
        "output_root": str(output),
        "platform_filter": list(platforms),
        "max_age_hours": max_age,
        "limit": item_limit,
        "execute": bool(execute),
        "write_receipt": bool(write_receipt and execute),
        "summary": summary,
        "worker_items": worker_items,
        "controlled_apply": {
            "network_call": False,
            "model_call": False,
            "artifact_write": bool(receipt_written_count),
            "db_write": False,
            "writes_database": False,
            "writes_database_count": 0,
            "secret_read": False,
            "raw_cookie_or_session_read": False,
            "credentialed_session_allowed": False,
            "login_allowed": False,
            "production_price_write": False,
            "primary_human_gate_count": 0,
        },
        "promotion_boundary": {
            "writes_ai_insights": False,
            "writes_price_tables": False,
            "adapter_dry_run_only": True,
            "requires_public_source_boundary": True,
            "requires_rate_limit_contract": True,
            "requires_provenance_contract": True,
            "requires_identity_matcher_replay": True,
            "requires_promotion_gate": True,
            "requires_embedding_signature_guard": True,
        },
        "next_machine_action": next_action,
    }


__all__ = [
    "ADAPTER_DRY_RUN_VERSION",
    "DEFAULT_OUTPUT_ROOT",
    "POLICY",
    "run_pixelrag_marketplace_source_contract_adapter_dry_run",
]
