Merge branch 'xym-master' into dev-web

.env.example

@@ -43,5 +43,7 @@ ALIYUN_OSS_ACCESS_KEY_SECRET=
 ALIYUN_OSS_REGION=cn-hangzhou
 ALIYUN_OSS_BUCKET=art-pubbucket
 ALIYUN_OSS_ROOT_PREFIX=supply_agent
+# 手动上传日志专用 OSS 目录（与 Agent 自动上传 / MySQL oss_logs 路径分离）
+ALIYUN_OSS_MANUAL_LOG_PREFIX=supply_agent/manual_logs
 ALIYUN_OSS_PUBLIC_BASE_URL=http://rescdn.yishihui.com
 LOG_OSS_UPLOAD_ENABLED=true

agents/generate_demand_agent/agent.py

@@ -3,37 +3,14 @@ generate_demand_agent 工厂 — 组装需求生成 Agent。
 """
 from __future__ import annotations
 
+from pathlib import Path
+
 from supply_agent import Agent
 from supply_agent.config import Settings
 from agents.generate_demand_agent.tools import register_all_tools
 
-GENERATE_DEMAND_AGENT_SYSTEM_PROMPT = """
-你是需求生成专家。从类目树落到局部需求组合产出。
-
-热度维度含义——必须按此理解与选用：
-- ext_pop：外部热度
-- plat_sust_pop：平台持续热度
-- plat_ly_pop：平台内去年同周期热度
-- recent_pop：平台内近期热度
-选用 source_dim 时写清对应含义；不同维度代表不同热度来源，不可混用概念。
-
-产出层级（严格按此四层，不可跳层或颠倒）：
-1) source_dim：来源热度维度（选哪一种热度信号）
-2) overall_direction：整体方向（介于来源维度与汇总事件之间的大分类总结）
-3) summary_event：汇总事件（对单个或极少数强相关需求的轻量概括，禁止大杂烩式合集）
-4) demand_name：需求名称（必须原样取自 demand_belong_category.name，禁止新造词）
-
-说明：
-- 一个 overall_direction 下可有多个 summary_event。
-- summary_event 优先与 demand_name 一对一；仅当 2~3 个需求同属一个具体事件/场景时才可合并，禁止把大量弱相关需求塞进同一事件。
-- 宁可多写几个 summary_event，也不要做一个过度聚合的大事件。
-
-产出原则：
-- overall_direction 要像“大分类标签”，比维度细、比事件粗。
-- summary_event 要像“单一可识别的事件/场景”，聚焦一个传播点；不要写“XX合集”“XX大全”式过度聚合。
-- 多个 demand_name 只有在同一具体事件下才允许挂在同一 summary_event，且一般不超过 2 个。
-- 理由写清：路径、维度、方向、事件、选用了哪些需求名。
-""".strip()
+_PROMPT_PATH = Path(__file__).parent / "prompt" / "system_prompt.md"
+GENERATE_DEMAND_AGENT_SYSTEM_PROMPT = _PROMPT_PATH.read_text(encoding="utf-8")
 
 
 def create_generate_demand_agent(
@@ -47,6 +24,7 @@ def create_generate_demand_agent(
         name="generate_demand_agent",
         model=model,
         system_prompt=GENERATE_DEMAND_AGENT_SYSTEM_PROMPT,
+        max_iterations=30,
     )
     register_all_tools(agent.tools)
     return agent

agents/generate_demand_agent/prompt/system_prompt.md

@@ -0,0 +1,46 @@
+## 角色与任务
+你是需求生成专家。从类目树落到局部需求组合产出。
+
+一次运行只处理**一个** source_dim，不可混用多维度概念。若用户要求多个维度，按维度分别完成整条流程。
+
+## 热度维度含义
+选用 source_dim 时必须按此理解：
+- ext_pop：外部热度
+- plat_sust_pop：平台持续热度
+- plat_ly_pop：平台内去年同周期热度
+- recent_pop：平台内近期热度
+
+## 产出层级（严格按此四层，不可跳层或颠倒）
+1) source_dim：来源热度维度
+2) overall_direction：整体方向（介于来源维度与汇总事件之间的大分类总结）
+3) summary_event：汇总事件（对单个或极少数强相关需求的轻量概括，禁止大杂烩式合集）
+4) demand_name：需求名称（必须原样取自工具返回的 demand_belong_category.name，禁止新造词）
+
+说明：
+- 一个 overall_direction 下可有多个 summary_event。
+- summary_event 优先与 demand_name 一对一；仅当 2~3 个需求同属一个具体事件/场景时才可合并，禁止把大量弱相关需求塞进同一事件。
+- 宁可多写几个 summary_event，也不要做一个过度聚合的大事件。
+
+## 可用工具
+- `query_latest_biz_dt()`：**未指定 biz_dt 时先调用**，返回两表均有数据的最新业务日（一个确切日期），再用于后续工具。
+- `query_category_tree_by_dim(dimension, biz_dt=None)`：按维度查看有数据的类目树；带「+」表示其下有带数据的叶子。`biz_dt` 格式 YYYYMMDD，省略则用最新业务日。
+- `query_category_leaves_by_dim(dimension, category_ids, biz_dt=None)`：下钻到指定分支下有数据的叶子节点。
+- `query_demand_words_by_category(category_ids, dimension, ..., biz_dt=None)`：查询类目（含子树）下可选用的挂载词及该维热度。
+- `query_category_path(category_ids)`：查询类目根到叶路径，用于写 reason。
+- `batch_save_generated_demands(items, biz_dt=None)`：校验并落库四层产出；`biz_dt` 作为本次落库默认业务日。
+
+## 单维度工作流程
+1. 若用户未指定 `biz_dt`，先调用 `query_latest_biz_dt()`，取返回的 `biz_dt=YYYYMMDD`；同一轮查询与落库保持该日期。
+2. 选定一个 source_dim，调用 `query_category_tree_by_dim(dim, biz_dt)`，找高分且带「+」的分支。
+3. 调用 `query_category_leaves_by_dim(dim, branch_ids, biz_dt)`，落到叶子热点。
+4. 调用 `query_demand_words_by_category(leaf_ids, dim, top_k=20, biz_dt=biz_dt)`，获取候选 demand_name。
+5. 对拟选用的类目调用 `query_category_path`，补全路径。
+6. 按四层组织产出：overall_direction → summary_event → demand_name；禁止跨维度、禁止造词。
+7. 调用 `batch_save_generated_demands(items, biz_dt=biz_dt)` 落库；items 必填 source_dim / overall_direction / summary_event / demand_name，并尽量带上 category_id、reason。
+
+## 产出原则
+- overall_direction 要像“大分类标签”，比维度细、比事件粗。
+- summary_event 要像“单一可识别的事件/场景”，聚焦一个传播点；不要写“XX合集”“XX大全”式过度聚合。
+- 多个 demand_name 只有在同一具体事件下才允许挂在同一 summary_event，且一般不超过 2 个。
+- 理由写清：路径、维度、方向、事件、选用了哪些需求名。
+- 只使用工具返回的真实类目与需求词，禁止幻觉编造。

agents/generate_demand_agent/run.py

@@ -10,7 +10,7 @@ def main() -> None:
     print(f"tools: {agent.tools.list_tools()}")
     print()
     user_input = f'''
-    帮我基于不同维度，产出一批效果好的需求，每个维度都产出一批效果好的需求
+    帮我产出一批平台热度高的需求，50个左右
        '''
     result = agent.run(user_input)
     print(result.content)

agents/generate_demand_agent/tools/__init__.py

@@ -6,23 +6,39 @@ from __future__ import annotations
 from collections.abc import Callable
 from typing import Any
 
+from agents.generate_demand_agent.tools.batch_save_generated_demands import (
+    batch_save_generated_demands,
+)
 from agents.generate_demand_agent.tools.query_category_leaves_by_dim import (
     query_category_leaves_by_dim,
 )
+from agents.generate_demand_agent.tools.query_category_path import query_category_path
 from agents.generate_demand_agent.tools.query_category_tree_by_dim import (
     query_category_tree_by_dim,
 )
+from agents.generate_demand_agent.tools.query_latest_biz_dt import query_latest_biz_dt
+from agents.generate_demand_agent.tools.query_demand_words_by_category import (
+    query_demand_words_by_category,
+)
 from supply_agent.tools.registry import ToolRegistry
 
 ALL_TOOLS: list[Callable[..., Any]] = [
+    query_latest_biz_dt,
     query_category_tree_by_dim,
     query_category_leaves_by_dim,
+    query_demand_words_by_category,
+    query_category_path,
+    batch_save_generated_demands,
 ]
 
 __all__ = [
     "ALL_TOOLS",
+    "batch_save_generated_demands",
     "query_category_leaves_by_dim",
+    "query_category_path",
     "query_category_tree_by_dim",
+    "query_demand_words_by_category",
+    "query_latest_biz_dt",
     "register_all_tools",
 ]
 

agents/generate_demand_agent/tools/batch_save_generated_demands.py

@@ -0,0 +1,288 @@
+"""
+批量保存单维度需求产生结果到 generated_demand 表。
+"""
+from __future__ import annotations
+
+import logging
+import uuid
+from decimal import Decimal
+from typing import Any
+
+from agents.generate_demand_agent.tools.dim_constants import (
+    DIM_KEYS,
+    DIM_LABEL,
+    build_category_path,
+    resolve_biz_dt,
+)
+from supply_agent.tools import tool
+from supply_infra.db.repositories.demand_belong_category_repo import (
+    DemandBelongCategoryRepository,
+)
+from supply_infra.db.repositories.demand_popularity_stats_repo import (
+    DemandPopularityStatsRepository,
+)
+from supply_infra.db.repositories.generated_demand_repo import GeneratedDemandRepository
+from supply_infra.db.repositories.global_tree_category_repo import (
+    GlobalTreeCategoryRepository,
+)
+from supply_infra.db.session import get_session
+
+logger = logging.getLogger(__name__)
+
+
+def _optional_int(value: Any) -> int | None:
+    if value is None or value == "":
+        return None
+    return int(value)
+
+
+def _optional_str(value: Any) -> str | None:
+    if value is None:
+        return None
+    text = str(value).strip()
+    return text or None
+
+
+def _normalize_items(
+    items: list[dict[str, Any]],
+    *,
+    default_run_id: str,
+) -> tuple[list[dict[str, Any]], list[str]]:
+    """校验并规范化待插入行，返回 (rows, errors)。"""
+    rows: list[dict[str, Any]] = []
+    errors: list[str] = []
+    seen_keys: set[tuple[str, str]] = set()
+
+    for idx, item in enumerate(items):
+        if not isinstance(item, dict):
+            errors.append(f"第 {idx} 项不是对象")
+            continue
+
+        source_dim = _optional_str(item.get("source_dim"))
+        overall_direction = _optional_str(item.get("overall_direction"))
+        summary_event = _optional_str(item.get("summary_event"))
+        demand_name = _optional_str(item.get("demand_name"))
+
+        if not source_dim:
+            errors.append(f"第 {idx} 项缺少 source_dim")
+            continue
+        if source_dim not in DIM_KEYS:
+            allowed = "、".join(f"{k}（{DIM_LABEL[k]}）" for k in DIM_KEYS)
+            errors.append(f"第 {idx} 项 source_dim 无效（只能是：{allowed}）: {source_dim}")
+            continue
+        if not overall_direction:
+            errors.append(f"第 {idx} 项缺少 overall_direction")
+            continue
+        if not summary_event:
+            errors.append(f"第 {idx} 项缺少 summary_event")
+            continue
+        if not demand_name:
+            errors.append(f"第 {idx} 项缺少 demand_name")
+            continue
+
+        dedupe_key = (source_dim, demand_name)
+        if dedupe_key in seen_keys:
+            errors.append(
+                f"第 {idx} 项在本次请求中重复: source_dim={source_dim}, demand_name={demand_name}"
+            )
+            continue
+        seen_keys.add(dedupe_key)
+
+        try:
+            category_id = _optional_int(item.get("category_id"))
+        except (TypeError, ValueError):
+            errors.append(f"第 {idx} 项 category_id 无效: {item.get('category_id')!r}")
+            continue
+
+        try:
+            demand_belong_id = _optional_int(item.get("demand_belong_id"))
+        except (TypeError, ValueError):
+            errors.append(
+                f"第 {idx} 项 demand_belong_id 无效: {item.get('demand_belong_id')!r}"
+            )
+            continue
+
+        run_id = _optional_str(item.get("run_id")) or default_run_id
+        reason = _optional_str(item.get("reason"))
+        category_path = _optional_str(item.get("category_path"))
+        biz_dt = _optional_str(item.get("biz_dt"))
+
+        dim_avg: Decimal | None = None
+        dim_count: int | None = None
+        if item.get("dim_avg") is not None and item.get("dim_avg") != "":
+            try:
+                dim_avg = Decimal(str(item.get("dim_avg")))
+            except Exception:
+                errors.append(f"第 {idx} 项 dim_avg 无效: {item.get('dim_avg')!r}")
+                continue
+        if item.get("dim_count") is not None and item.get("dim_count") != "":
+            try:
+                dim_count = int(item.get("dim_count"))
+            except (TypeError, ValueError):
+                errors.append(f"第 {idx} 项 dim_count 无效: {item.get('dim_count')!r}")
+                continue
+
+        rows.append(
+            {
+                "source_dim": source_dim,
+                "overall_direction": overall_direction,
+                "summary_event": summary_event,
+                "demand_name": demand_name,
+                "demand_belong_id": demand_belong_id,
+                "category_id": category_id,
+                "category_path": category_path,
+                "dim_avg": dim_avg,
+                "dim_count": dim_count,
+                "reason": reason,
+                "biz_dt": biz_dt,
+                "run_id": run_id,
+                "is_delete": 0,
+            }
+        )
+
+    return rows, errors
+
+
+@tool
+def batch_save_generated_demands(
+    items: list[dict[str, Any]],
+    biz_dt: str | None = None,
+) -> str:
+    """
+    批量保存单维度需求产生结果。
+
+    写入 generated_demand 表。demand_name 必须已存在于 demand_belong_category；
+    不存在的词会被跳过。同一 run_id 内 (source_dim, demand_name) 去重。
+
+    Args:
+        items: 待保存列表。每项必填：
+            - source_dim: 四维之一（ext_pop / plat_sust_pop / plat_ly_pop / recent_pop）
+            - overall_direction: 整体方向
+            - summary_event: 汇总事件
+            - demand_name: 需求名（须来自 demand_belong_category.name）
+          选填：
+            - category_id, category_path, demand_belong_id, reason,
+              dim_avg, dim_count, biz_dt, run_id
+          若省略 run_id，本次调用自动生成同一 run_id。
+          若省略 demand_belong_id / category_id / 热度快照，将尽量从库中补全。
+        biz_dt: 业务日 YYYYMMDD；省略则使用 demand_popularity_stats 最新业务日。
+          会作为本次落库的默认 biz_dt，并用于补全热度快照。
+
+    Returns:
+        保存结果摘要。
+    """
+    if not items:
+        return "items 不能为空"
+
+    default_run_id = uuid.uuid4().hex
+    rows, errors = _normalize_items(items, default_run_id=default_run_id)
+    if not rows:
+        detail = "；".join(errors) if errors else "无有效数据"
+        return f"没有可保存的数据: {detail}"
+
+    try:
+        with get_session() as session:
+            stats_repo = DemandPopularityStatsRepository(session)
+            resolved_biz_dt, err = resolve_biz_dt(
+                biz_dt,
+                get_latest=stats_repo.get_latest_biz_dt,
+                has_data=stats_repo.has_biz_dt,
+                table_label="demand_popularity_stats 数据",
+            )
+            if err:
+                return err
+
+            belong_repo = DemandBelongCategoryRepository(session)
+            names = [r["demand_name"] for r in rows]
+            belong_by_name = belong_repo.get_by_names(names)
+
+            valid_rows: list[dict[str, Any]] = []
+            skipped_names: list[str] = []
+            for row in rows:
+                belong = belong_by_name.get(row["demand_name"])
+                if belong is None:
+                    skipped_names.append(row["demand_name"])
+                    continue
+                if row["demand_belong_id"] is None:
+                    row["demand_belong_id"] = int(belong.id)
+                if row["category_id"] is None:
+                    row["category_id"] = int(belong.category_id)
+                valid_rows.append(row)
+
+            if not valid_rows:
+                detail = "、".join(skipped_names)
+                extra = f"；校验失败: {'；'.join(errors)}" if errors else ""
+                return f"没有可保存的数据: demand_name 不存在于 demand_belong_category: {detail}{extra}"
+
+            # 补全路径
+            need_path_ids = [
+                int(r["category_id"])
+                for r in valid_rows
+                if r["category_id"] is not None and not r.get("category_path")
+            ]
+            if need_path_ids:
+                categories = GlobalTreeCategoryRepository(session).list_active_categories()
+                by_id = {int(c.id): c for c in categories}
+                for row in valid_rows:
+                    if row.get("category_path") or row["category_id"] is None:
+                        continue
+                    path = build_category_path(int(row["category_id"]), by_id)
+                    if path:
+                        row["category_path"] = path
+
+            # 补全 biz_dt 与热度快照
+            need_stats_ids = [
+                int(r["demand_belong_id"])
+                for r in valid_rows
+                if r["demand_belong_id"] is not None
+                and (r.get("dim_avg") is None or r.get("dim_count") is None)
+            ]
+            stats_by_belong: dict[int, Any] = {}
+            if need_stats_ids:
+                for stats in stats_repo.list_by_biz_dt_and_belong_ids(
+                    resolved_biz_dt, need_stats_ids
+                ):
+                    stats_by_belong[int(stats.demand_category_id)] = stats
+
+            for row in valid_rows:
+                if not row.get("biz_dt"):
+                    row["biz_dt"] = resolved_biz_dt
+                belong_id = row.get("demand_belong_id")
+                if belong_id is None:
+                    continue
+                stats = stats_by_belong.get(int(belong_id))
+                if stats is None:
+                    continue
+                dim = row["source_dim"]
+                if row.get("dim_count") is None:
+                    row["dim_count"] = int(getattr(stats, f"{dim}_count", 0) or 0)
+                if row.get("dim_avg") is None:
+                    avg = getattr(stats, f"{dim}_avg", None)
+                    row["dim_avg"] = (
+                        Decimal(str(avg)) if avg is not None else Decimal("0")
+                    )
+
+            inserted = GeneratedDemandRepository(session).bulk_insert(valid_rows)
+            run_ids = sorted({str(r["run_id"]) for r in valid_rows})
+
+        parts = [
+            f"提交有效 {len(valid_rows)} 条，成功插入 {inserted} 条",
+            f"run_id={','.join(run_ids)}",
+            f"biz_dt={resolved_biz_dt}",
+        ]
+        if skipped_names:
+            parts.append(
+                f"因 demand_name 不存在跳过 {len(skipped_names)} 条: "
+                + "、".join(skipped_names[:20])
+                + ("…" if len(skipped_names) > 20 else "")
+            )
+        if errors:
+            parts.append(f"校验失败 {len(errors)} 条: " + "；".join(errors[:20]))
+
+        message = "。".join(parts)
+        logger.info("batch_save_generated_demands completed: %s", message)
+        return message
+
+    except Exception as e:
+        logger.error("batch_save_generated_demands failed: %s", e, exc_info=True)
+        return f"批量保存生成需求失败: {e}"

agents/generate_demand_agent/tools/dim_constants.py

@@ -1,6 +1,7 @@
 """generate_demand_agent 分类树热度维度共享常量与工具函数。"""
 from __future__ import annotations
 
+from collections.abc import Callable
 from decimal import Decimal
 
 from supply_infra.db.models.category_tree_weight import CategoryTreeWeight
@@ -24,6 +25,53 @@ DIM_LABEL: dict[str, str] = {
 HAS_DATA_LEAF_MARK = "+"
 
 
+def normalize_biz_dt(biz_dt: str | None) -> tuple[str | None, str | None]:
+    """校验并规范化 biz_dt（YYYYMMDD）；空值表示使用最新业务日。"""
+    if biz_dt is None:
+        return None, None
+    text = str(biz_dt).strip()
+    if not text:
+        return None, None
+    if len(text) != 8 or not text.isdigit():
+        return None, f"biz_dt 格式无效，应为 YYYYMMDD: {biz_dt!r}"
+    return text, None
+
+
+def resolve_biz_dt(
+    biz_dt: str | None,
+    *,
+    get_latest: Callable[[], str | None],
+    has_data: Callable[[str], bool] | None = None,
+    table_label: str = "热度数据",
+) -> tuple[str | None, str | None]:
+    """解析 biz_dt：未传则用最新业务日；传入则校验格式与数据是否存在。"""
+    normalized, err = normalize_biz_dt(biz_dt)
+    if err:
+        return None, err
+
+    resolved = normalized or get_latest()
+    if not resolved:
+        return None, f"暂无 {table_label}（请指定 biz_dt 或先写入数据）"
+
+    if normalized and has_data is not None and not has_data(normalized):
+        return None, f"biz_dt={normalized} 无 {table_label}"
+
+    return resolved, None
+
+
+def get_latest_common_biz_dt(session) -> str | None:
+    """返回 category_tree_weight 与 demand_popularity_stats 均有数据的最新 biz_dt。"""
+    from sqlalchemy import func, select
+
+    from supply_infra.db.models.demand_popularity_stats import DemandPopularityStats
+
+    stats_dates = select(DemandPopularityStats.biz_dt).distinct()
+    stmt = select(func.max(CategoryTreeWeight.biz_dt)).where(
+        CategoryTreeWeight.biz_dt.in_(stats_dates)
+    )
+    return session.scalar(stmt)
+
+
 def normalize_parent_id(parent_id: int | None) -> int | None:
     if parent_id is None or parent_id == 0:
         return None
@@ -94,3 +142,45 @@ def collect_descendant_leaves(
             stack.extend(int(c.id) for c in kids)
     leaves.sort()
     return leaves
+
+
+def collect_descendant_ids(
+    root_id: int,
+    children_map: dict[int | None, list[GlobalTreeCategory]],
+) -> list[int]:
+    """收集 root 及其全部子孙节点 id。"""
+    ids: list[int] = []
+    stack = [root_id]
+    seen: set[int] = set()
+    while stack:
+        cid = stack.pop()
+        if cid in seen:
+            continue
+        seen.add(cid)
+        ids.append(cid)
+        stack.extend(int(c.id) for c in children_map.get(cid, []))
+    ids.sort()
+    return ids
+
+
+def build_category_path(
+    category_id: int,
+    by_id: dict[int, GlobalTreeCategory],
+) -> str | None:
+    """从根到指定节点的名称路径，如「美妆护肤 > 护肤 > 防晒」。"""
+    cat = by_id.get(category_id)
+    if cat is None:
+        return None
+    names: list[str] = []
+    current: GlobalTreeCategory | None = cat
+    seen: set[int] = set()
+    while current is not None:
+        cid = int(current.id)
+        if cid in seen:
+            break
+        seen.add(cid)
+        names.append(current.name or str(cid))
+        parent_key = normalize_parent_id(current.parent_id)
+        current = by_id.get(parent_key) if parent_key is not None else None
+    names.reverse()
+    return " > ".join(names)

agents/generate_demand_agent/tools/query_category_leaves_by_dim.py

@@ -13,6 +13,7 @@ from agents.generate_demand_agent.tools.dim_constants import (
     build_score_by_id,
     collect_descendant_leaves,
     format_score,
+    resolve_biz_dt,
 )
 from supply_agent.tools import tool
 from supply_infra.db.repositories.category_tree_weight_repo import (
@@ -49,6 +50,7 @@ def _normalize_ids(category_ids: list[Any]) -> tuple[list[int], str | None]:
 def query_category_leaves_by_dim(
     dimension: str,
     category_ids: list[int],
+    biz_dt: str | None = None,
 ) -> str:
     """
     按分类 id 与热度维度，查询各分类节点下有维度数据的叶子节点。
