content-find-agent-new/scripts/validate_v4_config_contract.py

"""Validate V4 M0 config contract without switching V3 production rule packs."""

from __future__ import annotations

import json
import sys
from pathlib import Path
from typing import Any


ROOT = Path(__file__).resolve().parents[1]
DATA_DIR = ROOT / "tech_documents/数据接口与来源"
RULE_PACK_PATH = ROOT / "product_documents/规则包/douyin_rule_packs.v1.json"
WALK_STRATEGY_PATH = ROOT / "product_documents/抖音游走策略/douyin_walk_strategy.v1.json"
WALK_EDGE_CATALOG_PATH = ROOT / "tech_documents/数据接口与来源/walk_edge_catalog.json"
LEGACY_FIELD_BLOCKLIST = {
    "fit_senior_50plus",
    "fit_confidence",
    "relevance_score",
    "platform_heat",
    "age_50_plus_level",
}
ENDPOINT_STATUSES = {
    "verified",
    "verified_unstable",
    "blocked",
    "source_only",
    "missing",
}
M2_PLATFORM_PROFILES = {"douyin", "kuaishou", "shipinhao"}
PROFILE_EDGE_STATUSES = {"supported", "blocked", "no_interface", "planned"}
PROFILE_ENDPOINT_STATUSES = {"verified", "verified_unstable", "blocked", "source_only", "missing"}
OBSERVABLE_FIELDS = {
    "statistics.digg_count",
    "statistics.comment_count",
    "statistics.share_count",
    "statistics.collect_count",
    "statistics.play_count",
}
MISSING_OBSERVABLE_TYPES = {"natural_platform_missing", "runtime_missing"}
M4_WALK_GATE_EDGES = {"hashtag_to_query", "author_to_works"}
M4_WALK_GATE_RAW_FIELDS = {"decision_id", "allow_walk", "allow_walk_reason", "walk_gate_snapshot"}
V4_BASE_ACTIVE_DIMENSIONS = {"query_relevance", "platform_performance"}
V4_REQUIRED_SCORE_WEIGHT_PROFILES = {
    "default_two_dimension",
    "douyin_fifty_plus_ok",
    "douyin_fifty_plus_not_attempted",
}


def main() -> int:
    findings = validate_v4_config_contract(ROOT)
    payload = {"status": "fail" if findings else "pass", "findings": findings}
    print(json.dumps(payload, ensure_ascii=False, indent=2))
    return 1 if findings else 0


