"""Pure checks adapted from A2A operation knowledge and test readback boundaries.

Reference: learning/packages/contracts/python/penguin_learning_v1/operation_chain.py
(`_references`, `graph_integrity`, `validate_operation_candidate`) and
learning/src/learning_m0/test_readback.py. No source package or database is imported.
This reduced contract covers ordered, read-only steps, not the A2A operation DTO.
"""
from __future__ import annotations

from urllib.parse import urlsplit

from .db import digest
from .learning_contracts import valid_selector


def content_value_valid(value, field: dict) -> bool:
    if not isinstance(value, dict) or set(value) != {"kind", "text", "media", "media_truncated"}:
        return False
    kind, text, media, truncated = (value[key] for key in ("kind", "text", "media", "media_truncated"))
    if not isinstance(kind, str) or kind not in {"text", "media", "mixed"} or not (text is None or isinstance(text, str) and bool(text.strip())) or not isinstance(media, list) or len(media) > field.get("max_media", 4) or type(truncated) is not bool:
        return False
    if any(not isinstance(item, dict) or set(item) != {"kind", "selector"} or not isinstance(item.get("kind"), str) or item["kind"] not in {"image", "video"} or not isinstance(item.get("selector"), str) or not valid_selector(item["selector"], relative=True) for item in media):
        return False
    if len({item["selector"] for item in media}) != len(media) or (truncated and len(media) != field.get("max_media", 4)):
        return False
    return ((kind == "text" and text is not None and not media) or (kind == "media" and text is None and bool(media)) or (kind == "mixed" and text is not None and bool(media)))


def http_url(value) -> bool:
    try:
        parsed = urlsplit(value) if isinstance(value, str) else None
        return bool(parsed and parsed.scheme in {"http", "https"} and parsed.hostname and not parsed.username and not parsed.password)
    except ValueError:
        return False


def step_evidence_references(steps: list[dict]) -> set[str]:
    values = [step.get(key) for step in steps for key in ("evidence_id", "after_evidence_id", "selector_evidence_id")]
    values.extend(reference for step in steps if isinstance(step.get("batch_evidence_ids"), list) for reference in step["batch_evidence_ids"])
    return {reference for reference in values if isinstance(reference, str) and reference}


def evidence_graph_errors(steps: list[dict], bindings: list[dict], current: dict[str, dict]) -> list[str]:
    """Check the closed set of immutable before/after evidence references."""
    errors = []
    expected = step_evidence_references(steps)
    ids = [item.get("observation_id") for item in bindings]
    if any(not isinstance(value, str) or not value for value in ids) or len(ids) != len(set(ids)):
        errors.append("冻结证据标识缺失或重复")
    if set(ids) != expected:
        errors.append("候选步骤与冻结证据引用不完整对应")
    for position, step in enumerate(steps, 1):
        if not step.get("evidence_id") or not step.get("after_evidence_id"):
            errors.append(f"第 {position} 步缺少动作前后证据")
        elif step["evidence_id"] == step["after_evidence_id"]:
            errors.append(f"第 {position} 步动作前后必须是分别采集的证据")
        batches = step.get("batch_evidence_ids")
        if step.get("batch"):
            config = step["batch"] if isinstance(step["batch"], dict) else {}
            maximum = config.get("max_batches")
            if type(maximum) is not int or not 1 <= maximum <= 5 or not isinstance(batches, list) or not 1 <= len(batches) <= maximum or any(not isinstance(item, str) or not item for item in batches) or len(set(batches)) != len(batches):
                errors.append(f"第 {position} 步缺少真实批次证据")
            else:
                for index, reference in enumerate(batches, 1):
                    data = (current.get(reference) or {}).get("data", {})
                    batch = data.get("batch") or {}
                    if data.get("source") != "action_batch" or batch.get("batch_index") != index or any(type(batch.get(key)) is not int or batch[key] < 0 for key in ("added_count", "duplicate_count")):
                        errors.append(f"第 {position} 步批次证据不完整或来源不符")
        elif batches is not None:
            errors.append(f"第 {position} 步没有批次配置却引用批次证据")
    for binding in bindings:
        reference = binding.get("observation_id")
        actual = current.get(reference)
        if actual is None:
            errors.append("冻结证据缺少同任务的原始记录")
        elif digest(actual) != binding.get("content_digest"):
            errors.append("来源证据与候选冻结摘要不一致")
    return errors