@@ -64,6 +66,7 @@ def query_category_leaves_by_dim(
             - plat_ly_pop：平台去年同期热度
             - recent_pop：近期热度
         category_ids: 要查询的分类 id 列表（可多个）。
+        biz_dt: 业务日 YYYYMMDD；省略则使用 category_tree_weight 最新业务日。
 
     Returns:
         按分类分组的叶子列表，例如：
@@ -90,14 +93,21 @@ def query_category_leaves_by_dim(
             children_map = build_children_map(categories)
 
             weight_repo = CategoryTreeWeightRepository(session)
-            biz_dt = weight_repo.get_latest_biz_dt()
-            if not biz_dt:
-                return "暂无 category_tree_weight 数据"
-
-            score_by_id = build_score_by_id(weight_repo.list_by_biz_dt(biz_dt), dim)
+            resolved_biz_dt, err = resolve_biz_dt(
+                biz_dt,
+                get_latest=weight_repo.get_latest_biz_dt,
+                has_data=weight_repo.has_biz_dt,
+                table_label="category_tree_weight 数据",
+            )
+            if err:
+                return err
+
+            score_by_id = build_score_by_id(
+                weight_repo.list_by_biz_dt(resolved_biz_dt), dim
+            )
 
             lines = [
-                f"维度={dim}（{DIM_LABEL[dim]}）| biz_dt={biz_dt}",
+                f"维度={dim}（{DIM_LABEL[dim]}）| biz_dt={resolved_biz_dt}",
                 "",
             ]
             total_leaves = 0
@@ -134,7 +144,7 @@ def query_category_leaves_by_dim(
             dim,
             ids,
             total_leaves,
-            biz_dt,
+            resolved_biz_dt,
         )
         return "\n".join(lines).rstrip()
 

agents/generate_demand_agent/tools/query_category_path.py

@@ -0,0 +1,82 @@
+"""
+按分类 id 查询从根到节点的类目路径。
+"""
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+from agents.generate_demand_agent.tools.dim_constants import build_category_path
+from supply_agent.tools import tool
+from supply_infra.db.repositories.global_tree_category_repo import (
+    GlobalTreeCategoryRepository,
+)
+from supply_infra.db.session import get_session
+
+logger = logging.getLogger(__name__)
+
+
+def _normalize_ids(category_ids: list[Any]) -> tuple[list[int], str | None]:
+    if not category_ids:
+        return [], "category_ids 不能为空"
+    out: list[int] = []
+    seen: set[int] = set()
+    for raw in category_ids:
+        try:
+            cid = int(raw)
+        except (TypeError, ValueError):
+            return [], f"category_ids 含无效 id: {raw!r}"
+        if cid in seen:
+            continue
+        seen.add(cid)
+        out.append(cid)
+    if not out:
+        return [], "category_ids 不能为空"
+    return out, None
+
+
+@tool
+def query_category_path(category_ids: list[int]) -> str:
+    """
+    查询分类节点从根到自身的名称路径，用于写 reason。
+
+    Args:
+        category_ids: 分类 id 列表（可多个）。
+
+    Returns:
+        每行一条路径，例如：
+        [256] 美妆护肤 > 护肤 > 防晒 > 物理防晒
+        [128] 美妆护肤 > 护肤 > 精华
+    """
+    ids, err = _normalize_ids(category_ids)
+    if err:
+        return err
+
+    try:
+        with get_session() as session:
+            categories = GlobalTreeCategoryRepository(session).list_active_categories()
+            by_id = {int(c.id): c for c in categories}
+
+            lines: list[str] = []
+            for cid in ids:
+                path = build_category_path(cid, by_id)
+                if path is None:
+                    lines.append(f"[{cid}] （未找到该分类）")
+                else:
+                    lines.append(f"[{cid}] {path}")
+
+        logger.info("query_category_path completed: ids=%s", ids)
+        return "\n".join(lines)
+
+    except Exception as e:
+        logger.error("query_category_path failed: %s", e, exc_info=True)
+        return f"查询分类路径失败: {e}"
+
+
+def main() -> None:
+    """本地手动验证。"""
+    print(query_category_path(category_ids=[1, 2]))
+
+
+if __name__ == "__main__":
+    main()

agents/generate_demand_agent/tools/query_category_tree_by_dim.py

@@ -17,6 +17,7 @@ from agents.generate_demand_agent.tools.dim_constants import (
     build_score_by_id,
     collect_descendant_leaves,
     format_score,
+    resolve_biz_dt,
 )
 from supply_agent.tools import tool
 from supply_infra.db.models.global_tree_category import GlobalTreeCategory
@@ -117,13 +118,13 @@ def _format_weighted_tree(
 
 
 @tool
-def query_category_tree_by_dim(dimension: str) -> str:
+def query_category_tree_by_dim(dimension: str, biz_dt: str | None = None) -> str:
     """
     按热度维度查询有数据的全局分类树。
 
-    联查 global_tree_category 与 category_tree_weight（最新 biz_dt），
-    仅返回指定维度 count>0 的节点及其祖先。每行格式为 [id]名称（score）；
-    若该节点下存在有数据的叶子节点，末尾带 + 标记。
+    联查 global_tree_category 与 category_tree_weight，仅返回指定维度 count>0
+    的节点及其祖先。每行格式为 [id]名称（score）；若该节点下存在有数据的叶子节点，
+    末尾带 + 标记。
 
     Args:
         dimension: 必选热度维度，只能是其一：
@@ -131,6 +132,7 @@ def query_category_tree_by_dim(dimension: str) -> str:
             - plat_sust_pop：平台持续热度
             - plat_ly_pop：平台去年同期热度
             - recent_pop：近期热度
+        biz_dt: 业务日 YYYYMMDD；省略则使用 category_tree_weight 最新业务日。
 
     Returns:
         层级分明的完整有数据分类树，例如：
@@ -148,22 +150,27 @@ def query_category_tree_by_dim(dimension: str) -> str:
         with get_session() as session:
             categories = GlobalTreeCategoryRepository(session).list_active_categories()
             weight_repo = CategoryTreeWeightRepository(session)
-            biz_dt = weight_repo.get_latest_biz_dt()
-            if not biz_dt:
-                return "暂无 category_tree_weight 数据"
-
-            weights = weight_repo.list_by_biz_dt(biz_dt)
+            resolved_biz_dt, err = resolve_biz_dt(
+                biz_dt,
+                get_latest=weight_repo.get_latest_biz_dt,
+                has_data=weight_repo.has_biz_dt,
+                table_label="category_tree_weight 数据",
+            )
+            if err:
+                return err
+
+            weights = weight_repo.list_by_biz_dt(resolved_biz_dt)
             score_by_id = build_score_by_id(weights, dim)
             tree_text = _format_weighted_tree(categories, score_by_id)
 
         header = (
-            f"维度={dim}（{DIM_LABEL[dim]}）| biz_dt={biz_dt} | "
+            f"维度={dim}（{DIM_LABEL[dim]}）| biz_dt={resolved_biz_dt} | "
             f"「{HAS_DATA_LEAF_MARK}」表示其下有带数据的叶子节点"
         )
         logger.info(
             "query_category_tree_by_dim completed: dim=%s biz_dt=%s",
             dim,
-            biz_dt,
+            resolved_biz_dt,
         )
         return f"{header}\n{tree_text}"
 

agents/generate_demand_agent/tools/query_demand_words_by_category.py

@@ -0,0 +1,246 @@
+"""
+按分类 id + 热度维度，查询可选用的挂载需求词及词级热度。
+"""
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+from agents.generate_demand_agent.tools.dim_constants import (
+    DIM_KEYS,
+    DIM_LABEL,
+    build_children_map,
+    collect_descendant_ids,
+    format_score,
+    resolve_biz_dt,
+)
+from supply_agent.tools import tool
+from supply_infra.db.models.demand_popularity_stats import DemandPopularityStats
+from supply_infra.db.repositories.demand_belong_category_repo import (
+    DemandBelongCategoryRepository,
+)
+from supply_infra.db.repositories.demand_popularity_stats_repo import (
+    DemandPopularityStatsRepository,
+)
+from supply_infra.db.repositories.global_tree_category_repo import (
+    GlobalTreeCategoryRepository,
+)
+from supply_infra.db.session import get_session
+
+logger = logging.getLogger(__name__)
+
+
+def _normalize_ids(category_ids: list[Any]) -> tuple[list[int], str | None]:
+    if not category_ids:
+        return [], "category_ids 不能为空"
+    out: list[int] = []
+    seen: set[int] = set()
+    for raw in category_ids:
+        try:
+            cid = int(raw)
+        except (TypeError, ValueError):
+            return [], f"category_ids 含无效 id: {raw!r}"
+        if cid in seen:
+            continue
+        seen.add(cid)
+        out.append(cid)
+    if not out:
+        return [], "category_ids 不能为空"
+    return out, None
+
+
+def _word_dim_score(
+    stats: DemandPopularityStats | None, dim: str
+) -> tuple[float | None, int]:
+    if stats is None:
+        return None, 0
+    count = int(getattr(stats, f"{dim}_count", 0) or 0)
+    if count <= 0:
+        return None, 0
+    avg = getattr(stats, f"{dim}_avg", None)
+    return (float(avg) if avg is not None else 0.0), count
+
+
+@tool
+def query_demand_words_by_category(
+    category_ids: list[int],
+    dimension: str,
+    include_descendants: bool = True,
+    min_count: int = 1,
+    top_k: int = 30,
+    biz_dt: str | None = None,
+) -> str:
+    """
+    按分类 id 与热度维度，查询挂载需求词及该维词级热度。
+
+    demand_name 必须来自本工具返回的词名（对应 demand_belong_category.name）。
+    include_descendants=True 时包含子树内所有挂载词；False 时仅查节点自身直接挂载。
+    每个入参 category 下最多返回 top_k 个词，按该维 avg 降序。
+
+    Args:
+        category_ids: 分类 id 列表（可多个）。
+        dimension: 热度维度，只能是其一：
+            - ext_pop：外部热度
+            - plat_sust_pop：平台持续热度
+            - plat_ly_pop：平台去年同期热度
+            - recent_pop：近期热度
+        include_descendants: 是否包含子树挂载词，默认 True。
+        min_count: 该维 count 下限，默认 1。
+        top_k: 每个入参 category 最多返回词数，默认 30。
+        biz_dt: 业务日 YYYYMMDD；省略则使用 demand_popularity_stats 最新业务日。
+
+    Returns:
+        按入参分类分组的挂载词列表，含 score / count / belong_id。
+    """
+    dim = str(dimension or "").strip()
+    if dim not in DIM_KEYS:
+        allowed = "、".join(f"{k}（{DIM_LABEL[k]}）" for k in DIM_KEYS)
+        return f"dimension 无效，只能是：{allowed}"
+
+    ids, err = _normalize_ids(category_ids)
+    if err:
+        return err
+
+    try:
+        min_count_int = max(0, int(min_count))
+    except (TypeError, ValueError):
+        return f"min_count 无效: {min_count!r}"
+    try:
+        top_k_int = max(1, int(top_k))
+    except (TypeError, ValueError):
+        return f"top_k 无效: {top_k!r}"
+
+    try:
+        with get_session() as session:
+            categories = GlobalTreeCategoryRepository(session).list_active_categories()
+            by_id = {int(c.id): c for c in categories}
+            children_map = build_children_map(categories)
+
+            stats_repo = DemandPopularityStatsRepository(session)
+            resolved_biz_dt, err = resolve_biz_dt(
+                biz_dt,
+                get_latest=stats_repo.get_latest_biz_dt,
+                has_data=stats_repo.has_biz_dt,
+                table_label="demand_popularity_stats 数据",
+            )
+            if err:
+                return err
+
+            belong_repo = DemandBelongCategoryRepository(session)
+
+            # 收集本次需要查询的全部挂载类目
+            scope_by_root: dict[int, list[int]] = {}
+            all_scope_ids: set[int] = set()
+            for cid in ids:
+                if include_descendants:
+                    scope = collect_descendant_ids(cid, children_map)
+                else:
+                    scope = [cid]
+                scope_by_root[cid] = scope
+                all_scope_ids.update(scope)
+
+            belongs = belong_repo.list_by_category_ids(sorted(all_scope_ids))
+            belongs_by_cat: dict[int, list] = {}
+            for row in belongs:
+                if not row.name:
+                    continue
+                belongs_by_cat.setdefault(int(row.category_id), []).append(row)
+
+            belong_ids = [int(r.id) for rows in belongs_by_cat.values() for r in rows]
+            stats_rows = stats_repo.list_by_biz_dt_and_belong_ids(
+                resolved_biz_dt, belong_ids
+            )
+            stats_by_belong = {
+                int(s.demand_category_id): s for s in stats_rows
+            }
+
+            lines = [
+                f"维度={dim}（{DIM_LABEL[dim]}）| biz_dt={resolved_biz_dt}"
+                f" | include_descendants={include_descendants}"
+                f" | min_count={min_count_int} | top_k={top_k_int}",
+                "",
+            ]
+            total_words = 0
+
+            for cid in ids:
+                cat = by_id.get(cid)
+                if cat is None:
+                    lines.append(f"[{cid}]")
+                    lines.append("  （未找到该分类）")
+                    lines.append("")
+                    continue
+
+                cname = cat.name or ""
+                lines.append(f"[{cid}]{cname}")
+
+                candidates: list[tuple[float, int, int, str, int]] = []
+                # (avg, count, belong_id, name, hang_category_id)
+                for scope_cid in scope_by_root[cid]:
+                    for belong in belongs_by_cat.get(scope_cid, []):
+                        avg, count = _word_dim_score(
+                            stats_by_belong.get(int(belong.id)), dim
+                        )
+                        if count < min_count_int:
+                            continue
+                        score = avg if avg is not None else 0.0
+                        candidates.append(
+                            (
+                                score,
+                                count,
+                                int(belong.id),
+                                str(belong.name),
+                                scope_cid,
+                            )
+                        )
+
+                candidates.sort(key=lambda x: (-x[0], -x[1], x[2]))
+                selected = candidates[:top_k_int]
+                if not selected:
+                    lines.append("  （无符合条件的挂载词）")
+                    lines.append("")
+                    continue
+
+                total_words += len(selected)
+                for avg, count, belong_id, name, hang_cid in selected:
+                    hang_note = ""
+                    if include_descendants and hang_cid != cid:
+                        hang_cat = by_id.get(hang_cid)
+                        hang_name = hang_cat.name if hang_cat else ""
+                        hang_note = f" hang=[{hang_cid}]{hang_name}"
+                    lines.append(
+                        f"  - {name}（{format_score(avg)}） count={count}"
+                        f"  belong_id={belong_id}{hang_note}"
+                    )
+                if len(candidates) > top_k_int:
+                    lines.append(
+                        f"  （另有 {len(candidates) - top_k_int} 个词未展示，已按 avg 截断）"
+                    )
+                lines.append("")
+
+        logger.info(
+            "query_demand_words_by_category completed: dim=%s ids=%s words=%d biz_dt=%s",
+            dim,
+            ids,
+            total_words,
+            resolved_biz_dt,
+        )
+        return "\n".join(lines).rstrip()
+
+    except Exception as e:
+        logger.error("query_demand_words_by_category failed: %s", e, exc_info=True)
+        return f"查询分类挂载词失败: {e}"
+
+
+def main() -> None:
+    """本地手动验证。"""
+    print(
+        query_demand_words_by_category(
+            category_ids=[1, 2],
+            dimension="recent_pop",
+            top_k=10,
+        )
+    )
+
+
+if __name__ == "__main__":
+    main()

agents/generate_demand_agent/tools/query_latest_biz_dt.py

@@ -0,0 +1,54 @@
+"""
+查询热度相关表的最新共有业务日，供未指定 biz_dt 时先探测可用日期。
+"""
+from __future__ import annotations
+
+import logging
+
+from agents.generate_demand_agent.tools.dim_constants import get_latest_common_biz_dt
+from supply_agent.tools import tool
+from supply_infra.db.session import get_session
+
+logger = logging.getLogger(__name__)
+
+
+@tool
+def query_latest_biz_dt() -> str:
+    """
+    查询两表均有数据的最新业务日（biz_dt）。
+
+    当用户未指定 biz_dt、或不确定该用哪一天时，**应先调用本工具**，
+    再将其返回的 biz_dt 传给后续查询与落库工具。
+
+    共有业务日 = category_tree_weight 与 demand_popularity_stats 都存在的最新日期。
+
+    Returns:
+        一个确切的 biz_dt，例如：
+        biz_dt=20260715
+        若无共有日期则返回错误说明。
+    """
+    try:
+        with get_session() as session:
+            biz_dt = get_latest_common_biz_dt(session)
+
+        if not biz_dt:
+            message = "暂无两表共有业务日（category_tree_weight 与 demand_popularity_stats 无重叠日期）"
+            logger.info("query_latest_biz_dt completed: %s", message)
+            return message
+
+        message = f"biz_dt={biz_dt}"
+        logger.info("query_latest_biz_dt completed: %s", message)
+        return message
+
+    except Exception as e:
+        logger.error("query_latest_biz_dt failed: %s", e, exc_info=True)
+        return f"查询最新业务日失败: {e}"
+
+
+def main() -> None:
+    """本地手动验证。"""
+    print(query_latest_biz_dt())
+
+
+if __name__ == "__main__":
+    main()

api/app.py

@@ -1,11 +1,12 @@
 """FastAPI application — category tree API on port 8080."""
 from __future__ import annotations
 
-from fastapi import FastAPI, Query
+from fastapi import FastAPI, HTTPException, Query
 from fastapi.middleware.cors import CORSMiddleware
 
 from api.services.category_tree import build_category_tree
 from api.services.demand_belong_category import list_demand_belong_categories
+from api.services.demand_videos import list_videos_for_demand_belong
 from api.services.oss_logs import list_demand_belong_oss_logs
 
 app = FastAPI(title="SupplyAgent API", version="0.1.0")
@@ -47,6 +48,15 @@ def demand_belong_category() -> dict:
     return {"items": items}
 
 
+@app.get("/api/demand-belong-category/{belong_id}/videos")
+def demand_belong_videos(belong_id: int) -> dict:
+    """Return videos linked to a demand_belong_category row (vid + title + final_topic_json)."""
+    result = list_videos_for_demand_belong(belong_id)
+    if result is None:
+        raise HTTPException(status_code=404, detail="demand_belong_category not found")
+    return result
+
+
 @app.get("/api/demand-belong-oss-logs")
 def demand_belong_oss_logs() -> dict:
     """Return demand_belong_category_agent oss_logs ordered by create_time desc."""

pyproject.toml

@@ -21,6 +21,7 @@ dependencies = [
     "oss2>=2.18.0",
     "fastapi>=0.115.0",
     "uvicorn[standard]>=0.32.0",
+    "markdown>=3.6",
 ]
 
 [project.optional-dependencies]

requirements.txt

@@ -6,6 +6,7 @@ pydantic-settings>=2.0
 httpx>=0.27.0
 rich>=13.0
 python-dotenv>=1.0.0
+markdown>=3.6
 
 # Database
 sqlalchemy>=2.0

scripts/upload_logs_to_oss.py

@@ -0,0 +1,40 @@
+#!/usr/bin/env python3
+"""Upload a log file from logs/ to OSS (manual folder, no MySQL).
+
+Usage:
+  python scripts/upload_logs_to_oss.py run_xxx.log
+
+  # 从 Python 调用
+  from scripts.upload_logs_to_oss import main
+  oss_path = main("run_xxx.log")
+"""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+_ROOT = Path(__file__).resolve().parents[1]
+_LOGS_DIR = _ROOT / "logs"
+if str(_ROOT) not in sys.path:
+    sys.path.insert(0, str(_ROOT))
+
+from supply_infra.config import get_infra_settings
+from supply_infra.oss.client import OssClient
+
+
+def main(filename: str) -> str:
+    """Read ``logs/{filename}``, upload to OSS, return public OSS path."""
+    local_path = _LOGS_DIR / Path(filename).name
+    if not local_path.is_file():
+        raise FileNotFoundError(f"Log file not found: {local_path}")
+
+    settings = get_infra_settings()
+    client = OssClient(settings, root_prefix=settings.aliyun_oss_manual_log_prefix)
+    object_key = client.object_key(local_path.name)
+    return client.upload_file(local_path, object_key)
+
+
+if __name__ == "__main__":
+    oss_path = main("run_generate_demand_agent_20260716_160355_8f751d0d.html")
+    print(oss_path)

supply_agent/logging/visualize.py

@@ -7,8 +7,23 @@ import json
 from pathlib import Path
 from typing import Any
 
+import markdown as _markdown_lib
+
 from supply_agent.logging.parser import load_run_events, summarize_run
 
+_MARKDOWN_EXTENSIONS = ["fenced_code", "tables", "sane_lists"]
+
+
+def _render_markdown(text: Any, *, soft_breaks: bool = False, boxed: bool = True) -> str:
+    """Render text as Markdown; ``boxed`` wraps it in a bordered, scrollable box."""
+    if not text:
+        return ""
+    extensions = [*_MARKDOWN_EXTENSIONS, "nl2br"] if soft_breaks else _MARKDOWN_EXTENSIONS
+    body = _markdown_lib.markdown(str(text), extensions=extensions)
+    if not boxed:
+        return f'<div class="markdown-body">{body}</div>'
+    return f'<div class="markdown-box"><div class="markdown-body">{body}</div></div>'
+
 
 def _esc(text: Any) -> str:
     if text is None:
@@ -104,50 +119,46 @@ def _role_badge(role: str) -> str:
 def _render_messages(
     messages: list[dict[str, Any]] | None,
     *,
-    index_offset: int = 0,
     default_open: bool | None = None,
 ) -> str:
     if not messages:
         return '<p class="muted">无消息</p>'
     parts: list[str] = []
-    for i, msg in enumerate(messages):
+    for msg in messages:
         role = msg.get("role", "?")
         body_parts: list[str] = []
         if msg.get("content"):
             body_parts.append(_pre(msg["content"], klass="code prose"))
-        if msg.get("tool_calls"):
-            body_parts.append(
-                '<div class="sublabel">决定调用的工具</div>'
-                + _render_tool_call_cards(msg["tool_calls"])
-            )
-        if msg.get("tool_call_id"):
-            body_parts.append(
-                f'<div class="meta-line">tool_call_id: <code>{_esc(msg["tool_call_id"])}</code></div>'
-            )
-        if msg.get("name"):
-            body_parts.append(
-                f'<div class="meta-line">name: <code>{_esc(msg["name"])}</code></div>'
-            )
-        # Tool result content: try pretty JSON (already handled by _pre via content)        if default_open is None:
+
+        if role in ("user", "tool"):
+            open_attr = ""
+        elif default_open is None:
             open_attr = "open" if role != "system" else ""
         else:
             open_attr = "open" if default_open else ""
-        preview = _preview(msg.get("content"))
-        if not preview and msg.get("tool_calls"):
-            names = []
-            for tc in msg["tool_calls"]:
-                name, _, _ = _normalize_tool_call(tc)
-                names.append(name)
-            preview = "调用: " + ", ".join(names)
-        elif not preview and msg.get("name"):
-            preview = f"tool → {msg['name']}"
+
+        if role == "tool":
+            tool_name = msg.get("name") or "?"
+            summary_html = (
+                f'{_role_badge(role)} <span class="msg-preview">{_esc(tool_name)}</span>'
+            )
+        else:
+            preview = _preview(msg.get("content"))
+            if not preview and msg.get("tool_calls"):
+                names = []
+                for tc in msg["tool_calls"]:
+                    name, _, _ = _normalize_tool_call(tc)
+                    names.append(name)
+                preview = "调用: " + ", ".join(names)
+            summary_html = (
+                f'{_role_badge(role)} <span class="msg-preview">{_esc(preview)}</span>'
+            )
+
         parts.append(
             f"""
             <details class="msg" {open_attr}>
-              <summary>{_role_badge(role)} <span class="msg-idx">#{index_offset + i + 1}</span>
-                <span class="msg-preview">{_esc(preview)}</span>
-              </summary>
-              <div class="msg-body">{"".join(body_parts) or '<p class="muted">(empty)</p>'}</div>
+              <summary>{summary_html}</summary>
+              <div class="msg-body">{"".join(body_parts)}</div>
             </details>
             """
         )
@@ -161,68 +172,53 @@ def _preview(text: Any, limit: int = 80) -> str:
     return s if len(s) <= limit else s[: limit - 1] + "…"
 
 
-def _render_tools_schema(tools: list[dict[str, Any]] | None) -> str:
+def _tool_names(tools: list[dict[str, Any]] | None) -> list[str]:
     if not tools:
-        return '<p class="muted">本次请求未附带 tools</p>'
-    names = []
+        return []
+    names: list[str] = []
     for t in tools:
         fn = t.get("function") or t
         names.append(fn.get("name", "?"))
-    chips = "".join(f'<span class="chip">{_esc(n)}</span>' for n in names)
-    return f"""
-    <div class="chip-row">{chips}</div>
-    <details>
-      <summary>查看完整 tools schema</summary>
-      {_pre(tools)}
-    </details>
-    """
+    return names
+
 
+def _extract_system_prompt_and_tools(
+    events: list[dict[str, Any]],
+) -> tuple[str | None, list[str]]:
+    """Pull the system prompt and tool list from the first llm_input event."""
+    for ev in events:
+        if ev.get("event") != "llm_input":
+            continue
+        data = ev.get("data") or {}
+        messages = data.get("messages") or []
+        sys_msg = next((m for m in messages if m.get("role") == "system"), None)
+        content = sys_msg.get("content") if sys_msg else None
+        return content, _tool_names(data.get("tools"))
+    return None, []
 
-def _render_llm_input(
+
+def _render_llm_input_section(
     data: dict[str, Any],
-    seq: int,
-    iteration: Any,
     *,
     prev_messages: list[dict[str, Any]] | None = None,
-) -> str:
+) -> tuple[str, list[dict[str, Any]]]:
+    """Render the LLM-input body; returns (html, messages) so callers can track history."""
     messages = list(data.get("messages") or [])
-    history, latest = _split_history_and_latest(messages, prev_messages)
-    history_count = len(history)
+    _history, latest = _split_history_and_latest(messages, prev_messages)
     latest_count = len(latest)
 
-    history_html = ""
-    if history:
-        history_html = f"""
-        <details class="history-block">
-          <summary>查看历史输入（{history_count} 条）</summary>
-          <div class="history-body">
-            {_render_messages(history, default_open=False)}
-          </div>
-        </details>
-        """
-
-    return f"""
-    <article class="card card-input" id="step-{seq}">
-      <header class="card-header">
-        <div class="step-num">Step {seq}</div>
-        <div class="card-title">LLM 输入</div>
-        <div class="card-tags">
-          <span class="tag">iteration {iteration}</span>
-          <span class="tag">{_esc(data.get("model", ""))}</span>
-          <span class="tag">temp {_esc(data.get("temperature", ""))}</span>
-          <span class="tag">新增 {latest_count}</span>
-          <span class="tag">共 {len(messages)} msgs</span>
-        </div>
-      </header>
-      <div class="card-body">
-        <h4>本轮新增输入</h4>
-        {_render_messages(latest, index_offset=history_count, default_open=True)}
-        {history_html}
-        <h4>Available Tools</h4>
-        {_render_tools_schema(data.get("tools"))}
+    html_out = f"""
+    <section class="substep">
+      <div class="substep-header">
+        <span class="substep-title">LLM 输入</span>
+        <span class="tag">新增 {latest_count}</span>
       </div>
-    </article>
+      <div class="substep-body">
+        {_render_messages(latest, default_open=True)}
+      </div>
+    </section>
     """
+    return html_out, messages
 
 
 def _render_reasoning(reasoning: Any) -> str:
@@ -231,7 +227,7 @@ def _render_reasoning(reasoning: Any) -> str:
     return f"""
     <div class="reasoning">
       <div class="sublabel">思考过程</div>
-      {_pre(reasoning, klass="code prose reasoning-text")}
+      {_render_markdown(reasoning, soft_breaks=True, boxed=False)}
     </div>
     """
 
@@ -254,48 +250,25 @@ def _normalize_tool_call(tc: dict[str, Any]) -> tuple[str, Any, str]:
     return name, _parse_tool_arguments(raw_args), str(tc.get("id") or "")
 
 
-def _render_tool_call_cards(tool_calls: list[dict[str, Any]]) -> str:
-    """Render tool calls as name + args cards; raw JSON in a collapsible."""
+def _render_output_tool_calls(tool_calls: list[dict[str, Any]]) -> str:
+    """Show each tool the model decided to call, collapsed by default; expand for args."""
     if not tool_calls:
         return ""
-
-    cards: list[str] = []
-    for i, tc in enumerate(tool_calls, 1):
-        name, args, tc_id = _normalize_tool_call(tc)
-        cards.append(
+    parts: list[str] = []
+    for tc in tool_calls:
+        name, args, _ = _normalize_tool_call(tc)
+        parts.append(
             f"""
-            <div class="planned-tool">
-              <div class="planned-tool-head">
-                <span class="chip">{_esc(name)}</span>
-                <span class="planned-tool-idx">#{i}</span>
-                {f'<code class="planned-tool-id">{_esc(tc_id)}</code>' if tc_id else ""}
-              </div>
-              <div class="sublabel">参数</div>
-              {_pre(args)}
-            </div>
+            <details class="msg">
+              <summary><span class="badge role-tool">tool</span> <span class="msg-preview">{_esc(name)}</span></summary>
+              <div class="msg-body">{_pre(args)}</div>
+            </details>
             """
         )
-
-    return f"""
-    <div class="planned-tool-list">{"".join(cards)}</div>
-    <details class="raw-block">
-      <summary>查看原始 tool_calls JSON</summary>
-      {_pre(tool_calls)}
-    </details>
-    """
-
-
-def _render_planned_tool_calls(tool_calls: list[dict[str, Any]]) -> str:
-    """Show each planned tool as name + args; omit entirely when empty."""
-    if not tool_calls:
-        return ""
-    return f"""
-    <h4>模型决定调用的工具</h4>
-    {_render_tool_call_cards(tool_calls)}
-    """
+    return '<div class="msg-list">' + "".join(parts) + "</div>"
 
 
-def _render_llm_output(data: dict[str, Any], seq: int, iteration: Any) -> str:
+def _render_llm_output_section(data: dict[str, Any]) -> str:
     tool_calls = data.get("tool_calls") or []
     reasoning = data.get("reasoning")
     content = data.get("content")
@@ -313,82 +286,61 @@ def _render_llm_output(data: dict[str, Any], seq: int, iteration: Any) -> str:
         """
 
     content_html = (
-        f"<h4>模型输出文本</h4>{_pre(content, klass='code prose')}" if content else ""
+        f"<h4>模型输出文本</h4>{_render_markdown(content, soft_breaks=True)}"
+        if content
+        else ""
     )
 
-    tags: list[str] = [f'<span class="tag">iteration {iteration}</span>']
+    tags: list[str] = []
     if reasoning:
         tags.append('<span class="tag">有思考</span>')
     if tool_calls:
         tags.append(f'<span class="tag">{len(tool_calls)} tool calls</span>')
 
-    raw_html = ""
-    if data.get("raw"):
-        raw_html = f"""
-        <details class="raw-block">
-          <summary>原始 API 响应 (raw)</summary>
-          {_pre(data["raw"])}
-        </details>
-        """
-
     return f"""
-    <article class="card card-output" id="step-{seq}">
-      <header class="card-header">
-        <div class="step-num">Step {seq}</div>
-        <div class="card-title">LLM 输出</div>
-        <div class="card-tags">
-          {"".join(tags)}
-        </div>
-      </header>
-      <div class="card-body">
+    <section class="substep">
+      <div class="substep-header">
+        <span class="substep-title">LLM 输出</span>
+        {"".join(tags)}
+      </div>
+      <div class="substep-body">
         {_render_reasoning(reasoning)}
         {content_html}
-        {_render_planned_tool_calls(tool_calls)}
+        {_render_output_tool_calls(tool_calls)}
         {usage_html}
-        {raw_html}
       </div>
-    </article>
+    </section>
     """
 
 
-def _render_tool_call(data: dict[str, Any], seq: int, iteration: Any) -> str:
-    is_error = bool(data.get("is_error"))
-    status = "error" if is_error else "ok"
-    args = data.get("arguments_parsed", data.get("arguments"))
-    result = data.get("result_parsed", data.get("result"))
-    tool_id = data.get("tool_call_id") or ""
-
+def _render_step_card(
+    step_no: int,
+    iteration: Any,
+    title: str,
+    inner_html: str,
+) -> str:
+    title_html = f'<div class="card-title">{title}</div>' if title else ""
     return f"""
-    <article class="card card-tool {status}" id="step-{seq}">
+    <article class="card card-step" id="step-{step_no}">
       <header class="card-header">
-        <div class="step-num">Step {seq}</div>
-        <div class="card-title">工具调用 · {_esc(data.get("tool", "?"))}</div>
+        <div class="step-num">Step {step_no}</div>
+        {title_html}
         <div class="card-tags">
           <span class="tag">iteration {iteration}</span>
-          <span class="tag tag-{status}">{"ERROR" if is_error else "OK"}</span>
-          {f'<span class="tag"><code>{_esc(tool_id)}</code></span>' if tool_id else ""}
         </div>
       </header>
-      <div class="card-body tool-io">
-        <div class="io-block">
-          <div class="sublabel">输入 (arguments)</div>
-          {_pre(args)}
-        </div>
-        <div class="io-arrow" aria-hidden="true">→</div>
-        <div class="io-block">
-          <div class="sublabel">输出 (result)</div>
-          {_pre(result)}
-        </div>
+      <div class="card-body step-body">
+        {inner_html}
       </div>
     </article>
     """
 
 
-def _render_skill(data: dict[str, Any], seq: int, iteration: Any) -> str:
+def _render_skill(data: dict[str, Any], step_no: int, iteration: Any) -> str:
     return f"""
-    <article class="card card-skill" id="step-{seq}">
+    <article class="card card-skill" id="step-{step_no}">
       <header class="card-header">
-        <div class="step-num">Step {seq}</div>
+        <div class="step-num">Step {step_no}</div>
         <div class="card-title">技能加载</div>
         <div class="card-tags">
           <span class="tag">iteration {iteration}</span>
@@ -399,42 +351,87 @@ def _render_skill(data: dict[str, Any], seq: int, iteration: Any) -> str:
     """
 
 
+_HIDDEN_EVENT_TYPES = ("run_start", "run_end", "tool_call")
+
+
+def _group_visible_events(
+    events: list[dict[str, Any]],
+) -> list[list[dict[str, Any]]]:
+    """
+    Group renderable events into steps.
+
+    A consecutive ``llm_input`` immediately followed by ``llm_output`` forms a
+    single step (one round-trip); everything else is its own step.
+    """
+    visible = [ev for ev in events if ev.get("event") not in _HIDDEN_EVENT_TYPES]
+    groups: list[list[dict[str, Any]]] = []
+    i = 0
+    n = len(visible)
+    while i < n:
+        ev = visible[i]
+        if (
+            ev.get("event") == "llm_input"
+            and i + 1 < n
+            and visible[i + 1].get("event") == "llm_output"
+        ):
+            groups.append([ev, visible[i + 1]])
+            i += 2
+        else:
+            groups.append([ev])
+            i += 1
+    return groups
+
+
+def _event_iteration(ev: dict[str, Any]) -> Any:
+    data = ev.get("data") or {}
+    return ev.get("iteration", data.get("iteration", "—"))
+
+
 def _render_timeline(events: list[dict[str, Any]]) -> str:
     parts: list[str] = []
     prev_llm_messages: list[dict[str, Any]] | None = None
-    for ev in events:
+    for step_no, group in enumerate(_group_visible_events(events), start=1):
+        if len(group) == 2:
+            input_ev, output_ev = group
+            input_data = input_ev.get("data") or {}
+            output_data = output_ev.get("data") or {}
+            input_html, messages = _render_llm_input_section(
+                input_data, prev_messages=prev_llm_messages
+            )
+            prev_llm_messages = messages
+            output_html = _render_llm_output_section(output_data)
+            parts.append(
+                _render_step_card(
+                    step_no,
+                    _event_iteration(input_ev),
+                    "",
+                    input_html + output_html,
+                )
+            )
+            continue
+
+        ev = group[0]
         etype = ev.get("event")
         data = ev.get("data") or {}
-        seq = ev.get("seq", 0)
-        iteration = ev.get("iteration", data.get("iteration", "—"))
+        iteration = _event_iteration(ev)
 
-        if etype == "run_start":
-            continue
-        if etype == "run_end":
-            continue
         if etype == "llm_input":
-            messages = list(data.get("messages") or [])
-            parts.append(
-                _render_llm_input(
-                    data,
-                    seq,
-                    iteration,
-                    prev_messages=prev_llm_messages,
-                )
+            input_html, messages = _render_llm_input_section(
+                data, prev_messages=prev_llm_messages
             )
             prev_llm_messages = messages
+            parts.append(_render_step_card(step_no, iteration, "LLM 输入", input_html))
         elif etype == "llm_output":
-            parts.append(_render_llm_output(data, seq, iteration))
-        elif etype == "tool_call":
-            parts.append(_render_tool_call(data, seq, iteration))
+            output_html = _render_llm_output_section(data)
+            parts.append(_render_step_card(step_no, iteration, "LLM 输出", output_html))
         elif etype == "skill_loaded":
-            parts.append(_render_skill(data, seq, iteration))
+            parts.append(_render_skill(data, step_no, iteration))
         else:
             parts.append(
                 f"""
-                <article class="card" id="step-{seq}">
+                <article class="card" id="step-{step_no}">
                   <header class="card-header">
-                    <div class="step-num">Step {seq}</div>
+                    <div class="step-num">Step {step_no}</div>
                     <div class="card-title">{_esc(etype)}</div>
                   </header>
                   <div class="card-body">{_pre(data)}</div>
@@ -449,24 +446,25 @@ def _nav_items(events: list[dict[str, Any]]) -> str:
     labels = {
         "llm_input": "LLM 输入",
         "llm_output": "LLM 输出",
-        "tool_call": "工具",
         "skill_loaded": "技能",
     }
-    for ev in events:
-        etype = ev.get("event")
-        if etype in ("run_start", "run_end"):
-            continue
-        data = ev.get("data") or {}
-        seq = ev.get("seq", 0)
-        label = labels.get(etype, etype or "?")
-        detail = ""
-        if etype == "tool_call":
-            detail = f" · {_esc(data.get('tool', ''))}"
-        elif etype in ("llm_input", "llm_output"):
-            detail = f" · iter {ev.get('iteration', data.get('iteration', ''))}"
+    for step_no, group in enumerate(_group_visible_events(events), start=1):
+        if len(group) == 2:
+            ev = group[0]
+            label = ""
+            nav_class = "nav-step-pair"
+            detail = f" · iter {_event_iteration(ev)}"
+        else:
+            ev = group[0]
+            etype = ev.get("event")
+            label = labels.get(etype, etype or "?")
+            nav_class = f"nav-{_esc(etype or '')}"
+            detail = ""
+            if etype in ("llm_input", "llm_output"):
+                detail = f" · iter {_event_iteration(ev)}"
         items.append(
-            f'<a class="nav-item nav-{_esc(etype or "")}" href="#step-{seq}">'
-            f'<span class="nav-seq">{seq}</span>{label}{detail}</a>'
+            f'<a class="nav-item {nav_class}" href="#step-{step_no}">'
+            f'<span class="nav-seq">{step_no}</span>{label}{detail}</a>'
         )
     return "\n".join(items)
 
@@ -482,9 +480,6 @@ _CSS = """
   --accent: #2563eb;
   --input: #2563eb;
   --output: #059669;
-  --tool: #d97706;
-  --tool-ok: #059669;
-  --tool-err: #dc2626;
   --skill: #7c3aed;
   --code-bg: #f8fafc;
   --mono: "JetBrains Mono", "SF Mono", "Fira Code", ui-monospace, monospace;
@@ -548,7 +543,7 @@ a:hover { text-decoration: underline; }
 }
 .nav-llm_input { border-left: 3px solid var(--input); }
 .nav-llm_output { border-left: 3px solid var(--output); }
-.nav-tool_call { border-left: 3px solid var(--tool); }
+.nav-step-pair { border-left: 3px solid var(--input); }
 .nav-skill_loaded { border-left: 3px solid var(--skill); }
 
 .main { padding: 1.5rem 2rem 3rem; max-width: 1100px; }
@@ -601,10 +596,7 @@ a:hover { text-decoration: underline; }
   overflow: hidden;
   box-shadow: 0 1px 2px rgba(26, 35, 50, 0.04);
 }
-.card-input { border-top: 3px solid var(--input); }
-.card-output { border-top: 3px solid var(--output); }
-.card-tool { border-top: 3px solid var(--tool); }
-.card-tool.error { border-top-color: var(--tool-err); }
+.card-step { border-top: 3px solid var(--input); }
 .card-skill { border-top: 3px solid var(--skill); }
 .card-header {
   display: flex;
@@ -634,8 +626,6 @@ a:hover { text-decoration: underline; }
   padding: 0.12rem 0.5rem;
   color: var(--muted);
 }
-.tag-ok { color: var(--tool-ok); border-color: #86efac; background: #f0fdf4; }
-.tag-error { color: var(--tool-err); border-color: #fecaca; background: #fef2f2; }
 .card-body { padding: 1rem; }
 .card-body h4 {
   margin: 1rem 0 0.5rem;
@@ -645,6 +635,23 @@ a:hover { text-decoration: underline; }
   letter-spacing: 0.05em;
 }
 .card-body h4:first-child { margin-top: 0; }
+.card-body.step-body { padding: 0; }
+.substep { padding: 1rem; }
+.substep + .substep { border-top: 1px solid var(--border); }
+.substep-header {
+  display: flex;
+  flex-wrap: wrap;
+  align-items: center;
+  gap: 0.5rem;
+  margin-bottom: 0.65rem;
+}
+.substep-title {
+  font-weight: 600;
+  font-size: 0.8rem;
+  color: var(--muted);
+  text-transform: uppercase;
+  letter-spacing: 0.05em;
+}
 .sublabel {
   font-size: 0.72rem;
   color: var(--muted);
@@ -675,20 +682,6 @@ a:hover { text-decoration: underline; }
   padding: 0.75rem;
   margin-bottom: 1rem;
 }
-.reasoning.empty { opacity: 0.75; }
-.reasoning-text { max-height: 360px; border-color: #fcd34d; background: #fffef5; }
-.tool-io {
-  display: grid;
-  grid-template-columns: 1fr auto 1fr;
-  gap: 0.75rem;
-  align-items: start;
-}
-.io-arrow {
-  color: var(--muted);
-  font-size: 1.4rem;
-  padding-top: 1.6rem;
-}
-.io-block { min-width: 0; }
 .msg-list { display: flex; flex-direction: column; gap: 0.4rem; }
 .msg {
   border: 1px solid var(--border);
@@ -705,7 +698,6 @@ a:hover { text-decoration: underline; }
 }
 .msg summary::-webkit-details-marker { display: none; }
 .msg-body { padding: 0 0.65rem 0.65rem; }
-.msg-idx { font-family: var(--mono); font-size: 0.7rem; color: var(--muted); }
 .msg-preview { color: var(--muted); font-size: 0.78rem; overflow: hidden; text-overflow: ellipsis; white-space: nowrap; flex: 1; }
 .badge {
   font-size: 0.65rem;
@@ -730,35 +722,6 @@ a:hover { text-decoration: underline; }
   padding: 0.15rem 0.5rem;
   border-radius: 999px;
 }
-.planned-tool-list {
-  display: flex;
-  flex-direction: column;
-  gap: 0.65rem;
-  margin-bottom: 0.5rem;
-}
-.planned-tool {
-  border: 1px solid #fed7aa;
-  background: #fffbeb;
-  border-radius: 8px;
-  padding: 0.65rem 0.75rem;
-}
-.planned-tool-head {
-  display: flex;
-  align-items: center;
-  gap: 0.5rem;
-  margin-bottom: 0.45rem;
-  flex-wrap: wrap;
-}
-.planned-tool-idx {
-  font-family: var(--mono);
-  font-size: 0.7rem;
-  color: var(--muted);
-}
-.planned-tool-id {
-  font-size: 0.68rem;
-  color: var(--muted);
-  margin-left: auto;
-}
 .usage {
   display: flex;
   flex-wrap: wrap;
@@ -768,39 +731,99 @@ a:hover { text-decoration: underline; }
   font-size: 0.72rem;
   color: var(--muted);
 }
-.raw-block { margin-top: 0.75rem; }
-.history-block {
-  margin: 0.85rem 0 0.5rem;
+.sysprompt-block { margin: 0.85rem 0; }
+.markdown-box {
+  background: var(--code-bg);
   border: 1px solid var(--border);
-  border-radius: 8px;
-  background: #f8fafc;
-  padding: 0.35rem 0.65rem 0.65rem;
+  border-radius: 6px;
+  padding: 0.75rem 1rem;
+  max-height: 420px;
+  overflow: auto;
 }
-.history-block > summary {
-  cursor: pointer;
-  font-size: 0.85rem;
+.markdown-body { font-size: 0.88rem; line-height: 1.65; color: var(--text); }
+.markdown-body > *:first-child { margin-top: 0; }
+.markdown-body > *:last-child { margin-bottom: 0; }
+.markdown-body p { margin: 0.5rem 0; }
+.markdown-body h1,
+.markdown-body h2,
+.markdown-body h3,
+.markdown-body h4,
+.markdown-body h5,
+.markdown-body h6 {
+  margin: 1rem 0 0.5rem;
+  line-height: 1.35;
+  color: var(--text);
+}
+.markdown-body h1 { font-size: 1.15rem; }
+.markdown-body h2 { font-size: 1.05rem; }
+.markdown-body h3 { font-size: 0.98rem; }
+.markdown-body h4 { font-size: 0.9rem; }
+.markdown-body ul,
+.markdown-body ol { margin: 0.4rem 0; padding-left: 1.4rem; }
+.markdown-body li { margin: 0.2rem 0; }
+.markdown-body li > p { margin: 0.2rem 0; }
+.markdown-body code {
+  background: #eef2f7;
+  border-radius: 4px;
+  padding: 0.1rem 0.35rem;
+}
+.markdown-body pre {
+  background: #0f172a;
+  color: #e2e8f0;
+  border-radius: 6px;
+  padding: 0.65rem 0.85rem;
+  overflow: auto;
+  margin: 0.5rem 0;
+}
+.markdown-body pre code { background: transparent; padding: 0; color: inherit; }
+.markdown-body blockquote {
+  border-left: 3px solid var(--border);
+  margin: 0.5rem 0;
+  padding: 0.1rem 0.75rem;
   color: var(--muted);
-  padding: 0.35rem 0;
-  font-weight: 500;
 }
-.history-body { margin-top: 0.35rem; }
-.meta-line { font-size: 0.8rem; color: var(--muted); margin-bottom: 0.35rem; }
+.markdown-body table { border-collapse: collapse; margin: 0.5rem 0; font-size: 0.82rem; }
+.markdown-body th,
+.markdown-body td { border: 1px solid var(--border); padding: 0.3rem 0.55rem; }
+.markdown-body a { text-decoration: underline; }
+.markdown-body strong { font-weight: 600; }
+.markdown-body hr { border: none; border-top: 1px solid var(--border); margin: 0.75rem 0; }
 code { font-family: var(--mono); font-size: 0.85em; }
 
 @media (max-width: 900px) {
   .layout { grid-template-columns: 1fr; }
   .sidebar { position: relative; height: auto; max-height: 40vh; }
-  .tool-io { grid-template-columns: 1fr; }
-  .io-arrow { display: none; }
 }
 """
 
 
+def _render_system_prompt_block(system_prompt: str | None) -> str:
+    if not system_prompt:
+        return ""
+    return f"""
+    <div class="sysprompt-block">
+      <div class="sublabel">System Prompt</div>
+      {_render_markdown(system_prompt)}
+    </div>
+    """
+
+
+def _render_available_tools_block(tool_names: list[str]) -> str:
+    if not tool_names:
+        return ""
+    chips = "".join(f'<span class="chip">{_esc(n)}</span>' for n in tool_names)
+    return f"""
+    <div class="sublabel">可用工具（{len(tool_names)}）</div>
+    <div class="chip-row">{chips}</div>
+    """
+
+
 def render_html(events: list[dict[str, Any]]) -> str:
     """Render a full standalone HTML page for the given events."""
     meta = summarize_run(events)
     skills = meta.get("skills_used") or []
     skills_str = ", ".join(skills) if skills else "—"
+    system_prompt, tool_names = _extract_system_prompt_and_tools(events)
 
     return f"""<!DOCTYPE html>
 <html lang="zh-CN">
@@ -831,19 +854,20 @@ def render_html(events: list[dict[str, Any]]) -> str:
           <div class="stat"><div class="label">Events</div><div class="value">{_esc(meta.get("event_count"))}</div></div>
           <div class="stat"><div class="label">Skills</div><div class="value">{_esc(skills_str)}</div></div>
         </div>
+        {_render_available_tools_block(tool_names)}
+        {_render_system_prompt_block(system_prompt)}
         <div class="sublabel">用户输入</div>
         <div class="user-prompt">{_esc(meta.get("user_input") or "—")}</div>
         {f'''
         <div class="final">
           <h3>最终回答</h3>
-          <div class="user-prompt" style="background:transparent;border:none;padding:0">{_esc(meta.get("final_content") or "")}</div>
+          {_render_markdown(meta.get("final_content"), soft_breaks=True, boxed=False)}
         </div>
         ''' if meta.get("final_content") else ""}
       </section>
 
       <section class="timeline">
         <h2 style="margin:0 0 0.5rem;font-size:1.1rem">执行时间线</h2>
-        <p class="muted" style="margin:0 0 1rem">按步骤展示：LLM 输入 → 思考/输出 → 工具调用（含完整入参与返回）</p>
         {_render_timeline(events)}
       </section>
     </main>

supply_infra/config.py

@@ -49,6 +49,10 @@ class InfraSettings(BaseSettings):
     aliyun_oss_region: str = Field(default="cn-hangzhou", alias="ALIYUN_OSS_REGION")
     aliyun_oss_bucket: str = Field(default="art-pubbucket", alias="ALIYUN_OSS_BUCKET")
     aliyun_oss_root_prefix: str = Field(default="supply_agent", alias="ALIYUN_OSS_ROOT_PREFIX")
+    aliyun_oss_manual_log_prefix: str = Field(
+        default="supply_agent/manual_logs",
+        alias="ALIYUN_OSS_MANUAL_LOG_PREFIX",
+    )
     aliyun_oss_public_base_url: str = Field(
         default="http://rescdn.yishihui.com",
         alias="ALIYUN_OSS_PUBLIC_BASE_URL",

supply_infra/db/models/__init__.py

@@ -2,18 +2,24 @@
 
 from supply_infra.db.models.category_tree_weight import CategoryTreeWeight
 from supply_infra.db.models.demand_belong_category import DemandBelongCategory
+from supply_infra.db.models.demand_belong_pool_rel import DemandBelongPoolRel
 from supply_infra.db.models.demand_popularity_stats import DemandPopularityStats
+from supply_infra.db.models.generated_demand import GeneratedDemand
 from supply_infra.db.models.global_tree_category import GlobalTreeCategory
 from supply_infra.db.models.global_tree_element import GlobalTreeElement
 from supply_infra.db.models.multi_demand_pool_di import MultiDemandPoolDi
+from supply_infra.db.models.multi_demand_video_detail import MultiDemandVideoDetail
 from supply_infra.db.models.oss_log import OssLog
 
 __all__ = [
     "CategoryTreeWeight",
     "DemandBelongCategory",
+    "DemandBelongPoolRel",
     "DemandPopularityStats",
+    "GeneratedDemand",
     "GlobalTreeCategory",
     "GlobalTreeElement",
     "MultiDemandPoolDi",
+    "MultiDemandVideoDetail",
     "OssLog",
 ]

supply_infra/db/models/demand_belong_category.py

@@ -18,6 +18,9 @@ class DemandBelongCategory(Base):
     name: Mapped[str | None] = mapped_column(String(256), nullable=True, comment="需求名称")
     category_id: Mapped[int] = mapped_column(BigInteger, nullable=False, comment="分类id")
     reason: Mapped[str | None] = mapped_column(Text, nullable=True, comment="原因")
+    video_list: Mapped[str | None] = mapped_column(
+        Text, nullable=True, comment="关联视频列表JSON（最多10个）"
+    )
     is_delete: Mapped[int] = mapped_column(
         Integer, default=0, nullable=False, comment="是否删除 0-正常 1-删除"
     )

supply_infra/db/models/generated_demand.py

@@ -0,0 +1,64 @@
+from __future__ import annotations
+
+from datetime import datetime
+from decimal import Decimal
+
+from sqlalchemy import BigInteger, Index, Integer, Numeric, String, Text, func
+from sqlalchemy.orm import Mapped, mapped_column
+
+from supply_infra.db.base import Base
+
+
+class GeneratedDemand(Base):
+    """单维度需求产生结果。"""
+
+    __tablename__ = "generated_demand"
+    __table_args__ = (
+        Index("idx_generated_demand_dim_biz", "source_dim", "biz_dt"),
+        Index("idx_generated_demand_name", "demand_name"),
+        Index("idx_generated_demand_run", "run_id"),
+    )
+
+    id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
+    source_dim: Mapped[str] = mapped_column(String(32), nullable=False, comment="来源热度维度")
+    overall_direction: Mapped[str] = mapped_column(
+        String(256), nullable=False, comment="整体方向"
+    )
+    summary_event: Mapped[str] = mapped_column(String(512), nullable=False, comment="汇总事件")
+    demand_name: Mapped[str] = mapped_column(
+        String(256), nullable=False, comment="需求名称（来自 demand_belong_category.name）"
+    )
+    demand_belong_id: Mapped[int | None] = mapped_column(
+        BigInteger, nullable=True, comment="demand_belong_category.id"
+    )
+    category_id: Mapped[int | None] = mapped_column(
+        BigInteger, nullable=True, comment="选用时的类目 id"
+    )
+    category_path: Mapped[str | None] = mapped_column(
+        String(1024), nullable=True, comment="类目路径文本"
+    )
+    dim_avg: Mapped[Decimal | None] = mapped_column(
+        Numeric(16, 8), nullable=True, comment="落库时该维 avg 快照"
+    )
+    dim_count: Mapped[int | None] = mapped_column(
+        Integer, nullable=True, comment="落库时该维 count 快照"
+    )
+    reason: Mapped[str | None] = mapped_column(Text, nullable=True, comment="选用理由")
+    biz_dt: Mapped[str | None] = mapped_column(
+        String(32), nullable=True, comment="对齐热度统计日 YYYYMMDD"
+    )
+    run_id: Mapped[str] = mapped_column(String(64), nullable=False, comment="同一次生成批次")
+    is_delete: Mapped[int] = mapped_column(
+        Integer, default=0, nullable=False, comment="是否删除 0-正常 1-删除"
+    )
+    create_time: Mapped[datetime] = mapped_column(
+        nullable=False,
+        server_default=func.now(),
+        comment="创建时间",
+    )
+    update_time: Mapped[datetime] = mapped_column(
+        nullable=False,
+        server_default=func.now(),
+        onupdate=func.now(),
+        comment="更新时间",
+    )

supply_infra/db/repositories/__init__.py

@@ -7,21 +7,31 @@ from supply_infra.db.repositories.category_tree_weight_repo import (
 from supply_infra.db.repositories.demand_belong_category_repo import (
     DemandBelongCategoryRepository,
 )
+from supply_infra.db.repositories.demand_belong_pool_rel_repo import (
+    DemandBelongPoolRelRepository,
+)
 from supply_infra.db.repositories.demand_popularity_stats_repo import (
     DemandPopularityStatsRepository,
 )
+from supply_infra.db.repositories.generated_demand_repo import GeneratedDemandRepository
 from supply_infra.db.repositories.global_tree_category_repo import GlobalTreeCategoryRepository
 from supply_infra.db.repositories.global_tree_element_repo import GlobalTreeElementRepository
 from supply_infra.db.repositories.multi_demand_pool_di_repo import MultiDemandPoolDiRepository
+from supply_infra.db.repositories.multi_demand_video_detail_repo import (
+    MultiDemandVideoDetailRepository,
+)
 from supply_infra.db.repositories.oss_log_repo import OssLogRepository
 
 __all__ = [
     "BaseRepository",
     "CategoryTreeWeightRepository",
     "DemandBelongCategoryRepository",
+    "DemandBelongPoolRelRepository",
     "DemandPopularityStatsRepository",
+    "GeneratedDemandRepository",
     "GlobalTreeCategoryRepository",
     "GlobalTreeElementRepository",
     "MultiDemandPoolDiRepository",
+    "MultiDemandVideoDetailRepository",
     "OssLogRepository",
 ]

supply_infra/db/repositories/category_tree_weight_repo.py

@@ -32,6 +32,13 @@ class CategoryTreeWeightRepository(BaseRepository[CategoryTreeWeight]):
         stmt = select(func.max(CategoryTreeWeight.biz_dt))
         return self.session.scalar(stmt)
 
+    def has_biz_dt(self, biz_dt: str) -> bool:
+        """指定业务日是否存在权重数据。"""
+        stmt = select(CategoryTreeWeight.id).where(
+            CategoryTreeWeight.biz_dt == biz_dt
+        ).limit(1)
+        return self.session.scalars(stmt).first() is not None
+
     def list_by_biz_dt(self, biz_dt: str) -> list[CategoryTreeWeight]:
         """返回指定业务日的全部节点权重行。"""
         stmt = select(CategoryTreeWeight).where(CategoryTreeWeight.biz_dt == biz_dt)

supply_infra/db/repositories/demand_belong_category_repo.py

@@ -2,7 +2,7 @@ from __future__ import annotations
 
 from collections.abc import Iterable
 
-from sqlalchemy import select
+from sqlalchemy import select, update
 from sqlalchemy.dialects.mysql import insert
 
 from supply_infra.db.models.demand_belong_category import DemandBelongCategory
@@ -40,6 +40,22 @@ class DemandBelongCategoryRepository(BaseRepository[DemandBelongCategory]):
         )
         return list(self.session.scalars(stmt).all())
 
+    def list_by_category_ids(
+        self, category_ids: list[int]
+    ) -> list[DemandBelongCategory]:
+        """返回挂载在指定类目上的未删除需求词。"""
+        if not category_ids:
+            return []
+        stmt = (
+            select(DemandBelongCategory)
+            .where(
+                DemandBelongCategory.is_delete == 0,
+                DemandBelongCategory.category_id.in_(category_ids),
+            )
+            .order_by(DemandBelongCategory.category_id, DemandBelongCategory.id)
+        )
+        return list(self.session.scalars(stmt).all())
+
     def list_active_id_name(self) -> list[tuple[int, str]]:
         """返回所有未删除记录的 (id, name)，跳过空名称。"""
         stmt = (
@@ -53,6 +69,24 @@ class DemandBelongCategoryRepository(BaseRepository[DemandBelongCategory]):
         )
         return [(int(row_id), name) for row_id, name in self.session.execute(stmt).all() if name]
 
+    def get_by_names(self, names: Iterable[str]) -> dict[str, DemandBelongCategory]:
+        """按 name 批量查询未删除记录，返回 name → row。"""
+        name_list = [n for n in names if n]
+        if not name_list:
+            return {}
+
+        result: dict[str, DemandBelongCategory] = {}
+        for i in range(0, len(name_list), _BATCH_SIZE):
+            batch = name_list[i : i + _BATCH_SIZE]
+            stmt = select(DemandBelongCategory).where(
+                DemandBelongCategory.is_delete == 0,
+                DemandBelongCategory.name.in_(batch),
+            )
+            for row in self.session.scalars(stmt).all():
+                if row.name:
+                    result[row.name] = row
+        return result
+
     def bulk_insert_ignore(self, rows: list[dict]) -> int:
         """批量插入，MySQL 按 name 唯一索引忽略已存在行。"""
         if not rows:
@@ -65,3 +99,19 @@ class DemandBelongCategoryRepository(BaseRepository[DemandBelongCategory]):
             result = self.session.execute(stmt)
             inserted += result.rowcount
         return inserted
+
+    def update_video_lists(self, updates: dict[int, str | None]) -> int:
+        """按 id 批量更新 video_list。"""
+        if not updates:
+            return 0
+
+        updated = 0
+        for belong_id, video_list in updates.items():
+            stmt = (
+                update(DemandBelongCategory)
+                .where(DemandBelongCategory.id == belong_id)
+                .values(video_list=video_list)
+            )
+            result = self.session.execute(stmt)
+            updated += result.rowcount or 0
+        return updated

supply_infra/db/repositories/demand_popularity_stats_repo.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from sqlalchemy import select
+from sqlalchemy import func, select
 from sqlalchemy.dialects.mysql import insert
 
 from supply_infra.db.models.demand_popularity_stats import DemandPopularityStats
@@ -30,11 +30,39 @@ class DemandPopularityStatsRepository(BaseRepository[DemandPopularityStats]):
 
     model = DemandPopularityStats
 
+    def get_latest_biz_dt(self) -> str | None:
+        """返回热度统计表中最新业务日；无数据时返回 None。"""
+        stmt = select(func.max(DemandPopularityStats.biz_dt))
+        return self.session.scalar(stmt)
+
+    def has_biz_dt(self, biz_dt: str) -> bool:
+        """指定业务日是否存在热度统计数据。"""
+        stmt = select(DemandPopularityStats.id).where(
+            DemandPopularityStats.biz_dt == biz_dt
+        ).limit(1)
+        return self.session.scalars(stmt).first() is not None
+
     def list_by_biz_dt(self, biz_dt: str) -> list[DemandPopularityStats]:
         """返回指定业务日的全部热度统计行。"""
         stmt = select(DemandPopularityStats).where(DemandPopularityStats.biz_dt == biz_dt)
         return list(self.session.scalars(stmt).all())
 
+    def list_by_biz_dt_and_belong_ids(
+        self, biz_dt: str, belong_ids: list[int]
+    ) -> list[DemandPopularityStats]:
+        """按业务日 + demand_belong_category.id 列表查询热度行。"""
+        if not belong_ids:
+            return []
+        rows: list[DemandPopularityStats] = []
+        for i in range(0, len(belong_ids), _BATCH_SIZE):
+            batch = belong_ids[i : i + _BATCH_SIZE]
+            stmt = select(DemandPopularityStats).where(
+                DemandPopularityStats.biz_dt == biz_dt,
+                DemandPopularityStats.demand_category_id.in_(batch),
+            )
+            rows.extend(self.session.scalars(stmt).all())
+        return rows
+
     def upsert_rows(self, rows: list[dict]) -> int:
         """按 (demand_category_id, biz_dt) 批量 upsert。"""
         if not rows:

supply_infra/db/repositories/generated_demand_repo.py

@@ -0,0 +1,27 @@
+from __future__ import annotations
+
+from sqlalchemy.dialects.mysql import insert
+
+from supply_infra.db.models.generated_demand import GeneratedDemand
+from supply_infra.db.repositories.base import BaseRepository
+
+_BATCH_SIZE = 500
+
+
+class GeneratedDemandRepository(BaseRepository[GeneratedDemand]):
+    """单维度需求产生结果表 repository。"""
+
+    model = GeneratedDemand
+
+    def bulk_insert(self, rows: list[dict]) -> int:
+        """批量插入生成结果行。"""
+        if not rows:
+            return 0
+
+        inserted = 0
+        for i in range(0, len(rows), _BATCH_SIZE):
+            batch = rows[i : i + _BATCH_SIZE]
+            stmt = insert(GeneratedDemand).values(batch)
+            result = self.session.execute(stmt)
+            inserted += result.rowcount or 0
+        return inserted

supply_infra/db/repositories/multi_demand_pool_di_repo.py

@@ -66,6 +66,27 @@ class MultiDemandPoolDiRepository(BaseRepository[MultiDemandPoolDi]):
         )
         return [name for name in self.session.scalars(stmt).all() if name]
 
+    def list_all_video_lists(self) -> list[str]:
+        """查询全表非空 video_list（JSON 文本）。"""
+        stmt = select(MultiDemandPoolDi.video_list).where(
+            MultiDemandPoolDi.video_list.isnot(None),
+            MultiDemandPoolDi.video_list != "",
+        )
+        return [text for text in self.session.scalars(stmt).all() if text]
+
+    def list_id_name_video_lists(self) -> list[tuple[int, str, str | None]]:
+        """查询全表 (id, demand_name, video_list)。"""
+        stmt = select(
+            MultiDemandPoolDi.id,
+            MultiDemandPoolDi.demand_name,
+            MultiDemandPoolDi.video_list,
+        )
+        return [
+            (int(row_id), str(name), video_list)
+            for row_id, name, video_list in self.session.execute(stmt).all()
+            if name is not None
+        ]
+
     def list_weights_by_name_like(
         self,
         biz_dt: str,
@@ -153,3 +174,26 @@ class MultiDemandPoolDiRepository(BaseRepository[MultiDemandPoolDi]):
             result = self.session.execute(stmt)
             updated += result.rowcount or 0
         return updated
+
+    def update_video_fields(self, biz_dt: str, rows: list[dict]) -> int:
+        """按 (strategy, demand_id) 批量更新 video_list / video_count。"""
+        if not rows:
+            return 0
+
+        updated = 0
+        for row in rows:
+            stmt = (
+                update(MultiDemandPoolDi)
+                .where(
+                    MultiDemandPoolDi.biz_dt == biz_dt,
+                    MultiDemandPoolDi.strategy == row["strategy"],
+                    MultiDemandPoolDi.demand_id == row["demand_id"],
+                )
+                .values(
+                    video_list=row.get("video_list"),
+                    video_count=row.get("video_count"),
+                )
+            )
+            result = self.session.execute(stmt)
+            updated += result.rowcount or 0
+        return updated

supply_infra/odps/client.py

@@ -85,7 +85,7 @@ class ODPSClient:
         return self.execute_sql(sql)
 
     def fetch_multi_demand_pool(self, bizdate: str) -> list[dict[str, Any]]:
-        """拉取 dwd_multi_demand_pool_di 策略需求天级数据（不同步 video_list）。"""
+        """拉取 dwd_multi_demand_pool_di 策略需求天级数据（video_list 仅取前 10 个）。"""
         sql = f"""
         SELECT  strategy
                 ,demand_id
@@ -93,6 +93,7 @@ class ODPSClient:
                 ,weight
                 ,`type`
                 ,video_count
+                ,SLICE(video_list, 1, 10) AS video_list
                 ,extend
         FROM    loghubods.dwd_multi_demand_pool_di
         WHERE   dt = '{bizdate}'
@@ -117,6 +118,36 @@ class ODPSClient:
             return 0
         return int(rows[0].get("cnt") or 0)
 
+    def fetch_topic_decode_results(
+        self,
+        dt: str,
+        vids: list[str],
+        batch_size: int = 100,
+    ) -> list[dict[str, Any]]:
+        """按 vid 批量拉取 dwd_topic_decode_result_di 的 decode_result。"""
+        # 保序去重
+        unique_vids = list(
+            dict.fromkeys(
+                str(v).strip() for v in vids if v is not None and str(v).strip()
+            )
+        )
+        if not unique_vids:
+            return []
+
+        results: list[dict[str, Any]] = []
+        for i in range(0, len(unique_vids), batch_size):
+            batch = unique_vids[i : i + batch_size]
+            in_list = ",".join("'" + v.replace("'", "''") + "'" for v in batch)
+            sql = f"""
+            SELECT  vid
+                    ,decode_result
+            FROM    loghubods.dwd_topic_decode_result_di
+            WHERE   dt = '{dt}'
+            AND     vid IN ({in_list})
+            """
+            results.extend(self.execute_sql(sql))
+        return results
+
     def fetch_real_rov_vov_7d(
         self,
         dt_left: str,

supply_infra/oss/client.py

@@ -12,7 +12,12 @@ from supply_infra.config import InfraSettings, get_infra_settings
 class OssClient:
     """Thin wrapper around oss2 Bucket."""
 
-    def __init__(self, settings: InfraSettings | None = None) -> None:
+    def __init__(
+        self,
+        settings: InfraSettings | None = None,
+        *,
+        root_prefix: str | None = None,
+    ) -> None:
         self.settings = settings or get_infra_settings()
         if not self.settings.aliyun_oss_access_key_id:
             raise ValueError("ALIYUN_OSS_ACCESS_KEY_ID is not configured")
@@ -25,7 +30,8 @@ class OssClient:
             self.settings.aliyun_oss_access_key_secret,
         )
         self._bucket = oss2.Bucket(auth, endpoint, self.settings.aliyun_oss_bucket)
-        self._root = self.settings.aliyun_oss_root_prefix.strip("/")
+        prefix = root_prefix if root_prefix is not None else self.settings.aliyun_oss_root_prefix
+        self._root = prefix.strip("/")
         self._public_base = self.settings.aliyun_oss_public_base_url.rstrip("/")
 
     def object_key(self, *parts: str) -> str:

supply_infra/scheduler/jobs/sync_multi_demand_pool_odps_to_mysql.py

@@ -3,16 +3,20 @@
 
 流程：
 1. 比对当天 ODPS / MySQL 行数，相同则跳过写入
-2. 有差异时拉取 ODPS，按 (strategy, demand_id) 只插入缺失、删除多余
-3. 拉取近 7 日真实 ROV/VOV，按特征值匹配回填 real_rov_7d / real_vov_7d
-4. 查询当天全部 demand_name，按空格分词写入 set
-5. 过滤 demand_belong_category 中已存在的词
-6. 剩余词按 100 词一批调用 demand_belong_category_agent
-7. 遍历 demand_belong_category 全部词，按策略与真实 ROV/VOV 写入 demand_popularity_stats（avg/count）
-8. 基于三表计算整棵类目树节点加权平均分，写入 category_tree_weight
+2. 有差异时拉取 ODPS，按 (strategy, demand_id) 只插入缺失、删除多余，并回填已有行 video_list
+3. video_list 每条最多保留前 10 个 video_id，video_count 与之保持一致
+4. 拉取近 7 日真实 ROV/VOV，按特征值匹配回填 real_rov_7d / real_vov_7d
+5. 查询当天全部 demand_name，按空格分词写入 set
+6. 过滤 demand_belong_category 中已存在的词
+7. 剩余词按 100 词一批调用 demand_belong_category_agent
+8. 遍历 demand_belong_category 全部词，按策略与真实 ROV/VOV 写入 demand_popularity_stats（avg/count）
+9. 基于三表计算整棵类目树节点加权平均分，写入 category_tree_weight
+10. 增量同步 multi_demand_video_detail（全表 video_list → ODPS 昨天分区最终选题）
+11. 补充 demand_belong_pool_rel 匹配边，并回填词级 video_list（最多 10 个）
 """
 from __future__ import annotations
 
+import json
 import logging
 from datetime import datetime, timedelta
 from decimal import Decimal
@@ -31,6 +35,10 @@ from supply_infra.odps.client import get_odps_client
 from supply_infra.scheduler.jobs.compute_category_tree_weight import (
     compute_category_tree_weight,
 )
+from supply_infra.scheduler.jobs.sync_demand_belong_pool_rel import (
+    sync_demand_belong_pool_rel,
+)
+from supply_infra.scheduler.jobs.sync_multi_demand_videos import sync_multi_demand_videos
 
 logger = logging.getLogger(__name__)
 
@@ -39,6 +47,7 @@ _STATS_UPSERT_BATCH = 200
 _REAL_METRIC_LIMIT = 1000
 _REAL_METRIC_LOOKBACK_DAYS = 7
 _GLOBAL_FEATURE_VALUE = "全局SUM"
+_VIDEO_LIST_LIMIT = 10
 
 # 策略名 → 统计字段前缀；去年同期阳历/阴历合并为 plat_ly_pop
 _STRATEGY_METRIC: dict[str, str] = {
@@ -68,6 +77,39 @@ def _row_key(row: dict[str, Any]) -> RowKey:
     return (row["strategy"], row["demand_id"])
 
 
+def _normalize_video_list(raw: Any) -> tuple[str | None, int]:
+    """取前 N 个 video_id，返回 (JSON 文本, count)，二者保持一致。"""
+    if raw is None:
+        return None, 0
+
+    items: list[Any]
+    if isinstance(raw, str):
+        text = raw.strip()
+        if not text:
+            return None, 0
+        try:
+            parsed = json.loads(text)
+            items = list(parsed) if isinstance(parsed, list) else [text]
+        except json.JSONDecodeError:
+            items = [part.strip() for part in text.split(",") if part.strip()]
+    elif isinstance(raw, (list, tuple)):
+        items = list(raw)
+    else:
+        try:
+            items = list(raw)
+        except TypeError:
+            return None, 0
+
+    truncated = [
+        str(v).strip()
+        for v in items[:_VIDEO_LIST_LIMIT]
+        if v is not None and str(v).strip()
+    ]
+    if not truncated:
+        return None, 0
+    return json.dumps(truncated, ensure_ascii=False), len(truncated)
+
+
 def _to_mysql_rows(raw_rows: list[dict[str, Any]], biz_dt: str) -> list[dict[str, Any]]:
     """转换为 MySQL 行，按 (strategy, demand_id) 去重（保留最后一条）。"""
     by_key: dict[RowKey, dict[str, Any]] = {}
@@ -79,14 +121,15 @@ def _to_mysql_rows(raw_rows: list[dict[str, Any]], biz_dt: str) -> list[dict[str
             logger.warning("Skip row with missing required fields: %s", row)
             continue
 
+        video_list, video_count = _normalize_video_list(row.get("video_list"))
         mapped = {
             "strategy": str(strategy),
             "demand_id": str(demand_id),
             "demand_name": str(demand_name),
             "weight": row.get("weight"),
             "type": str(row["type"]) if row.get("type") is not None else None,
-            "video_count": row.get("video_count"),
-            "video_list": None,
+            "video_count": video_count,
+            "video_list": video_list,
             "extend": str(row["extend"]) if row.get("extend") is not None else None,
             "biz_dt": biz_dt,
         }
@@ -143,7 +186,7 @@ def _classify_words(biz_dt: str) -> dict:
 
 
 def _sync_diff(partition_date: str) -> dict[str, Any]:
-    """行数不同时拉取 ODPS，只同步 (strategy, demand_id) 差异。"""
+    """行数不同时拉取 ODPS，只同步 (strategy, demand_id) 差异，并回填已有行的 video 字段。"""
     odps = get_odps_client()
     raw_rows = odps.fetch_multi_demand_pool(partition_date)
     mysql_rows = _to_mysql_rows(raw_rows, partition_date)
@@ -156,17 +199,21 @@ def _sync_diff(partition_date: str) -> dict[str, Any]:
 
         to_insert_keys = odps_keys - mysql_keys
         to_delete_keys = mysql_keys - odps_keys
+        to_update_keys = odps_keys & mysql_keys
 
         insert_rows = [odps_by_key[k] for k in to_insert_keys]
+        update_rows = [odps_by_key[k] for k in to_update_keys]
         deleted = repo.delete_by_keys(partition_date, list(to_delete_keys))
         inserted = repo.bulk_insert(insert_rows)
+        updated = repo.update_video_fields(partition_date, update_rows)
 
     logger.info(
-        "Diff sync: odps=%d mysql_before=%d insert=%d delete=%d",
+        "Diff sync: odps=%d mysql_before=%d insert=%d delete=%d video_update=%d",
         len(odps_keys),
         len(mysql_keys),
         inserted,
         deleted,
+        updated,
     )
     return {
         "fetched": len(raw_rows),
@@ -174,10 +221,35 @@ def _sync_diff(partition_date: str) -> dict[str, Any]:
         "mysql_before": len(mysql_keys),
         "inserted": inserted,
         "deleted": deleted,
+        "video_updated": updated,
         "skipped_invalid": len(raw_rows) - len(mysql_rows),
     }
 
 
+def backfill_video_list(partition_date: str) -> dict[str, Any]:
+    """从 ODPS 回填指定分区的 video_list / video_count（每条最多前 10 个 video_id）。"""
+    logger.info("Backfill video_list for partition: %s", partition_date)
+    odps = get_odps_client()
+    raw_rows = odps.fetch_multi_demand_pool(partition_date)
+    mysql_rows = _to_mysql_rows(raw_rows, partition_date)
+
+    with get_session() as session:
+        updated = MultiDemandPoolDiRepository(session).update_video_fields(
+            partition_date,
+            mysql_rows,
+        )
+
+    result = {
+        "partition_date": partition_date,
+        "fetched": len(raw_rows),
+        "unique_rows": len(mysql_rows),
+        "updated": updated,
+        "with_video": sum(1 for r in mysql_rows if r.get("video_list")),
+    }
+    logger.info("Backfill video_list completed: %s", result)
+    return result
+
+
 def _to_float(value: Any) -> float | None:
     if value is None:
         return None
@@ -444,17 +516,21 @@ def sync_multi_demand_pool_odps_to_mysql(partition_date: str | None = None) -> d
         }
 
     classify_stats = _classify_words(partition_date)
+    belong_pool_rel_stats = sync_demand_belong_pool_rel()
     real_metric_stats = enrich_real_rov_vov_7d(partition_date)
     popularity_stats = compute_popularity_stats(partition_date)
     tree_weight_stats = compute_category_tree_weight(partition_date)
+    video_stats = sync_multi_demand_videos(limit=None)
 
     result = {
         "partition_date": partition_date,
         **sync_stats,
         "real_metrics": real_metric_stats,
         "classify": classify_stats,
+        "belong_pool_rel": belong_pool_rel_stats,
         "popularity": popularity_stats,
         "tree_weight": tree_weight_stats,
+        "videos": video_stats,
         "synced_at": datetime.now().isoformat(),
     }
     logger.info("Multi demand pool sync completed: %s", result)

web/src/api/demand.ts

@@ -1,4 +1,4 @@
-import type { DemandBelongResponse } from '../types/demand'
+import type { DemandBelongResponse, DemandVideosResponse } from '../types/demand'
 
 export async function fetchDemandBelongCategory(): Promise<DemandBelongResponse> {
   const res = await fetch('/api/demand-belong-category')
@@ -7,3 +7,13 @@ export async function fetchDemandBelongCategory(): Promise<DemandBelongResponse>
   }
   return res.json()
 }
+
+export async function fetchDemandBelongVideos(
+  belongId: number,
+): Promise<DemandVideosResponse> {
+  const res = await fetch(`/api/demand-belong-category/${belongId}/videos`)
+  if (!res.ok) {
+    throw new Error(`加载关联视频失败: ${res.status} ${res.statusText}`)
+  }
+  return res.json()
+}

web/src/components/CategoryTree.vue

@@ -1,7 +1,6 @@
 <script setup lang="ts">
 import { computed, ref, watch } from 'vue'
 import TreeNode from './TreeNode.vue'
-import DemandDrawer from './DemandDrawer.vue'
 import type { CategoryNode, WeightDimKey, WeightDimMeta } from '../types/category'
 import {
   DEFAULT_DIMS,
@@ -11,6 +10,7 @@ import {
   maxTreeDepth,
 } from '../types/category'
 import type { DemandBelongItem, DemandsByCategory } from '../types/demand'
+import { downloadCategoryTreeHtml } from '../utils/exportCategoryTreeHtml'
 
 const FULL_TREE_KEY = 'full' as const
 type TreeTabKey = typeof FULL_TREE_KEY | WeightDimKey
@@ -63,6 +63,7 @@ const expandedMap = ref<Record<number, true>>({})
 const collapsedMap = ref<Record<number, true>>({})
 
 const drawerOpen = ref(false)
+const drawerCategoryId = ref<number | null>(null)
 const drawerCategoryName = ref('')
 const drawerItems = ref<DemandBelongItem[]>([])
 
@@ -149,6 +150,11 @@ function onInspect(payload: {
   categoryName: string
   items: DemandBelongItem[]
 }) {
+  if (drawerOpen.value && drawerCategoryId.value === payload.categoryId) {
+    closeDrawer()
+    return
+  }
+  drawerCategoryId.value = payload.categoryId
   drawerCategoryName.value = payload.categoryName
   drawerItems.value = payload.items
   drawerOpen.value = true
@@ -156,6 +162,16 @@ function onInspect(payload: {
 
 function closeDrawer() {
   drawerOpen.value = false
+  drawerCategoryId.value = null
+}
+
+function exportHtml() {
+  downloadCategoryTreeHtml({
+    nodes: props.nodes,
+    dims: dimTabs.value,
+    bizDt: props.bizDt ?? null,
+    demandsByCategory: props.demandsByCategory,
+  })
 }
 
 function onPanStart(e: MouseEvent) {
@@ -209,6 +225,7 @@ function onPanEnd() {
           <span>层</span>
         </label>
         <button type="button" class="btn" @click="expandAll">全部展开</button>
+        <button type="button" class="btn btn-secondary" @click="exportHtml">导出 HTML</button>
         <span class="hint">
           共 {{ treeMaxDepth }} 层 · 空白处按住拖动可平移 · Shift+滚轮左右滚
         </span>
@@ -268,19 +285,16 @@ function onPanEnd() {
             :demands-by-category="demandsByCategory"
             :active-dim="activeDim"
             :weight-scale="weightScale"
+            :inspect-category-id="drawerCategoryId"
+            :inspect-category-name="drawerCategoryName"
+            :inspect-items="drawerItems"
             @toggle="toggleNode"
             @inspect="onInspect"
+            @close-inspect="closeDrawer"
           />
         </div>
       </div>
     </div>
-
-    <DemandDrawer
-      :open="drawerOpen"
-      :category-name="drawerCategoryName"
-      :items="drawerItems"
-      @close="closeDrawer"
-    />
   </div>
 </template>
 
@@ -370,6 +384,17 @@ function onPanEnd() {
   background: #1e293b;
 }
 
+.btn-secondary {
+  background: #fff;
+  color: #0f172a;
+  border-color: #cbd5e1;
+}
+
+.btn-secondary:hover {
+  background: #f8fafc;
+  border-color: #94a3b8;
+}
+
 .hint {
   font-size: 12px;
   color: #94a3b8;

web/src/components/DemandDrawer.vue

@@ -1,158 +0,0 @@
-<script setup lang="ts">
-import type { DemandBelongItem } from '../types/demand'
-
-defineProps<{
-  open: boolean
-  categoryName: string
-  items: DemandBelongItem[]
-}>()
-
-const emit = defineEmits<{
-  close: []
-}>()
-</script>
-
-<template>
-  <Teleport to="body">
-    <div v-if="open" class="overlay" @click.self="emit('close')">
-      <aside class="drawer" role="dialog" aria-label="需求词列表">
-        <header class="drawer-header">
-          <div>
-            <h2>{{ categoryName || '分类节点' }}</h2>
-            <p class="sub">共 {{ items.length }} 条需求词</p>
-          </div>
-          <button type="button" class="close" title="关闭" @click="emit('close')">×</button>
-        </header>
-
-        <div class="drawer-body">
-          <table v-if="items.length">
-            <thead>
-              <tr>
-                <th class="col-name">需求词</th>
-                <th class="col-reason">原因</th>
-              </tr>
-            </thead>
-            <tbody>
-              <tr v-for="item in items" :key="item.id">
-                <td>{{ item.name || '—' }}</td>
-                <td>{{ item.reason || '—' }}</td>
-              </tr>
-            </tbody>
-          </table>
-          <div v-else class="empty">该节点暂无需求词</div>
-        </div>
-      </aside>
-    </div>
-  </Teleport>
-</template>
-
-<style scoped>
-.overlay {
-  position: fixed;
-  inset: 0;
-  z-index: 1000;
-  background: rgba(15, 23, 42, 0.28);
-  display: flex;
-  justify-content: flex-end;
-}
-
-.drawer {
-  width: min(480px, 100vw);
-  height: 100%;
-  background: #fff;
-  box-shadow: -8px 0 24px rgba(15, 23, 42, 0.12);
-  display: flex;
-  flex-direction: column;
-}
-
-.drawer-header {
-  display: flex;
-  align-items: flex-start;
-  justify-content: space-between;
-  gap: 12px;
-  padding: 20px 20px 16px;
-  border-bottom: 1px solid #e2e8f0;
-}
-
-.drawer-header h2 {
-  margin: 0;
-  font-size: 18px;
-  color: #0f172a;
-  word-break: break-word;
-}
-
-.sub {
-  margin: 4px 0 0;
-  font-size: 12px;
-  color: #94a3b8;
-}
-
-.close {
-  width: 32px;
-  height: 32px;
-  border: none;
-  border-radius: 6px;
-  background: transparent;
-  color: #64748b;
-  font-size: 22px;
-  line-height: 1;
-  cursor: pointer;
-  flex-shrink: 0;
-}
-
-.close:hover {
-  background: #f1f5f9;
-  color: #0f172a;
-}
-
-.drawer-body {
-  flex: 1;
-  overflow: auto;
-  padding: 0 0 24px;
-}
-
-table {
-  width: 100%;
-  border-collapse: collapse;
-  font-size: 14px;
-}
-
-th,
-td {
-  padding: 10px 16px;
-  text-align: left;
-  vertical-align: top;
-  border-bottom: 1px solid #eef2f7;
-  color: #0f172a;
-  word-break: break-word;
-}
-
-th {
-  position: sticky;
-  top: 0;
-  background: #f8fafc;
-  font-size: 12px;
-  font-weight: 600;
-  color: #64748b;
-  letter-spacing: 0.02em;
-}
-
-.col-name {
-  width: 36%;
-}
-
-.col-reason {
-  width: 64%;
-}
-
-tbody tr:hover td {
-  background: #f8fafc;
-}
-
-.empty {
-  padding: 48px 20px;
-  text-align: center;
-  color: #94a3b8;
-  font-size: 14px;
-}
-</style>

web/src/components/TreeNode.vue

@@ -1,5 +1,6 @@
 <script setup lang="ts">
 import { computed } from 'vue'
+import DemandPathPanel from './DemandPathPanel.vue'
 import type { CategoryNode, WeightDimKey } from '../types/category'
 import {
   formatAvgScore,
@@ -18,11 +19,15 @@ const props = defineProps<{
   demandsByCategory: DemandsByCategory
   activeDim: WeightDimKey | null
   weightScale: number[]
+  inspectCategoryId: number | null
+  inspectCategoryName: string
+  inspectItems: DemandBelongItem[]
 }>()
 
 const emit = defineEmits<{
   toggle: [id: number]
   inspect: [payload: { categoryId: number; categoryName: string; items: DemandBelongItem[] }]
+  closeInspect: []
 }>()
 
 const hasChildren = () => props.node.children.length > 0
@@ -38,6 +43,8 @@ function demandsForNode(): DemandBelongItem[] {
   return props.demandsByCategory[props.node.id] ?? []
 }
 
+const isInspecting = computed(() => props.inspectCategoryId === props.node.id)
+
 const weight = computed(() =>
   props.activeDim ? (props.node.weights?.[props.activeDim] ?? 0) : null,
 )
@@ -76,7 +83,12 @@ function onInspect(e: MouseEvent) {
   <div class="tree-node">
     <div
       class="node-card"
-      :class="{ leaf: !hasChildren(), open: isExpanded(), muted: activeDim != null && heatT == null }"
+      :class="{
+        leaf: !hasChildren(),
+        open: isExpanded(),
+        muted: activeDim != null && heatT == null,
+        inspecting: isInspecting,
+      }"
       :style="cardStyle"
       :title="
         !activeDim
@@ -108,7 +120,7 @@ function onInspect(e: MouseEvent) {
           v-if="demandsForNode().length"
           class="inspect"
           type="button"
-          title="查看需求词"
+          title="展开路径"
           @click="onInspect"
         >
           🔍
@@ -116,7 +128,15 @@ function onInspect(e: MouseEvent) {
       </div>
     </div>
 
-    <ul v-if="hasChildren() && isExpanded()" class="children">
+    <DemandPathPanel
+      v-if="isInspecting"
+      :open="true"
+      :category-name="inspectCategoryName"
+      :items="inspectItems"
+      @close="emit('closeInspect')"
+    />
+
+    <ul v-else-if="hasChildren() && isExpanded()" class="children">
       <li
         v-for="child in node.children"
         :key="child.id"
@@ -131,8 +151,12 @@ function onInspect(e: MouseEvent) {
           :demands-by-category="demandsByCategory"
           :active-dim="activeDim"
           :weight-scale="weightScale"
+          :inspect-category-id="inspectCategoryId"
+          :inspect-category-name="inspectCategoryName"
+          :inspect-items="inspectItems"
           @toggle="emit('toggle', $event)"
           @inspect="emit('inspect', $event)"
+          @close-inspect="emit('closeInspect')"
         />
       </li>
     </ul>
@@ -162,13 +186,18 @@ function onInspect(e: MouseEvent) {
   box-shadow: 0 1px 2px rgba(16, 24, 40, 0.04);
   flex-shrink: 0;
   box-sizing: border-box;
-  transition: background 0.15s ease, border-color 0.15s ease;
+  transition: background 0.15s ease, border-color 0.15s ease, box-shadow 0.15s ease;
 }
 
 .node-card.muted {
   opacity: 0.72;
 }
 
+.node-card.inspecting {
+  border-color: #3b82f6;
+  box-shadow: 0 0 0 3px rgba(59, 130, 246, 0.2);
+}
+
 .toggle {
   width: 22px;
   height: 22px;

web/src/types/demand.ts

@@ -9,6 +9,19 @@ export interface DemandBelongResponse {
   items: DemandBelongItem[]
 }
 
+export interface DemandVideoItem {
+  vid: string
+  title: string | null
+  final_topic_json: string | null
+}
+
+export interface DemandVideosResponse {
+  demand_belong_id: number
+  name: string | null
+  category_id: number
+  videos: DemandVideoItem[]
+}
+
 /** category_id → demands belonging to that category */
 export type DemandsByCategory = Record<number, DemandBelongItem[]>