from __future__ import annotations import json from dataclasses import dataclass from pathlib import Path from typing import Any from content_agent.findings import fail as _fail WALK_STRATEGY_PATH = Path("product_documents/抖音游走策略/douyin_walk_strategy.v1.json") RULE_PACK_PATH = Path("product_documents/规则包/douyin_rule_packs.v1.json") EDGE_CATALOG_PATH = Path("tech_documents/数据接口与来源/walk_edge_catalog.json") # V5 M1: 边目录迁到平台无关 catalog；strategy 只保留运行时实际消费的绑定/门/事实契约。 REQUIRED_SECTIONS = [ "walk_rule_pack_binding", "v4_walk_gate", "walk_fact_contract", ] @dataclass(frozen=True) class WalkStrategyStore: root_dir: Path = Path(".") strategy_path: Path = WALK_STRATEGY_PATH rule_pack_path: Path = RULE_PACK_PATH def load_walk_strategy(self) -> dict[str, Any]: from content_agent.integrations import config_store path = self.root_dir / self.strategy_path strategy, _ = config_store.load_json(path) findings = validate_walk_strategy_config( strategy, root_dir=self.root_dir, strategy_path=self.strategy_path, rule_pack_path=self.rule_pack_path, edge_catalog_path=EDGE_CATALOG_PATH, ) failures = [finding for finding in findings if finding["level"] == "fail"] if failures: raise ValueError(f"invalid walk strategy config: {failures}") return { **strategy, "walk_strategy_version": strategy.get("strategy_version"), "walk_strategy_source_ref": { "file": str(self.strategy_path), "strategy_id": strategy.get("strategy_id"), "walk_strategy_version": strategy.get("strategy_version"), "source_of_truth": strategy.get("source_of_truth"), }, } def validate_walk_strategy_config( strategy: dict[str, Any], *, root_dir: Path = Path("."), strategy_path: Path = WALK_STRATEGY_PATH, rule_pack_path: Path = RULE_PACK_PATH, edge_catalog_path: Path = EDGE_CATALOG_PATH, ) -> list[dict[str, Any]]: findings: list[dict[str, Any]] = [] _check_identity(strategy, strategy_path, findings) _check_required_sections(strategy, findings) if any(finding["level"] == "fail" for finding in findings): return findings edge_ids = _load_edge_catalog_ids(root_dir / edge_catalog_path, findings) if any(finding["level"] == "fail" for finding in findings): return findings _check_edge_refs(strategy, edge_ids, findings) _check_v4_walk_gate(strategy["v4_walk_gate"], edge_ids, findings) _check_fact_contract(strategy["walk_fact_contract"], findings) _check_rule_pack_bindings( strategy["walk_rule_pack_binding"], root_dir / rule_pack_path, findings, ) return findings def _check_identity( strategy: dict[str, Any], strategy_path: Path, findings: list[dict[str, Any]], ) -> None: if not strategy.get("strategy_id"): _fail(findings, "strategy_id", "strategy_id must be non-empty") if strategy.get("strategy_version") != "V1.0": _fail(findings, "strategy_version", "walk strategy config version must be V1.0") if strategy.get("source_of_truth") != str(strategy_path): _fail(findings, "source_of_truth", f"source_of_truth must be {strategy_path}") def _check_required_sections(strategy: dict[str, Any], findings: list[dict[str, Any]]) -> None: for section in REQUIRED_SECTIONS: value = strategy.get(section) if not isinstance(value, list) or not value: _fail(findings, "section_missing", f"{section} must be a non-empty list") def _check_edge_refs( strategy: dict[str, Any], edge_ids: set[str], findings: list[dict[str, Any]], ) -> None: for section in ["walk_rule_pack_binding"]: for row in strategy.get(section, []): if row.get("edge_id") not in edge_ids: _fail( findings, "edge_ref", f"{section} references unknown edge_id: {row.get('edge_id')}", ) def _check_fact_contract( contracts: list[dict[str, Any]], findings: list[dict[str, Any]] ) -> None: by_file = {contract.get("runtime_file"): contract for contract in contracts} walk_actions = by_file.get("walk_actions.jsonl") if not walk_actions: _fail(findings, "walk_actions_contract", "walk_actions.jsonl contract is required") else: unique_key = walk_actions.get("unique_key") if unique_key != ["run_id", "policy_run_id", "walk_action_id"]: _fail(findings, "walk_actions_unique_key", "walk_actions unique key is invalid") search_clues = by_file.get("search_clues.jsonl") if not search_clues: _fail(findings, "search_clues_contract", "search_clues.jsonl contract is required") elif search_clues.get("unique_key") != ["run_id", "policy_run_id", "clue_id"]: _fail(findings, "search_clues_unique_key", "search_clues unique key must use clue_id") def _check_v4_walk_gate( gates: list[dict[str, Any]], edge_ids: set[str], findings: list[dict[str, Any]], ) -> None: by_id = {gate.get("gate_id"): gate for gate in gates} gate = by_id.get("allow_walk_required") if not gate: _fail(findings, "v4_walk_gate_missing", "allow_walk_required gate is required") return if gate.get("requires_allow_walk") is not True: _fail(findings, "v4_walk_gate_requires_allow_walk", "allow_walk_required must require allow_walk") if gate.get("deny_reason_code") != "v4_allow_walk_denied": _fail(findings, "v4_walk_gate_deny_reason", "allow_walk_required deny reason is invalid") applies_to = gate.get("applies_to_edges") expected_edges = {"hashtag_to_query", "author_to_works"} if set(applies_to or []) != expected_edges: _fail(findings, "v4_walk_gate_edges", "allow_walk_required must cover M4 expansion edges") for edge_id in applies_to or []: if edge_id not in edge_ids: _fail(findings, "v4_walk_gate_edge_ref", f"allow_walk_required unknown edge_id: {edge_id}") raw_payload_fields = set(gate.get("raw_payload_fields") or []) expected_fields = {"decision_id", "allow_walk", "allow_walk_reason", "walk_gate_snapshot"} if not expected_fields <= raw_payload_fields: _fail(findings, "v4_walk_gate_raw_fields", "allow_walk_required raw payload fields incomplete") def _check_rule_pack_bindings( bindings: list[dict[str, Any]], rule_pack_path: Path, findings: list[dict[str, Any]], ) -> None: rule_package = json.loads(rule_pack_path.read_text(encoding="utf-8")) enabled_packs = { (pack.get("rule_pack_id"), pack.get("version")) for pack in rule_package.get("rule_packs", []) if pack.get("enabled") } for binding in bindings: key = (binding.get("rule_pack_id"), binding.get("rule_pack_version")) if key not in enabled_packs: _fail( findings, "rule_pack_binding", f"{binding.get('binding_id')} references missing enabled rule pack: {key}", ) def _ids(rows: list[dict[str, Any]], field: str) -> set[str]: return {str(row[field]) for row in rows if row.get(field)} def _load_edge_catalog_ids(path: Path, findings: list[dict[str, Any]]) -> set[str]: try: catalog = json.loads(path.read_text(encoding="utf-8")) except FileNotFoundError: _fail(findings, "edge_catalog_missing", f"missing edge catalog: {path}") return set() except json.JSONDecodeError as exc: _fail(findings, "edge_catalog_parse", f"edge catalog cannot parse: {exc}") return set() rows = catalog.get("walk_edge_catalog") if not isinstance(rows, list) or not rows: _fail(findings, "edge_catalog_invalid", "walk_edge_catalog must be a non-empty list") return set() return _ids(rows, "edge_id")