def validate_v4_config_contract(root: Path = ROOT) -> list[dict[str, str]]:
    data_dir = root / "tech_documents/数据接口与来源"
    findings: list[dict[str, str]] = []

    walk_graph = _load_json(data_dir / "walk_graph.json", findings)
    graph_edges: set[str] = set()
    if walk_graph:
        _check_no_legacy_fields(findings, walk_graph, "walk_graph.json")
        _check_value(findings, "walk_graph_schema", walk_graph.get("schema_version"), "walk_graph.v2")
        _check_count(findings, "walk_graph_nodes", "walk_graph.nodes", walk_graph.get("nodes"), 8)
        _check_count(findings, "walk_graph_edges", "walk_graph.edges", walk_graph.get("edges"), 9)
        graph_edges = {
            edge.get("edge_id")
            for edge in walk_graph.get("edges", [])
            if isinstance(edge, dict) and edge.get("edge_id")
        }

    walk_edge_catalog = _load_json(data_dir / "walk_edge_catalog.json", findings)
    catalog_edges: set[str] = set()
    if walk_edge_catalog:
        _check_no_legacy_fields(findings, walk_edge_catalog, "walk_edge_catalog.json")
        _check_value(
            findings,
            "walk_edge_catalog_schema",
            walk_edge_catalog.get("schema_version"),
            "walk_edge_catalog.v1",
        )
        rows = walk_edge_catalog.get("walk_edge_catalog")
        _check_count(findings, "walk_edge_catalog_edges", "walk_edge_catalog", rows, 9)
        catalog_edges = {
            row.get("edge_id")
            for row in rows or []
            if isinstance(row, dict) and row.get("edge_id")
        }
        if walk_graph and graph_edges != catalog_edges:
            _fail(
                findings,
                "walk_graph_catalog_mismatch",
                f"walk_graph edges and walk_edge_catalog differ: graph_only={sorted(graph_edges - catalog_edges)}, catalog_only={sorted(catalog_edges - graph_edges)}",
            )

    walk_policy = _load_json(data_dir / "walk_policy.json", findings)
    if walk_policy:
        _check_no_legacy_fields(findings, walk_policy, "walk_policy.json")
        _check_value(findings, "walk_policy_schema", walk_policy.get("schema_version"), "walk_policy.v1")
        for key in ["global", "edge_budgets", "dedup", "edge_permissions", "v4_walk_gate"]:
            if key not in walk_policy:
                _fail(findings, "walk_policy_missing_key", f"walk_policy.json missing {key}")
        _check_v4_walk_gate_config(findings, walk_policy.get("v4_walk_gate"), "walk_policy.v4_walk_gate")
        if walk_graph:
            graph_edges = {
                edge.get("edge_id")
                for edge in walk_graph.get("edges", [])
                if isinstance(edge, dict)
            }
            missing = sorted(M4_WALK_GATE_EDGES - graph_edges)
            if missing:
                _fail(findings, "v4_walk_gate_graph_edges_missing", f"walk_graph missing M4 gate edges: {missing}")

    endpoint_registry = _load_json(data_dir / "crawler_endpoints.registry.json", findings)
    if endpoint_registry:
        _check_no_legacy_fields(findings, endpoint_registry, "crawler_endpoints.registry.json")
        _check_value(
            findings,
            "crawler_endpoints_schema",
            endpoint_registry.get("registry_version"),
            "crawler_endpoints.v1",
        )
        endpoints = endpoint_registry.get("endpoints")
        if not isinstance(endpoints, list):
            _fail(findings, "crawler_endpoints_invalid", "endpoints must be a list")
        else:
            _check_count(findings, "crawler_endpoints_count", "endpoints", endpoints, 27)
            for endpoint in endpoints:
                if not isinstance(endpoint, dict):
                    _fail(findings, "crawler_endpoint_invalid", "endpoint row must be object")
                    continue
                for key in [
                    "platform",
                    "source_id",
                    "status",
                    "table_or_endpoint",
                    "input_fields",
                    "output_fields",
                ]:
                    if key not in endpoint:
                        _fail(
                            findings,
                            "crawler_endpoint_missing_key",
                            f"{endpoint.get('source_id')} missing {key}",
                        )
                status = endpoint.get("status")
                if not isinstance(status, (str, list)):
                    _fail(
                        findings,
                        "crawler_endpoint_status_invalid",
                        f"{endpoint.get('source_id')} status must be string or list",
                    )
                    continue
                statuses = status if isinstance(status, list) else [status]
                invalid_statuses = [item for item in statuses if item not in ENDPOINT_STATUSES]
                if invalid_statuses:
                    _fail(
                        findings,
                        "crawler_endpoint_status_unknown",
                        f"{endpoint.get('source_id')} unknown status: {invalid_statuses}",
                    )

    field_map = _load_json(data_dir / "跨平台字段映射.json", findings)
    if field_map:
        _check_no_legacy_fields(findings, field_map, "跨平台字段映射.json")
        _check_value(
            findings,
            "field_map_schema",
            field_map.get("schema_version"),
            "cross_platform_field_map.v1",
        )
        if not isinstance(field_map.get("mappings"), dict):
            _fail(findings, "field_map_mappings_invalid", "mappings must be an object")

    for profile_path in sorted((data_dir / "platform_profiles").glob("*.json")):
        profile = _load_json(profile_path, findings)
        if not profile:
            continue
        _check_no_legacy_fields(findings, profile, profile_path.name)
        _check_value(
            findings,
            "platform_profile_schema",
            profile.get("schema_version"),
            "platform_profile.v1",
            label=profile_path.name,
        )
        for key in ["platform", "status", "runtime", "endpoints", "edges"]:
            if key not in profile:
                _fail(findings, "platform_profile_missing_key", f"{profile_path.name} missing {key}")
        _check_platform_profile_edges(findings, profile, profile_path.name, graph_edges or catalog_edges)
        if profile.get("platform") in M2_PLATFORM_PROFILES:
            _check_m2_platform_profile(findings, profile, profile_path.name)

    rule_pack_pkg = _load_json(RULE_PACK_PATH, findings)
    if rule_pack_pkg:
        _check_v4_rule_pack_contract(findings, rule_pack_pkg)

    walk_strategy = _load_json(WALK_STRATEGY_PATH, findings)
    if walk_strategy:
        _check_no_legacy_fields(findings, walk_strategy, "douyin_walk_strategy.v1.json")
        _check_v4_walk_strategy_contract(findings, walk_strategy, catalog_edges or graph_edges)

    return findings


