from __future__ import annotations import copy import json import os import time from collections import Counter from typing import Any from content_agent.dashboard_service import ( DATA_ORIGIN_PRODUCTION_DB, DATA_ORIGIN_RUNTIME_EXPORT, DashboardService, ) from content_agent.integrations.database_runtime import DatabaseRuntimeStore from content_agent.interfaces import RuntimeStore FIRST_ROUND_QUERY_METHODS = { "seed_term", "piaoquan_topic_point", "query_seed_point", "category_leaf_element", } EXTENSION_QUERY_METHODS = { "tag_query", "author_works", } SOURCE_LABELS = { "seed_term": "从需求单的 pattern 词来", "piaoquan_topic_point": "票圈帖子具体的点", "query_seed_point": "票圈帖子具体的点", "category_leaf_element": "分类叶子元素", } QUERY_METHOD_LABELS = { "seed_term": "从需求单的 pattern 词来", "piaoquan_topic_point": "票圈帖子具体的点", "query_seed_point": "票圈帖子具体的点", "category_leaf_element": "分类叶子元素", "tag_query": "从视频标签继续游走", "author_works": "从作者继续游走", } ACTION_LABELS = { "ADD_TO_CONTENT_POOL": "入池", "KEEP_CONTENT_FOR_REVIEW": "待复看", "REJECT_CONTENT": "淘汰", "TECHNICAL_RETRY_REQUIRED": "技术问题", } MEDIA_STATUS_LABELS = { "oss_uploaded": "已保存", "oss_upload_pending": "待补传", "oss_upload_failed": "补传失败", "metadata_only": "尚未拿到视频", "unavailable": "无可用视频", } WALK_EDGE_LABELS = { "hashtag_to_query": "从视频标签继续游走", "tag_query": "从视频标签继续游走", "author_to_works": "从作者继续游走", "video_to_author": "从作者继续游走", "author_works": "从作者继续游走", "path_stop": "到这里停止", "decision_to_asset": "入池沉淀", "budget_downgrade": "转入待复看", "terminal": "到这里停止", } REASON_LABELS = { "content_pattern_recall_required": "内容和本次需求没有形成有效关联", "blocked_by_rule_decision": "上一步内容未通过判断", "missing_content_portrait": "缺少判断所需的内容画像", "category_or_element_binding_required": "没有命中本次需求的类目或要素", "missing_score": "没有足够信息进入打分", "v4_query_and_platform_pass": "搜索词和平台表现都符合要求", "v4_query_or_score_below_threshold": "相关性或平台表现未达到要求", "v4_score_review_needed": "分数处在复看区间，需要人工确认", "v4_technical_retry_needed": "系统需要重试后再判断", "v4_allow_walk_denied": "不适合继续向外扩展", "budget_exhausted": "达到该类游走次数上限", "global_action_cap_reached": "已达全 run 游走次数上限", "edge_blocked_by_platform_profile": "该平台 profile 阻断这种游走", "portrait_unavailable": "拉不到作者 50+ 画像(热点宝接口失败),无法判定", "portrait_incomplete": "作者 50+ 画像字段缺失,无法判定", "query>=65/platform>=65/score>=65": "评分达到继续扩展门槛", "query>=70/platform>=65/score>=70": "评分达到继续扩展门槛", "content_decision_reused_for_walk_gate": "复用本条视频的内容判断作为扩展门槛", "no_decision": "暂未产生判断结果", } # 向外游走边(标签/作者),用于统计游走次数(不含 path_stop/decision_to_asset 等终端结局)。 _WALK_EXPANSION_EDGES = {"hashtag_to_query", "author_to_works"} _FLOW_LEDGER_CACHE: dict[tuple[str, str], tuple[float, dict[str, Any]]] = {} _DEFAULT_CACHE_TTL_SECONDS = 300 def _reason_label(reason: str, default: str) -> str: if reason.startswith("query>=") and "/platform>=" in reason and "/score>=" in reason: return "评分达到继续扩展门槛" return REASON_LABELS.get(reason, default) class FlowLedgerService: def __init__(self, runtime: RuntimeStore) -> None: dashboard = DashboardService.from_runtime(runtime) self.runtime = dashboard.runtime self.export_runtime = dashboard.export_runtime self.data_origin = ( DATA_ORIGIN_PRODUCTION_DB if isinstance(self.runtime, DatabaseRuntimeStore) else DATA_ORIGIN_RUNTIME_EXPORT ) def ledger(self, run_id: str, *, debug: bool = False) -> dict[str, Any]: cache_key = ("ledger", run_id) if not debug: cached = _cache_get(cache_key) if cached is not None: return cached bundle = self._bundle(run_id, cache=not debug) walk_out = _walk_out_counter(bundle) rows = [ self._ledger_row(query, bundle, walk_out, debug=debug) for query in bundle["queries"] if _query_method(query) in FIRST_ROUND_QUERY_METHODS ] extension_queries = [ self._extension_query(query, bundle, debug=debug) for query in bundle["queries"] if _query_method(query) in EXTENSION_QUERY_METHODS ] media_counts = Counter(_media_status(media) for media in bundle["media"]) action_counts = Counter(_decision_action(decision) for decision in bundle["decisions"]) gemini_counts = Counter(_text(evidence.get("decode_status") or evidence.get("gemini_status"), "unknown") for evidence in bundle["evidence"]) payload = { "run_id": run_id, "data_origin": self.data_origin, "data_origin_label": "生产数据库" if self.data_origin == DATA_ORIGIN_PRODUCTION_DB else "本地调试缓存", "summary": self._summary(bundle, rows, extension_queries, media_counts, action_counts, gemini_counts), "demand_summary": _demand_summary(bundle), "rows": rows, "extension_queries": extension_queries, "debug": bundle if debug else None, } if not debug: _cache_set(cache_key, payload) return payload def query_videos(self, run_id: str, query_id: str, *, debug: bool = False) -> dict[str, Any]: bundle = self._bundle(run_id, cache=not debug) query = self._query_by_id(bundle, query_id) related = self._query_content(query_id, bundle, first_round_only=True) videos = [self._video(content, bundle, debug=debug) for content in related] return { "run_id": run_id, "query": self._query_summary(query, bundle, debug=debug) if query else None, "videos": videos, "summary": self._video_summary(videos), "data_origin": self.data_origin, "data_origin_label": "生产数据库" if self.data_origin == DATA_ORIGIN_PRODUCTION_DB else "本地调试缓存", } def query_walk(self, run_id: str, query_id: str, *, debug: bool = False) -> dict[str, Any]: bundle = self._bundle(run_id, cache=not debug) query = self._query_by_id(bundle, query_id) related_ids = self._query_related_content_ids(query_id, bundle) related_decision_ids = self._query_related_decision_ids(query_id, bundle) actions = [ self._walk_action(action, query, bundle, debug=debug) for action in bundle["walk_actions"] if _walk_action_related(action, query_id, related_ids) or self._walk_action_related_to_decisions(action, related_decision_ids) ] lanes: dict[str, list[dict[str, Any]]] = {key: [] for key in ["tag_query", "author_works", "path_stop"]} for action in actions: lanes[_walk_lane(action["edge_id"])].append(action) target_query_ids = {_text(_record(action.get("target_query")).get("id")) for action in actions} extension_queries = [ self._extension_query(item, bundle, debug=debug) for item in bundle["queries"] if _query_method(item) in EXTENSION_QUERY_METHODS and ( _text(item.get("search_query_id")) in target_query_ids or _text(_record(item.get("raw_payload")).get("parent_search_query_id") or item.get("llm_variant_of")) == query_id ) ] return { "run_id": run_id, "query": self._query_summary(query, bundle, debug=debug) if query else None, "actions": actions, "lanes": lanes, "extension_queries": extension_queries, "summary": { "total_actions": len(actions), "success_count": sum(1 for item in actions if item["status"] == "success"), "skipped_count": sum(1 for item in actions if item["status"] == "skipped"), "extension_query_count": len(extension_queries), }, "data_origin": self.data_origin, "data_origin_label": "生产数据库" if self.data_origin == DATA_ORIGIN_PRODUCTION_DB else "本地调试缓存", } def run_walk(self, run_id: str, *, debug: bool = False) -> dict[str, Any]: # 全 run 游走图:不按 query 过滤,富化全部 walk_actions(full=True 不采样带回视频),并给出游走预算总览。 bundle = self._bundle(run_id, cache=not debug) actions = [self._walk_action(action, None, bundle, debug=debug, full=True) for action in bundle["walk_actions"]] lanes: dict[str, list[dict[str, Any]]] = {key: [] for key in ["tag_query", "author_works", "path_stop"]} for action in actions: lanes[_walk_lane(action["edge_id"])].append(action) extension_queries = [ self._extension_query(item, bundle, debug=debug) for item in bundle["queries"] if _query_method(item) in EXTENSION_QUERY_METHODS ] summary = _walk_budget_overview(bundle, actions, extension_queries) # 全run游走图按所属搜索词分组用:首轮 query id → 文案。 summary["query_labels"] = { _text(item.get("search_query_id")): _text(item.get("search_query"), "搜索词待确认") for item in bundle["queries"] if _query_method(item) in FIRST_ROUND_QUERY_METHODS and _text(item.get("search_query_id")) } return { "run_id": run_id, "actions": actions, "lanes": lanes, "extension_queries": extension_queries, "summary": summary, "data_origin": self.data_origin, "data_origin_label": "生产数据库" if self.data_origin == DATA_ORIGIN_PRODUCTION_DB else "本地调试缓存", } def video_walk(self, run_id: str, content_id: str, *, debug: bool = False) -> dict[str, Any]: # 单视频游走子树:从该视频出发 BFS,只保留从它向外走出去的那一支(沿 source_video.id → target_videos.id 递归)。 bundle = self._bundle(run_id, cache=not debug) root_content = next( ( item for item in bundle["content_items"] if content_id in {_text(item.get("platform_content_id")), _text(item.get("content_discovery_id"))} ), None, ) origin_label = "生产数据库" if self.data_origin == DATA_ORIGIN_PRODUCTION_DB else "本地调试缓存" if not root_content: return { "run_id": run_id, "root_video": None, "actions": [], "summary": {"total_actions": 0, "success_count": 0, "skipped_count": 0}, "data_origin": self.data_origin, "data_origin_label": origin_label, } root_video = self._video(root_content, bundle, debug=debug) enriched = [self._walk_action(action, None, bundle, debug=debug, full=True) for action in bundle["walk_actions"]] by_source: dict[str, list[dict[str, Any]]] = {} for item in enriched: source_id = _text(_record(item.get("source_video")).get("id")) if source_id: by_source.setdefault(source_id, []).append(item) root_id = _text(root_video.get("id")) seen_videos = {root_id} seen_actions: set[str] = set() subtree: list[dict[str, Any]] = [] frontier = [root_id] while frontier: nxt: list[str] = [] for video_id in frontier: for item in by_source.get(video_id, []): action_id = _text(item.get("id")) or str(id(item)) if action_id in seen_actions: continue seen_actions.add(action_id) subtree.append(item) for target in item.get("target_videos") or []: target_id = _text(_record(target).get("id")) if target_id and target_id not in seen_videos: seen_videos.add(target_id) nxt.append(target_id) frontier = nxt statuses = Counter(_text(item.get("status"), "unknown") for item in subtree) return { "run_id": run_id, "root_video": root_video, "actions": subtree, "summary": { "total_actions": len(subtree), "success_count": statuses["success"], "skipped_count": statuses["skipped"], }, "data_origin": self.data_origin, "data_origin_label": origin_label, } def video_detail(self, run_id: str, content_id: str, *, debug: bool = False) -> dict[str, Any]: bundle = self._bundle(run_id, cache=not debug) content = next( ( item for item in bundle["content_items"] if content_id in { _text(item.get("platform_content_id")), _text(item.get("content_discovery_id")), } ), None, ) if not content: return { "run_id": run_id, "video": None, "data_origin": self.data_origin, "data_origin_label": "生产数据库" if self.data_origin == DATA_ORIGIN_PRODUCTION_DB else "本地调试缓存", } video = self._video(content, bundle, debug=debug) platform_id = _text(content.get("platform_content_id")) decision = _decision_for_content(content, bundle["decisions"]) source_paths = [ path for path in bundle["source_paths"] if platform_id and platform_id in {_text(path.get("from_node_id")), _text(path.get("to_node_id")), _text(path.get("decision_target_id"))} ] return { "run_id": run_id, "video": video, "source_paths": [_source_path(path, debug=debug) for path in source_paths], "decision": self._decision_summary(decision, debug=debug) if decision else None, "data_origin": self.data_origin, "data_origin_label": "生产数据库" if self.data_origin == DATA_ORIGIN_PRODUCTION_DB else "本地调试缓存", } def _bundle(self, run_id: str, *, cache: bool = True) -> dict[str, Any]: cache_key = ("bundle", run_id) if cache: cached = _cache_get(cache_key) if cached is not None: return cached payload = { "source_context": self._read_json(run_id, "source_context.json") or {}, "pattern_seed_pack": self._read_json(run_id, "pattern_seed_pack.json") or {}, "queries": self._read_jsonl(run_id, "search_queries.jsonl"), "content_items": self._read_jsonl(run_id, "discovered_content_items.jsonl"), "media": self._read_jsonl(run_id, "content_media_records.jsonl"), "evidence": self._read_jsonl(run_id, "pattern_recall_evidence.jsonl"), "decisions": self._read_jsonl(run_id, "rule_decisions.jsonl"), "walk_actions": self._read_jsonl(run_id, "walk_actions.jsonl"), "source_paths": self._read_jsonl(run_id, "source_path_records.jsonl"), "final_output": self._read_json(run_id, "final_output.json") or {}, } if cache: _cache_set(cache_key, payload) return payload def _read_json(self, run_id: str, filename: str) -> dict[str, Any] | None: try: return self.runtime.read_json(run_id, filename) except Exception: if self.export_runtime: try: return self.export_runtime.read_json(run_id, filename) except Exception: return None return None def _read_jsonl(self, run_id: str, filename: str) -> list[dict[str, Any]]: try: return self.runtime.read_jsonl(run_id, filename) except Exception: if self.export_runtime: try: return self.export_runtime.read_jsonl(run_id, filename) except Exception: return [] return [] def _query_by_id(self, bundle: dict[str, Any], query_id: str) -> dict[str, Any] | None: return next((query for query in bundle["queries"] if _text(query.get("search_query_id")) == query_id), None) def _ledger_row(self, query: dict[str, Any], bundle: dict[str, Any], walk_out: dict[str, int], *, debug: bool) -> dict[str, Any]: query_id = _text(query.get("search_query_id")) contents = self._query_content(query_id, bundle, first_round_only=True) videos = [self._video(content, bundle, debug=debug) for content in contents] # 每条首轮视频是否真正向外走过(success 扩展边),供首页「看扩展过程」按钮判断显示。 for video in videos: video["walk_out_count"] = walk_out.get(_text(video.get("platform_content_id")), 0) decisions = [ decision for decision in (_decision_for_content(content, bundle["decisions"]) for content in contents) if decision ] related_ids = self._query_related_content_ids(query_id, bundle) actions = [action for action in bundle["walk_actions"] if _walk_action_related(action, query_id, related_ids)] source = self._source_summary(query, bundle, debug=debug) return { "id": query_id, "source": source, "query": self._query_summary(query, bundle, debug=debug), "first_round_videos": videos, "first_round_video_total": len(videos), "video_stats": self._video_summary(videos), "rule_summary": self._rule_summary(decisions, debug=debug), "walk_summary": self._walk_summary(actions), "asset_summary": self._asset_summary(decisions, bundle["final_output"]), "debug": {"query": query} if debug else None, } def _query_content(self, query_id: str, bundle: dict[str, Any], *, first_round_only: bool) -> list[dict[str, Any]]: items = [item for item in bundle["content_items"] if _content_belongs_to_query(item, query_id)] if first_round_only: walk_query_ids = _walk_query_ids(bundle) items = [item for item in items if _is_first_round_content(item, walk_query_ids)] return items def _query_related_content_ids(self, query_id: str, bundle: dict[str, Any]) -> set[str]: ids: set[str] = set() for item in self._query_content(query_id, bundle, first_round_only=True): ids.update(_content_ids(item)) return {item for item in ids if item} def _query_related_decision_ids(self, query_id: str, bundle: dict[str, Any]) -> set[str]: ids: set[str] = set() for content in self._query_content(query_id, bundle, first_round_only=True): decision = _decision_for_content(content, bundle["decisions"]) if decision: ids.add(_text(decision.get("decision_id"))) return {item for item in ids if item} def _walk_action_related_to_decisions(self, action: dict[str, Any], decision_ids: set[str]) -> bool: if not decision_ids: return False raw = _record(action.get("raw_payload")) values = { _text(action.get("decision_id")), _text(raw.get("decision_id")), _text(action.get("from_node_id")), _text(raw.get("from_node_id")), } return bool(values & decision_ids) def _source_summary(self, query: dict[str, Any], bundle: dict[str, Any], *, debug: bool) -> dict[str, Any]: method = _query_method(query) seed_ref = _record(query.get("pattern_seed_ref")) raw_payload = _record(query.get("raw_payload")) refs = _list(raw_payload.get("query_source_refs")) or _list(seed_ref.get("query_source_refs")) source_text = ( _text(seed_ref.get("query_source_text")) or _text(seed_ref.get("seed_term")) or _text(query.get("search_query")) ) terms = _text_list(query.get("query_source_terms")) source_ref = _query_source_ref(seed_ref, refs) details = _source_details(method, source_text, seed_ref, source_ref, bundle) return { "type": method, "label": SOURCE_LABELS.get(method, "来源待确认"), "evidence": source_text or (terms[0] if terms else "来源证据待确认"), "source_ref": _source_ref_label(method, seed_ref, source_ref, bundle), "terms": terms, "details": details, "query_source_refs": refs, "debug": {"pattern_seed_ref": seed_ref} if debug else None, } def _query_summary(self, query: dict[str, Any] | None, bundle: dict[str, Any], *, debug: bool) -> dict[str, Any]: if not query: return { "id": "", "text": "搜索词待确认", "method": "unknown", "method_label": "生成方式待确认", "source_terms": [], "effect_label": "处理状态待确认", } method = _query_method(query) return { "id": _text(query.get("search_query_id")), "text": _text(query.get("search_query"), "搜索词待确认"), "method": method, "method_label": QUERY_METHOD_LABELS.get(method, "生成方式待确认"), "source_terms": _text_list(query.get("query_source_terms")), "effect_status": _text(query.get("search_query_effect_status"), "pending"), "effect_label": _effect_label(_text(query.get("search_query_effect_status"), "pending")), "debug": {"query": query} if debug else None, } def _extension_query(self, query: dict[str, Any], bundle: dict[str, Any], *, debug: bool) -> dict[str, Any]: summary = self._query_summary(query, bundle, debug=debug) raw_payload = _record(query.get("raw_payload")) summary.update({ "parent_query_id": _text(raw_payload.get("parent_search_query_id") or query.get("llm_variant_of")), "trigger": _text(raw_payload.get("hashtag") or raw_payload.get("author_id") or raw_payload.get("cursor")), }) return summary def _video(self, content: dict[str, Any], bundle: dict[str, Any], *, debug: bool) -> dict[str, Any]: platform_id = _text(content.get("platform_content_id")) media = _media_for_content(content, bundle["media"]) evidence = _evidence_for_content(content, bundle["evidence"]) decision = _decision_for_content(content, bundle["decisions"]) status = _media_status(media) stats = _record(content.get("statistics")) decision_summary = self._decision_summary(decision, debug=debug) if decision else None technical_retry_detail = _technical_retry_detail(decision, evidence, media) # 内容审核拦截从"技术问题"里单独分出来:重试无效、原因不同,展示成独立类别。 if ( decision_summary and decision_summary.get("action") == "TECHNICAL_RETRY_REQUIRED" and _is_content_inspection_blocked(technical_retry_detail, evidence) ): decision_summary = { **decision_summary, "label": "模型内容审核拦截", "reason": "content_inspection_blocked", "reason_label": "视频被大模型内容安全审核拦截(非系统故障,重试无效)", "content_inspection_blocked": True, } return { "id": platform_id or _text(content.get("content_discovery_id"), "content"), "content_discovery_id": _text(content.get("content_discovery_id")), "platform_content_id": platform_id, "search_query_id": _text(content.get("search_query_id")), "progressive_batch_kind": _text(content.get("progressive_batch_kind") or _record(content.get("raw_payload")).get("progressive_batch_kind")), "progressive_page_number": _number(content.get("progressive_page_number") or _record(content.get("raw_payload")).get("progressive_page_number")), "progressive_item_rank_in_page": _number(content.get("progressive_item_rank_in_page") or _record(content.get("raw_payload")).get("progressive_item_rank_in_page")), "title": _text(content.get("description") or content.get("title"), "无标题视频"), "author": _text(content.get("author_display_name") or content.get("platform_author_id"), "未知作者"), "platform": _text(content.get("platform"), "unknown"), "platform_label": _platform_label(_text(content.get("platform"), "unknown")), "platform_url": _platform_content_url(content), "tags": _text_list(content.get("tags")), "statistics": {key: _number(value) for key, value in stats.items()}, "published_at": _text(content.get("published_at") or content.get("publish_time")), "oss_url": _text(media.get("oss_url")) if media else "", "playable_url": _text(media.get("oss_url")) if media else "", "media_status": status, "media_status_label": MEDIA_STATUS_LABELS.get(status, "视频状态待确认"), "media_failure_reason": _media_failure_reason(media), "decision": decision_summary, "gemini": self._evidence_summary(evidence, debug=debug) if evidence else None, "technical_retry_detail": technical_retry_detail, "debug": {"content": content, "media": media, "evidence": evidence} if debug else None, } def _decision_summary(self, decision: dict[str, Any] | None, *, debug: bool) -> dict[str, Any]: if not decision: return { "action": "NOT_JUDGED", "label": "未进入判断", "reason": "no_decision", "reason_label": REASON_LABELS["no_decision"], } reason = _text(decision.get("decision_reason_code"), "no_decision") scorecard = _record(decision.get("scorecard")) score = decision.get("score") if decision.get("score") is not None else scorecard.get("total_score") action = _decision_action(decision) return { "id": _text(decision.get("decision_id")), "action": action, "label": ACTION_LABELS.get(action, "待判断"), "reason": reason, "reason_label": _reason_label(reason, "系统原因待补充"), "score": score, "score_label": f"{round(float(score))} 分" if isinstance(score, (int, float)) else "", "score_items": _decision_score_items(scorecard, score), "score_thresholds": _record(scorecard.get("score_thresholds")) or None, "allow_walk": bool(_record(decision.get("decision_replay_data")).get("allow_walk")), "v4_explanation": _record(decision.get("v4_explanation") or _record(decision.get("raw_payload")).get("v4_explanation")), "debug": {"decision": decision} if debug else None, } def _evidence_summary(self, evidence: dict[str, Any], *, debug: bool) -> dict[str, Any]: status = _text(evidence.get("decode_status") or evidence.get("gemini_status"), "unknown") return { "status": status, "label": "Gemini 已看过" if status in {"success", "ok", "passed"} else "Gemini 未完成判断", "summary": _text(evidence.get("reason") or evidence.get("evidence_summary") or evidence.get("summary")), "debug": {"evidence": evidence} if debug else None, } def _rule_summary(self, decisions: list[dict[str, Any]], *, debug: bool) -> dict[str, Any]: actions = Counter(_decision_action(item) for item in decisions) reasons = Counter(_text(item.get("decision_reason_code") or item.get("reason"), "no_decision") for item in decisions) primary_reason = reasons.most_common(1)[0][0] if reasons else "no_decision" return { "total": len(decisions), "pool_count": actions["ADD_TO_CONTENT_POOL"], "review_count": actions["KEEP_CONTENT_FOR_REVIEW"], "technical_retry_count": actions["TECHNICAL_RETRY_REQUIRED"], "reject_count": actions["REJECT_CONTENT"], "primary_reason": primary_reason, "primary_reason_label": _reason_label(primary_reason, "系统原因待补充"), "score_summary": _score_summary(decisions), "headline": _rule_headline(actions, len(decisions)), "debug": {"decisions": decisions} if debug else None, } def _walk_summary(self, actions: list[dict[str, Any]]) -> dict[str, Any]: edges = Counter(_walk_lane_id(action) for action in actions) reasons = Counter(_text(action.get("reason_code")) for action in actions if _text(action.get("reason_code"))) statuses = Counter(_text(action.get("walk_status"), "unknown") for action in actions) expansion = [a for a in actions if _text(a.get("edge_id")) in _WALK_EXPANSION_EDGES] return { "total": len(actions), "success_count": statuses["success"], "skipped_count": statuses["skipped"], "expansion_count": len(expansion), "expansion_success": sum(1 for a in expansion if _text(a.get("walk_status")) == "success"), "max_depth": max((_number(a.get("depth")) or 0 for a in actions), default=0), "edge_counts": dict(edges), "edge_labels": {key: WALK_EDGE_LABELS.get(key, "系统原因待补充") for key in edges}, "stop_reason": reasons.most_common(1)[0][0] if reasons else "", "stop_reason_label": _reason_label(reasons.most_common(1)[0][0], "本轮没有继续扩展") if reasons else "本轮没有继续扩展", "headline": f"{len(actions)} 次扩展动作" if actions else "本轮没有继续扩展", } def _asset_summary(self, decisions: list[dict[str, Any]], final_output: dict[str, Any]) -> dict[str, Any]: decision_ids = {_text(item.get("decision_id") or item.get("id")) for item in decisions} if not decision_ids: return { "total": 0, "pool_count": 0, "review_count": 0, "technical_retry_count": 0, "reject_count": 0, "headline": _asset_headline(0, 0, 0, 0), } sections = { "content_assets": "入池", "review_records": "待复看", "technical_retry_records": "技术问题", "reject_records": "淘汰", } counts = {} total = 0 for section, label in sections.items(): rows = [ item for item in _list(final_output.get(section)) if _text(item.get("decision_id")) in decision_ids ] counts[label] = len(rows) total += len(rows) return { "total": total, "pool_count": counts.get("入池", 0), "review_count": counts.get("待复看", 0), "technical_retry_count": counts.get("技术问题", 0), "reject_count": counts.get("淘汰", 0), "headline": _asset_headline( counts.get("入池", 0), counts.get("待复看", 0), counts.get("技术问题", 0), counts.get("淘汰", 0), ), } def _video_summary(self, videos: list[dict[str, Any]]) -> dict[str, Any]: actions = Counter(_record(video.get("decision")).get("action") or "NOT_JUDGED" for video in videos) media = Counter(video.get("media_status") or "unknown" for video in videos) return { "total": len(videos), "pool_count": actions["ADD_TO_CONTENT_POOL"], "review_count": actions["KEEP_CONTENT_FOR_REVIEW"], "technical_retry_count": actions["TECHNICAL_RETRY_REQUIRED"], "reject_count": actions["REJECT_CONTENT"], "not_judged_count": actions["NOT_JUDGED"], "oss_uploaded_count": media["oss_uploaded"], "oss_pending_count": media["oss_upload_pending"], "oss_failed_count": media["oss_upload_failed"], "unavailable_count": media["unavailable"] + media["metadata_only"] + media["oss_upload_failed"], } def _walk_action(self, action: dict[str, Any], query: dict[str, Any] | None, bundle: dict[str, Any], *, debug: bool, full: bool = False) -> dict[str, Any]: edge = _walk_lane_id(action) reason = _text(action.get("reason_code")) status = _text(action.get("walk_status"), "unknown") gate_reason = _text(action.get("walk_gate_reason_code") or action.get("allow_walk_reason") or reason) source_content = _content_for_action(action, bundle) source_video = self._video(source_content, bundle, debug=debug) if source_content else None target_query = _target_query_for_action(action, bundle) target_query_summary = self._query_summary(target_query, bundle, debug=debug) if target_query else None target_contents = _target_contents_for_action(action, bundle, source_content, target_query) # full=True(全run游走图 / 单视频子树)不再采样,展示全部带回视频;per-query 视图仍取前 4 条。 shown_contents = target_contents if full else target_contents[:4] target_videos = [self._video(content, bundle, debug=debug) for content in shown_contents] trigger_label = _walk_trigger_label(action, source_content, target_query_summary) from_label = _walk_from_label(action, query, bundle, source_content) score_items = _walk_score_items(action) result_label = _walk_result_label(action, edge, status, reason, source_video, target_query_summary, target_videos) reason_label = _walk_reason_label(action, reason) operator_display_status = _walk_operator_display_status(status, reason) return { "id": _text(action.get("walk_action_id")), "edge_id": edge, "edge_label": WALK_EDGE_LABELS.get(edge, "系统原因待补充"), "depth": _number(action.get("depth")), "source_video": source_video, "target_query": target_query_summary, "target_videos": target_videos, "trigger_label": trigger_label, "status": status, "status_label": _walk_status_label(status), "operator_display_status": operator_display_status, "from_label": from_label, "result": result_label, "reason": reason, "reason_label": reason_label, "impact": _walk_impact(status, reason, gate_reason), "score_items": score_items, "summary": _walk_step_summary(edge, status, source_video, trigger_label, result_label, reason_label), "detail_lines": _walk_detail_lines(action, edge, from_label, trigger_label, score_items, result_label), "debug": {"action": action} if debug else None, } def _summary( self, bundle: dict[str, Any], rows: list[dict[str, Any]], extension_queries: list[dict[str, Any]], media_counts: Counter[str], action_counts: Counter[str], gemini_counts: Counter[str], ) -> dict[str, Any]: query_failures = _query_failures(bundle["queries"]) # 游走指标(run 级):游走带回视频、其中入池、向外游走次数、最深层数。 walk_query_ids = _walk_query_ids(bundle) walk_items = [c for c in bundle["content_items"] if not _is_first_round_content(c, walk_query_ids)] pooled_targets = { _text(d.get("decision_target_id")) for d in bundle["decisions"] if _decision_action(d) == "ADD_TO_CONTENT_POOL" } walk_pooled = sum(1 for c in walk_items if _text(c.get("platform_content_id")) in pooled_targets) expansion = [a for a in bundle["walk_actions"] if _text(a.get("edge_id")) in _WALK_EXPANSION_EDGES] walk_max_depth = max((_number(a.get("depth")) or 0 for a in bundle["walk_actions"]), default=0) return { "first_round_query_count": len(rows), "walk_video_total": len(walk_items), "walk_pooled_count": walk_pooled, "walk_action_total": len(expansion), "walk_action_success": sum(1 for a in expansion if _text(a.get("walk_status")) == "success"), "walk_max_depth": walk_max_depth, "extension_query_count": len(extension_queries), "all_query_count": len(bundle["queries"]), "query_failure_count": len(query_failures), "query_failure_examples": [ { "search_query_id": _text(query.get("search_query_id")), "search_query": _text(query.get("search_query")), "message": _text(_query_failure(query).get("message")), "error_code": _text(_query_failure(query).get("error_code")), } for query in query_failures[:5] ], "content_count": len(bundle["content_items"]), "oss_uploaded_count": media_counts["oss_uploaded"], "oss_pending_count": media_counts["oss_upload_pending"], "oss_failed_count": media_counts["oss_upload_failed"], "gemini_completed_count": gemini_counts["ok"] + gemini_counts["success"] + gemini_counts["passed"], "gemini_failed_count": gemini_counts["failed"], "pool_count": action_counts["ADD_TO_CONTENT_POOL"], "review_count": action_counts["KEEP_CONTENT_FOR_REVIEW"], "technical_retry_count": action_counts["TECHNICAL_RETRY_REQUIRED"], "reject_count": action_counts["REJECT_CONTENT"], } def _query_method(query: dict[str, Any]) -> str: return _text(query.get("search_query_generation_method"), "unknown") def _query_failure(query: dict[str, Any]) -> dict[str, Any]: failure = query.get("query_failure") if isinstance(failure, dict): return failure raw_failure = _record(query.get("raw_payload")).get("query_failure") return raw_failure if isinstance(raw_failure, dict) else {} def _query_failures(queries: list[dict[str, Any]]) -> list[dict[str, Any]]: return [query for query in queries if _query_failure(query)] def _content_belongs_to_query(content: dict[str, Any], query_id: str) -> bool: if not query_id: return False if _text(content.get("search_query_id")) == query_id: return True if query_id in _text_list(content.get("matched_search_query_ids")): return True raw_payload = _record(content.get("raw_payload")) if query_id in _text_list(raw_payload.get("matched_search_query_ids")): return True for source in _list(content.get("query_sources")) + _list(raw_payload.get("query_sources")): if _text(source.get("search_query_id")) == query_id: return True return False def _is_first_round_content(content: dict[str, Any], walk_query_ids: set[str] | None = None) -> bool: # 优先用「内容所属 query 是不是扩展(标签/作者)query」判定——这条 query→内容 的关联即使在 # labeling 修复前的旧 run 里也正确(旧 run 把 previous_discovery_step 误标成 search_query_direct)。 if walk_query_ids and _text(content.get("search_query_id")) in walk_query_ids: return False previous = _text(content.get("previous_discovery_step") or _record(content.get("raw_payload")).get("previous_discovery_step")) if not previous: return True return not any(token in previous for token in ["tag", "hashtag", "author", "next_page", "walk", "pagination"]) def _walk_query_ids(bundle: dict[str, Any]) -> set[str]: return { _text(query.get("search_query_id")) for query in bundle["queries"] if _query_method(query) in EXTENSION_QUERY_METHODS and _text(query.get("search_query_id")) } def _content_ids(content: dict[str, Any]) -> set[str]: return { _text(content.get("content_discovery_id")), _text(content.get("platform_content_id")), _text(content.get("decision_target_id")), } def _decision_for_content(content: dict[str, Any], decisions: list[dict[str, Any]]) -> dict[str, Any] | None: ids = _content_ids(content) return next((decision for decision in decisions if _text(decision.get("decision_target_id")) in ids), None) def _media_for_content(content: dict[str, Any], media_rows: list[dict[str, Any]]) -> dict[str, Any] | None: platform_id = _text(content.get("platform_content_id")) return next((media for media in media_rows if _text(media.get("platform_content_id")) == platform_id), None) def _evidence_for_content(content: dict[str, Any], evidence_rows: list[dict[str, Any]]) -> dict[str, Any] | None: platform_id = _text(content.get("platform_content_id")) discovery_id = _text(content.get("content_discovery_id")) return next( ( evidence for evidence in evidence_rows if _text(evidence.get("platform_content_id")) == platform_id or _text(evidence.get("content_discovery_id")) == discovery_id ), None, ) def _walk_out_counter(bundle: dict[str, Any]) -> dict[str, int]: # 每个源视频 platform_content_id → 它成功向外游走(标签/作者)的次数;0 表示这条没真正走过。 counter: Counter[str] = Counter() for action in bundle["walk_actions"]: if _text(action.get("edge_id")) not in _WALK_EXPANSION_EDGES: continue if _text(action.get("walk_status")) != "success": continue source = _content_for_action(action, bundle) platform_id = _text((source or {}).get("platform_content_id")) if platform_id: counter[platform_id] += 1 return dict(counter) def _load_walk_caps() -> dict[str, Any]: # 游走预算上限(best-effort,只读 walk_policy.json;env 覆盖时数值可能与真实 run 略有出入)。 path = os.path.join("tech_documents", "数据接口与来源", "walk_policy.json") try: with open(path, encoding="utf-8") as handle: data = json.load(handle) except Exception: return {} def _value(node: Any) -> Any: return node.get("value") if isinstance(node, dict) else node global_caps = data.get("global", {}) if isinstance(data.get("global"), dict) else {} edges = { _text(edge.get("edge_id")): edge for edge in data.get("edge_budgets", []) if isinstance(edge, dict) } return { "max_depth": _value(global_caps.get("max_depth")), "hashtag_to_query": (edges.get("hashtag_to_query") or {}).get("max_total_actions"), "author_to_works": (edges.get("author_to_works") or {}).get("max_total_actions"), } def _walk_budget_overview( bundle: dict[str, Any], actions: list[dict[str, Any]], extension_queries: list[dict[str, Any]], ) -> dict[str, Any]: caps = _load_walk_caps() statuses = Counter(_text(action.get("status"), "unknown") for action in actions) tag_used = sum(1 for a in actions if a.get("edge_id") == "hashtag_to_query" and a.get("status") == "success") author_used = sum(1 for a in actions if a.get("edge_id") == "author_to_works" and a.get("status") == "success") walk_query_ids = _walk_query_ids(bundle) walk_items = [c for c in bundle["content_items"] if not _is_first_round_content(c, walk_query_ids)] return { "total_actions": len(actions), "success_count": statuses["success"], "skipped_count": statuses["skipped"], "extension_query_count": len(extension_queries), "tag_used": tag_used, "tag_cap": caps.get("hashtag_to_query"), "author_used": author_used, "author_cap": caps.get("author_to_works"), "max_depth_reached": max((_number(a.get("depth")) or 0 for a in actions), default=0), "max_depth_cap": caps.get("max_depth"), "walk_video_total": len(walk_items), } def _walk_action_related(action: dict[str, Any], query_id: str, related_ids: set[str]) -> bool: values = { _text(action.get("from_node_id")), _text(action.get("to_node_id")), _text(action.get("decision_target_id")), } raw = _record(action.get("raw_payload")) values.update({ _text(raw.get("parent_search_query_id")), _text(raw.get("search_query_id")), _text(raw.get("decision_target_id")), }) return query_id in values or bool(values & related_ids) def _walk_lane_id(action: dict[str, Any]) -> str: return _text(action.get("edge_id") or action.get("edge_type"), "unknown") def _walk_lane(edge: str) -> str: if edge in {"hashtag_to_query", "tag_query"}: return "tag_query" if edge in {"author_to_works", "video_to_author", "author_works"}: return "author_works" return "path_stop" def _source_path(path: dict[str, Any], *, debug: bool) -> dict[str, Any]: return { "type": _text(path.get("source_path_type")), "label": "内容来源路径", "status": _text(path.get("status"), "unknown"), "debug": {"source_path": path} if debug else None, } def _decision_action(decision: dict[str, Any]) -> str: action = _text(decision.get("decision_action") or decision.get("action"), "NOT_JUDGED") if action == "ADD_CONTENT_TO_POOL": return "ADD_TO_CONTENT_POOL" return action def _media_status(media: dict[str, Any] | None) -> str: if not media: return "unavailable" return _text(media.get("content_media_status"), "metadata_only") def _media_failure_reason(media: dict[str, Any] | None) -> str: if not media: return "" raw = _record(media.get("raw_payload")) return _text(media.get("failure_reason") or raw.get("upload_failure_reason") or raw.get("failure_reason")) def _is_content_inspection_blocked( detail: dict[str, Any] | None, evidence: dict[str, Any] | None = None, ) -> bool: # 内容审核拦截:新数据 failure_type=content_inspection_blocked;旧数据从响应体文本兜底(DataInspectionFailed)。 if isinstance(detail, dict) and _text(detail.get("failure_type")) == "content_inspection_blocked": return True evidence_raw = _record((evidence or {}).get("raw_payload")) if _text(evidence_raw.get("failure_type")) == "content_inspection_blocked": return True texts = [ _text(_record(evidence_raw.get("response_body_summary")).get("text_excerpt")), _text(_record(_record((detail or {}).get("openrouter")).get("response_summary")).get("text_excerpt")), ] blob = " ".join(texts).lower() return "datainspection" in blob or "inappropriate content" in blob def _technical_retry_detail( decision: dict[str, Any] | None, evidence: dict[str, Any] | None, media: dict[str, Any] | None, ) -> dict[str, Any] | None: if _decision_action(decision or {}) != "TECHNICAL_RETRY_REQUIRED": return None scorecard = _record((decision or {}).get("scorecard")) replay_data = _record((decision or {}).get("decision_replay_data")) evidence_summary = _record((evidence or {}).get("evidence_summary")) evidence_raw = _record((evidence or {}).get("raw_payload")) media_raw = _record((media or {}).get("raw_payload")) merged = {**scorecard, **replay_data, **evidence_summary, **evidence_raw} failure_type = _text( merged.get("failure_type") or media_raw.get("oss_archive_last_error") or media_raw.get("failure_reason") or _media_failure_reason(media) ) if not failure_type: return None timing = _record(merged.get("timing_metrics")) video_fetch = _record(timing.get("video_fetch")) gemini_request = _record(timing.get("gemini_request")) attempts = [ { "attempt": item.get("attempt"), "status": _text(item.get("status")), "duration_seconds": _seconds(item.get("duration_ms")), "failure_type": _text(item.get("failure_type")), "exception_type": _text(item.get("exception_type")), "http_status_code": item.get("http_status_code"), "response_body_summary": _record(item.get("response_body_summary")), } for item in _list(gemini_request.get("attempts")) ] detail = { "brief_reason": _technical_retry_brief_reason(failure_type, merged, media_raw), "stage": _technical_retry_stage(failure_type), "stage_label": _technical_retry_stage_label(failure_type), "failure_type": failure_type, "failure_label": _technical_retry_failure_label(failure_type), "exception_type": _text(merged.get("exception_type")), "http_status_code": merged.get("http_status_code"), "error_message": _text(merged.get("error_message")), "retry_count": merged.get("retry_count"), "video_source": _text(video_fetch.get("gemini_video_source")), "timings": { "download_seconds": _seconds(video_fetch.get("download_duration_ms")), "ffmpeg_seconds": _seconds(video_fetch.get("ffmpeg_duration_ms")), "video_fetch_total_seconds": _seconds(video_fetch.get("total_duration_ms")), "gemini_seconds": _seconds(gemini_request.get("total_duration_ms")), }, "sizes": { "downloaded_mb": _megabytes(video_fetch.get("downloaded_bytes")), "compressed_mb": _megabytes(video_fetch.get("compressed_bytes")), }, "attempts": attempts, "openrouter": { "response_summary": _record(merged.get("response_body_summary")), }, "oss": { "status": _text((media or {}).get("content_media_status")), "last_error": _text(media_raw.get("oss_archive_last_error") or media_raw.get("failure_reason")), "last_exception_type": _text(media_raw.get("oss_archive_last_exception_type")), "last_http_status_code": media_raw.get("oss_archive_last_http_status_code"), "attempt_count": media_raw.get("oss_archive_attempt_count"), "duration_seconds": _seconds(_record(media_raw.get("oss_timing_metrics")).get("oss_upload_duration_ms")), "payload_mode": _text(media_raw.get("oss_payload_mode")), "response_summary": _record(media_raw.get("oss_response_summary")), }, "video_url": { "selected_path": _text(media_raw.get("selected_video_url_path")), "selected_host": _text(media_raw.get("selected_video_url_host")), "probe_status": _text(media_raw.get("selected_video_url_probe_status")), "content_type": _text(media_raw.get("selected_video_url_content_type")), "content_range": _text(media_raw.get("selected_video_url_content_range")), "candidate_counts": _record(media_raw.get("video_url_candidate_counts")), "reject_reasons": _record(media_raw.get("video_url_reject_reasons")), }, } return detail def _technical_retry_brief_reason( failure_type: str, payload: dict[str, Any], media_raw: dict[str, Any], ) -> str: if failure_type == "content_inspection_blocked": return "视频被大模型内容安全审核拦截(非系统故障,重试无效)" if failure_type in {"portrait_unavailable", "portrait_incomplete"}: reason = _text(payload.get("fifty_plus_failure_reason")) base = "拉不到作者 50+ 画像(热点宝)" if failure_type == "portrait_unavailable" else "作者 50+ 画像数据不全" return f"{base}:{reason}" if reason else base if failure_type == "no_valid_play_url": return "平台结果未找到可用正片 URL" if failure_type == "no_play_url": return "平台结果没有返回可用 play_url" if failure_type == "qwen_client_timeout": return "通义千问请求写入超时" if failure_type == "qwen_http_error": status = payload.get("http_status_code") return f"通义千问返回 HTTP {status}" if status else "通义千问请求失败" if failure_type == "qwen_response_invalid": return "通义千问返回格式无法解析" if failure_type == "gemini_client_timeout": return "OpenRouter/Gemini 请求写入超时" if failure_type == "gemini_http_error": status = payload.get("http_status_code") return f"OpenRouter/Gemini 返回 HTTP {status}" if status else "OpenRouter/Gemini 请求失败" if failure_type == "gemini_response_invalid": return "OpenRouter/Gemini 返回格式无法解析" if failure_type == "video_fetch_failed": return "视频下载或压缩失败" if failure_type == "video_judge_timeout": return "视频判定 worker 超时未返回(已超兜底上限,跳过本条)" if failure_type == "oss_worker_timeout": return "OSS 归档 worker 超时未返回(已超兜底上限,跳过本条)" if failure_type.startswith("oss_"): return "OSS 转存未拿到可用视频地址" if media_raw.get("oss_archive_last_error"): return "OSS 归档失败" return failure_type def _technical_retry_stage(failure_type: str) -> str: if failure_type == "content_inspection_blocked": return "content_inspection" if failure_type == "video_judge_timeout": return "video_judge" if failure_type.startswith("portrait"): return "portrait" if failure_type.startswith("oss_"): return "oss" if failure_type in {"no_play_url", "no_valid_play_url"}: return "play_url" if "ffmpeg" in failure_type: return "ffmpeg" if failure_type in {"video_fetch_failed", "video_download_failed"}: return "video_download" if failure_type.startswith("qwen_"): return "dashscope" if failure_type.startswith("gemini_"): return "openrouter" return "unknown" def _technical_retry_stage_label(failure_type: str) -> str: return { "oss": "OSS", "play_url": "平台 play_url", "video_download": "视频下载", "ffmpeg": "ffmpeg", "dashscope": "通义千问", "openrouter": "OpenRouter/Gemini", "content_inspection": "内容审核拦截", "portrait": "热点宝画像", "video_judge": "视频判定调度", "unknown": "未知阶段", }[_technical_retry_stage(failure_type)] def _technical_retry_failure_label(failure_type: str) -> str: return { "content_inspection_blocked": "模型内容审核拦截", "qwen_client_timeout": "通义千问 客户端超时", "qwen_http_error": "通义千问 HTTP 错误", "qwen_response_invalid": "通义千问 响应无法解析", "gemini_client_timeout": "Gemini 客户端超时", "gemini_http_error": "Gemini HTTP 错误", "gemini_response_invalid": "Gemini 响应无法解析", "video_fetch_failed": "视频获取失败", "no_play_url": "缺少 play_url", "no_valid_play_url": "未找到可用正片 URL", "oss_upload_response_invalid": "OSS 响应无效", "oss_upload_http_error": "OSS HTTP 错误", "oss_worker_timeout": "OSS 归档 worker 超时", "portrait_unavailable": "拉不到作者 50+ 画像", "portrait_incomplete": "作者 50+ 画像数据不全", "video_judge_timeout": "视频判定调度超时", }.get(failure_type, failure_type) def _seconds(value: Any) -> float | None: try: return round(float(value) / 1000.0, 3) except (TypeError, ValueError): return None def _megabytes(value: Any) -> float | None: try: return round(float(value) / 1024.0 / 1024.0, 2) except (TypeError, ValueError): return None def _rule_headline(actions: Counter[str], total: int) -> str: if not total: return "首轮内容暂未进入判断" if actions["ADD_TO_CONTENT_POOL"]: return f"有 {actions['ADD_TO_CONTENT_POOL']} 条内容入池" if actions["KEEP_CONTENT_FOR_REVIEW"]: return f"有 {actions['KEEP_CONTENT_FOR_REVIEW']} 条内容待复看" if actions["TECHNICAL_RETRY_REQUIRED"]: return f"有 {actions['TECHNICAL_RETRY_REQUIRED']} 条内容出现技术问题" return "本轮没有可沉淀内容" def _asset_headline( pool_count: int, review_count: int, technical_retry_count: int, reject_count: int, ) -> str: parts = [] if pool_count: parts.append(f"入池 {pool_count} 条") if review_count: parts.append(f"待复看 {review_count} 条") if technical_retry_count: parts.append(f"技术问题 {technical_retry_count} 条") if parts: return "，".join(parts) if reject_count: return f"没有入池，淘汰 {reject_count} 条" return "暂无可沉淀内容" def _effect_label(status: str) -> str: return { "success": "已找到可用结果", "rule_blocked": "找到结果，但未通过内容判断", "pending": "处理中", "failed": "搜索失败", "skipped": "已跳过", }.get(status, "处理状态待确认") def _walk_status_label(status: str) -> str: return { "success": "已执行", "skipped": "未继续", "failed": "执行失败", "pending": "处理中", }.get(status, "状态待确认") def _action_value(action: dict[str, Any], key: str) -> Any: raw = _record(action.get("raw_payload")) return action.get(key) if action.get(key) not in (None, "") else raw.get(key) def _find_content_by_id(content_id: str, bundle: dict[str, Any]) -> dict[str, Any] | None: if not content_id: return None return next((item for item in bundle["content_items"] if content_id in _content_ids(item)), None) def _decision_by_id(decision_id: str, bundle: dict[str, Any]) -> dict[str, Any] | None: if not decision_id: return None return next((item for item in bundle["decisions"] if _text(item.get("decision_id")) == decision_id), None) def _decision_for_action(action: dict[str, Any], bundle: dict[str, Any]) -> dict[str, Any] | None: decision_id = _text(_action_value(action, "decision_id")) if not decision_id and _text(_action_value(action, "from_node_type")).lower() == "ruledecision": decision_id = _text(_action_value(action, "from_node_id")) return _decision_by_id(decision_id, bundle) def _content_for_action(action: dict[str, Any], bundle: dict[str, Any]) -> dict[str, Any] | None: candidate_ids = [ _text(_action_value(action, "decision_target_id")), _text(_action_value(action, "to_node_id")), _text(_action_value(action, "from_node_id")), ] for candidate_id in candidate_ids: content = _find_content_by_id(candidate_id, bundle) if content: return content decision = _decision_for_action(action, bundle) if decision: return _find_content_by_id(_text(decision.get("decision_target_id")), bundle) return None def _target_query_for_action(action: dict[str, Any], bundle: dict[str, Any]) -> dict[str, Any] | None: query_id = _text(_action_value(action, "to_node_id")) return next((item for item in bundle["queries"] if _text(item.get("search_query_id")) == query_id), None) def _target_contents_for_action( action: dict[str, Any], bundle: dict[str, Any], source_content: dict[str, Any] | None, target_query: dict[str, Any] | None, ) -> list[dict[str, Any]]: if target_query: query_id = _text(target_query.get("search_query_id")) return [item for item in bundle["content_items"] if _content_belongs_to_query(item, query_id)] edge = _walk_lane_id(action) if edge in {"author_to_works", "video_to_author", "author_works"}: author_id = _text(_action_value(action, "author_id") or _action_value(action, "from_node_id") or (source_content or {}).get("platform_author_id")) if not author_id: return [] return [ item for item in bundle["content_items"] if _text(item.get("platform_author_id")) == author_id and "author" in _text(item.get("previous_discovery_step") or _record(item.get("raw_payload")).get("previous_discovery_step")) ] target_content = _find_content_by_id(_text(_action_value(action, "to_node_id")), bundle) return [target_content] if target_content else [] def _walk_from_label( action: dict[str, Any], query: dict[str, Any] | None, bundle: dict[str, Any], source_content: dict[str, Any] | None = None, ) -> str: from_type = _text(action.get("from_node_type")).lower() from_node_id = _text(action.get("from_node_id")) edge = _walk_lane_id(action) if edge in {"hashtag_to_query", "tag_query", "author_to_works", "video_to_author", "author_works"} and source_content: title = _text(source_content.get("description") or source_content.get("title")) return f"从视频《{_short_text(title, 34)}》触发" if title else "从视频内容触发" if "query" in from_type: source_query = next((item for item in bundle["queries"] if _text(item.get("search_query_id")) == from_node_id), None) or query return f"从搜索词《{_text((source_query or {}).get('search_query'), '搜索词待确认')}》继续" if "ruledecision" in from_type: content = _content_for_action(action, bundle) title = _text((content or {}).get("description") or (content or {}).get("title")) return f"判断视频《{_short_text(title, 34)}》后触发" if title else "由内容判断结果触发" if "content" in from_type or "video" in from_type or source_content: title = _text((source_content or {}).get("description") or (source_content or {}).get("title")) return f"从视频《{_short_text(title, 34)}》触发" if title else "从视频内容触发" if "author" in from_type: content = _content_for_action(action, bundle) author = _text((content or {}).get("author_display_name") or from_node_id, "作者待确认") title = _text((content or {}).get("description") or (content or {}).get("title")) return f"从作者《{author}》触发，来源视频《{_short_text(title, 24)}》" if title else f"从作者《{author}》触发" if "tag" in from_type: return "从视频标签触发" return "从上一环节出发" def _walk_impact(status: str, reason: str, gate_reason: str) -> str: if reason == "blocked_by_rule_decision" or gate_reason == "v4_allow_walk_denied": return "这条内容不会继续带出更多视频。" if reason == "budget_exhausted": return "达到游走次数上限，这一步不会继续新增搜索词。" if status == "success": return "这一步已经执行，结果会进入后续判断。" if status == "skipped": return "这一步没有继续执行，不影响最终沉淀。" return "这一步暂不影响最终沉淀。" def _walk_operator_display_status(status: str, reason: str) -> str: return "budget_blocked" if status == "skipped" and reason == "budget_exhausted" else status def _walk_trigger_label( action: dict[str, Any], source_content: dict[str, Any] | None = None, target_query: dict[str, Any] | None = None, ) -> str: hashtag = _text(_action_value(action, "hashtag")) if hashtag: return f"标签：#{hashtag.lstrip('#')}" edge = _walk_lane_id(action) if edge in {"hashtag_to_query", "tag_query"} and target_query: return f"标签搜索词：{_text(target_query.get('text'), '搜索词待确认')}" if edge in {"hashtag_to_query", "tag_query"} and source_content: tags = _text_list(source_content.get("tags")) return f"候选标签：{' / '.join(tags[:3])}" if tags else "标签未生成：这条视频未达到扩展门槛" author = _text(_action_value(action, "author_id") or _action_value(action, "platform_author_id")) if not author and edge in {"author_to_works", "video_to_author", "author_works"} and source_content: author = _text(source_content.get("author_display_name") or source_content.get("platform_author_id")) if author: return f"作者：{author}" if edge in {"author_to_works", "video_to_author", "author_works"}: return "作者：未进入作者作品扩展" cursor = _text(_action_value(action, "page_cursor") or _action_value(action, "cursor")) if cursor: return f"翻下一页：游标 {cursor}" to_node = _text(_action_value(action, "to_node_id")) if source_content and edge in {"path_stop", "decision_to_asset", "budget_downgrade"}: return f"视频：{_short_text(_text(source_content.get('description') or source_content.get('title'), '视频待确认'), 34)}" if to_node and to_node not in {"tag_query_skipped", "author_works_skipped"}: return f"结果对象：{to_node}" return "触发对象待确认" def _walk_reason_label(action: dict[str, Any], reason: str) -> str: gate_reason = _text(_action_value(action, "walk_gate_reason_code") or _action_value(action, "allow_walk_reason")) if gate_reason == "v4_allow_walk_denied": allow_reason = _text(_action_value(action, "allow_walk_reason")) if allow_reason and allow_reason != gate_reason: return f"不适合继续向外扩展：{_reason_label(allow_reason, '评分未达到继续扩展门槛')}" return "不适合继续向外扩展" if reason: return _reason_label(reason, "这一步没有记录额外原因") return "这一步没有记录额外原因" def _walk_result_label( action: dict[str, Any], edge: str, status: str, reason: str, source_video: dict[str, Any] | None, target_query: dict[str, Any] | None, target_videos: list[dict[str, Any]], ) -> str: if edge == "decision_to_asset": return "这条视频已进入内容池。" if edge == "path_stop": return "这条视频到这里停止，不再向外扩展。" if edge == "budget_downgrade": return "这条视频进入待复看，先不作为强信号继续扩展。" if status == "skipped" and reason == "budget_exhausted": return "内容达到扩展门槛，但已经达到游走次数上限。" if status == "skipped": return "没有继续扩展。" if edge in {"hashtag_to_query", "tag_query"}: query_text = _text((target_query or {}).get("text")) return f"已生成标签搜索词《{query_text}》，带回 {len(target_videos)} 条视频。" if query_text else f"已从视频标签继续游走，带回 {len(target_videos)} 条视频。" if edge in {"author_to_works", "video_to_author", "author_works"}: author = _text((source_video or {}).get("author"), "该作者") return f"已查看作者《{author}》更多作品，带回 {len(target_videos)} 条视频。" return "这一步已经执行，结果会进入后续判断。" def _walk_step_summary( edge: str, status: str, source_video: dict[str, Any] | None, trigger_label: str, result_label: str, reason_label: str, ) -> str: title = _text((source_video or {}).get("title")) if edge in {"hashtag_to_query", "tag_query"}: action = "从视频标签继续游走" elif edge in {"author_to_works", "video_to_author", "author_works"}: action = "从作者继续游走" elif edge == "decision_to_asset": action = "入池沉淀" elif edge == "budget_downgrade": action = "转入待复看" elif edge == "path_stop": action = "停止这条线" else: action = WALK_EDGE_LABELS.get(edge, "扩展动作") prefix = f"从视频《{_short_text(title, 28)}》" if title else "从当前搜索词" if status == "success": return f"{prefix}{action}：{result_label}" return f"{prefix}{action}：{reason_label}" def _walk_detail_lines( action: dict[str, Any], edge: str, from_label: str, trigger_label: str, score_items: list[dict[str, Any]], result_label: str, ) -> list[str]: lines = [ f"动作：{WALK_EDGE_LABELS.get(edge, '扩展动作')}", f"起点：{from_label}", ] if trigger_label != "触发对象待确认": lines.append(f"扩展对象：{trigger_label}") if score_items: lines.append("扩展门槛：" + " / ".join(f"{item['label']} {item['score_label']}" for item in score_items[:3])) reason = _walk_reason_label(action, _text(_action_value(action, "reason_code"))) if reason != "这一步没有记录额外原因": lines.append(f"判断：{reason}") if result_label: lines.append(f"结果：{result_label}") return lines def _walk_score_items(action: dict[str, Any]) -> list[dict[str, Any]]: snapshot = _record(action.get("walk_gate_snapshot") or _record(action.get("raw_payload")).get("walk_gate_snapshot")) return _score_items_from_values( total=snapshot.get("score"), query=snapshot.get("query_relevance_score"), platform=snapshot.get("platform_performance_score"), components=[], ) def _decision_score_items(scorecard: dict[str, Any], score: Any) -> list[dict[str, Any]]: weights, fifty = _v4_display_weights(scorecard) return _score_items_from_values( total=score if score is not None else scorecard.get("total_score"), query=scorecard.get("query_relevance_score"), platform=scorecard.get("platform_performance_score"), components=_list(scorecard.get("platform_performance_components")), weights=weights, fifty=fifty, ) def _v4_display_weights(scorecard: dict[str, Any]) -> tuple[dict[str, float], dict[str, Any] | None]: """展示 evaluator 已落在 scorecard 的权重;旧记录无 score_weights 时保留历史推断。""" weights = _display_weights_from_scorecard(scorecard) if "fifty_plus_status" not in scorecard: return weights or {"query": 0.5, "platform": 0.5}, None status = scorecard.get("fifty_plus_status") fifty: dict[str, Any] = {"status": status} if status == "ok": fifty["score"] = scorecard.get("fifty_plus_score") fifty["components"] = _record(scorecard.get("fifty_plus_components")) return weights or {"query": 0.35, "platform": 0.35, "fifty": 0.30}, fifty def _display_weights_from_scorecard(scorecard: dict[str, Any]) -> dict[str, float]: weights = scorecard.get("score_weights") if not isinstance(weights, dict): return {} display = {} mapping = { "query_relevance": "query", "platform_performance": "platform", "fifty_plus": "fifty", } for source_key, target_key in mapping.items(): value = weights.get(source_key) if isinstance(value, (int, float)) and not isinstance(value, bool): display[target_key] = float(value) return display def _fifty_status_note(status: Any) -> str: return { "not_attempted": "相关性未达标(query<65),未拉取画像", "unavailable": "作者画像拉取失败,未评估", "incomplete": "作者画像数据不全,未评估", }.get(str(status), "未评估") def _score_summary(decisions: list[dict[str, Any]]) -> dict[str, Any]: scorecards = [_record(item.get("scorecard")) for item in decisions] return { "items": [ _avg_score_item("总分", [item.get("score") if item.get("score") is not None else _record(item.get("scorecard")).get("total_score") for item in decisions]), _avg_score_item("是否契合需求", [item.get("query_relevance_score") for item in scorecards]), _avg_score_item("平台表现", [item.get("platform_performance_score") for item in scorecards]), ], } def _score_items_from_values( total: Any, query: Any, platform: Any, components: list[dict[str, Any]], *, weights: dict[str, float] | None = None, fifty: dict[str, Any] | None = None, ) -> list[dict[str, Any]]: weights = weights or {} items = [ _score_item("总分", total, "最终用于入池 / 复看 / 淘汰的总分"), _score_item("是否契合需求", query, "视频内容和本次搜索词、需求的匹配程度", weight=weights.get("query"), group="main"), _score_item("平台表现", platform, "点赞、评论、转发、收藏等互动表现", weight=weights.get("platform"), group="main"), ] if fifty is not None: # 50+老年 维度卡(仅抖音):status=ok 出真实分,否则出灰卡「未评估」。 if fifty.get("status") == "ok": items.append(_score_item( "50+老年受众", fifty.get("score"), "作者粉丝 50+ 画像(TGI × 占比)", weight=weights.get("fifty"), group="main", status="ok", )) else: items.append(_score_item( "50+老年受众", None, _fifty_status_note(fifty.get("status")), weight=weights.get("fifty"), group="main", status=fifty.get("status"), allow_none=True, )) for component in components: if component.get("type"): # M11:新打分 component 自带 label(中文)+ type(absolute/ratio) label = _text(component.get("label")) or _platform_component_label(_text(component.get("field"))) detail = _platform_component_detail(component) else: # 旧 run 的 component(无 type):沿用绝对值展示 field = _text(component.get("field")) label = _platform_component_label(field) detail = _platform_component_raw_text(field, component.get("value")) items.append(_score_item( label, component.get("normalized_score"), detail, weight=_float_or_none(component.get("weight")), group="platform", )) if fifty is not None and fifty.get("status") == "ok": # 50+老年怎么算:TGI 表现(60%)+ 占比表现(40%),取自 fifty_plus_components。 comps = _record(fifty.get("components")) items.append(_score_item( "TGI 表现", comps.get("tgi_score"), f"50+ 粉丝 TGI≈{_num_text(comps.get('band_tgi'))}(>100 即老年粉丝偏好高于大盘;TGI 200=满分,按此折算得分)", weight=0.6, group="fifty", )) items.append(_score_item( "占比表现", comps.get("percent_score"), f"50+ 粉丝占比≈{_num_text(comps.get('band_percent'))}%", weight=0.4, group="fifty", )) return [item for item in items if item] def _score_item( label: str, value: Any, detail: str, *, weight: Any = None, group: str | None = None, status: Any = None, allow_none: bool = False, ) -> dict[str, Any]: score = _float_or_none(value) if score is None and not allow_none: return {} item: dict[str, Any] = { "label": label, "score": round(score, 2) if score is not None else None, "score_label": f"{round(score, 1):g} 分" if score is not None else "未评估", "detail": detail, } weight_num = _float_or_none(weight) if weight_num is not None: item["weight"] = round(weight_num, 4) if group: item["group"] = group if status is not None: item["status"] = str(status) return item def _num_text(value: Any) -> str: number = _float_or_none(value) return f"{round(number):g}" if number is not None else "—" def _avg_score_item(label: str, values: list[Any]) -> dict[str, Any]: nums = [value for value in (_float_or_none(item) for item in values) if value is not None] if not nums: return {"label": label, "score": None, "score_label": "暂无评分"} avg = sum(nums) / len(nums) return { "label": label, "score": round(avg, 2), "score_label": f"平均 {round(avg, 1):g} 分", } def _platform_component_label(field: str) -> str: return { "statistics.digg_count": "点赞表现", "statistics.like_count": "点赞表现", "statistics.comment_count": "评论表现", "statistics.share_count": "转发表现", "statistics.collect_count": "收藏表现", "statistics.play_count": "播放表现", }.get(field, field or "互动表现") def _platform_component_raw_text(field: str, value: Any) -> str: """该视频这项指标的原始量,自然语言(如「该视频原始 1784 赞」)。""" number = _num_text(value) unit = { "statistics.digg_count": "赞", "statistics.like_count": "赞", "statistics.comment_count": "条评论", "statistics.share_count": "次转发", "statistics.collect_count": "次收藏", "statistics.play_count": "次播放", }.get(field) return f"该视频原始 {number} {unit}" if unit else f"该视频原始值 {number}" def _metric_unit(field: str) -> str: """指标字段 → 中文短词(用于 ratio detail,如 转发/赞/播放)。""" return { "digg_count": "赞", "like_count": "赞", "share_count": "转发", "collect_count": "收藏", "comment_count": "评论", "play_count": "播放", }.get(_text(field).removeprefix("statistics."), _text(field)) def _ratio_pct(value: Any) -> str: num = _float_or_none(value) return f"{round(num * 100, 1):g}%" if num is not None else "—" def _platform_component_detail(component: dict[str, Any]) -> str: """M11 平台分项的展示文案:ratio 出「转发 783 / 赞 1532 = 51% · 收缩后 X% · 目标 27%」; absolute 出体量/原始量。""" if _text(component.get("type")) == "ratio": nu = _metric_unit(component.get("numerator_field")) du = _metric_unit(component.get("denominator_field")) return ( f"{nu} {_num_text(component.get('numerator_value'))} / {du} {_num_text(component.get('denominator_value'))}" f" = {_ratio_pct(component.get('raw_ratio'))}" f" · 收缩后 {_ratio_pct(component.get('shrunk_ratio'))}" f" · 目标 {_ratio_pct(component.get('target'))}" ) field = _text(component.get("field")) value = component.get("value") if field == "total_interaction": return f"内容体量(赞+评+转+藏 共 {_num_text(value)})" unit = _metric_unit(field) return f"该视频原始 {_num_text(value)} {unit}" if unit else f"该视频原始值 {_num_text(value)}" def _percent_text(value: Any) -> str: number = _float_or_none(value) if number is None: return "待确认" return f"{round(number * 100):g}%" def _float_or_none(value: Any) -> float | None: try: if value is None or value == "": return None return float(value) except (TypeError, ValueError): return None def _short_text(value: str, limit: int) -> str: if len(value) <= limit: return value return value[: limit - 1] + "…" def _platform_label(platform: str) -> str: return { "douyin": "抖音", "tiktok": "TikTok", "bilibili": "B 站", "kuaishou": "快手", "xhs": "小红书", }.get(platform.lower(), platform or "平台待确认") def _platform_content_url(content: dict[str, Any]) -> str: raw = _record(content.get("raw_payload")) platform_raw = _record(content.get("platform_raw_payload") or raw.get("platform_raw_payload")) direct = _text( content.get("platform_content_url") or content.get("share_url") or content.get("url") or content.get("aweme_url") or raw.get("platform_content_url") or raw.get("share_url") or raw.get("url") or raw.get("aweme_url") ) if direct: return direct platform = _text(content.get("platform") or raw.get("platform")).lower() content_id = _text( content.get("platform_content_id") or raw.get("platform_content_id") or platform_raw.get("aweme_id") ) if platform == "douyin" and content_id: return f"https://www.douyin.com/video/{content_id}" return "" def _query_source_ref(seed_ref: dict[str, Any], refs: list[dict[str, Any]]) -> dict[str, Any]: for ref in refs: nested = _record(ref.get("source_ref")) if nested: return nested if ref: return ref return _record(seed_ref.get("source_ref")) def _source_details( method: str, source_text: str, seed_ref: dict[str, Any], source_ref: dict[str, Any], bundle: dict[str, Any], ) -> list[str]: if method == "seed_term": return _dedupe_texts([ _detail("需求单号", _demand_id(bundle)), _detail("种子词", source_text), _detail("来源帖子", _source_post_id(seed_ref, source_ref, bundle)), ]) if method in {"piaoquan_topic_point", "query_seed_point"}: point_text = _text(source_ref.get("point_text") or source_ref.get("cleaned_text") or source_text) return _dedupe_texts([ _detail("帖子 ID", source_ref.get("post_id") or _source_post_id(seed_ref, source_ref, bundle)), _detail("内容点 ID", source_ref.get("point_id") or source_ref.get("topic_point_id")), _detail("内容点类型", source_ref.get("point_type")), _detail("内容点", point_text), ]) if method == "category_leaf_element": binding = _category_binding(bundle, source_ref) category_path = _text( source_ref.get("category_path") or binding.get("category_full_path") or binding.get("category_path") ) category_level = _text(source_ref.get("category_level") or binding.get("category_level")) or _level_from_path(category_path) execution_id = _text( seed_ref.get("pattern_execution_id") or _evidence_pack(bundle).get("pattern_execution_id") or _record(bundle.get("pattern_seed_pack")).get("pattern_execution_id") ) category_id = _text(source_ref.get("category_id") or binding.get("category_id")) element_name = _text(source_ref.get("element_name") or binding.get("element_name") or source_text) return _dedupe_texts([ f"分类树：Pattern 执行 {execution_id}" if execution_id else "分类树：当前运行记录未写入执行编号", _detail("分类路径", category_path), f"所在层级：第 {category_level} 层" if category_level else "所在层级：当前运行记录未写入，需要接分类树快照反查", _detail("叶子元素", element_name), _detail("分类节点", category_id), ]) return _dedupe_texts([_detail("来源", source_text)]) def _source_ref_label( method: str, seed_ref: dict[str, Any], source_ref: dict[str, Any], bundle: dict[str, Any], ) -> str: if method == "seed_term": return _text(_demand_id(bundle) or _source_post_id(seed_ref, source_ref, bundle), "来源编号待确认") if method in {"piaoquan_topic_point", "query_seed_point"}: return _text(source_ref.get("post_id") or _source_post_id(seed_ref, source_ref, bundle), "帖子编号待确认") if method == "category_leaf_element": binding = _category_binding(bundle, source_ref) path = _text(source_ref.get("category_path") or binding.get("category_full_path") or binding.get("category_path")) return path or _text(source_ref.get("category_id") or binding.get("category_id"), "分类节点待确认") return _text(seed_ref.get("source_post_id") or seed_ref.get("pattern_execution_id"), "来源编号待确认") def _demand_id(bundle: dict[str, Any]) -> str: source_context = _record(bundle.get("source_context")) return _text(source_context.get("demand_content_id") or source_context.get("id") or _evidence_pack(bundle).get("demand_content_id")) def _demand_summary(bundle: dict[str, Any]) -> dict[str, Any]: source_context = _record(bundle.get("source_context")) raw = _record(source_context.get("raw_demand_content")) or source_context ext_data = _record(raw.get("ext_data")) or _record(source_context.get("ext_data")) evidence = _evidence_pack(bundle) category_paths = _dedupe_texts([ _text(item.get("category_full_path") or item.get("category_path")) for item in _list(evidence.get("category_bindings") or evidence.get("itemset_items")) ]) extracted_points: list[str] = [] for binding in _list(evidence.get("element_bindings")): for sample in _list(_record(binding).get("sample_elements")): sample_record = _record(sample) point_text = _text(sample_record.get("point_text") or sample_record.get("name")) if not point_text: continue post_id = _text(sample_record.get("post_id")) point_type = _text(sample_record.get("point_type")) suffix = " / ".join(item for item in [point_type, f"帖子 {post_id}" if post_id else ""] if item) extracted_points.append(f"{point_text}（{suffix}）" if suffix else point_text) # 分类叶子来源的候选搜索词(element_bindings 元素名)。前端用它对比首轮 query, # 讲清"分类来源的某词与前面哪条来源的同名词重复、被去重未单独显示"。 category_leaf_terms: list[str] = [] for binding in _list(evidence.get("element_bindings")): for sample in _list(_record(binding).get("sample_elements")): name = _text(_record(sample).get("name")) if name: category_leaf_terms.append(name) return { "id": _demand_id(bundle), "name": _text(raw.get("name") or source_context.get("name")), "description": _text(ext_data.get("desc") or raw.get("suggestion") or source_context.get("suggestion")), "reason": _text(ext_data.get("reason") or raw.get("reason") or source_context.get("reason")), "source": _text(raw.get("merge_leve2") or source_context.get("merge_leve2")), "score": raw.get("score") if raw.get("score") is not None else source_context.get("score"), "date": _text(raw.get("dt") or source_context.get("dt")), "type": _text(ext_data.get("type") or evidence.get("source_kind")), "itemset_ids": _text_list(evidence.get("itemset_ids")), "seed_terms": _text_list(evidence.get("seed_terms")), "source_post_id": _text(evidence.get("source_post_id")), "support": evidence.get("absolute_support") if evidence.get("absolute_support") is not None else evidence.get("support"), "pattern_execution_id": _text(evidence.get("pattern_execution_id")), "mining_config_id": _text(evidence.get("mining_config_id")), "category_paths": category_paths, "extracted_points": _dedupe_texts(extracted_points), "category_leaf_terms": _dedupe_texts(category_leaf_terms), } def _source_post_id(seed_ref: dict[str, Any], source_ref: dict[str, Any], bundle: dict[str, Any]) -> str: return _text( source_ref.get("post_id") or seed_ref.get("source_post_id") or _evidence_pack(bundle).get("source_post_id") or _record(bundle.get("pattern_seed_pack")).get("source_post_id") ) def _evidence_pack(bundle: dict[str, Any]) -> dict[str, Any]: source_context = _record(bundle.get("source_context")) ext_data = _record(source_context.get("ext_data")) return _record(ext_data.get("evidence_pack") or source_context.get("evidence_pack")) def _category_binding(bundle: dict[str, Any], source_ref: dict[str, Any]) -> dict[str, Any]: category_id = _text(source_ref.get("category_id")) itemset_item_id = _text(source_ref.get("itemset_item_id")) element_name = _text(source_ref.get("element_name")) evidence = _evidence_pack(bundle) for key in ("category_bindings", "itemset_items"): matched = _find_category_record(_list(evidence.get(key)), category_id, itemset_item_id, element_name) if matched: return matched seed_pack = _record(bundle.get("pattern_seed_pack")) matched = _find_category_record(_list(seed_pack.get("category_bindings")), category_id, itemset_item_id, element_name) if matched: return matched for binding in _list(seed_pack.get("element_bindings")) + _list(evidence.get("element_bindings")): matched = _find_category_record([binding], category_id, itemset_item_id, element_name) if matched: return binding for sample in _list(binding.get("sample_elements")): matched = _find_category_record([sample], category_id, "", element_name) if matched: sample_with_binding = dict(sample) sample_with_binding.setdefault("itemset_item_id", binding.get("itemset_item_id")) sample_with_binding.setdefault("category_path", binding.get("category_path")) return sample_with_binding return {} def _find_category_record( rows: list[dict[str, Any]], category_id: str, itemset_item_id: str, element_name: str, ) -> dict[str, Any]: for row in rows: if itemset_item_id and _text(row.get("itemset_item_id")) == itemset_item_id: return row if category_id and _text(row.get("category_id")) == category_id: return row if element_name and _text(row.get("element_name") or row.get("name")) == element_name: return row return {} def _level_from_path(path: str) -> str: parts = [part for part in path.split("/") if part.strip()] return str(len(parts)) if parts else "" def _detail(label: str, value: Any) -> str: text = _text(value) return f"{label}：{text}" if text else "" def _dedupe_texts(values: list[str]) -> list[str]: result: list[str] = [] seen: set[str] = set() for value in values: text = _text(value) if text and text not in seen: seen.add(text) result.append(text) return result def _text(value: Any, default: str = "") -> str: if value is None: return default text = str(value).strip() return text or default def _text_list(value: Any) -> list[str]: if isinstance(value, list): return [_text(item) for item in value if _text(item)] if isinstance(value, str) and value.strip(): return [value.strip()] return [] def _record(value: Any) -> dict[str, Any]: return value if isinstance(value, dict) else {} def _list(value: Any) -> list[dict[str, Any]]: return [item for item in value if isinstance(item, dict)] if isinstance(value, list) else [] def _number(value: Any) -> int: try: return int(float(value)) except (TypeError, ValueError): return 0 def _cache_ttl_seconds() -> int: raw = os.environ.get("CONTENT_AGENT_FLOW_LEDGER_CACHE_TTL_SECONDS") if raw is None: return _DEFAULT_CACHE_TTL_SECONDS try: return max(0, int(raw)) except ValueError: return _DEFAULT_CACHE_TTL_SECONDS def _cache_get(key: tuple[str, str]) -> dict[str, Any] | None: ttl = _cache_ttl_seconds() if ttl <= 0: return None cached = _FLOW_LEDGER_CACHE.get(key) if not cached: return None expires_at, payload = cached if expires_at <= time.monotonic(): _FLOW_LEDGER_CACHE.pop(key, None) return None return copy.deepcopy(payload) def _cache_set(key: tuple[str, str], payload: dict[str, Any]) -> None: ttl = _cache_ttl_seconds() if ttl <= 0: return _FLOW_LEDGER_CACHE[key] = (time.monotonic() + ttl, copy.deepcopy(payload))