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

import json
from pathlib import Path


ROOT = Path(__file__).resolve().parents[1]
DATA_DIR = ROOT / "tech_documents/数据接口与来源"


def test_v4_query_source_refs_support_three_sources():
    query = {
        "pattern_seed_ref": {"query_source_type": "seed"},
        "raw_payload": {
            "query_source_refs": [
                {"query_source_type": "seed", "source_rank": 1},
                {"query_source_type": "topic_point", "source_rank": 2},
                {"query_source_type": "terminal_element", "source_rank": 3},
            ]
        },
    }

    source_types = {ref["query_source_type"] for ref in query["raw_payload"]["query_source_refs"]}

    assert source_types == {"seed", "topic_point", "terminal_element"}


def test_v4_tag_walk_contract_uses_80_threshold_and_raw_tag_query():
    tag_action = {
        "edge_id": "hashtag_to_query",
        "raw_payload": {
            "tag_text": "父爱感悟",
            "next_query": "父爱感悟",
            "tag_query_relevance_score": 80,
            "tag_walk_status": "allow",
        },
    }

    payload = tag_action["raw_payload"]

    assert payload["tag_query_relevance_score"] >= 80
    assert payload["next_query"] == payload["tag_text"]
    assert payload["tag_walk_status"] == "allow"


def test_v4_douyin_author_gate_contract_uses_elderly_ratio_and_tgi():
    author_asset = {
        "platform": "douyin",
        "elderly_ratio": 0.61,
        "elderly_tgi": 121,
        "eligible_as_source": 1,
        "profile_snapshot": {"source": "hotspotbao_account_profile"},
        "evidence_refs": {"decision_ids": ["decision_001"]},
    }

    assert author_asset["elderly_ratio"] > 0.6
    assert author_asset["elderly_tgi"] > 120
    assert author_asset["eligible_as_source"] == 1
    assert author_asset["profile_snapshot"]
    assert author_asset["evidence_refs"]


def test_v4_three_short_video_platform_author_work_edges_match_m3_contract():
    douyin = _json(DATA_DIR / "platform_profiles/douyin.json")
    kuaishou = _json(DATA_DIR / "platform_profiles/kuaishou.json")
    shipinhao = _json(DATA_DIR / "platform_profiles/shipinhao.json")

    assert douyin["edges"]["author_to_works"]["status"] == "supported"
    assert douyin["edges"]["author_work_to_content"]["status"] == "supported"
    assert kuaishou["edges"]["author_to_works"]["status"] == "supported"
    assert kuaishou["edges"]["author_work_to_content"]["status"] == "supported"
    assert kuaishou["endpoints"]["blogger"]["path"] == "/crawler/kuai_shou/blogger"
    assert shipinhao["edges"]["author_to_works"]["status"] == "blocked"
    assert shipinhao["edges"]["author_work_to_content"]["status"] == "blocked"
    assert shipinhao["endpoints"]["account_info"]["status"] == "blocked"


def test_v4_m2_platform_profiles_define_observable_fields():
    for platform in ["douyin", "kuaishou", "shipinhao"]:
        profile = _json(DATA_DIR / f"platform_profiles/{platform}.json")
        covered = {
            item["field"]
            for item in profile["observable_fields"] + profile["missing_observable_fields"]
        }

        assert covered == {
            "statistics.digg_count",
            "statistics.comment_count",
            "statistics.share_count",
            "statistics.collect_count",
            "statistics.play_count",
        }
        for item in profile["missing_observable_fields"]:
            assert item["missing_type"] in {"natural_platform_missing", "runtime_missing"}


def test_v4_video_and_author_dedup_keys_are_separate():
    dedup = _json(DATA_DIR / "walk_policy.json")["dedup"]

    assert "platform_content_id" in dedup["content_key"]
    assert "platform_author_id" in dedup["author_key"]
    assert dedup["content_key"] != dedup["author_key"]


def test_v4_walk_policy_defines_allow_walk_gate():
    gate = _json(DATA_DIR / "walk_policy.json")["v4_walk_gate"]

    assert gate["requires_allow_walk"] is True
    assert gate["deny_reason_code"] == "v4_allow_walk_denied"
    assert set(gate["applies_to_edges"]) == {
        "hashtag_to_query",
        "author_to_works",
    }
    assert {
        "decision_id",
        "allow_walk",
        "allow_walk_reason",
        "walk_gate_snapshot",
    } <= set(gate["raw_payload_fields"])


def test_v4_walk_strategy_json_and_excel_define_allow_walk_gate():
    import pytest

    openpyxl = pytest.importorskip("openpyxl")
    strategy = _json(ROOT / "product_documents/抖音游走策略/douyin_walk_strategy.v1.json")
    gate = strategy["v4_walk_gate"][0]

    assert gate["gate_id"] == "allow_walk_required"
    assert gate["deny_reason_code"] == "v4_allow_walk_denied"

    workbook = openpyxl.load_workbook(
        ROOT / "tech_documents/游走策略/游走策略配置表.xlsx",
        read_only=True,
        data_only=True,
    )
    assert "v4_walk_gate" in workbook.sheetnames
    rows = list(workbook["v4_walk_gate"].iter_rows(values_only=True))
    assert rows[0][:4] == (
        "gate_id",
        "requires_allow_walk",
        "source_field",
        "deny_reason_code",
    )
    assert rows[2][0] == "allow_walk_required"
    assert rows[2][3] == "v4_allow_walk_denied"


def _json(path: Path):
    return json.loads(path.read_text(encoding="utf-8"))