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

"""V3 游走配置入口(M4A):读 walk_graph / walk_policy / platform_profiles 并校验。

镜像 walk_strategy_json.WalkStrategyStore 的 load+validate+raise 结构。
walk_policy 的拍板值可能带 {value, provenance, tbd} 包裹(留痕),load_policy 解包成纯值。
"""

from __future__ import annotations

import os
from dataclasses import dataclass
from pathlib import Path
from typing import Any

from content_agent.findings import fail as _fail

WALK_GRAPH_PATH = Path("tech_documents/数据接口与来源/walk_graph.json")
WALK_POLICY_PATH = Path("tech_documents/数据接口与来源/walk_policy.json")
PROFILE_DIR = Path("tech_documents/数据接口与来源/platform_profiles")
EDGE_CATALOG_PATH = Path("tech_documents/数据接口与来源/walk_edge_catalog.json")
PROFILE_EDGE_STATUSES = {"supported", "blocked", "no_interface", "planned"}

PERMISSION_ACTIONS = [
    "ADD_TO_CONTENT_POOL",
    "KEEP_CONTENT_FOR_REVIEW",
    "TECHNICAL_RETRY_REQUIRED",
    "REJECT_CONTENT",
]
_PERMISSION_META_KEYS = {"_meaning", "_provenance", "note", "tbd"}


def _unwrap(value: Any) -> Any:
    if isinstance(value, dict) and "value" in value:
        return value["value"]
    return value