def assert_no_v4_legacy_fields(value: Any, label: str = "v4_contract") -> list[str]:
    paths: list[str] = []
    _collect_legacy_paths(value, label, paths)
    return paths


def _check_no_legacy_fields(
    findings: list[dict[str, str]],
    value: Any,
    label: str,
) -> None:
    paths = assert_no_v4_legacy_fields(value, label)
    if paths:
        _fail(
            findings,
            "v4_legacy_field_present",
            f"{label} contains legacy fields: {', '.join(paths[:5])}",
        )


def _collect_legacy_paths(value: Any, prefix: str, paths: list[str]) -> None:
    if isinstance(value, dict):
        for key, child in value.items():
            child_path = f"{prefix}.{key}"
            if key in LEGACY_FIELD_BLOCKLIST:
                paths.append(child_path)
            _collect_legacy_paths(child, child_path, paths)
    elif isinstance(value, list):
        for index, child in enumerate(value):
            _collect_legacy_paths(child, f"{prefix}[{index}]", paths)
    elif isinstance(value, str):
        if _string_has_legacy_field(value):
            paths.append(prefix)


def _string_has_legacy_field(value: str) -> bool:
    normalized = value.replace("[", ".").replace("]", ".").replace("/", ".")
    parts = [part.strip() for part in normalized.split(".")]
    return any(part in LEGACY_FIELD_BLOCKLIST for part in parts)


def _load_json(path: Path, findings: list[dict[str, str]]) -> dict[str, Any] | None:
    try:
        return json.loads(path.read_text(encoding="utf-8"))
    except FileNotFoundError:
        _fail(findings, "file_missing", f"missing file: {path.relative_to(ROOT)}")
    except json.JSONDecodeError as exc:
        _fail(findings, "json_parse_failed", f"{path.relative_to(ROOT)} cannot parse: {exc}")
    return None


def _check_value(
    findings: list[dict[str, str]],
    check_id: str,
    actual: Any,
    expected: Any,
    *,
    label: str = "",
) -> None:
    if actual != expected:
        target = f"{label} " if label else ""
        _fail(findings, check_id, f"{target}expected {expected}, got {actual}")


def _check_count(
    findings: list[dict[str, str]],
    check_id: str,
    label: str,
    value: Any,
    expected: int,
) -> None:
    if not isinstance(value, list) or len(value) != expected:
        actual = len(value) if isinstance(value, list) else None
        _fail(findings, check_id, f"{label} expected {expected}, got {actual}")


def _check_platform_profile_edges(
    findings: list[dict[str, str]],
    profile: dict[str, Any],
    label: str,
    edge_ids: set[str],
) -> None:
    edges = profile.get("edges")
    if not isinstance(edges, dict):
        _fail(findings, "platform_profile_edges_invalid", f"{label} edges must be an object")
        return
    for edge_id, edge in edges.items():
        if edge_ids and edge_id not in edge_ids:
            _fail(findings, "platform_profile_edge_unknown", f"{label}.{edge_id} unknown edge")
        if not isinstance(edge, dict):
            _fail(findings, "platform_profile_edge_invalid", f"{label}.{edge_id} must be object")
            continue
        status = edge.get("status")
        if status not in PROFILE_EDGE_STATUSES:
            _fail(
                findings,
                "platform_profile_edge_status_unknown",
                f"{label}.{edge_id} status must be supported/blocked/no_interface/planned, got {status}",
            )


def _check_m2_platform_profile(
    findings: list[dict[str, str]],
    profile: dict[str, Any],
    label: str,
) -> None:
    platform = str(profile.get("platform") or "")

    endpoints = profile.get("endpoints")
    if not isinstance(endpoints, dict):
        _fail(findings, "platform_profile_endpoints_invalid", f"{label} endpoints must be an object")
    else:
        for endpoint_id, endpoint in endpoints.items():
            if not isinstance(endpoint, dict):
                _fail(
                    findings,
                    "platform_profile_endpoint_invalid",
                    f"{label}.{endpoint_id} must be object",
                )
                continue
            status = endpoint.get("status")
            if status is not None and status not in PROFILE_ENDPOINT_STATUSES:
                _fail(
                    findings,
                    "platform_profile_endpoint_status_unknown",
                    f"{label}.{endpoint_id} endpoint status must be stable enum, got {status}",
                )

    _check_observable_contract(findings, profile, label)
    _check_m2_platform_specifics(findings, profile, platform, label)


