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

from fastapi.testclient import TestClient

from content_agent import api
from content_agent.flow_ledger_service import _FLOW_LEDGER_CACHE
from content_agent.run_service import RunService


def test_flow_ledger_api_returns_business_v4_ledger(tmp_path, monkeypatch):
    _FLOW_LEDGER_CACHE.clear()
    service = RunService(runtime_root=tmp_path / "runtime" / "v1")
    run_id = "run_flow_ledger"
    policy_run_id = "policy_flow_ledger"
    service.runtime.prepare_run(run_id)
    service.runtime.write_json(
        run_id,
        "source_context.json",
        {
            "run_id": run_id,
            "demand_content_id": "demand_123",
            "ext_data": {
                "evidence_pack": {
                    "source_post_id": "post_001",
                    "pattern_execution_id": 581,
                    "category_bindings": [
                        {
                            "category_id": 76006,
                            "category_path": "/理念/观念/个人观念/人生观",
                            "category_level": 4,
                            "category_name": "人生观",
                            "itemset_item_id": 6055,
                        }
                    ],
                }
            },
        },
    )
    service.runtime.write_json(
        run_id,
        "pattern_seed_pack.json",
        {
            "run_id": run_id,
            "policy_run_id": policy_run_id,
            "pattern_execution_id": 581,
            "source_post_id": "post_001",
            "element_bindings": [],
        },
    )
    service.runtime.append_jsonl(
        run_id,
        "search_queries.jsonl",
        [
            {
                "run_id": run_id,
                "policy_run_id": policy_run_id,
                "search_query_id": "q_001",
                "search_query": "睡前拉伸",
                "search_query_generation_method": "seed_term",
                "pattern_seed_ref": {"query_source_text": "睡前拉伸", "source_post_id": "seed_001"},
                "query_source_terms": ["睡前拉伸"],
                "raw_payload": {"query_source_refs": [{"query_source_type": "seed_term"}]},
            },
            {
                "run_id": run_id,
                "policy_run_id": policy_run_id,
                "search_query_id": "q_002",
                "search_query": "肩颈放松",
                "search_query_generation_method": "tag_query",
                "raw_payload": {"parent_search_query_id": "q_001", "hashtag": "肩颈放松"},
            },
            {
                "run_id": run_id,
                "policy_run_id": policy_run_id,
                "search_query_id": "q_001_page_002",
                "search_query": "睡前拉伸",
                "search_query_generation_method": "query_next_page",
                "raw_payload": {"parent_search_query_id": "q_001"},
            },
            {
                "run_id": run_id,
                "policy_run_id": policy_run_id,
                "search_query_id": "q_003",
                "search_query": "情绪放松练习",
                "search_query_generation_method": "piaoquan_topic_point",
                "pattern_seed_ref": {"query_source_text": "情绪放松练习", "source_post_id": "post_001"},
                "raw_payload": {
                    "query_source_refs": [
                        {
                            "query_source_type": "piaoquan_point_text",
                            "source_ref": {
                                "post_id": "post_888",
                                "point_id": "point_456",
                                "point_type": "灵感点",
                                "point_text": "情绪放松练习",
                            },
                        }
                    ]
                },
            },
            {
                "run_id": run_id,
                "policy_run_id": policy_run_id,
                "search_query_id": "q_004",
                "search_query": "人生观",
                "search_query_generation_method": "category_leaf_element",
                "pattern_seed_ref": {
                    "query_source_text": "人生观",
                    "source_post_id": "post_001",
                    "pattern_execution_id": 581,
                },
                "raw_payload": {
                    "query_source_refs": [
                        {
                            "query_source_type": "category_terminal_element",
                            "source_ref": {
                                "category_id": 76006,
                                "element_name": "人生观",
                                "itemset_item_id": 6055,
                            },
                        }
                    ]
                },
            },
        ],
    )
    service.runtime.append_jsonl(
        run_id,
        "discovered_content_items.jsonl",
        [
            {
                "run_id": run_id,
                "policy_run_id": policy_run_id,
                "content_discovery_id": "content_001",
                "platform_content_id": "douyin_001",
                "search_query_id": "q_001",
                "platform": "douyin",
                "description": "睡前 5 分钟肩颈拉伸",
                "author_display_name": "健康教练",
                "tags": ["肩颈", "拉伸"],
                "statistics": {"digg_count": 100, "comment_count": 8, "share_count": 3},
                "raw_payload": {},
            },
            {
                "run_id": run_id,
                "policy_run_id": policy_run_id,
                "content_discovery_id": "content_002",
                "platform_content_id": "douyin_002",
                "search_query_id": "q_002",
                "previous_discovery_step": "hashtag_to_query",
                "platform": "douyin",
                "description": "扩展搜到的视频",
                "statistics": {},
                "raw_payload": {},
            },
            {
                "run_id": run_id,
                "policy_run_id": policy_run_id,
                "content_discovery_id": "content_003",
                "platform_content_id": "douyin_003",
                "search_query_id": "q_003",
                "platform": "douyin",
                "description": "情绪放松练习视频",
                "author_display_name": "心理教练",
                "statistics": {},
                "raw_payload": {},
            },
            {
                "run_id": run_id,
                "policy_run_id": policy_run_id,
                "content_discovery_id": "content_004",
                "platform_content_id": "douyin_004",
                "search_query_id": "q_001_page_002",
                "previous_discovery_step": "query_next_page",
                "platform": "douyin",
                "description": "翻页搜到的视频",
                "author_display_name": "拉伸助手",
                "statistics": {},
                "raw_payload": {},
            },
        ],
    )
    service.runtime.append_jsonl(
        run_id,
        "content_media_records.jsonl",
        [
            {
                "run_id": run_id,
                "policy_run_id": policy_run_id,
                "platform": "douyin",
                "platform_content_id": "douyin_001",
                "play_url": "https://example.com/raw.mp4",
                "oss_url": "https://oss.example.com/video.mp4",
                "local_path": None,
                "content_media_status": "oss_uploaded",
                "raw_payload": {"oss_object_key": "video.mp4"},
            }
        ],
    )
    service.runtime.append_jsonl(
        run_id,
        "pattern_recall_evidence.jsonl",
        [
            {
                "run_id": run_id,
                "policy_run_id": policy_run_id,
                "recall_evidence_id": "evidence_001",
                "platform_content_id": "douyin_001",
                "decode_status": "ok",
                "raw_payload": {},
            }
        ],
    )
    service.runtime.append_jsonl(
        run_id,
        "rule_decisions.jsonl",
        [
            {
                "run_id": run_id,
                "policy_run_id": policy_run_id,
                "decision_id": "decision_001",
                "search_query_id": "q_001",
                "decision_target_id": "content_001",
                "decision_action": "ADD_TO_CONTENT_POOL",
                "decision_reason_code": "v4_query_and_platform_pass",
                "score": 82,
                "scorecard": {
                    "total_score": 82,
                    "query_relevance_score": 90,
                    "platform_performance_score": 74,
                    "platform_performance_components": [
                        {"field": "statistics.digg_count", "value": 100, "weight": 0.4, "normalized_score": 80},
                        {"field": "statistics.comment_count", "value": 8, "weight": 0.2, "normalized_score": 70},
                    ],
                },
                "decision_replay_data": {
                    "allow_walk": True,
                    "walk_gate_snapshot": {
                        "score": 82,
                        "query_relevance_score": 90,
                        "platform_performance_score": 74,
                    },
                },
                "raw_payload": {},
            },
            {
                "run_id": run_id,
                "policy_run_id": policy_run_id,
                "decision_id": "decision_003",
                "search_query_id": "q_003",
                "decision_target_id": "content_003",
                "decision_action": "KEEP_CONTENT_FOR_REVIEW",
                "decision_reason_code": "v4_score_review_needed",
                "score": 62,
                "scorecard": {
                    "total_score": 62,
                    "query_relevance_score": 95,
                    "platform_performance_score": 29,
                },
                "decision_replay_data": {"allow_walk": False},
                "raw_payload": {},
            }
        ],
    )
    service.runtime.append_jsonl(
        run_id,
        "walk_actions.jsonl",
        [
            {
                "run_id": run_id,
                "policy_run_id": policy_run_id,
                "walk_action_id": "walk_001",
                "edge_id": "hashtag_to_query",
                "hashtag": "拉伸",
                "depth": 1,
                "from_node_type": "Content",
                "from_node_id": "douyin_001",
                "to_node_id": "q_002",
                "walk_status": "success",
                "reason_code": "v4_query_and_platform_pass",
                "walk_gate_snapshot": {
                    "score": 82,
                    "query_relevance_score": 90,
                    "platform_performance_score": 74,
                },
                "raw_payload": {},
            },
            {
                "run_id": run_id,
                "policy_run_id": policy_run_id,
                "walk_action_id": "walk_page_001",
                "edge_id": "query_next_page",
                "depth": 1,
                "from_node_type": "SearchQuery",
                "from_node_id": "q_001",
                "to_node_type": "SearchQuery",
                "to_node_id": "q_001_page_002",
                "decision_id": "decision_001",
                "walk_status": "success",
                "walk_action": "fetch_next_page",
                "page_cursor": "10",
                "reason_code": "v4_query_and_platform_pass",
                "walk_gate_snapshot": {
                    "score": 82,
                    "query_relevance_score": 90,
                    "platform_performance_score": 74,
                },
                "raw_payload": {},
            },
        ],
    )
    service.runtime.write_json(
        run_id,
        "final_output.json",
        {
            "run_id": run_id,
            "policy_run_id": policy_run_id,
            "content_assets": [{"decision_id": "decision_001", "platform_content_id": "douyin_001"}],
            "review_records": [{"decision_id": "decision_003", "platform_content_id": "douyin_003"}],
            "reject_records": [],
            "summary": {},
        },
    )
    monkeypatch.setattr(api, "service", service)
    client = TestClient(api.app)

    ledger = client.get(f"/runs/{run_id}/flow-ledger").json()
    assert ledger["data_origin_label"] == "本地调试缓存"
    assert len(ledger["rows"]) == 3
    assert ledger["summary"]["first_round_query_count"] == 3
    assert ledger["summary"]["extension_query_count"] == 2
    row = next(item for item in ledger["rows"] if item["id"] == "q_001")
    assert row["query"]["method_label"] == "Pattern 种子词"
    assert row["source"]["details"] == ["需求单号：demand_123", "种子词：睡前拉伸", "来源帖子：seed_001"]
    assert row["first_round_video_total"] == 1
    assert row["video_stats"]["oss_uploaded_count"] == 1
    assert row["rule_summary"]["pool_count"] == 1
    assert row["rule_summary"]["score_summary"]["items"][0]["label"] == "总分"
    assert row["rule_summary"]["score_summary"]["items"][0]["score_label"] == "平均 82 分"
    assert row["rule_summary"]["score_summary"]["items"][1]["label"] == "是否契合需求"
    assert row["rule_summary"]["score_summary"]["items"][1]["score_label"] == "平均 90 分"
    assert row["first_round_videos"][0]["decision"]["label"] == "入池"
    assert row["first_round_videos"][0]["media_status_label"] == "已保存"
    assert row["first_round_videos"][0]["platform_url"] == "https://www.douyin.com/video/douyin_001"
    assert row["first_round_videos"][0]["decision"]["score_items"][1]["label"] == "是否契合需求"
    assert row["first_round_videos"][0]["decision"]["score_items"][3]["label"] == "点赞表现"
    piaoquan = next(item for item in ledger["rows"] if item["id"] == "q_003")
    assert piaoquan["source"]["label"] == "票圈帖子具体的点"
    assert "帖子 ID：post_888" in piaoquan["source"]["details"]
    assert "内容点 ID：point_456" in piaoquan["source"]["details"]
    assert piaoquan["asset_summary"]["pool_count"] == 0
    assert piaoquan["asset_summary"]["review_count"] == 1
    assert piaoquan["asset_summary"]["headline"] == "待复看 1 条"
    category = next(item for item in ledger["rows"] if item["id"] == "q_004")
    assert "分类树：Pattern 执行 581" in category["source"]["details"]
    assert "分类路径：/理念/观念/个人观念/人生观" in category["source"]["details"]
    assert "所在层级：第 4 层" in category["source"]["details"]

    videos = client.get(f"/runs/{run_id}/flow-ledger/queries/q_001/videos").json()
    assert videos["summary"]["total"] == 1
    assert videos["videos"][0]["oss_url"] == "https://oss.example.com/video.mp4"

    walk = client.get(f"/runs/{run_id}/flow-ledger/queries/q_001/walk").json()
    assert walk["summary"]["total_actions"] == 2
    assert walk["summary"]["extension_query_count"] == 2
    assert walk["lanes"]["tag_query"][0]["edge_label"] == "从视频标签继续游走"
    assert walk["lanes"]["tag_query"][0]["source_video"]["title"] == "睡前 5 分钟肩颈拉伸"
    assert walk["lanes"]["tag_query"][0]["trigger_label"] == "标签：#拉伸"
    assert walk["lanes"]["tag_query"][0]["target_query"]["text"] == "肩颈放松"
    assert walk["lanes"]["tag_query"][0]["target_videos"][0]["title"] == "扩展搜到的视频"
    assert "起点：从视频《睡前 5 分钟肩颈拉伸》触发" in walk["lanes"]["tag_query"][0]["detail_lines"]
    assert "结果：已生成标签搜索词《肩颈放松》，带回 1 条视频。" in walk["lanes"]["tag_query"][0]["detail_lines"]
    assert walk["lanes"]["tag_query"][0]["score_items"][0]["score_label"] == "82 分"
    page_action = walk["lanes"]["query_next_page"][0]
    assert page_action["edge_label"] == "翻页继续找"
    assert page_action["source_video"] is None
    assert page_action["from_label"] == "从搜索词《睡前拉伸》继续"
    assert page_action["target_query"]["id"] == "q_001_page_002"
    assert page_action["target_videos"][0]["title"] == "翻页搜到的视频"
    assert "起点：从搜索词《睡前拉伸》继续" in page_action["detail_lines"]

    detail = client.get(f"/runs/{run_id}/flow-ledger/videos/douyin_001").json()
    assert detail["video"]["playable_url"] == "https://oss.example.com/video.mp4"
    assert detail["decision"]["label"] == "入池"

    service.runtime.append_jsonl(
        run_id,
        "search_queries.jsonl",
        [
            {
                "run_id": run_id,
                "policy_run_id": policy_run_id,
                "search_query_id": "q_005",
                "search_query": "新搜索词",
                "search_query_generation_method": "seed_term",
                "raw_payload": {},
            }
        ],
    )
    cached = client.get(f"/runs/{run_id}/flow-ledger").json()
    assert cached["summary"]["first_round_query_count"] == 3
    debug_fresh = client.get(f"/runs/{run_id}/flow-ledger?debug=true").json()
    assert debug_fresh["summary"]["first_round_query_count"] == 4