content-find-agent-new/tests/p6_walk_helpers.py

from __future__ import annotations

from pathlib import Path
from typing import Any

from content_agent.business_modules import content_discovery, rule_judgment, source_seed
from content_agent.business_modules.content_discovery import pattern_recall
from content_agent.constants import RUNTIME_RECORD_SCHEMA_VERSION
from content_agent.integrations.policy_json import JsonPolicyBundleStore
from content_agent.integrations.runtime_files import LocalRuntimeFileStore
from content_agent.record_payload import with_raw_payload
from tests.gemini_helpers import FakeGeminiVideoClient
from tests.p1_helpers import real_source_payload


class FakeWalkPlatformClient:
    def __init__(self, fail_tag: bool = False, tags: list[str] | None = None) -> None:
        self.fail_tag = fail_tag
        self.tags = tags or ["#人物故事"]
        self.search_calls: list[dict[str, Any]] = []
        self.author_calls: list[dict[str, Any]] = []
        self.portrait_calls: list[str] = []

    def search(self, query: dict[str, Any]) -> list[dict[str, Any]]:
        self.search_calls.append(dict(query))
        method = query.get("search_query_generation_method")
        if method == "tag_query" and self.fail_tag:
            raise RuntimeError("tag search unavailable")
        if method == "tag_query":
            return [_platform_result(query, "7390000000000000201", "标签内容", [])]
        return []

    def fetch_author_works(self, query: dict[str, Any]) -> list[dict[str, Any]]:
        self.author_calls.append(dict(query))
        return [_platform_result(query, "7390000000000000301", "作者作品", [])]

    def fetch_account_fans_portrait(self, account_id: str) -> dict[str, Any]:
        # 强 50+ 画像(纯 50- 桶,占比60%/TGI200 → 50+子分高)→ 不改变 walk 既有 allow_walk/入池 行为。
        self.portrait_calls.append(account_id)
        return {
            "fans": {
                "age": {
                    "data": {
                        "50-": {"percentage": "60%", "preference": "200"},
                    }
                }
            }
        }


def build_initial_walk_context(tmp_path: Path, *, tags: list[str] | None = None) -> dict[str, Any]:
    run_id = "run_001"
    policy_run_id = "policy_run_001"
    runtime = LocalRuntimeFileStore(tmp_path / "runtime")
    runtime.prepare_run(run_id)
    seed = source_seed.run(run_id, policy_run_id, real_source_payload(), runtime)
    search_query = with_raw_payload(
        {
            "record_schema_version": RUNTIME_RECORD_SCHEMA_VERSION,
            "run_id": run_id,
            "policy_run_id": policy_run_id,
            "search_query_id": "q_001",
            "search_query": "人物故事",
            "search_query_generation_method": "item_single",
            "discovery_start_source": "pattern_itemset",
            "previous_discovery_step": "search_query_generated",
            "pattern_seed_ref": {"seed_term": "人物故事"},
        }
    )
    runtime.append_jsonl(run_id, "search_queries.jsonl", [search_query])
    initial_result = _platform_result(search_query, "7390000000000000001", "首轮内容", tags or ["#人物故事"])
    initial_result["has_more"] = True
    initial_result["next_cursor"] = "10"
    discovered = content_discovery.run(
        run_id,
        policy_run_id,
        [initial_result],
        seed["source_context"],
        runtime,
    )
    recalled = pattern_recall.run(
        run_id,
        policy_run_id,
        discovered["discovered_content_items"],
        discovered["content_media_records"],
        discovered["evidence_bundles"],
        seed["source_context"],
        runtime,
        FakeGeminiVideoClient(),
    )
    policy_bundle = JsonPolicyBundleStore(Path(".")).load_policy_bundle("V4")
    decisions = rule_judgment.run(
        run_id,
        policy_run_id,
        recalled["evidence_bundles"],
        policy_bundle,
        runtime,
    )
    return {
        "run_id": run_id,
        "policy_run_id": policy_run_id,
        "runtime": runtime,
        "source_context": seed["source_context"],
        "pattern_seed_pack": seed["pattern_seed_pack"],
        "search_queries": [search_query],
        "discovered_content_items": recalled["discovered_content_items"],
        "content_media_records": discovered["content_media_records"],
        "evidence_bundles": recalled["evidence_bundles"],
        "rule_decisions": decisions,
        "policy_bundle": policy_bundle,
        "gemini_video_client": FakeGeminiVideoClient(),
    }


def set_v4_allow_walk(decision: dict[str, Any], allow_walk: bool) -> None:
    query_score = 80
    # M11 re-baseline:walk fixture(digg=100k,高共鸣占比)新平台分由 rule_judgment 实算为 88.41。
    # True 分支必须与实算一致,否则 final_output v4_explanation 校验报 platform_performance_score 不一致。
    # False 分支仅用于"显式拒绝游走"测试(不跑 validate_run),沿用低分占位即可。
    platform_score = 88.41 if allow_walk else 60
    total_score = round(query_score * 0.5 + platform_score * 0.5, 2)
    decision["decision_action"] = "ADD_TO_CONTENT_POOL"
    decision["decision_reason_code"] = "v4_query_and_platform_pass"
    decision["search_query_effect_status"] = "success"
    decision["score"] = total_score
    decision.setdefault("scorecard", {})["schema_version"] = "v4_scorecard.v1"
    decision["scorecard"]["query_relevance_score"] = query_score
    decision["scorecard"]["platform_performance_score"] = platform_score
    decision["scorecard"]["total_score"] = total_score
    replay = decision.setdefault("decision_replay_data", {})
    replay["allow_walk"] = allow_walk
    replay["allow_walk_reason"] = (
        "query>=65/platform>=65/score>=65"
        if allow_walk
        else "v4_query_and_platform_pass"
    )
    replay["walk_gate_snapshot"] = {
        "query_relevance_score": query_score,
        "platform_performance_score": platform_score,
        "score": total_score,
    }


def _platform_result(
    query: dict[str, Any],
    platform_content_id: str,
    description: str,
    tags: list[str],
) -> dict[str, Any]:
    return {
        "content_discovery_id": f"{query['search_query_id']}_content_{platform_content_id[-3:]}",
        "search_query_id": query["search_query_id"],
        "platform": "douyin",
        "platform_content_id": platform_content_id,
        "platform_content_format": "video",
        "description": description,
        "platform_author_id": "MS4wLjABAAAA001",
        "author_display_name": "作者",
        "statistics": {
            # M11:平台分改 ratio 后,光大点赞不够;给"高共鸣"占比(转/赞30%、藏/赞50%、评/赞8%)
            # 让该 fixture 在新打分下平台分 ~88 ≥ 65 门槛(walk 测试本意:成功内容能继续游走)。
            "digg_count": 100_000,
            "comment_count": 8_000,
            "share_count": 30_000,
            "collect_count": 50_000,
        },
        "tags": tags,
        "score": 72,
        "risk_level": "low",
        "availability": "available",
        "discovery_relation": "fake_walk",
        "discovery_start_source": query.get("discovery_start_source", "pattern_itemset"),
        "previous_discovery_step": query.get("previous_discovery_step", "search_query_direct"),
        "content_metadata_source": "fake_platform_search",
        "platform_raw_payload": {"content_id": platform_content_id},
    }