"""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", } 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") dimensions = [row for row in scorecard.get("dimensions", []) if row.get("runtime_status") == "active"] keys = [row.get("key") for row in dimensions] if keys != ["query_relevance", "platform_performance"]: _fail(findings, "v4_scorecard_dimensions", f"{pid} active dimensions must be query_relevance/platform_performance, got {keys}") for row in dimensions: if row.get("max_score") != 50 or row.get("weight_percent") != 50: _fail(findings, "v4_scorecard_weight", f"{pid}/{row.get('key')} must be max_score=50 weight_percent=50") 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 _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())