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_first_round_detection_robust_to_old_labeling(): from content_agent.flow_ledger_service import _is_first_round_content # 旧 run:标签带回视频被误标 search_query_direct;只看 previous_discovery_step 会误判成首轮。 old = {"search_query_id": "tag_q", "previous_discovery_step": "search_query_direct"} assert _is_first_round_content(old) is True # 传入扩展 query 集合后(query→内容关联在旧 run 里仍正确),正确识别为游走带回。 assert _is_first_round_content(old, {"tag_q"}) is False # 真首轮内容不受影响。 assert _is_first_round_content({"search_query_id": "q1", "previous_discovery_step": "seed_term"}, {"tag_q"}) is True 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", "name": "睡前拉伸需求", "suggestion": "用户希望看到睡前放松内容", "reason": "来自 itemset 123", "merge_leve2": "PG Pattern", "score": 0.8, "dt": "20260615", "ext_data": { "desc": "用户希望看到睡前放松内容", "reason": "来自 itemset 123", "type": "pattern", "evidence_pack": { "source_post_id": "post_001", "pattern_execution_id": 581, "itemset_ids": [1608101], "seed_terms": ["睡前拉伸"], "absolute_support": 33, "category_bindings": [ { "category_id": 76006, "category_path": "/理念/观念/个人观念/人生观", "category_level": 4, "category_name": "人生观", "itemset_item_id": 6055, } ], "element_bindings": [ { "itemset_id": 1608101, "category_id": 76006, "sample_elements": [ { "post_id": "post_001", "point_text": "睡前肩颈放松", "point_type": "目的点", } ], } ], } }, "raw_demand_content": { "id": "demand_123", "name": "睡前拉伸需求", "suggestion": "用户希望看到睡前放松内容", "reason": "来自 itemset 123", "merge_leve2": "PG Pattern", "score": 0.8, "dt": "20260615", "ext_data": { "desc": "用户希望看到睡前放松内容", "reason": "来自 itemset 123", "type": "pattern", }, }, }, ) 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_003", "search_query": "情绪放松练习", "search_query_generation_method": "query_seed_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_tech", "platform_content_id": "douyin_tech", "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_no_valid", "platform_content_id": "shipinhao_no_valid", "search_query_id": "q_003", "platform": "shipinhao", "description": "无可用正片 URL", "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"}, }, { "run_id": run_id, "policy_run_id": policy_run_id, "platform": "douyin", "platform_content_id": "douyin_tech", "play_url": "https://example.com/raw-tech.mp4", "oss_url": "https://oss.example.com/tech.mp4", "local_path": None, "content_media_status": "oss_uploaded", "raw_payload": { "oss_archive_attempt_count": 1, "oss_timing_metrics": {"oss_upload_duration_ms": 1234}, "oss_payload_mode": "no_referer", "oss_response_summary": { "status": 10000, "msg": "bad", "oss_object_present": False, "oss_object_has_cdn_url": False, }, "selected_video_url_path": "$.search.video_url_list[0].video_url", "selected_video_url_host": "findermp.video.qq.com", "selected_video_url_probe_status": "verified", "selected_video_url_content_type": "video/mp4", "selected_video_url_content_range": "bytes 0-99/100", "video_url_candidate_counts": {"video": 1, "image": 1}, "video_url_reject_reasons": {"image_path": 1}, }, }, { "run_id": run_id, "policy_run_id": policy_run_id, "platform": "shipinhao", "platform_content_id": "shipinhao_no_valid", "play_url": None, "oss_url": None, "local_path": None, "content_media_status": "unavailable", "failure_reason": "no_valid_play_url", "raw_payload": { "failure_reason": "no_valid_play_url", "no_valid_play_url": True, "selected_video_url_path": "$.search.video_url_list[0].video_url", "selected_video_url_host": "findermp.video.qq.com", "selected_video_url_probe_status": "failed_fallback", "video_url_candidate_counts": {"video": 1, "image": 1}, "video_url_reject_reasons": {"image_path": 1, "probe_failed": 1}, }, }, ], ) 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": {}, }, { "run_id": run_id, "policy_run_id": policy_run_id, "recall_evidence_id": "evidence_tech", "platform_content_id": "douyin_tech", "recall_status": "failed", "evidence_summary": { "schema_version": "v4_gemini_query_relevance.v1", "failure_type": "gemini_client_timeout", "exception_type": "WriteTimeout", "error_message": "The write operation timed out", "retry_count": 2, "final_status": "failed", "response_body_summary": { "http_status_code": 502, "content_type": "application/json", "json_top_level_keys": ["error"], "text_excerpt": "upstream overloaded", }, "timing_metrics": { "video_fetch": { "gemini_video_source": "oss_url", "download_duration_ms": 1500, "ffmpeg_duration_ms": 2600, "total_duration_ms": 4100, "downloaded_bytes": 10485760, "compressed_bytes": 1048576, }, "gemini_request": { "total_duration_ms": 180000, "attempts": [ { "attempt": 1, "status": "failed", "duration_ms": 90000, "failure_type": "gemini_client_timeout", "exception_type": "WriteTimeout", "http_status_code": None, "response_body_summary": { "http_status_code": 502, "text_excerpt": "upstream overloaded", }, } ], }, }, }, "raw_payload": {}, }, { "run_id": run_id, "policy_run_id": policy_run_id, "recall_evidence_id": "evidence_no_valid", "platform_content_id": "shipinhao_no_valid", "recall_status": "failed", "evidence_summary": { "schema_version": "v4_gemini_query_relevance.v1", "failure_type": "no_valid_play_url", "final_status": "failed", "retry_count": 1, "timing_metrics": {"video_fetch": {"skipped": "no_valid_play_url"}}, }, "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": {}, }, { "run_id": run_id, "policy_run_id": policy_run_id, "decision_id": "decision_tech", "search_query_id": "q_003", "decision_target_id": "content_tech", "decision_action": "TECHNICAL_RETRY_REQUIRED", "decision_reason_code": "v4_technical_retry_needed", "score": None, "scorecard": { "failure_type": "gemini_client_timeout", "exception_type": "WriteTimeout", "retry_count": 2, "final_status": "failed", "total_score": None, }, "decision_replay_data": {"allow_walk": False}, "raw_payload": {}, }, { "run_id": run_id, "policy_run_id": policy_run_id, "decision_id": "decision_no_valid", "search_query_id": "q_003", "decision_target_id": "content_no_valid", "decision_action": "TECHNICAL_RETRY_REQUIRED", "decision_reason_code": "v4_technical_retry_needed", "score": None, "scorecard": { "failure_type": "no_valid_play_url", "final_status": "failed", "total_score": None, }, "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": {}, }, ], ) 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 ledger["demand_summary"]["id"] == "demand_123" assert ledger["demand_summary"]["name"] == "睡前拉伸需求" assert ledger["demand_summary"]["description"] == "用户希望看到睡前放松内容" assert ledger["demand_summary"]["itemset_ids"] == ["1608101"] assert ledger["demand_summary"]["category_paths"] == ["/理念/观念/个人观念/人生观"] assert ledger["demand_summary"]["extracted_points"] == ["睡前肩颈放松（目的点 / 帖子 post_001）"] assert len(ledger["rows"]) == 3 assert ledger["summary"]["first_round_query_count"] == 3 assert ledger["summary"]["extension_query_count"] == 1 # M9 游走指标(run 级):键齐全且为数值。 for key in ("walk_video_total", "walk_pooled_count", "walk_action_total", "walk_action_success", "walk_max_depth"): assert isinstance(ledger["summary"][key], int) assert ledger["summary"]["walk_action_total"] >= 1 # fixture 有 1 个 hashtag_to_query success 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" # 这条视频真正向外走过 1 次(walk_001 hashtag success),首页据此显示「看扩展过程」。 assert row["first_round_videos"][0]["walk_out_count"] == 1 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"] assert category["asset_summary"]["pool_count"] == 0 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" tech_videos = client.get(f"/runs/{run_id}/flow-ledger/queries/q_003/videos").json() tech_video = next(item for item in tech_videos["videos"] if item["platform_content_id"] == "douyin_tech") assert tech_video["decision"]["label"] == "技术问题" assert tech_video["technical_retry_detail"]["stage_label"] == "OpenRouter/Gemini" assert tech_video["technical_retry_detail"]["failure_type"] == "gemini_client_timeout" assert tech_video["technical_retry_detail"]["timings"]["download_seconds"] == 1.5 assert tech_video["technical_retry_detail"]["timings"]["ffmpeg_seconds"] == 2.6 assert tech_video["technical_retry_detail"]["timings"]["gemini_seconds"] == 180.0 assert tech_video["technical_retry_detail"]["sizes"]["downloaded_mb"] == 10.0 assert tech_video["technical_retry_detail"]["attempts"][0]["exception_type"] == "WriteTimeout" assert tech_video["technical_retry_detail"]["attempts"][0]["response_body_summary"]["http_status_code"] == 502 assert tech_video["technical_retry_detail"]["openrouter"]["response_summary"]["text_excerpt"] == "upstream overloaded" assert tech_video["technical_retry_detail"]["oss"]["payload_mode"] == "no_referer" assert tech_video["technical_retry_detail"]["oss"]["response_summary"]["status"] == 10000 assert tech_video["technical_retry_detail"]["video_url"]["selected_host"] == "findermp.video.qq.com" assert tech_video["technical_retry_detail"]["video_url"]["probe_status"] == "verified" assert tech_video["technical_retry_detail"]["video_url"]["candidate_counts"] == {"video": 1, "image": 1} no_valid_video = next(item for item in tech_videos["videos"] if item["platform_content_id"] == "shipinhao_no_valid") assert no_valid_video["technical_retry_detail"]["failure_type"] == "no_valid_play_url" assert no_valid_video["technical_retry_detail"]["failure_label"] == "未找到可用正片 URL" assert no_valid_video["technical_retry_detail"]["brief_reason"] == "平台结果未找到可用正片 URL" assert no_valid_video["technical_retry_detail"]["video_url"]["probe_status"] == "failed_fallback" walk = client.get(f"/runs/{run_id}/flow-ledger/queries/q_001/walk").json() assert walk["summary"]["total_actions"] == 1 assert walk["summary"]["extension_query_count"] == 1 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 分" # M8:query_next_page 边已删,原翻页 lane 断言随之移除。 # M9+ 全 run 游走图:不按 query 过滤,附游走预算总览。 run_walk = client.get(f"/runs/{run_id}/flow-ledger/walk").json() assert run_walk["summary"]["total_actions"] >= 1 assert run_walk["summary"]["tag_used"] == 1 # walk_001 hashtag success assert run_walk["summary"]["tag_cap"] == 5 assert run_walk["summary"]["author_used"] == 0 assert run_walk["summary"]["author_cap"] == 5 assert run_walk["summary"]["max_depth_reached"] == 1 assert run_walk["summary"]["max_depth_cap"] == 5 assert run_walk["summary"]["walk_video_total"] == 1 # douyin_002 是游走带回 assert "q_001" in run_walk["summary"]["query_labels"] assert any(a["source_video"]["title"] == "睡前 5 分钟肩颈拉伸" for a in run_walk["lanes"]["tag_query"]) # M9+ 单视频子树:从 douyin_001 出发,只保留它走出去的那一支并递归。 video_walk = client.get(f"/runs/{run_id}/flow-ledger/videos/douyin_001/walk").json() assert video_walk["root_video"]["id"] == "douyin_001" assert video_walk["summary"]["total_actions"] >= 1 assert video_walk["actions"][0]["edge_id"] == "hashtag_to_query" assert video_walk["actions"][0]["target_videos"][0]["title"] == "扩展搜到的视频" # 没向外走过的视频(douyin_003):子树为空。 empty_walk = client.get(f"/runs/{run_id}/flow-ledger/videos/douyin_003/walk").json() assert empty_walk["root_video"]["id"] == "douyin_003" assert empty_walk["summary"]["total_actions"] == 0 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