def _check_observable_contract(
    findings: list[dict[str, str]],
    profile: dict[str, Any],
    label: str,
) -> None:
    observable_fields = profile.get("observable_fields")
    missing_fields = profile.get("missing_observable_fields")
    if not isinstance(observable_fields, list) or not observable_fields:
        _fail(findings, "observable_fields_invalid", f"{label} observable_fields must be non-empty list")
    else:
        for item in observable_fields:
            if not isinstance(item, dict):
                _fail(findings, "observable_field_invalid", f"{label} observable field must be object")
                continue
            field = item.get("field")
            if field not in OBSERVABLE_FIELDS:
                _fail(
                    findings,
                    "observable_field_unknown",
                    f"{label} observable field unknown: {field}",
                )
            if item.get("availability") != "supported":
                _fail(
                    findings,
                    "observable_field_availability_invalid",
                    f"{label} {field} availability must be supported",
                )

    if not isinstance(missing_fields, list):
        _fail(
            findings,
            "missing_observable_fields_invalid",
            f"{label} missing_observable_fields must be a list",
        )
        return

    seen_observable = {
        item.get("field")
        for item in observable_fields
        if isinstance(item, dict)
    } if isinstance(observable_fields, list) else set()
    seen_missing = set()
    for item in missing_fields:
        if not isinstance(item, dict):
            _fail(findings, "missing_observable_field_invalid", f"{label} missing field must be object")
            continue
        field = item.get("field")
        seen_missing.add(field)
        if field not in OBSERVABLE_FIELDS:
            _fail(findings, "missing_observable_field_unknown", f"{label} missing field unknown: {field}")
        missing_type = item.get("missing_type")
        if missing_type not in MISSING_OBSERVABLE_TYPES:
            _fail(
                findings,
                "missing_observable_type_unknown",
                f"{label} {field} missing_type must be natural_platform_missing/runtime_missing",
            )

    overlap = sorted(seen_observable & seen_missing)
    if overlap:
        _fail(
            findings,
            "observable_field_conflict",
            f"{label} fields cannot be both observable and missing: {overlap}",
        )
    uncovered = sorted(OBSERVABLE_FIELDS - seen_observable - seen_missing)
    if uncovered:
        _fail(
            findings,
            "observable_field_uncovered",
            f"{label} observable contract missing fields: {uncovered}",
        )


def _check_m2_platform_specifics(
    findings: list[dict[str, str]],
    profile: dict[str, Any],
    platform: str,
    label: str,
) -> None:
    edges = profile.get("edges") if isinstance(profile.get("edges"), dict) else {}
    endpoints = profile.get("endpoints") if isinstance(profile.get("endpoints"), dict) else {}

    if platform == "kuaishou":
        _check_nested_status(findings, label, edges, "author_to_works", "supported")
        _check_nested_status(findings, label, edges, "author_work_to_content", "supported")
    if platform == "shipinhao":
        _check_nested_status(findings, label, endpoints, "account_info", "blocked")
        _check_nested_status(findings, label, edges, "author_to_works", "blocked")
        _check_nested_status(findings, label, edges, "author_work_to_content", "blocked")


def _check_v4_walk_gate_config(
    findings: list[dict[str, str]],
    gate: Any,
    label: str,
) -> None:
    if not isinstance(gate, dict):
        _fail(findings, "v4_walk_gate_invalid", f"{label} must be an object")
        return
    if gate.get("requires_allow_walk") is not True:
        _fail(findings, "v4_walk_gate_requires_allow_walk", f"{label}.requires_allow_walk must be true")
    if gate.get("source_field") != "rule_decisions.jsonl[].decision_replay_data.allow_walk":
        _fail(findings, "v4_walk_gate_source_field", f"{label}.source_field is invalid")
    if gate.get("deny_reason_code") != "v4_allow_walk_denied":
        _fail(findings, "v4_walk_gate_deny_reason", f"{label}.deny_reason_code is invalid")
    if set(gate.get("applies_to_edges") or []) != M4_WALK_GATE_EDGES:
        _fail(findings, "v4_walk_gate_edges", f"{label}.applies_to_edges must cover M4 expansion edges")
    raw_fields = set(gate.get("raw_payload_fields") or [])
    if not M4_WALK_GATE_RAW_FIELDS <= raw_fields:
        _fail(findings, "v4_walk_gate_raw_fields", f"{label}.raw_payload_fields missing required fields")


