content-find-agent-new/content_agent/integrations/walk_strategy_json.py

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