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

"""Real + synthetic case replay tests (V2-M0D).

- real_id45: the harvested production baseline (demand_content.id=45). V4-M3 受控变化:
  画像门槛整体退役,改由 Gemini query relevance + 平台可观测表现 50/50 打分。
- syn_pool / syn_review: synthetic corpora (authored with high/low engagement)
  exercise the ADD / KEEP paths via the same V4 scorecard.

Snapshots lock the deterministic replay output; regenerate with UPDATE_SNAPSHOTS=1.
"""

from __future__ import annotations

import copy
import json
from collections import Counter
from pathlib import Path
from typing import Any

from tests.replay_harness import CASES_DIR, replay_case

_SUMMARY_KEYS = [
    "pooled_content_count",
    "review_content_count",
    "rejected_content_count",
    "pending_content_count",
]


def _decision_counts(artifacts) -> dict[str, int]:
    return dict(Counter(d.get("decision_action") for d in artifacts.decisions))


def _build_synthetic_corpus(cases_dir: Path, case_id: str, items: list[dict[str, Any]]) -> None:
    """Author a minimal corpus: real (validated) source_context + given items."""
    source_context = json.loads(
        (CASES_DIR / "real_id45" / "input" / "source_context.json").read_text(encoding="utf-8")
    )
    dest = cases_dir / case_id / "input"
    dest.mkdir(parents=True, exist_ok=True)
    (dest / "source_context.json").write_text(
        json.dumps(source_context, ensure_ascii=False, indent=2), encoding="utf-8"
    )
    (dest / "discovered_content_items.jsonl").write_text(
        json.dumps(items, ensure_ascii=False, indent=2), encoding="utf-8"
    )


def _synthetic_item(
    content_id: str,
    *,
    digg: int,
    comment: int = 800,
    share: int = 600,
    collect: int = 5000,
) -> dict[str, Any]:
    return {
        "content_discovery_id": f"syn_{content_id}",
        "search_query_id": "q_001",
        "platform": "douyin",
        "platform_content_id": content_id,
        "platform_content_format": "video",
        "description": "中医养生合成内容",
        "platform_author_id": "syn_author",
        "author_display_name": "养生作者",
        "statistics": {
            "digg_count": digg,
            "comment_count": comment,
            "share_count": share,
            "collect_count": collect,
        },
        "tags": ["#中医养生"],
        "score": 85,
        "risk_level": "low",
        "availability": "available",
        "discovery_start_source": "pattern_itemset",
        "previous_discovery_step": "search_query_direct",
        "content_metadata_source": "synthetic",
    }


def test_replay_id45_baseline_gemini_score(tmp_path):
    artifacts = replay_case("real_id45", runtime_root=tmp_path / "rt")
    assert artifacts.state["status"] == "success"
    # M11 re-baseline:平台分改"量级+收缩比例"后,7406990358799732018(中等比例)平台分 60.5、
    # 总分 70.25 ≥ 70 → 由原"待复看"升为"入池"(allow_walk=False);7577667864522907506 比例极低
    # (官方大号式)平台分 25.1 → 仍淘汰。结果:3 入池 / 0 复看 / 1 淘汰。
    assert artifacts.summary["pooled_content_count"] == 3
    assert artifacts.summary["review_content_count"] == 0
    assert artifacts.summary["rejected_content_count"] == 1
    assert artifacts.summary["pending_content_count"] == 0
    assert _decision_counts(artifacts) == {
        "ADD_TO_CONTENT_POOL": 3,
        "REJECT_CONTENT": 1,
    }
    assert {d.get("decision_reason_code") for d in artifacts.decisions} == {
        "v4_query_and_platform_pass",
        "v4_query_or_score_below_threshold",
    }


def test_replay_synthetic_pool_case(tmp_path):
    # M11:平台分改"收缩后比例"后,光大点赞不入池;高共鸣占比(转/赞0.3、藏/赞0.5、评/赞0.08)
    # 才是该入池的画像 → 给强比例,平台分 ~88、总分 ~84 ≥ 70 入池。
    _build_synthetic_corpus(
        tmp_path / "cases",
        "syn_pool",
        [
            _synthetic_item(
                "9000000000000000001",
                digg=100_000,
                comment=8_000,
                share=30_000,
                collect=50_000,
            )
        ],
    )
    artifacts = replay_case("syn_pool", runtime_root=tmp_path / "rt", cases_dir=tmp_path / "cases")
    assert artifacts.state["status"] == "success"
    assert artifacts.summary["pooled_content_count"] >= 1
    assert artifacts.summary["rejected_content_count"] == 0


def test_replay_synthetic_review_case(tmp_path):
    _build_synthetic_corpus(
        tmp_path / "cases",
        "syn_review",
        [_synthetic_item("9000000000000000002", digg=50_000)],
    )
    artifacts = replay_case("syn_review", runtime_root=tmp_path / "rt", cases_dir=tmp_path / "cases")
    assert artifacts.state["status"] == "success"
    assert artifacts.summary["review_content_count"] >= 1
    assert artifacts.summary["pooled_content_count"] == 0


def test_replay_id45_walk_obeys_decisions_after_m4(tmp_path):
    artifacts = replay_case("real_id45", runtime_root=tmp_path / "rt")
    walk_actions = artifacts.files["walk_actions.jsonl"]

    # M8:query_next_page 边已删,翻页归渐进筛选;游走只剩 tag/author。tag 预算 3→5。
    assert not [row for row in walk_actions if row["edge_id"] == "query_next_page"]

    tag_actions = [row for row in walk_actions if row["edge_id"] == "hashtag_to_query"]
    executed_tags = [row for row in tag_actions if row["walk_status"] == "success"]
    skipped_tags = [row for row in tag_actions if row["walk_status"] == "skipped"]
    assert executed_tags
    assert all(row["budget_tier"] == "normal" for row in executed_tags)
    assert sorted(row["reason_code"] for row in skipped_tags) == [
        "v4_allow_walk_denied",
        "v4_allow_walk_denied",
    ]
    assert len(executed_tags) == 5

    author_actions = [row for row in walk_actions if row["edge_id"] == "author_to_works"]
    assert author_actions
    assert all(row["walk_status"] == "success" for row in author_actions)
    assert all(row["budget_tier"] == "normal" for row in author_actions)

    # M11 re-baseline:新平台分下原"待复看"的 7406990358799732018 升为"入池"(allow_walk=False),
    # 已无 KEEP 内容触发预算降级 → budget_downgrade 边为空。
    downgrades = [row for row in walk_actions if row["edge_id"] == "budget_downgrade"]
    assert len(downgrades) == 0

    for row in walk_actions:
        execution = row["raw_payload"]["rule_pack_execution"]
        assert execution["executed"] is True
        assert execution["executed_rule_pack_id"] == "douyin_content_discovery_rule_pack_v1"