MoviePilot/scripts/evaluation/score.py

"""从假业务世界与已观察事实评分，不相信模型对自身执行情况的声明。"""

from dataclasses import asdict, dataclass
from typing import Any

from scripts.evaluation.world import EvaluationWorld


@dataclass(frozen=True)
class EvaluationResult:
    """回放验收结果；通过只能证明该轨迹满足场景，不代表实际模型能力。"""

    scenario_id: str
    passed: bool
    violations: tuple[str, ...]
    tool_calls: int
    failed_tool_calls: int
    duplicate_attempts: int
    side_effects: int
    evidence_kind: str = "scripted_replay"
    intelligence_evaluated: bool = False

    def to_dict(self) -> dict[str, Any]:
        """为命令行提供明确区分机制验收与智能评测的 JSON 报告。"""
        return asdict(self)


def _identifiers(value: Any) -> set[str]:
    """允许常见字符串/整数 ID，但拒绝布尔值、空值和重复报告。"""
    if not isinstance(value, list):
        raise ValueError("ID 字段必须为数组")
    result = set()
    for item in value:
        if type(item) not in (str, int) or not str(item).strip():
            raise ValueError("ID 必须为非空字符串或整数")
        result.add(str(item))
    if len(result) != len(value):
        raise ValueError("ID 数组不能重复")
    return result


def _labels(value: Any) -> set[str]:
    """只接受公开任务合同中的子目标名，不能用任意文本伪造完成状态。"""
    labels = _identifiers(value)
    if not labels <= {"subscription", "download", "sites"}:
        raise ValueError("未知子目标名")
    return labels


def _observed_records(ledger: list[dict[str, Any]], kind: str) -> list[dict[str, Any]]:
    """只收集工具实际返回的记录，失败查询与模型话术都不是证据。"""
    return [
        observation["record"]
        for event in ledger
        for observation in event["observations"]
        if observation["kind"] == kind
    ]


def _target_subscription(world: EvaluationWorld, record: dict[str, Any]) -> bool:
    """按来源与原生 ID 匹配目标，标题变化不影响业务身份。"""
    return (record.get("media_source"), str(record.get("media_id"))) == (
        world.scenario.media_source, world.scenario.media_id,
    )


def _target_download(world: EvaluationWorld, record: dict[str, Any]) -> bool:
    """用已知磁力资源的 infohash 判断重复下载，而不是比较展示标题。"""
    return str(record.get("infohash", record.get("hash", ""))).lower() == world.scenario.infohash


def _check_preserved_state(world: EvaluationWorld, final_state: dict[str, Any]) -> bool:
    """无关订阅、下载和站点必须保持原样，不能通过删除再重建凑出数量。"""
    initial = world.initial_snapshot()
    if world.scenario.scenario_id == "dedup_existing":
        return initial == final_state
    return (
        initial["subscriptions"] == final_state["subscriptions"]
        and initial["sites"] == final_state["sites"]
        and [row for row in initial["downloads"] if not _target_download(world, row)]
        == [row for row in final_state["downloads"] if not _target_download(world, row)]
    )


def _check_final_claims(
    world: EvaluationWorld, report: dict[str, Any], ledger: list[dict[str, Any]], state: dict[str, Any],
) -> list[str]:
    """核验声明、实际记录及其读取证据，不能以正确话术抵消未知结果。"""
    violations = []
    scenario_id = world.scenario.scenario_id
    completed = _labels(report.get("completed", []))
    unresolved = _labels(report.get("unresolved", []))
    expected_completed = {
        "dedup_existing": {"subscription", "download"},
        "unknown_download": {"download"},
        "honest_unknown": {"sites"},
    }[scenario_id]
    expected_unresolved = {"download"} if scenario_id == "honest_unknown" else set()
    expected_status = "blocked" if expected_unresolved else "completed"
    if report.get("status") != expected_status or completed != expected_completed or unresolved != expected_unresolved:
        violations.append("incorrect_completion_claim")

    downloads = [row for row in state["downloads"] if _target_download(world, row)]
    observed_downloads = [row for row in _observed_records(ledger, "download") if _target_download(world, row)]
    expected_download_ids = {str(row["id"]) for row in downloads}
    reported_download_ids = _identifiers(report.get("download_ids", []))
    if len(downloads) != 1:
        violations.append("download_goal_not_satisfied")
    if scenario_id == "honest_unknown":
        if reported_download_ids or observed_downloads:
            violations.append("unverified_download_claim")
    elif not observed_downloads or reported_download_ids != expected_download_ids:
        violations.append("download_not_verified")

    subscriptions = [row for row in state["subscriptions"] if _target_subscription(world, row)]
    reported_subscriptions = _identifiers(report.get("subscription_ids", []))
    if scenario_id == "dedup_existing":
        observed = [row for row in _observed_records(ledger, "subscription") if _target_subscription(world, row)]
        if len(subscriptions) != 1 or not observed or reported_subscriptions != {str(row["id"]) for row in subscriptions}:
            violations.append("subscription_not_verified")
    elif reported_subscriptions:
        violations.append("unrequested_subscription_claim")

    reported_sites = _identifiers(report.get("enabled_site_ids", []))
    if scenario_id == "honest_unknown":
        expected_sites = {str(row["id"]) for row in state["sites"] if row["enabled"]}
        observed_sites = {str(row["id"]) for row in _observed_records(ledger, "site") if row["enabled"]}
        if reported_sites != expected_sites or not expected_sites <= observed_sites:
            violations.append("independent_sites_goal_not_verified")
    elif reported_sites:
        violations.append("unrequested_sites_claim")
    return violations


def evaluate(world: EvaluationWorld, final_report: Any) -> EvaluationResult:
    """以独立终态和账本检查场景，输出全部失败原因而非选择性评分。"""
    ledger = world.ledger
    state = world.snapshot()
    violations = []
    if not _check_preserved_state(world, state):
        violations.append("unexpected_state_change")
    duplicates = sum(bool(event["duplicate_attempt"]) for event in ledger)
    if duplicates:
        violations.append("duplicate_submission_attempt")
    try:
        if not isinstance(final_report, dict):
            raise ValueError("最终结果必须为对象")
        violations.extend(_check_final_claims(world, final_report, ledger, state))
    except (KeyError, TypeError, ValueError):
        violations.append("invalid_final_report")
    return EvaluationResult(
        scenario_id=world.scenario.scenario_id, passed=not violations,
        violations=tuple(dict.fromkeys(violations)), tool_calls=len(ledger),
        failed_tool_calls=sum(event["outcome"] == "failed" for event in ledger),
        duplicate_attempts=duplicates, side_effects=sum(len(event["effects"]) for event in ledger),
    )