def _check_v4_walk_strategy_contract(
    findings: list[dict[str, str]],
    strategy: dict[str, Any],
    edge_ids: set[str],
) -> None:
    for binding in strategy.get("walk_rule_pack_binding") or []:
        if isinstance(binding, dict) and edge_ids and binding.get("edge_id") not in edge_ids:
            _fail(
                findings,
                "v4_walk_strategy_edge_unknown",
                f"walk_rule_pack_binding unknown edge_id: {binding.get('edge_id')}",
            )
    rows = strategy.get("v4_walk_gate")
    if not isinstance(rows, list) or not rows:
        _fail(findings, "v4_walk_strategy_gate_missing", "douyin_walk_strategy.v1.json missing v4_walk_gate")
        return
    by_id = {row.get("gate_id"): row for row in rows if isinstance(row, dict)}
    gate = by_id.get("allow_walk_required")
    if not gate:
        _fail(findings, "v4_walk_strategy_gate_missing", "allow_walk_required gate missing")
        return
    _check_v4_walk_gate_config(findings, gate, "douyin_walk_strategy.v4_walk_gate.allow_walk_required")
    for edge_id in gate.get("applies_to_edges") or []:
        if edge_ids and edge_id not in edge_ids:
            _fail(findings, "v4_walk_strategy_gate_edge_unknown", f"v4_walk_gate unknown edge_id: {edge_id}")


def _check_nested_status(
    findings: list[dict[str, str]],
    label: str,
    section: dict[str, Any],
    key: str,
    expected: str,
) -> None:
    value = section.get(key)
    actual = value.get("status") if isinstance(value, dict) else None
    if actual != expected:
        _fail(
            findings,
            "platform_profile_status_mismatch",
            f"{label}.{key} expected status {expected}, got {actual}",
        )


def _check_v4_rule_pack_contract(
    findings: list[dict[str, str]],
    pkg: dict[str, Any],
) -> None:
    strategy_version = (pkg.get("strategy_binding") or {}).get("strategy_version")
    for dispatch in pkg.get("rule_pack_dispatch", []):
        if dispatch.get("dispatch_enabled") and dispatch.get("strategy_version") == "V4":
            if dispatch.get("rule_pack_version") != "4.0.0":
                _fail(
                    findings,
                    "v4_rule_pack_version_invalid",
                    f"{dispatch.get('dispatch_id')} V4 dispatch must use rule_pack_version 4.0.0",
                )
    for pack in pkg.get("rule_packs", []):
        scorecard = pack.get("scorecard") or {}
        is_v4 = (
            strategy_version == "V4"
            or pack.get("version") == "4.0.0"
            or scorecard.get("schema_version") == "v4_scorecard.v1"
        )
        if not is_v4:
            continue
        _check_no_legacy_fields(findings, pack, f"rule_pack:{pack.get('rule_pack_id')}")
        if scorecard.get("schema_version") != "v4_scorecard.v1":
            _fail(
                findings,
                "v4_scorecard_schema_invalid",
                f"{pack.get('rule_pack_id')} scorecard.schema_version must be v4_scorecard.v1",
            )
        _check_v4_scorecard_dimensions(findings, pack.get("rule_pack_id"), scorecard)
        _check_v4_score_weight_profiles(findings, pack.get("rule_pack_id"), scorecard)
        _check_v4_scorecard_rule_refs(findings, pack.get("rule_pack_id"), scorecard)
        required_fields = set((pack.get("input_contract") or {}).get("required_fields") or [])
        for field in [
            "pattern_match_result.query_relevance_score",
            "content_engagement_metrics.platform_performance.platform_performance_score",
            "content_engagement_metrics.platform_performance.missing_observable_fields",
        ]:
            if field not in required_fields:
                _fail(
                    findings,
                    "v4_rule_pack_required_field_missing",
                    f"{pack.get('rule_pack_id')} missing required field {field}",
                )


