"""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())