|
|
@@ -28,12 +28,12 @@ EXTENSION_QUERY_METHODS = {
|
|
|
|
|
|
SOURCE_LABELS = {
|
|
|
"seed_term": "Pattern 种子词",
|
|
|
- "piaoquan_topic_point": "票圈内容点",
|
|
|
+ "piaoquan_topic_point": "票圈帖子具体的点",
|
|
|
"category_leaf_element": "分类叶子元素",
|
|
|
}
|
|
|
QUERY_METHOD_LABELS = {
|
|
|
"seed_term": "Pattern 种子词",
|
|
|
- "piaoquan_topic_point": "票圈内容点",
|
|
|
+ "piaoquan_topic_point": "票圈帖子具体的点",
|
|
|
"category_leaf_element": "分类叶子元素",
|
|
|
"query_next_page": "翻页继续找",
|
|
|
"tag_query": "根据标签继续搜",
|
|
|
@@ -58,6 +58,8 @@ WALK_EDGE_LABELS = {
|
|
|
"video_to_author": "查看作者更多作品",
|
|
|
"author_works": "查看作者更多作品",
|
|
|
"path_stop": "到这里停止",
|
|
|
+ "decision_to_asset": "入池沉淀",
|
|
|
+ "budget_downgrade": "转入待复看",
|
|
|
"terminal": "到这里停止",
|
|
|
}
|
|
|
REASON_LABELS = {
|
|
|
@@ -73,6 +75,7 @@ REASON_LABELS = {
|
|
|
"v4_allow_walk_denied": "不适合继续向外扩展",
|
|
|
"budget_exhausted": "本轮扩展名额已用完",
|
|
|
"query>=70/platform>=65/score>=70": "评分达到继续扩展门槛",
|
|
|
+ "content_decision_reused_for_walk_gate": "复用本条视频的内容判断作为扩展门槛",
|
|
|
"no_decision": "暂未产生判断结果",
|
|
|
}
|
|
|
|
|
|
@@ -142,18 +145,25 @@ class FlowLedgerService:
|
|
|
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 ["query_next_page", "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,
|
|
|
@@ -292,6 +302,26 @@ class FlowLedgerService:
|
|
|
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"))
|
|
|
@@ -485,24 +515,36 @@ class FlowLedgerService:
|
|
|
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)
|
|
|
- trigger_label = _walk_trigger_label(action)
|
|
|
- from_label = _walk_from_label(action, query, bundle)
|
|
|
+ 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)
|
|
|
+ target_videos = [self._video(content, bundle, debug=debug) for content in target_contents[:4]]
|
|
|
+ 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)
|
|
|
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),
|
|
|
"from_label": from_label,
|
|
|
- "result": "已带入后续流程" if status == "success" else "没有继续执行",
|
|
|
+ "result": result_label,
|
|
|
"reason": reason,
|
|
|
- "reason_label": _walk_reason_label(action, reason),
|
|
|
+ "reason_label": reason_label,
|
|
|
"impact": _walk_impact(status, reason, gate_reason),
|
|
|
"score_items": score_items,
|
|
|
- "detail_lines": _walk_detail_lines(action, edge, from_label, trigger_label, 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,
|
|
|
}
|
|
|
|
|
|
@@ -676,16 +718,98 @@ def _walk_status_label(status: str) -> str:
|
|
|
}.get(status, "状态待确认")
|
|
|
|
|
|
|
|
|
-def _walk_from_label(action: dict[str, Any], query: dict[str, Any] | None, bundle: dict[str, Any]) -> str:
|
|
|
+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"))
|
|
|
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 "content" in from_type or "video" in from_type:
|
|
|
- content = next((item for item in bundle["content_items"] if from_node_id in _content_ids(item)), None)
|
|
|
+ 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"从视频“{_short_text(title, 34)}”触发" if title else "从视频内容触发"
|
|
|
+ return f"从作者《{author}》触发,来源视频《{_short_text(title, 24)}》" if title else f"从作者《{author}》触发"
|
|
|
if "tag" in from_type:
|
|
|
return "从视频标签触发"
|
|
|
return "从上一环节出发"
|
|
|
@@ -703,32 +827,42 @@ def _walk_impact(status: str, reason: str, gate_reason: str) -> str:
|
|
|
return "这一步暂不影响最终沉淀。"
|
|
|
|
|
|
|
|
|
-def _walk_trigger_label(action: dict[str, Any]) -> str:
|
|
|
- raw = _record(action.get("raw_payload"))
|
|
|
- hashtag = _text(action.get("hashtag") or raw.get("hashtag"))
|
|
|
+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"}:
|
|
|
- return "标签:未生成(先被扩展门槛拦下)"
|
|
|
- author = _text(action.get("author_id") or raw.get("author_id") or action.get("platform_author_id") or raw.get("platform_author_id"))
|
|
|
+ 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.get("page_cursor") or raw.get("page_cursor") or action.get("cursor") or raw.get("cursor"))
|
|
|
+ cursor = _text(_action_value(action, "page_cursor") or _action_value(action, "cursor"))
|
|
|
if cursor:
|
|
|
- return f"翻页游标:{cursor}"
|
|
|
- to_node = _text(action.get("to_node_id"))
|
|
|
+ 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 f"结果对象:{to_node}"
|
|
|
return "触发对象待确认"
|
|
|
|
|
|
|
|
|
def _walk_reason_label(action: dict[str, Any], reason: str) -> str:
|
|
|
- gate_reason = _text(action.get("walk_gate_reason_code") or action.get("allow_walk_reason"))
|
|
|
+ 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.get("allow_walk_reason"))
|
|
|
+ allow_reason = _text(_action_value(action, "allow_walk_reason"))
|
|
|
if allow_reason and allow_reason != gate_reason:
|
|
|
return f"不适合继续向外扩展:{REASON_LABELS.get(allow_reason, '评分未达到继续扩展门槛')}"
|
|
|
return "不适合继续向外扩展"
|
|
|
@@ -737,24 +871,86 @@ def _walk_reason_label(action: dict[str, Any], reason: str) -> str:
|
|
|
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 == "query_next_page":
|
|
|
+ return f"已翻到下一页,带回 {len(target_videos)} 条视频。"
|
|
|
+ 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 == "query_next_page":
|
|
|
+ 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}",
|
|
|
+ f"起点:{from_label}",
|
|
|
]
|
|
|
if trigger_label != "触发对象待确认":
|
|
|
- lines.append(f"触发对象:{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.get("reason_code")))
|
|
|
+ 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
|
|
|
|
|
|
|