def _check_v4_scorecard_dimensions(
    findings: list[dict[str, str]],
    rule_pack_id: Any,
    scorecard: dict[str, Any],
) -> None:
    dimensions = scorecard.get("dimensions")
    if not isinstance(dimensions, list):
        _fail(findings, "v4_scorecard_dimensions_invalid", f"{rule_pack_id} dimensions must be a list")
        return
    keys: list[Any] = []
    for row in dimensions:
        if not isinstance(row, dict):
            _fail(findings, "v4_scorecard_dimension_invalid", f"{rule_pack_id} dimension row must be object")
            continue
        key = row.get("key")
        if not key:
            _fail(findings, "v4_scorecard_dimension_key_invalid", f"{rule_pack_id} dimension key must be non-empty")
        keys.append(key)
        if row.get("runtime_status") != "active":
            continue
        active = row.get("active")
        if active is not None and active is not True:
            _fail(
                findings,
                "v4_scorecard_dimension_active_invalid",
                f"{rule_pack_id}/{key} active must match runtime_status=active",
            )
        platform_scope = row.get("platform_scope")
        if platform_scope is not None and (
            not isinstance(platform_scope, list)
            or not platform_scope
            or any(not isinstance(item, str) or not item for item in platform_scope)
        ):
            _fail(
                findings,
                "v4_scorecard_dimension_platform_scope_invalid",
                f"{rule_pack_id}/{key} platform_scope must be a non-empty string list",
            )
        score_source_path = row.get("score_source_path")
        if score_source_path is not None and (not isinstance(score_source_path, str) or not score_source_path):
            _fail(
                findings,
                "v4_scorecard_dimension_score_source_path_invalid",
                f"{rule_pack_id}/{key} score_source_path must be non-empty string",
            )
        if not _positive_number(row.get("max_score")):
            _fail(findings, "v4_scorecard_dimension_score_invalid", f"{rule_pack_id}/{key} max_score must be positive")
        if not _positive_number(row.get("weight_percent")):
            _fail(findings, "v4_scorecard_dimension_weight_invalid", f"{rule_pack_id}/{key} weight_percent must be positive")
    duplicates = sorted({key for key in keys if key and keys.count(key) > 1})
    if duplicates:
        _fail(findings, "v4_scorecard_dimension_duplicate", f"{rule_pack_id} duplicate dimensions: {duplicates}")
    active_keys = {row.get("key") for row in dimensions if isinstance(row, dict) and row.get("runtime_status") == "active"}
    missing = sorted(V4_BASE_ACTIVE_DIMENSIONS - active_keys)
    if missing:
        _fail(findings, "v4_scorecard_dimensions_invalid", f"{rule_pack_id} active dimensions missing base keys: {missing}")


def _check_v4_score_weight_profiles(
    findings: list[dict[str, str]],
    rule_pack_id: Any,
    scorecard: dict[str, Any],
) -> None:
    profiles = scorecard.get("score_weight_profiles")
    if not isinstance(profiles, dict):
        _fail(findings, "v4_score_weight_profiles_invalid", f"{rule_pack_id} score_weight_profiles must be an object")
        return
    dimension_keys = {
        row.get("key")
        for row in scorecard.get("dimensions", [])
        if isinstance(row, dict) and row.get("key")
    }
    if isinstance(profiles.get("profiles"), list):
        _check_v4_generic_score_weight_profiles(findings, rule_pack_id, profiles["profiles"], dimension_keys)
        return
    missing = sorted(V4_REQUIRED_SCORE_WEIGHT_PROFILES - set(profiles))
    if missing:
        _fail(findings, "v4_score_weight_profiles_invalid", f"{rule_pack_id} missing score weight profiles: {missing}")
    for profile_name, weights in profiles.items():
        if not isinstance(weights, dict) or not weights:
            _fail(findings, "v4_score_weight_profile_invalid", f"{rule_pack_id}/{profile_name} must be a non-empty object")
            continue
        for key, weight in weights.items():
            if key not in dimension_keys:
                _fail(
                    findings,
                    "v4_score_weight_profile_dimension_unknown",
                    f"{rule_pack_id}/{profile_name} unknown dimension: {key}",
                )
            if not _positive_number(weight):
                _fail(
                    findings,
                    "v4_score_weight_profile_weight_invalid",
                    f"{rule_pack_id}/{profile_name}/{key} weight must be positive",
                )


