content-find-agent-new/content_agent/flow_ledger_service.py

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))