def low_budget(budget: int) -> int:
    """low_budget 档:减半向下取整,至少 1(2026-06-11 拍板,policy 值 halve_min_1)。"""
    return max(1, budget // 2)


def edge_permission(policy: dict[str, Any], decision_action: str | None, edge_id: str) -> str:
    """判定结果 → 该出边通行证;无判定/未列入的组合一律 deny(从严)。"""
    row = policy["edge_permissions"].get(decision_action or "", {})
    return row.get(edge_id, "deny")


def edge_budgets_for_platform(policy: dict[str, Any], platform: str | None) -> dict[str, dict[str, Any]]:
    """Return edge budgets for a platform, falling back to default then legacy rows."""
    selected = dict(policy.get("edge_budgets_by_id") or {})
    by_platform = policy.get("platform_edge_budgets_by_platform_edge") or {}
    for edge_id in selected:
        explicit = by_platform.get((platform or "", edge_id))
        default = by_platform.get(("default", edge_id))
        if explicit:
            selected[edge_id] = {**selected[edge_id], **explicit}
        elif default:
            selected[edge_id] = {**selected[edge_id], **default}
    return selected


def edge_supported(profile: dict[str, Any], edge_id: str) -> bool:
    """profile 未列出的边 = 平台无关控制边(终端边),恒 supported。"""
    spec = profile["edges"].get(edge_id, {"status": "supported"})
    return isinstance(spec, dict) and spec.get("status") == "supported"


@dataclass(frozen=True)
class WalkGraphStore:
    root_dir: Path = Path(".")

    def load_graph(self) -> dict[str, Any]:
        from content_agent.integrations import config_store

        graph, _ = config_store.load_json(self.root_dir / WALK_GRAPH_PATH)
        catalog, _ = config_store.load_json(self.root_dir / EDGE_CATALOG_PATH)
        catalog_ids = _catalog_edge_ids(catalog)
        _raise_on_fail(_validate_graph(graph, catalog_ids), "walk graph")
        return graph

    def load_policy(self) -> dict[str, Any]:
        from content_agent.integrations import config_store

        raw_policy, _ = config_store.load_json(self.root_dir / WALK_POLICY_PATH)
        policy = _unwrap_policy(raw_policy)
        edge_ids = {edge["edge_id"] for edge in self.load_graph()["edges"]}
        _raise_on_fail(_validate_policy(policy, edge_ids), "walk policy")
        return policy

    def load_profile(self, platform: str) -> dict[str, Any]:
        from content_agent.integrations import config_store

        profile, _ = config_store.load_json(self.root_dir / PROFILE_DIR / f"{platform}.json")
        edge_ids = {edge["edge_id"] for edge in self.load_graph()["edges"]}
        _raise_on_fail(_validate_profile(profile, edge_ids), f"platform profile {platform}")
        return profile


def _unwrap_policy(raw: dict[str, Any]) -> dict[str, Any]:
    policy = dict(raw)
    policy["global"] = {key: _unwrap(value) for key, value in raw["global"].items()}
    _apply_global_env_overrides(policy["global"])
    policy["edge_permissions"] = {
        action: {edge: _unwrap(perm) for edge, perm in row.items() if edge not in _PERMISSION_META_KEYS}
        for action, row in raw["edge_permissions"].items()
        if action in PERMISSION_ACTIONS
    }
    policy["budget_tiers"] = {key: _unwrap(value) for key, value in raw["budget_tiers"].items()}
    policy["edge_budgets_by_id"] = {row["edge_id"]: row for row in raw["edge_budgets"]}
    policy["platform_edge_budgets_by_platform_edge"] = {
        (row.get("platform"), row.get("edge_id")): row
        for row in raw.get("platform_edge_budgets", [])
        if row.get("platform") and row.get("edge_id")
    }
    return policy


def _apply_global_env_overrides(global_policy: dict[str, Any]) -> None:
    overrides = {
        "CONTENT_AGENT_WALK_MAX_TOTAL_ACTIONS_PER_RUN": "max_total_actions_per_run",
        "CONTENT_AGENT_WALK_MAX_DEPTH": "max_depth",
        "CONTENT_AGENT_GEMINI_MAX_WORKERS": "gemini_max_workers",
    }
    for env_key, policy_key in overrides.items():
        value = _optional_positive_int(os.environ.get(env_key))
        if value is not None:
            global_policy[policy_key] = value


def _optional_positive_int(value: str | None) -> int | None:
    if value is None or value == "":
        return None
    try:
        parsed = int(value)
    except ValueError:
        return None
    return parsed if parsed > 0 else None


def _validate_graph(graph: dict[str, Any], catalog_ids: set[str]) -> list[dict[str, Any]]:
    findings: list[dict[str, Any]] = []
    node_types = {node["node_type"] for node in graph.get("nodes", [])}
    graph_edge_ids = set()
    for edge in graph.get("edges", []):
        edge_id = edge.get("edge_id")
        graph_edge_ids.add(edge_id)
        if edge_id not in catalog_ids:
            _fail(findings, "edge_not_in_catalog", f"{edge_id} not in walk_edge_catalog")
        for field in ["from_node", "to_node"]:
            if edge.get(field) not in node_types:
                _fail(findings, "edge_node_ref", f"{edge_id} unknown {field}: {edge.get(field)}")
    for edge_id in sorted(catalog_ids - graph_edge_ids):
        _fail(findings, "catalog_edge_missing_from_graph", f"{edge_id} missing from walk_graph.edges")
    return findings


def _validate_policy(policy: dict[str, Any], edge_ids: set[str]) -> list[dict[str, Any]]:
    findings: list[dict[str, Any]] = []
    for row in policy.get("edge_budgets", []):
        if row.get("edge_id") not in edge_ids:
            _fail(findings, "budget_edge_ref", f"edge_budgets unknown edge_id: {row.get('edge_id')}")
    platform_budget_keys = set()
    has_default_global = False
    has_kuaishou_author = False
    for row in policy.get("platform_edge_budgets_by_platform_edge", {}).values():
        key = (row.get("platform"), row.get("edge_id"))
        if key in platform_budget_keys:
            _fail(findings, "platform_budget_duplicate", f"duplicate platform budget: {key}")
        platform_budget_keys.add(key)
        if row.get("edge_id") != "__global__" and row.get("edge_id") not in edge_ids:
            _fail(findings, "platform_budget_edge_ref", f"platform budget unknown edge_id: {row.get('edge_id')}")
        has_default_global = has_default_global or key == ("default", "__global__")
        has_kuaishou_author = has_kuaishou_author or key == ("kuaishou", "author_to_works")
    if policy.get("platform_edge_budgets_by_platform_edge"):
        if not has_default_global:
            _fail(findings, "platform_budget_default_global", "platform_edge_budgets missing default/__global__")
        if not has_kuaishou_author:
            _fail(findings, "platform_budget_kuaishou_author", "platform_edge_budgets missing kuaishou/author_to_works")
    permission_rows = policy.get("edge_permissions", {})
    key_sets = []
    for action in PERMISSION_ACTIONS:
        if action not in permission_rows:
            _fail(findings, "permission_action_missing", f"edge_permissions missing action: {action}")
            continue
        keys = set(permission_rows[action])
        key_sets.append(keys)
        for edge_id in keys - edge_ids:
            _fail(findings, "permission_edge_ref", f"{action} unknown edge_id: {edge_id}")
    if key_sets and any(keys != key_sets[0] for keys in key_sets[1:]):
        _fail(findings, "permission_closure", "edge_permissions edge keys differ across actions")
    return findings


def _validate_profile(profile: dict[str, Any], edge_ids: set[str]) -> list[dict[str, Any]]:
    findings: list[dict[str, Any]] = []
    edges = profile.get("edges")
    if not isinstance(edges, dict):
        _fail(findings, "profile_edges_invalid", "profile edges must be an object")
        return findings
    for edge_id, spec in edges.items():
        if edge_id not in edge_ids:
            _fail(findings, "profile_edge_ref", f"profile unknown edge_id: {edge_id}")
        if not isinstance(spec, dict):
            _fail(findings, "profile_edge_invalid", f"{edge_id} must be an object")
            continue
        if spec.get("status") not in PROFILE_EDGE_STATUSES:
            _fail(findings, "profile_edge_status", f"{edge_id} invalid status: {spec.get('status')}")
    return findings


def _catalog_edge_ids(catalog: dict[str, Any]) -> set[str]:
    return {str(row["edge_id"]) for row in catalog.get("walk_edge_catalog", []) if row.get("edge_id")}



def _raise_on_fail(findings: list[dict[str, Any]], label: str) -> None:
    failures = [finding for finding in findings if finding["level"] == "fail"]
    if failures:
        raise ValueError(f"invalid {label} config: {failures}")