def output_assertion_checks(actual: dict, expected: dict, contracts: dict | None = None) -> list[dict]:
    """Deterministic assertions on saved outputs; this is not an independent read."""
    checks = []
    for name, expectation in expected.items():
        value = actual.get(name) if isinstance(actual, dict) else None
        contract = (contracts or {}).get(name) or {}
        if contract.get("kind") == "extract":
            fields = contract.get("fields") or []
            names = {field["name"] for field in fields}
            items = value.get("items") if isinstance(value, dict) else None
            complete_only = contract.get("complete_only") is True
            batch = contract.get("batch")
            allow_empty = contract.get("allow_empty") is True
            has_empty = isinstance(value, dict) and "empty_state" in value
            expected_keys = {"items", "count", "truncated", "source_url", "coverage"} | ({"skipped_incomplete_count"} if complete_only else set())
            if batch:
                expected_keys.add("batch_summary")
            if has_empty:
                expected_keys.add("empty_state")
            skipped = value.get("skipped_incomplete_count", 0) if isinstance(value, dict) else 0
            valid = (isinstance(value, dict) and set(value) == expected_keys
                and isinstance(items, list) and len(items) <= contract.get("limit", 50)
                and type(value.get("count")) is int and value["count"] == len(items)
                and type(value.get("truncated")) is bool
                and type(skipped) is int and skipped >= 0
                and (not skipped or value["truncated"])
                and isinstance(value.get("coverage"), str) and value["coverage"] in {"bounded", "limit_reached", "empty"}
                and http_url(value.get("source_url"))
                and ((value["coverage"] == "empty") == (not items))
                and (value["coverage"] != "limit_reached" or (value["truncated"] and len(items) == contract.get("limit", 50)))
                and all(isinstance(item, dict) and set(item) == names and all(content_value_valid(item[field["name"]], field) if field.get("source") == "content" else isinstance(item[field["name"]], str) or item[field["name"]] is None for field in fields) for item in items))
            if valid:
                valid = all(not field.get("required") or all(content_value_valid(item[field["name"]], field) if field.get("source") == "content" else isinstance(item[field["name"]], str) and item[field["name"]].strip() for item in items) for field in fields)
            if valid and any(item[field["name"]]["media_truncated"] for item in items for field in fields if field.get("source") == "content"):
                valid = value["truncated"] is True
            if valid and batch:
                summary = value.get("batch_summary")
                identities = [item.get(batch["dedupe_field"]) for item in items]
                valid = (value["truncated"] is True and isinstance(summary, dict) and set(summary) == {"batches_read", "duplicates_removed", "stop_reason"}
                    and type(summary.get("batches_read")) is int and 1 <= summary["batches_read"] <= batch["max_batches"]
                    and type(summary.get("duplicates_removed")) is int and summary["duplicates_removed"] >= 0
                    and isinstance(summary.get("stop_reason"), str) and summary["stop_reason"] in {"limit_reached", "boundary", "batch_budget", "time_budget"}
                    and (summary["stop_reason"] != "limit_reached" or len(items) == contract.get("limit", 50))
                    and all(isinstance(item, str) and item.strip() for item in identities) and len(set(identities)) == len(identities))
            if valid and has_empty:
                proof = value["empty_state"]
                expected_proof = contract.get("empty_state") or {}
                valid = (allow_empty and not batch and not items and not skipped and value["truncated"] is False
                    and isinstance(proof, dict) and set(proof) == {"confirmed", "selector", "text"} and proof.get("confirmed") is True
                    and proof.get("selector") == expected_proof.get("selector") and proof.get("text") == expected_proof.get("text"))
            if valid and allow_empty and not items and not has_empty:
                valid = False
            values = [item[field["name"]].get("text") if field.get("source") == "content" else item[field["name"]] for item in items for field in fields if field.get("source", "text") != "href"] if valid else []
            text = "\n".join(value for value in values if isinstance(value, str))
        else:
            valid = isinstance(value, list) and all(isinstance(item, dict) and isinstance(item.get("text"), str) and set(item) <= {"text", "href"} and ("href" not in item or isinstance(item["href"], str)) for item in value)
            text = "\n".join(item["text"] for item in value) if valid else ""
        checks.append({"name": f"output_schema:{name}", "passed": valid, "message": "读取输出结构符合固定契约" if valid else f"输出 {name} 不符合文本读取结构"})
        if isinstance(expectation, dict):
            if expectation == {"state": "nonempty"}:
                nonempty = valid and contract.get("kind") == "extract" and isinstance(value, dict) and type(value.get("count")) is int and value["count"] > 0
                checks.append({"name": f"output_nonempty:{name}", "passed": nonempty, "message": "结构化结果非空，不代表语义判断正确" if nonempty else "未取得有效的非空结构化结果"})
                continue
            confirmed = valid and expectation == {"state": "empty"} and contract.get("allow_empty") is True and isinstance(value, dict) and value.get("empty_state", {}).get("confirmed") is True and value.get("count") == 0
            checks.append({"name": f"output_empty:{name}", "passed": confirmed, "message": "已核对明确空态证据" if confirmed else "未取得符合预期的真实空态证据"})
            continue
        # A URL containing the keyword is not evidence that the rendered result
        # text contains it. Only the explicitly read text participates.
        checks.append({"name": f"output_present:{name}", "passed": bool(text.strip()), "message": f"输出 {name} 未读到内容" if not text.strip() else "已读到输出"})
        for item in [expectation] if isinstance(expectation, str) else expectation:
            checks.append({"name": f"output_contains:{name}", "passed": item in text, "message": "预期文本匹配" if item in text else f"输出 {name} 未包含预期文本：{item[:100]}"})
    return checks