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

"""Validate referential integrity inside douyin_rule_packs.v1.json (V2-M1D).

Mirrors the walk-side checks in walk_strategy_json (which already cover walk FKs).
Here we close the rule-pack side: every decision_action / decision_reason_code /
scoring_rule dimension_key / dispatch rule_pack_id must resolve to an
authoritative catalog entry within the same file.
"""

from __future__ import annotations

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

ROOT = Path(__file__).resolve().parents[1]
RULE_PACK_PATH = Path("product_documents/规则包/douyin_rule_packs.v1.json")
LEGACY_FIELD_BLOCKLIST = {
    "fit_senior_50plus",
    "fit_confidence",
    "relevance_score",
    "platform_heat",
    "age_50_plus_level",
}
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",
}
V4_SCORE_WEIGHT_ALIAS_TARGETS = {
    "base_video_two_dimension_v1": "default_two_dimension",
    "audience_50plus_available_v1": "douyin_fifty_plus_ok",
    "audience_50plus_skipped_by_query_gate_v1": "douyin_fifty_plus_not_attempted",
}


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


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


def validate_rule_pack_config(pkg: dict[str, Any]) -> list[dict[str, Any]]:
    findings: list[dict[str, Any]] = []
    rule_packs = pkg.get("rule_packs", [])
    rule_pack_ids = {p.get("rule_pack_id") for p in rule_packs}
    actions: set[str] = set()
    for entry in pkg.get("decision_action_catalog", []):
        actions.update(entry.get("allowed_actions", []))
    reason_codes = {r.get("decision_reason_code") for r in pkg.get("decision_reason_codes", [])}

    enabled_by_group: dict[tuple[Any, ...], list[str]] = {}
    for dispatch in pkg.get("rule_pack_dispatch", []):
        if dispatch.get("rule_pack_id") not in rule_pack_ids:
            _fail(findings, "dispatch_rule_pack_id",
                  f"{dispatch.get('dispatch_id')} references unknown rule_pack_id: {dispatch.get('rule_pack_id')}")
        if dispatch.get("dispatch_enabled"):
            group = (dispatch.get("platform"), dispatch.get("strategy_version"), dispatch.get("runtime_stage"),
                     dispatch.get("target_entity"), dispatch.get("content_format"))
            enabled_by_group.setdefault(group, []).append(dispatch.get("rule_pack_id"))
    for group, pack_ids in enabled_by_group.items():
        if len(pack_ids) > 1:
            _fail(findings, "dispatch_conflict",
                  f"CONFIG_RULE_PACK_DISPATCH_CONFLICT: multiple enabled dispatches for group {group}: {pack_ids}")

    for pack in rule_packs:
        pid = pack.get("rule_pack_id")
        dimension_keys = {d.get("key") for d in pack.get("scorecard", {}).get("dimensions", [])}
        for gate in pack.get("hard_gates", []):
            if gate.get("decision_action") not in actions:
                _fail(findings, "hard_gate_action", f"{pid}/{gate.get('gate_id')} unknown decision_action: {gate.get('decision_action')}")
            if gate.get("decision_reason_code") not in reason_codes:
                # decision_reason_codes is a curated subset, not an exhaustive enum -> warn only.
                _warn(findings, "hard_gate_reason", f"{pid}/{gate.get('gate_id')} reason_code not in catalog: {gate.get('decision_reason_code')}")
        for rule in pack.get("scorecard", {}).get("scoring_rules", []):
            if rule.get("dimension_key") not in dimension_keys:
                _fail(findings, "scoring_rule_dimension", f"{pid}/{rule.get('scoring_rule_id')} unknown dimension_key: {rule.get('dimension_key')}")
        for i, threshold in enumerate(pack.get("thresholds", [])):
            if threshold.get("decision_action") not in actions:
                _fail(findings, "threshold_action", f"{pid}/threshold[{i}] unknown decision_action: {threshold.get('decision_action')}")
            if threshold.get("decision_reason_code") not in reason_codes:
                _warn(findings, "threshold_reason", f"{pid}/threshold[{i}] reason_code not in catalog: {threshold.get('decision_reason_code')}")
        if _is_v4_pack(pack, pkg):
            _check_v4_rule_pack(findings, pack)
    return findings


def _is_v4_pack(pack: dict[str, Any], pkg: dict[str, Any]) -> bool:
    return (
        (pkg.get("strategy_binding") or {}).get("strategy_version") == "V4"
        or (pack.get("scorecard") or {}).get("schema_version") == "v4_scorecard.v1"
        or pack.get("version") == "4.0.0"
    )


def _check_v4_rule_pack(findings: list[dict[str, Any]], pack: dict[str, Any]) -> None:
    pid = pack.get("rule_pack_id")
    legacy_paths = _legacy_paths(pack, f"rule_pack:{pid}")
    if legacy_paths:
        _fail(findings, "v4_rule_pack_legacy_field", f"{pid} contains legacy fields: {legacy_paths[:5]}")
    scorecard = pack.get("scorecard") or {}
    if scorecard.get("schema_version") != "v4_scorecard.v1":
        _fail(findings, "v4_scorecard_schema", f"{pid} scorecard.schema_version must be v4_scorecard.v1")
    _check_scorecard_dimensions(findings, pid, scorecard)
    _check_score_weight_profiles(findings, pid, scorecard)
    thresholds = pack.get("thresholds", [])
    expected_reasons = {
        "v4_query_and_platform_pass",
        "v4_score_review_needed",
        "v4_query_or_score_below_threshold",
    }
    if {row.get("decision_reason_code") for row in thresholds} != expected_reasons:
        _fail(findings, "v4_threshold_reasons", f"{pid} thresholds must use V4 reason codes")
    pool = [row for row in thresholds if row.get("decision_action") == "ADD_TO_CONTENT_POOL"]
    review = [row for row in thresholds if row.get("decision_action") == "KEEP_CONTENT_FOR_REVIEW"]
    reject = [row for row in thresholds if row.get("decision_action") == "REJECT_CONTENT"]
    if not pool or pool[0].get("min_score") != 70:
        _fail(findings, "v4_pool_threshold", f"{pid} pool threshold min_score must be 70")
    if not review or review[0].get("min_score") != 55:
        _fail(findings, "v4_review_threshold", f"{pid} review threshold min_score must be 55")
    if not reject or reject[0].get("max_score", 0) > 55:
        _fail(findings, "v4_reject_threshold", f"{pid} reject threshold must stay below 55")