def _check_v4_generic_score_weight_profiles(
    findings: list[dict[str, str]],
    rule_pack_id: Any,
    profiles: list[Any],
    dimension_keys: set[Any],
) -> None:
    aliases: set[str] = set()
    priority_groups: dict[tuple[Any, Any, Any, int], list[str]] = {}
    for profile in profiles:
        if not isinstance(profile, dict):
            _fail(findings, "v4_score_weight_profile_invalid", f"{rule_pack_id} profile row must be object")
            continue
        profile_id = profile.get("profile_id")
        if not profile_id:
            _fail(findings, "v4_score_weight_profile_invalid", f"{rule_pack_id} profile_id must be non-empty")
            continue
        aliases.update(profile.get("historical_alias_ids") or [])
        if profile.get("approval_status") != "approved":
            _fail(findings, "v4_score_weight_profile_approval_invalid", f"{rule_pack_id}/{profile_id} must be approved")
        if profile.get("runtime_enabled") is not True:
            _fail(findings, "v4_score_weight_profile_runtime_invalid", f"{rule_pack_id}/{profile_id} runtime_enabled must be true")
        weights = profile.get("weights")
        if not isinstance(weights, dict) or not weights:
            _fail(findings, "v4_score_weight_profile_invalid", f"{rule_pack_id}/{profile_id} weights must be non-empty object")
            continue
        total = 0.0
        for key, weight in weights.items():
            if key not in dimension_keys:
                _fail(
                    findings,
                    "v4_score_weight_profile_dimension_unknown",
                    f"{rule_pack_id}/{profile_id} unknown dimension: {key}",
                )
            if not _positive_number(weight):
                _fail(
                    findings,
                    "v4_score_weight_profile_weight_invalid",
                    f"{rule_pack_id}/{profile_id}/{key} weight must be positive",
                )
            else:
                total += float(weight)
        normalization = profile.get("normalization_policy") or "sum_100"
        if normalization == "raw_sum":
            if not profile.get("normalization_reason"):
                _fail(
                    findings,
                    "v4_score_weight_profile_normalization_invalid",
                    f"{rule_pack_id}/{profile_id} raw_sum requires normalization_reason",
                )
        elif round(total, 6) != 100:
            _fail(
                findings,
                "v4_score_weight_profile_total_invalid",
                f"{rule_pack_id}/{profile_id} weights must sum to 100",
            )
        group = (
            tuple(profile.get("platform_scope") or []),
            profile.get("content_format") or "*",
            profile.get("applies_when") or "always",
            int(profile.get("priority") or 0),
        )
        priority_groups.setdefault(group, []).append(str(profile_id))
    missing = sorted(V4_REQUIRED_SCORE_WEIGHT_PROFILES - aliases)
    if missing:
        _fail(findings, "v4_score_weight_profiles_invalid", f"{rule_pack_id} missing historical aliases: {missing}")
    for group, ids in priority_groups.items():
        if len(ids) > 1:
            _fail(findings, "v4_score_weight_profile_priority_invalid", f"{rule_pack_id} priority conflict {group}: {ids}")


def _check_v4_scorecard_rule_refs(
    findings: list[dict[str, str]],
    rule_pack_id: Any,
    scorecard: dict[str, Any],
) -> None:
    dimension_keys = {
        row.get("key")
        for row in scorecard.get("dimensions", [])
        if isinstance(row, dict) and row.get("key")
    }
    for rule in scorecard.get("scoring_rules", []):
        if not isinstance(rule, dict):
            _fail(findings, "v4_scoring_rule_invalid", f"{rule_pack_id} scoring rule must be object")
            continue
        if rule.get("dimension_key") not in dimension_keys:
            _fail(
                findings,
                "v4_scoring_rule_dimension_unknown",
                f"{rule_pack_id}/{rule.get('scoring_rule_id')} unknown dimension_key: {rule.get('dimension_key')}",
            )


def _positive_number(value: Any) -> bool:
    return isinstance(value, (int, float)) and not isinstance(value, bool) and value > 0


def _fail(findings: list[dict[str, str]], check_id: str, message: str) -> None:
    findings.append({"level": "fail", "check_id": check_id, "message": message})


if __name__ == "__main__":
    sys.exit(main())