def _legacy_paths(value: Any, prefix: str) -> list[str]:
    paths: list[str] = []
    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)
            paths.extend(_legacy_paths(child, child_path))
    elif isinstance(value, list):
        for index, child in enumerate(value):
            paths.extend(_legacy_paths(child, f"{prefix}[{index}]"))
    elif isinstance(value, str):
        if _string_has_legacy_field(value):
            paths.append(prefix)
    return paths


def _check_scorecard_dimensions(
    findings: list[dict[str, Any]],
    pid: Any,
    scorecard: dict[str, Any],
) -> None:
    dimensions = scorecard.get("dimensions")
    if not isinstance(dimensions, list):
        _fail(findings, "v4_scorecard_dimensions", f"{pid} scorecard.dimensions must be a list")
        return
    keys: list[Any] = []
    for row in dimensions:
        if not isinstance(row, dict):
            _fail(findings, "v4_scorecard_dimension", f"{pid} dimension row must be object")
            continue
        key = row.get("key")
        if not key:
            _fail(findings, "v4_scorecard_dimension_key", f"{pid} 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", f"{pid}/{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", f"{pid}/{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", f"{pid}/{key} score_source_path must be non-empty string")
        if not _positive_number(row.get("max_score")):
            _fail(findings, "v4_scorecard_dimension_score", f"{pid}/{key} max_score must be positive")
        if not _positive_number(row.get("weight_percent")):
            _fail(findings, "v4_scorecard_dimension_weight", f"{pid}/{key} weight_percent must be positive")
    duplicate_keys = sorted({key for key in keys if key and keys.count(key) > 1})
    if duplicate_keys:
        _fail(findings, "v4_scorecard_dimension_duplicate", f"{pid} duplicate dimensions: {duplicate_keys}")
    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", f"{pid} active dimensions missing base keys: {missing}")


def _check_score_weight_profiles(
    findings: list[dict[str, Any]],
    pid: Any,
    scorecard: dict[str, Any],
) -> None:
    profiles = scorecard.get("score_weight_profiles")
    if not isinstance(profiles, dict):
        _fail(findings, "v4_score_weight_profiles", f"{pid} 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_generic_score_weight_profiles(findings, pid, profiles["profiles"], dimension_keys)
        return
    missing_profiles = sorted(V4_REQUIRED_SCORE_WEIGHT_PROFILES - set(profiles))
    if missing_profiles:
        _fail(findings, "v4_score_weight_profiles", f"{pid} missing score weight profiles: {missing_profiles}")
    for profile_name, weights in profiles.items():
        if not isinstance(weights, dict) or not weights:
            _fail(findings, "v4_score_weight_profile", f"{pid}/{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", f"{pid}/{profile_name} unknown dimension: {key}")
            if not _positive_number(weight):
                _fail(findings, "v4_score_weight_profile_weight", f"{pid}/{profile_name}/{key} weight must be positive")


def _check_generic_score_weight_profiles(
    findings: list[dict[str, Any]],
    pid: 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", f"{pid} profile row must be object")
            continue
        profile_id = profile.get("profile_id")
        if not profile_id:
            _fail(findings, "v4_score_weight_profile", f"{pid} 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", f"{pid}/{profile_id} runtime profile must be approved")
        if profile.get("runtime_enabled") is not True:
            _fail(findings, "v4_score_weight_profile_runtime", f"{pid}/{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", f"{pid}/{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", f"{pid}/{profile_id} unknown dimension: {key}")
            if not _positive_number(weight):
                _fail(findings, "v4_score_weight_profile_weight", f"{pid}/{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", f"{pid}/{profile_id} raw_sum requires normalization_reason")
        elif round(total, 6) != 100:
            _fail(findings, "v4_score_weight_profile_total", f"{pid}/{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", f"{pid} missing historical aliases: {missing}")
    for group, ids in priority_groups.items():
        if len(ids) > 1:
            _fail(findings, "v4_score_weight_profile_priority", f"{pid} priority conflict {group}: {ids}")


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


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 main() -> int:
    args = _parse_args()
    path = args.config_path if args.config_path.is_absolute() else ROOT / args.config_path
    pkg = json.loads(path.read_text(encoding="utf-8"))
    findings = validate_rule_pack_config(pkg)
    result = {
        "status": "fail" if any(f["level"] == "fail" for f in findings) else "pass",
        "config_path": str(path.relative_to(ROOT)),
        "findings": findings,
    }
    print(json.dumps(result, ensure_ascii=False, indent=2))
    return 1 if result["status"] == "fail" else 0


def _parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("config_path", nargs="?", type=Path, default=RULE_PACK_PATH)
    return parser.parse_args()


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