增加需求分级流程

agents/demand_grade_agent/__init__.py

@@ -0,0 +1,10 @@
+"""
+demand_grade_agent — 需求分级评估 Agent
+
+职责：对 multi_demand_pool_di 中的现有需求，结合全局树先验热度
+（category_tree_weight）与后验真实效果（real_rov_7d），以及需求词粒度效果
+（demand_popularity_stats），划分 S/A/B/C/D 等级并落库到 demand_grade 表。
+"""
+from agents.demand_grade_agent.agent import create_demand_grade_agent
+
+__all__ = ["create_demand_grade_agent"]

agents/demand_grade_agent/agent.py

@@ -0,0 +1,30 @@
+"""
+demand_grade_agent 工厂 — 组装需求分级评估 Agent。
+"""
+from __future__ import annotations
+
+from pathlib import Path
+
+from supply_agent import Agent
+from supply_agent.config import Settings
+from agents.demand_grade_agent.tools import register_all_tools
+
+_PROMPT_PATH = Path(__file__).parent / "prompt" / "system_prompt.md"
+DEMAND_GRADE_AGENT_SYSTEM_PROMPT = _PROMPT_PATH.read_text(encoding="utf-8")
+
+
+def create_demand_grade_agent(
+    settings: Settings | None = None,
+    *,
+    model: str | None = None,
+) -> Agent:
+    """创建 demand_grade_agent 实例，注册专属工具。"""
+    agent = Agent(
+        settings=settings,
+        name="demand_grade_agent",
+        model=model,
+        system_prompt=DEMAND_GRADE_AGENT_SYSTEM_PROMPT,
+        max_iterations=40,
+    )
+    register_all_tools(agent.tools)
+    return agent

agents/demand_grade_agent/prompt/system_prompt.md

@@ -0,0 +1,66 @@
+## 角色与任务
+你是需求优先级分级专家。你会在用户消息中收到一批需求词（来自 `multi_demand_pool_di` 策略需求池）
+和对应的 biz_dt，任务是结合其归属树节点的**先验热度**与**后验真实效果**，逐一划分 S/A/B/C/D
+五档优先级，并调用工具落库到 `demand_grade` 表，供下游选题/投放决策参考。
+
+你只做分级判断，不生成新需求词，也不修改需求池原始数据；只处理消息中给定的这些需求词，
+不需要自行查找或列举其他待处理需求。
+
+## 先验 / 后验含义
+- **先验**：`category_tree_weight.total_score`（四个分维度 ext_pop / plat_sust_pop / plat_ly_pop / recent_pop 的排名分之和）。反映该需求所在树节点在类目树里的历史热度排名，是"没有真实上线数据时"的兜底依据。
+- **后验**：`real_rov_7d_avg` + `real_rov_7d_count`。
+  - `real_rov_7d_count > 0`：说明该节点/需求已有真实上线验证数据，**这是高置信信息，判级时应优先参考**，可以据此给出全档位（包括 S 或 D）。
+  - `real_rov_7d_count = 0`（或数据缺失）：说明效果未知，只能用先验兜底判断。**无论先验多高，都不建议给到 S 级**（因为没有真实验证支撑），一般封顶在 A。
+
+## 分级参考准则（非硬编码规则，需结合 `query_score_distribution` 自主定阈值）
+- 建议在每个批次开始时调用一次 `query_score_distribution`，参考 total_score 与 real_rov_7d_avg 的分位数（p25/p50/p75/p90），自行制定本批次统一的分档阈值，避免同一批次内前后标准漂移。
+- 有后验数据的需求：
+  - 后验效果处于同类中高位（如 real_rov_7d_avg ≥ p75）→ 可评 S 或 A
+  - 中等 → B
+  - 明显偏低（如低于 p25）→ C 或 D（即使先验很高，也应如实按后验降级，说明"先验热但验证效果不佳"）
+- 无后验数据的需求：
+  - 先验 total_score 很高（如 ≥ p75）→ A（不给 S，注明"无验证数据"）
+  - 中等 → B
+  - 偏低 → C
+  - 先验也很低、几乎无信号 → D
+- 一个需求可能挂在多个树节点上：以最相关/得分最高的节点为主要依据，reason 中说明取用了哪个节点。
+
+## 同义/相似需求合并
+同一语义的需求可能因措辞不同而在需求池里表现为多条独立记录（例如「减脂期加餐」与
+「减脂加餐」）。判级前应调用 `search_related_pool_demands(biz_dt, keywords=[...])`
+**批量**搜索本批各需求词，把找到的相关记录一并纳入参考（尤其是它们各自的 weight / real_rov_7d），
+不要只看单条记录就下结论；返回的 `[id=...]` 就是 `multi_demand_pool_di.id`，落库时必须原样
+收集进 `related_pool_ids`（**必填字段**，用于把分级结果关联回原始需求行）。
+
+`video_list`（关联视频列表）与 `strategies`（来源策略）会由 `batch_save_demand_grades` 根据
+`related_pool_ids` 自动从对应的原始需求行推导写入，你不需要手工整理这两个字段——但必须保证
+`related_pool_ids` 完整、准确，否则这两个字段会推导缺失或不完整。`category_ids` 除了写入
+`demand_grade` 的展示快照字段，也会同步写入 `demand_grade_category_rel` 映射表，供前端按分类
+高效查询已分级需求，因此尽量把该需求真实归属的所有节点都列全。
+
+## 可用工具
+- `query_latest_biz_dt()`：若用户消息未给出明确 biz_dt 时调用，返回需求池/权重表/热度统计表各自最新业务日。
+- `search_related_pool_demands(biz_dt, keywords)`：按同名/包含关系搜索需求池，**可一次传入多个 keyword** 批量查找同语义需求。
+- `query_demand_category_and_weight(demand_names, biz_dt=None)`：核心取数工具，**可一次传入多个 demand_name** 批量查询归属树节点 → 先验四维度 + total_score，及后验 real_rov_7d_avg/count。
+- `query_category_path(category_ids)`：查询类目根到叶路径文本，用于写 reason。
+- `query_demand_popularity_by_word(demand_word_names, biz_dt=None)`：按需求词粒度直接查热度统计，**可一次传入多个词** 交叉验证树节点级结论。
+- `query_score_distribution(biz_dt=None)`：查询先验/后验分数分布，制定本批次统一分档标准。
+- `batch_save_demand_grades(items, biz_dt=None)`：批量落库分级结果，可重复调用按 (biz_dt, demand_name) upsert 覆盖修正。`related_pool_ids` 必填，`video_list`/`strategies` 自动推导。
+
+## 工作流程
+1. 若用户消息未给出 biz_dt，先调用 `query_latest_biz_dt()` 确定用哪个业务日。
+2. 调用一次 `query_score_distribution(biz_dt)`，确定本批次判级用的分档阈值（该分布来自数据表，不依赖对话历史，每轮调用结果一致，可保证跨批次标准统一）。
+3. **优先批量调用取数工具以减少往返**：
+   - 一次 `search_related_pool_demands(biz_dt, keywords=[...])` 覆盖本批所有需求词（或按 5~10 个一组分批）；
+   - 一次 `query_demand_category_and_weight(demand_names=[...], biz_dt=...)` 批量取归属与权重；
+   - 必要时一次 `query_demand_popularity_by_word(demand_word_names=[...])` 做词粒度交叉验证。
+   各工具返回结果每段均标注原始查询词（如 `--- demand_name: xxx ---`），便于对应落库。
+4. 对给定列表中的每一个需求词，结合上述批量结果判定 S/A/B/C/D，reason 写清引用的具体数值（如 "total_score=3.2，位于本批p80，无后验数据"）；`related_pool_ids` 取自 `search_related_pool_demands` 返回的 `[id=...]`。
+5. 全部处理完后，调用一次（或分 2~3 次）`batch_save_demand_grades` 落库，覆盖这批给定的所有需求词。
+6. 简要汇报本批次的分级结果后结束本轮任务。
+
+## 原则
+- 禁止幻觉：只能引用工具真实返回的数值和类目路径，不能编造 category_id、分数或树节点名称。
+- reason 必须具体：写清引用的先验/后验数值、是否合并了同义词、依据哪个树节点。
+- 找不到归属树节点或权重数据的需求：如实说明"无法评级/数据缺失"，不要强行给出等级去凑数。
+- 有后验数据始终优先于纯先验判断；无后验数据时保持谨慎，不给最高档。

agents/demand_grade_agent/run.py

@@ -0,0 +1,33 @@
+#!/usr/bin/env python3
+"""Run demand_grade_agent on one batch of demand names.
+
+批次选择（哪些词、多少个）由调用方（通常是调度任务）负责，本入口只处理
+传入的这一批，不做分页/自行遍历需求池。
+"""
+from __future__ import annotations
+
+from agents.demand_grade_agent import create_demand_grade_agent
+
+
+def main(demand_names: list[str], biz_dt: str | None = None) -> None:
+    agent = create_demand_grade_agent()
+    print(f"demand_grade_agent ready | model={agent.model}")
+    print(f"tools: {agent.tools.list_tools()}")
+    print()
+
+    names_str = "\n".join(f"- {name}" for name in demand_names)
+    biz_dt_note = f"biz_dt={biz_dt}" if biz_dt else "未指定 biz_dt，请先调用 query_latest_biz_dt() 确定"
+    user_input = f"""
+    请对以下 {len(demand_names)} 个需求词逐一评级（S/A/B/C/D），{biz_dt_note}。
+    只处理这一批，不要尝试查找或列举更多需求词。判级完成后调用 batch_save_demand_grades 落库。
+
+    需求词列表:
+    {names_str}
+    """
+    result = agent.run(user_input)
+    print(result.content)
+    print(f"\n[iterations={result.iterations}, tool_calls={result.tool_calls_made}]")
+
+
+if __name__ == "__main__":
+    main(["因果报应", "降半旗"], biz_dt="20260714")

agents/demand_grade_agent/tools/__init__.py

@@ -0,0 +1,52 @@
+"""
+demand_grade_agent 工具包
+
+批次选词（哪些需求词、多少个）由外部调度任务负责，agent 只处理调用方在
+用户消息里显式给出的需求词，不提供“列出/分页遍历需求池”类工具。
+"""
+from __future__ import annotations
+
+from collections.abc import Callable
+from typing import Any
+
+from agents.demand_grade_agent.tools.batch_save_demand_grades import batch_save_demand_grades
+from agents.demand_grade_agent.tools.query_category_path import query_category_path
+from agents.demand_grade_agent.tools.query_demand_category_and_weight import (
+    query_demand_category_and_weight,
+)
+from agents.demand_grade_agent.tools.query_demand_popularity_by_word import (
+    query_demand_popularity_by_word,
+)
+from agents.demand_grade_agent.tools.query_latest_biz_dt import query_latest_biz_dt
+from agents.demand_grade_agent.tools.query_score_distribution import query_score_distribution
+from agents.demand_grade_agent.tools.search_related_pool_demands import (
+    search_related_pool_demands,
+)
+from supply_agent.tools.registry import ToolRegistry
+
+ALL_TOOLS: list[Callable[..., Any]] = [
+    query_latest_biz_dt,
+    search_related_pool_demands,
+    query_demand_category_and_weight,
+    query_category_path,
+    query_demand_popularity_by_word,
+    query_score_distribution,
+    batch_save_demand_grades,
+]
+
+__all__ = [
+    "ALL_TOOLS",
+    "batch_save_demand_grades",
+    "query_category_path",
+    "query_demand_category_and_weight",
+    "query_demand_popularity_by_word",
+    "query_latest_biz_dt",
+    "query_score_distribution",
+    "register_all_tools",
+    "search_related_pool_demands",
+]
+
+
+def register_all_tools(registry: ToolRegistry) -> ToolRegistry:
+    """将 demand_grade_agent 包内的所有工具注册到 ToolRegistry。"""
+    return registry.from_decorated(*ALL_TOOLS)

agents/demand_grade_agent/tools/batch_save_demand_grades.py

@@ -0,0 +1,248 @@
+"""
+批量保存需求分级结果到 demand_grade 表。
+"""
+from __future__ import annotations
+
+import logging
+from decimal import Decimal
+from typing import Any, Optional
+
+from agents.demand_grade_agent.tools.shared import (
+    VALID_GRADES,
+    collect_strategies,
+    dump_int_list,
+    merge_video_ids,
+    normalize_biz_dt,
+)
+from supply_agent.tools import tool
+from supply_infra.db.repositories.demand_grade_category_rel_repo import (
+    DemandGradeCategoryRelRepository,
+)
+from supply_infra.db.repositories.demand_grade_repo import DemandGradeRepository
+from supply_infra.db.repositories.multi_demand_pool_di_repo import MultiDemandPoolDiRepository
+from supply_infra.db.session import get_session
+
+logger = logging.getLogger(__name__)
+
+
+def _optional_decimal(value: Any, field: str, idx: int, errors: list[str]) -> Decimal | None:
+    if value is None or value == "":
+        return None
+    try:
+        return Decimal(str(value))
+    except Exception:
+        errors.append(f"第 {idx} 项 {field} 无效: {value!r}")
+        return None
+
+
+def _optional_int_list(value: Any, field: str, idx: int, errors: list[str]) -> list[int]:
+    if value is None:
+        return []
+    if not isinstance(value, list):
+        errors.append(f"第 {idx} 项 {field} 必须是数组: {value!r}")
+        return []
+    out: list[int] = []
+    for v in value:
+        try:
+            out.append(int(v))
+        except (TypeError, ValueError):
+            errors.append(f"第 {idx} 项 {field} 含无效元素: {v!r}")
+            return []
+    return out
+
+
+def _normalize_items(
+    items: list[dict[str, Any]],
+    *,
+    default_biz_dt: str | None,
+) -> tuple[list[dict[str, Any]], list[list[int]], list[list[int]], list[str]]:
+    """校验并规范化待落库行，返回 (rows, related_pool_id_lists, category_id_lists, errors)。"""
+    rows: list[dict[str, Any]] = []
+    related_pool_id_lists: list[list[int]] = []
+    category_id_lists: list[list[int]] = []
+    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
+
+        demand_name = str(item.get("demand_name") or "").strip()
+        if not demand_name:
+            errors.append(f"第 {idx} 项缺少 demand_name")
+            continue
+
+        grade = str(item.get("grade") or "").strip().upper()
+        if grade not in VALID_GRADES:
+            errors.append(f"第 {idx} 项 grade 无效（只能是 {'/'.join(VALID_GRADES)}）: {item.get('grade')!r}")
+            continue
+
+        reason = str(item.get("reason") or "").strip()
+        if not reason:
+            errors.append(f"第 {idx} 项缺少 reason")
+            continue
+
+        item_biz_dt, err = normalize_biz_dt(item.get("biz_dt"))
+        if err:
+            errors.append(f"第 {idx} 项 {err}")
+            continue
+        resolved_biz_dt = item_biz_dt or default_biz_dt
+        if not resolved_biz_dt:
+            errors.append(f"第 {idx} 项缺少 biz_dt（且未传入默认 biz_dt）")
+            continue
+
+        dedupe_key = (resolved_biz_dt, demand_name)
+        if dedupe_key in seen_keys:
+            errors.append(f"第 {idx} 项在本次请求中重复: biz_dt={resolved_biz_dt}, demand_name={demand_name}")
+            continue
+
+        related_pool_ids = _optional_int_list(item.get("related_pool_ids"), "related_pool_ids", idx, errors)
+        if not related_pool_ids:
+            errors.append(
+                f"第 {idx} 项缺少 related_pool_ids（必填，需先用 search_related_pool_demands 找到对应的 "
+                f"multi_demand_pool_di.id）"
+            )
+            continue
+
+        seen_keys.add(dedupe_key)
+
+        score = _optional_decimal(item.get("score"), "score", idx, errors)
+        prior_total_score = _optional_decimal(item.get("prior_total_score"), "prior_total_score", idx, errors)
+        posterior_rov_avg = _optional_decimal(item.get("posterior_rov_avg"), "posterior_rov_avg", idx, errors)
+
+        posterior_rov_count = 0
+        if item.get("posterior_rov_count") is not None:
+            try:
+                posterior_rov_count = int(item.get("posterior_rov_count"))
+            except (TypeError, ValueError):
+                errors.append(f"第 {idx} 项 posterior_rov_count 无效: {item.get('posterior_rov_count')!r}")
+                continue
+
+        category_ids = _optional_int_list(item.get("category_ids"), "category_ids", idx, errors)
+
+        has_posterior = 1 if posterior_rov_count > 0 else 0
+
+        rows.append(
+            {
+                "biz_dt": resolved_biz_dt,
+                "demand_name": demand_name,
+                "category_ids": dump_int_list(category_ids),
+                "grade": grade,
+                "score": score,
+                "prior_total_score": prior_total_score,
+                "posterior_rov_avg": posterior_rov_avg,
+                "posterior_rov_count": posterior_rov_count,
+                "has_posterior": has_posterior,
+                "related_pool_ids": dump_int_list(related_pool_ids),
+                "reason": reason,
+            }
+        )
+        related_pool_id_lists.append(related_pool_ids)
+        category_id_lists.append(category_ids)
+
+    return rows, related_pool_id_lists, category_id_lists, errors
+
+
+@tool
+def batch_save_demand_grades(items: list[dict[str, Any]], biz_dt: Optional[str] = None) -> str:
+    """
+    批量保存需求分级结果到 demand_grade 表（按 biz_dt+demand_name upsert，可重复调用覆盖修正）。
+
+    video_list（关联视频列表）与 strategies（来源策略列表）会自动从 related_pool_ids 对应的
+    multi_demand_pool_di 原始行推导写入，无需手工传入。category_ids 除了写入展示快照字段，
+    也会同步写入 demand_grade_category_rel 映射表，供前端按分类高效查询。
+
+    Args:
+        items: 待保存列表，每项字段：
+            - demand_name (必填): 需求名称
+            - grade (必填): S/A/B/C/D 之一
+            - reason (必填): 判断依据，需引用具体的先验/后验数值
+            - related_pool_ids (必填): 该需求对应的 multi_demand_pool_di.id 列表，需先调用
+              search_related_pool_demands 找到；用于关联原始需求，并自动推导 video_list/strategies
+            - score (可选): 0-100 数值分，辅助同级排序
+            - category_ids (可选): 归属的树节点 id 列表，会写入 demand_grade_category_rel 映射表
+            - prior_total_score (可选): 落库时的先验 total_score 快照
+            - posterior_rov_avg / posterior_rov_count (可选): 落库时的后验 real_rov_7d 快照；
+              count>0 时自动标记为「有后验数据」
+            - biz_dt (可选): 覆盖本项使用的业务日，不传则用调用时的 biz_dt 参数
+        biz_dt: 本次调用的默认业务日期 YYYYMMDD；items 内每项也可单独指定 biz_dt 覆盖。
+
+    Returns:
+        保存结果摘要，包含成功条数与校验失败说明。
+    """
+    if not items:
+        return "items 不能为空"
+
+    default_biz_dt, err = normalize_biz_dt(biz_dt)
+    if err:
+        return err
+
+    rows, related_pool_id_lists, category_id_lists, errors = _normalize_items(
+        items, default_biz_dt=default_biz_dt
+    )
+    if not rows:
+        detail = "；".join(errors) if errors else "无有效数据"
+        return f"没有可保存的数据: {detail}"
+
+    try:
+        with get_session() as session:
+            pool_repo = MultiDemandPoolDiRepository(session)
+            all_pool_ids = sorted({pid for ids in related_pool_id_lists for pid in ids})
+            pool_rows = pool_repo.get_by_ids(all_pool_ids) if all_pool_ids else []
+            pool_by_id = {int(r.id): r for r in pool_rows}
+
+            final_rows: list[dict[str, Any]] = []
+            saved_indices: list[int] = []
+            for i, (row, pool_ids) in enumerate(zip(rows, related_pool_id_lists)):
+                matched = [pool_by_id[pid] for pid in pool_ids if pid in pool_by_id]
+                missing = [pid for pid in pool_ids if pid not in pool_by_id]
+                if not matched:
+                    errors.append(
+                        f"demand_name={row['demand_name']!r} 的 related_pool_ids={pool_ids} "
+                        f"均未在 multi_demand_pool_di 中找到，跳过该项"
+                    )
+                    continue
+                if missing:
+                    errors.append(
+                        f"demand_name={row['demand_name']!r} 的 related_pool_ids 中 {missing} 未找到，已忽略"
+                    )
+                row["video_list"] = merge_video_ids(matched)
+                row["strategies"] = collect_strategies(matched)
+                final_rows.append(row)
+                saved_indices.append(i)
+
+            if not final_rows:
+                detail = "；".join(errors) if errors else "无有效数据"
+                return f"没有可保存的数据: {detail}"
+
+            grade_repo = DemandGradeRepository(session)
+            affected = grade_repo.bulk_upsert(final_rows)
+
+            names_by_biz_dt: dict[str, list[str]] = {}
+            for i in saved_indices:
+                names_by_biz_dt.setdefault(rows[i]["biz_dt"], []).append(rows[i]["demand_name"])
+
+            id_by_biz_dt_name: dict[tuple[str, str], int] = {}
+            for bd, names in names_by_biz_dt.items():
+                for name, demand_grade_id in grade_repo.get_ids_by_names(bd, names).items():
+                    id_by_biz_dt_name[(bd, name)] = demand_grade_id
+
+            rel_repo = DemandGradeCategoryRelRepository(session)
+            for i in saved_indices:
+                key = (rows[i]["biz_dt"], rows[i]["demand_name"])
+                demand_grade_id = id_by_biz_dt_name.get(key)
+                if demand_grade_id is not None:
+                    rel_repo.replace_for_demand_grade(demand_grade_id, category_id_lists[i])
+
+        parts = [f"提交 {len(rows)} 条，成功写入/更新 {affected} 条（{len(final_rows)} 条通过校验）"]
+        if errors:
+            parts.append(f"校验失败/警告 {len(errors)} 条: " + "；".join(errors))
+
+        message = "。".join(parts)
+        logger.info("batch_save_demand_grades completed: %s", message)
+        return message
+
+    except Exception as e:
+        logger.error("batch_save_demand_grades failed: %s", e, exc_info=True)
+        return f"批量保存需求分级失败: {e}"

agents/demand_grade_agent/tools/query_category_path.py

@@ -0,0 +1,81 @@
+"""
+按分类 id 查询从根到节点的类目路径，用于写 reason。
+"""
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+from agents.demand_grade_agent.tools.shared 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:
+        每行一条路径，例如：
+        [88] 美食 > 减脂饮食 > 加餐
+        [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/demand_grade_agent/tools/query_demand_category_and_weight.py

@@ -0,0 +1,167 @@
+"""
+核心取数工具：需求名 → 归属树节点 → 该节点的先验热度与后验真实效果。
+"""
+from __future__ import annotations
+
+import logging
+from typing import Optional
+
+from sqlalchemy.orm import Session
+
+from agents.demand_grade_agent.tools.shared import (
+    build_category_path,
+    format_score,
+    normalize_biz_dt,
+    normalize_str_list,
+)
+from supply_agent.tools import tool
+from supply_infra.db.models.category_tree_weight import CategoryTreeWeight
+from supply_infra.db.models.global_tree_category import GlobalTreeCategory
+from supply_infra.db.repositories.category_tree_weight_repo import CategoryTreeWeightRepository
+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.global_tree_category_repo import GlobalTreeCategoryRepository
+from supply_infra.db.repositories.multi_demand_pool_di_repo import MultiDemandPoolDiRepository
+from supply_infra.db.session import get_session
+
+logger = logging.getLogger(__name__)
+
+
+def _format_weight_row(weight: CategoryTreeWeight, path: str | None) -> str:
+    posterior_note = "有后验验证数据" if weight.real_rov_7d_count > 0 else "无后验验证数据(效果未知)"
+    return (
+        f"[category_id={weight.category_id}] {path or '（未知路径）'}\n"
+        f"  biz_dt={weight.biz_dt}  先验total_score={format_score(weight.total_score)}\n"
+        f"  先验分维度: 外部热度={format_score(weight.ext_pop_avg)}"
+        f" 平台持续热度={format_score(weight.plat_sust_pop_avg)}"
+        f" 平台去年同期={format_score(weight.plat_ly_pop_avg)}"
+        f" 近期热度={format_score(weight.recent_pop_avg)}\n"
+        f"  后验real_rov_7d: avg={format_score(weight.real_rov_7d_avg)}"
+        f" count={weight.real_rov_7d_count} ({posterior_note})"
+    )
+
+
+def _query_one_demand_category_and_weight(
+    session: Session,
+    demand_name: str,
+    normalized_dt: str | None,
+    by_id: dict[int, GlobalTreeCategory],
+) -> str:
+    belong_ids: set[int] = set()
+
+    pool_repo = MultiDemandPoolDiRepository(session)
+    if normalized_dt:
+        pool_rows = pool_repo.search_rows_by_name_fragment(normalized_dt, demand_name)
+        pool_ids = [row["id"] for row in pool_rows if row["demand_name"] == demand_name]
+    else:
+        pool_ids = []
+
+    if pool_ids:
+        rel_map = DemandBelongPoolRelRepository(session).get_belong_ids_by_pool_ids(pool_ids)
+        for ids in rel_map.values():
+            belong_ids.update(ids)
+
+    belong_repo = DemandBelongCategoryRepository(session)
+    if not belong_ids:
+        fuzzy = belong_repo.search_by_name_like(demand_name)
+        belong_ids.update(int(row.id) for row in fuzzy)
+
+    if not belong_ids:
+        return f"「{demand_name}」未找到归属的树节点（demand_belong_pool_rel 与 demand_belong_category 均无匹配）"
+
+    belong_rows = belong_repo.get_by_ids(list(belong_ids))
+    category_ids = sorted({row.category_id for row in belong_rows if row.category_id})
+    if not category_ids:
+        return f"「{demand_name}」匹配到需求归属记录，但均未挂载 category_id"
+
+    weight_repo = CategoryTreeWeightRepository(session)
+    weights = weight_repo.get_by_category_ids(category_ids, normalized_dt)
+    if not weights:
+        dt_note = f"biz_dt={normalized_dt}" if normalized_dt else "任意业务日"
+        return (
+            f"「{demand_name}」归属树节点 {category_ids}，"
+            f"但 category_tree_weight 在 {dt_note} 无数据"
+        )
+
+    lines = [f"「{demand_name}」归属 {len(weights)} 个树节点:"]
+    for weight in sorted(weights, key=lambda w: w.total_score, reverse=True):
+        path = build_category_path(int(weight.category_id), by_id)
+        lines.append(_format_weight_row(weight, path))
+    return "\n".join(lines)
+
+
+@tool
+def query_demand_category_and_weight(
+    demand_names: list[str],
+    biz_dt: Optional[str] = None,
+) -> str:
+    """
+    需求名 → 归属树节点 → 先验热度(total_score等) + 后验真实效果(real_rov_7d)。
+
+    支持批量传入多个需求名，一次调用返回各词的归属与权重；每段结果前会标注原始 demand_name。
+
+    查找顺序（对每个 demand_name）：
+    1. 先在 multi_demand_pool_di 中按需求名精确匹配找到池表行 id；
+    2. 经 demand_belong_pool_rel 反查这些池表行归属的 demand_belong_category；
+    3. 若第 2 步查不到关系，退化为按需求名模糊匹配 demand_belong_category.name；
+    4. 用归属到的 category_id 查询 category_tree_weight 的先验/后验数据。
+
+    一个需求可能挂在多个树节点上，会全部列出。
+
+    Args:
+        demand_names: 需求名称列表（通常来自 multi_demand_pool_di.demand_name），可一次传多个。
+        biz_dt: 业务日期 YYYYMMDD，可选。传入时会先在该日期的需求池里精确匹配需求名，
+                再经 demand_belong_pool_rel 反查归属；不传时直接对 demand_belong_category.name
+                做模糊匹配（跳过第 1/2 步），权重也取每个树节点各自的最新业务日数据。
+
+    Returns:
+        每个 demand_name 一段，段首标注 `--- demand_name: xxx ---`，例如：
+        --- demand_name: 减脂期加餐 ---
+        「减脂期加餐」归属 1 个树节点:
+        [category_id=88] 美食 > 减脂饮食 > 加餐
+          biz_dt=20260716  先验total_score=3.42
+          ...
+    """
+    normalized_dt, err = normalize_biz_dt(biz_dt)
+    if err:
+        return err
+
+    names, err = normalize_str_list(demand_names, "demand_names")
+    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}
+
+            sections: list[str] = []
+            for demand_name in names:
+                result = _query_one_demand_category_and_weight(
+                    session, demand_name, normalized_dt, by_id
+                )
+                sections.append(f"--- demand_name: {demand_name} ---\n{result}")
+
+        message = "\n\n".join(sections)
+        logger.info(
+            "query_demand_category_and_weight completed: count=%d biz_dt=%s",
+            len(names),
+            normalized_dt,
+        )
+        return message
+
+    except Exception as e:
+        logger.error("query_demand_category_and_weight failed: %s", e, exc_info=True)
+        return f"查询需求归属与权重失败: {e}"
+
+
+def main() -> None:
+    print(query_demand_category_and_weight(demand_names=["减脂期加餐", "减脂加餐"]))
+
+
+if __name__ == "__main__":
+    main()

agents/demand_grade_agent/tools/query_demand_popularity_by_word.py

@@ -0,0 +1,107 @@
+"""
+按需求词粒度直接查询 demand_popularity_stats，作为树节点粒度数据的交叉验证。
+"""
+from __future__ import annotations
+
+import logging
+from typing import Optional
+
+from sqlalchemy.orm import Session
+
+from agents.demand_grade_agent.tools.shared import (
+    format_score,
+    normalize_biz_dt,
+    normalize_str_list,
+)
+from supply_agent.tools import tool
+from supply_infra.db.repositories.demand_popularity_stats_repo import (
+    DemandPopularityStatsRepository,
+)
+from supply_infra.db.session import get_session
+
+logger = logging.getLogger(__name__)
+
+
+def _query_one_demand_popularity_by_word(
+    session: Session,
+    keyword: str,
+    normalized_dt: str | None,
+) -> str:
+    rows = DemandPopularityStatsRepository(session).search_by_word_name(keyword, normalized_dt)
+
+    if not rows:
+        dt_note = f"biz_dt={normalized_dt}" if normalized_dt else "任意业务日"
+        return f"「{keyword}」在 demand_popularity_stats（{dt_note}）无匹配数据"
+
+    lines = []
+    for row in rows:
+        posterior_note = "有后验数据" if row.real_rov_7d_count > 0 else "无后验数据(效果未知)"
+        lines.append(
+            f"[biz_dt={row.biz_dt}] {row.demand_word_name}: "
+            f"外部热度={format_score(row.ext_pop_avg)} "
+            f"平台持续热度={format_score(row.plat_sust_pop_avg)} "
+            f"平台去年同期={format_score(row.plat_ly_pop_avg)} "
+            f"近期热度={format_score(row.recent_pop_avg)}\n"
+            f"  后验real_rov_7d: avg={format_score(row.real_rov_7d_avg)} "
+            f"count={row.real_rov_7d_count} ({posterior_note})"
+        )
+    return "\n".join(lines)
+
+
+@tool
+def query_demand_popularity_by_word(
+    demand_word_names: list[str],
+    biz_dt: Optional[str] = None,
+) -> str:
+    """
+    按需求词名（精确+模糊）查询 demand_popularity_stats 的词粒度效果数据。
+
+    支持批量传入多个需求词，一次调用返回各词的统计；每段结果前会标注原始 demand_word_name。
+
+    demand_popularity_stats 比 category_tree_weight 粒度更细（按具体需求词而非整个树节点），
+    可用于交叉验证 query_demand_category_and_weight 给出的树节点级结论，
+    也可用于发现措辞相近但独立统计的近似词数据。
+
+    Args:
+        demand_word_names: 需求词名称或关键片段列表，可一次传多个。
+        biz_dt: 业务日期 YYYYMMDD，可选；不传则不限日期，按业务日降序列出（可能有多天历史数据）。
+
+    Returns:
+        每个 demand_word_name 一段，段首标注 `--- demand_word_name: xxx ---`，例如：
+        --- demand_word_name: 减脂期加餐 ---
+        [biz_dt=20260716] 减脂期加餐: 外部热度=1.10 ...
+    """
+    normalized_dt, err = normalize_biz_dt(biz_dt)
+    if err:
+        return err
+
+    names, err = normalize_str_list(demand_word_names, "demand_word_names")
+    if err:
+        return err
+
+    try:
+        with get_session() as session:
+            sections: list[str] = []
+            for keyword in names:
+                result = _query_one_demand_popularity_by_word(session, keyword, normalized_dt)
+                sections.append(f"--- demand_word_name: {keyword} ---\n{result}")
+
+        message = "\n\n".join(sections)
+        logger.info(
+            "query_demand_popularity_by_word completed: count=%d biz_dt=%s",
+            len(names),
+            normalized_dt,
+        )
+        return message
+
+    except Exception as e:
+        logger.error("query_demand_popularity_by_word failed: %s", e, exc_info=True)
+        return f"查询需求词热度统计失败: {e}"
+
+
+def main() -> None:
+    print(query_demand_popularity_by_word(demand_word_names=["加餐", "减脂期加餐"]))
+
+
+if __name__ == "__main__":
+    main()

agents/demand_grade_agent/tools/query_latest_biz_dt.py

@@ -0,0 +1,60 @@
+"""
+查询需求分级相关三张表各自的最新业务日，供未指定 biz_dt 时先探测可用日期。
+"""
+from __future__ import annotations
+
+import logging
+
+from supply_agent.tools import tool
+from supply_infra.db.repositories.category_tree_weight_repo import CategoryTreeWeightRepository
+from supply_infra.db.repositories.demand_popularity_stats_repo import (
+    DemandPopularityStatsRepository,
+)
+from supply_infra.db.repositories.multi_demand_pool_di_repo import MultiDemandPoolDiRepository
+from supply_infra.db.session import get_session
+
+logger = logging.getLogger(__name__)
+
+
+@tool
+def query_latest_biz_dt() -> str:
+    """
+    查询需求池、类目树权重、需求词热度统计三张表各自的最新业务日。
+
+    三张表由不同 job 产出，可能不完全同步（权重/热度统计可能滞后于需求池）。
+    未指定 biz_dt 时应先调用本工具，再根据返回结果决定分级使用哪个业务日
+    （通常以需求池的最新日为准；若权重/热度统计当日还没跑出来，可退到它们各自的最新日）。
+
+    Returns:
+        三张表各自的最新业务日，例如：
+        multi_demand_pool_di=20260716
+        category_tree_weight=20260716
+        demand_popularity_stats=20260715
+        （某表暂无数据时显示为 无数据）
+    """
+    try:
+        with get_session() as session:
+            pool_dt = MultiDemandPoolDiRepository(session).get_latest_biz_dt()
+            weight_dt = CategoryTreeWeightRepository(session).get_latest_biz_dt()
+            stats_dt = DemandPopularityStatsRepository(session).get_latest_biz_dt()
+
+        lines = [
+            f"multi_demand_pool_di={pool_dt or '无数据'}",
+            f"category_tree_weight={weight_dt or '无数据'}",
+            f"demand_popularity_stats={stats_dt or '无数据'}",
+        ]
+        message = "\n".join(lines)
+        logger.info("query_latest_biz_dt completed: %s", message.replace("\n", " "))
+        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()

agents/demand_grade_agent/tools/query_score_distribution.py

@@ -0,0 +1,104 @@
+"""
+查询指定业务日 category_tree_weight 的分数分布，供批量分级前统一分档阈值。
+"""
+from __future__ import annotations
+
+import logging
+from typing import Optional
+
+from agents.demand_grade_agent.tools.shared import distribution_summary, normalize_biz_dt, to_float
+from supply_agent.tools import tool
+from supply_infra.db.repositories.category_tree_weight_repo import CategoryTreeWeightRepository
+from supply_infra.db.session import get_session
+
+logger = logging.getLogger(__name__)
+
+_DIM_FIELDS: tuple[tuple[str, str], ...] = (
+    ("total_score", "先验total_score(四维排名分之和)"),
+    ("ext_pop_avg", "外部热度"),
+    ("plat_sust_pop_avg", "平台持续热度"),
+    ("plat_ly_pop_avg", "平台去年同期热度"),
+    ("recent_pop_avg", "近期热度"),
+)
+
+
+def _format_dist(label: str, dist: dict) -> str:
+    if dist["count"] == 0:
+        return f"{label}: 无数据"
+    return (
+        f"{label} (n={dist['count']}): "
+        f"min={dist['min']:.4f} p25={dist['p25']:.4f} p50={dist['p50']:.4f} "
+        f"p75={dist['p75']:.4f} p90={dist['p90']:.4f} max={dist['max']:.4f}"
+    )
+
+
+@tool
+def query_score_distribution(biz_dt: Optional[str] = None) -> str:
+    """
+    查询指定业务日 category_tree_weight 的分数分布（先验四维度+total_score，及后验real_rov_7d）。
+
+    建议在批量分级任务开始时调用一次，参考分位数自行制定本批次统一的分档阈值
+    （例如 total_score 前 10% 视为先验很高），避免同一批次内多次判断标准漂移。
+    后验 real_rov_7d_avg 的分布只统计 real_rov_7d_count>0（有真实验证数据）的子集，
+    因为无验证数据的行 avg 无意义。
+
+    Args:
+        biz_dt: 业务日期 YYYYMMDD，可选；不传则使用 category_tree_weight 最新业务日。
+
+    Returns:
+        每个维度一行分布摘要，例如：
+        先验total_score(四维排名分之和) (n=1500): min=0.0000 p25=0.8500 p50=1.6000 p75=2.9000 p90=4.1000 max=8.2000
+        后验real_rov_7d_avg(仅count>0子集) (n=210): ...
+    """
+    normalized_dt, err = normalize_biz_dt(biz_dt)
+    if err:
+        return err
+
+    try:
+        with get_session() as session:
+            repo = CategoryTreeWeightRepository(session)
+            resolved_dt = normalized_dt or repo.get_latest_biz_dt()
+            if not resolved_dt:
+                return "category_tree_weight 暂无数据"
+
+            weights = repo.list_by_biz_dt(resolved_dt)
+            if not weights:
+                return f"biz_dt={resolved_dt} category_tree_weight 无数据"
+
+            node_count = len(weights)
+            dim_values: dict[str, list[float]] = {
+                field: [v for v in (to_float(getattr(w, field)) for w in weights) if v is not None]
+                for field, _ in _DIM_FIELDS
+            }
+            posterior_values = [
+                v
+                for v in (to_float(w.real_rov_7d_avg) for w in weights if w.real_rov_7d_count > 0)
+                if v is not None
+            ]
+
+        lines = [f"biz_dt={resolved_dt} 共 {node_count} 个树节点"]
+        for field, label in _DIM_FIELDS:
+            lines.append(_format_dist(label, distribution_summary(dim_values[field])))
+
+        lines.append(
+            _format_dist(
+                f"后验real_rov_7d_avg(仅count>0子集，共{len(posterior_values)}个节点有验证数据)",
+                distribution_summary(posterior_values),
+            )
+        )
+
+        message = "\n".join(lines)
+        logger.info("query_score_distribution completed: biz_dt=%s nodes=%d", resolved_dt, node_count)
+        return message
+
+    except Exception as e:
+        logger.error("query_score_distribution failed: %s", e, exc_info=True)
+        return f"查询分数分布失败: {e}"
+
+
+def main() -> None:
+    print(query_score_distribution())
+
+
+if __name__ == "__main__":
+    main()

agents/demand_grade_agent/tools/search_related_pool_demands.py

@@ -0,0 +1,101 @@
+"""
+在需求池中按同名/包含关系搜索需求词，用于合并同语义、措辞不同的需求一起判断。
+"""
+from __future__ import annotations
+
+import logging
+
+from sqlalchemy.orm import Session
+
+from agents.demand_grade_agent.tools.shared import (
+    format_score,
+    normalize_biz_dt,
+    normalize_str_list,
+)
+from supply_agent.tools import tool
+from supply_infra.db.repositories.multi_demand_pool_di_repo import MultiDemandPoolDiRepository
+from supply_infra.db.session import get_session
+
+logger = logging.getLogger(__name__)
+
+
+def _search_one_related_pool_demands(
+    session: Session,
+    normalized: str,
+    keyword: str,
+) -> str:
+    rows = MultiDemandPoolDiRepository(session).search_rows_by_name_fragment(normalized, keyword)
+
+    if not rows:
+        return f"biz_dt={normalized} 未找到与「{keyword}」同名/包含关系的需求词"
+
+    lines = []
+    for row in rows:
+        rov = format_score(row["real_rov_7d"])
+        vov = format_score(row["real_vov_7d"])
+        weight = format_score(row["weight"])
+        video_count = row["video_count"] if row["video_count"] is not None else "—"
+        lines.append(
+            f"[id={row['id']}|{row['strategy']}|weight={weight}"
+            f"|视频数={video_count}|真实ROV={rov}|真实VOV={vov}] {row['demand_name']}"
+        )
+    return "\n".join(lines)
+
+
+@tool
+def search_related_pool_demands(biz_dt: str, keywords: list[str]) -> str:
+    """
+    按同名/包含关系搜索需求池，合并同语义需求的多条记录一起判断。
+
+    支持批量传入多个 keyword，同一 biz_dt 下一次调用返回各关键词的匹配结果；
+    每段结果前会标注原始 keyword。
+
+    匹配规则为双向包含：keyword 是 demand_name 的子串，或 demand_name 是 keyword 的子串。
+    用于发现措辞不同但语义相同/高度相关的需求词（例如「减脂期加餐」与「减脂加餐」），
+    在判级时应把这些记录一起纳入参考，而不是只看单条记录。
+
+    Args:
+        biz_dt: 业务日期，格式 YYYYMMDD（必填）。
+        keywords: 需求词或关键片段列表，可一次传多个。
+
+    Returns:
+        每个 keyword 一段，段首标注 `--- keyword: xxx ---`，例如：
+        --- keyword: 加餐 ---
+        [id=101|strategy_a|weight=3.20|视频数=12|真实ROV=0.0410|真实VOV=0.0021] 减脂期加餐怎么吃
+    """
+    normalized, err = normalize_biz_dt(biz_dt)
+    if err:
+        return err
+    if not normalized:
+        return "biz_dt 不能为空"
+
+    keyword_list, err = normalize_str_list(keywords, "keywords")
+    if err:
+        return err
+
+    try:
+        with get_session() as session:
+            sections: list[str] = []
+            for keyword in keyword_list:
+                result = _search_one_related_pool_demands(session, normalized, keyword)
+                sections.append(f"--- keyword: {keyword} ---\n{result}")
+
+        message = "\n\n".join(sections)
+        logger.info(
+            "search_related_pool_demands completed: biz_dt=%s count=%d",
+            normalized,
+            len(keyword_list),
+        )
+        return message
+
+    except Exception as e:
+        logger.error("search_related_pool_demands failed: %s", e, exc_info=True)
+        return f"搜索关联需求词失败: {e}"
+
+
+def main() -> None:
+    print(search_related_pool_demands(biz_dt="20260716", keywords=["加餐", "减脂"]))
+
+
+if __name__ == "__main__":
+    main()

agents/demand_grade_agent/tools/shared.py

@@ -0,0 +1,207 @@
+"""demand_grade_agent 工具共享辅助函数（非 @tool，不对外暴露为工具）。"""
+from __future__ import annotations
+
+import json
+from decimal import Decimal
+from typing import Any
+
+from supply_infra.db.models.global_tree_category import GlobalTreeCategory
+from supply_infra.db.models.multi_demand_pool_di import MultiDemandPoolDi
+
+_VIDEO_LIST_LIMIT = 10
+
+VALID_GRADES: tuple[str, ...] = ("S", "A", "B", "C", "D")
+
+
+def normalize_biz_dt(biz_dt: str | None) -> tuple[str | None, str | None]:
+    """校验并规范化 biz_dt（YYYYMMDD）；空值返回 (None, None) 表示未指定。"""
+    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 to_float(value: Any) -> float | None:
+    """将 Decimal/None/数值统一转换为 float，None 原样返回。"""
+    if value is None:
+        return None
+    return float(value)
+
+
+def format_score(avg: Decimal | float | int | None) -> str:
+    """格式化分数为易读文本，None 显示为 —。"""
+    if avg is None:
+        return "—"
+    value = float(avg)
+    if value >= 100:
+        return f"{value:.0f}"
+    if value >= 1:
+        return f"{value:.2f}"
+    return f"{value:.4f}"
+
+
+def percentile(sorted_values: list[float], pct: float) -> float | None:
+    """对已排序（升序）的数值列表求分位数（线性插值），pct 取 0~100。"""
+    if not sorted_values:
+        return None
+    if len(sorted_values) == 1:
+        return sorted_values[0]
+
+    rank = (pct / 100) * (len(sorted_values) - 1)
+    lower_idx = int(rank)
+    upper_idx = min(lower_idx + 1, len(sorted_values) - 1)
+    frac = rank - lower_idx
+    return sorted_values[lower_idx] + (sorted_values[upper_idx] - sorted_values[lower_idx]) * frac
+
+
+def distribution_summary(values: list[float]) -> dict[str, float | int | None]:
+    """返回一组数值的 min/p25/p50/p75/p90/max/count 分布摘要。"""
+    if not values:
+        return {"count": 0, "min": None, "p25": None, "p50": None, "p75": None, "p90": None, "max": None}
+    ordered = sorted(values)
+    return {
+        "count": len(ordered),
+        "min": ordered[0],
+        "p25": percentile(ordered, 25),
+        "p50": percentile(ordered, 50),
+        "p75": percentile(ordered, 75),
+        "p90": percentile(ordered, 90),
+        "max": ordered[-1],
+    }
+
+
+def _normalize_parent_id(parent_id: int | None) -> int | None:
+    if parent_id is None or parent_id == 0:
+        return None
+    return parent_id
+
+
+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)
+
+
+def normalize_str_list(raw: Any, field_name: str = "items") -> tuple[list[str], str | None]:
+    """将 JSON 文本/列表/单个字符串统一解析为非空 str 列表（去重保序）。"""
+    if raw is None:
+        return [], f"{field_name} 不能为空"
+
+    items: list[Any]
+    if isinstance(raw, str):
+        text = raw.strip()
+        if not text:
+            return [], f"{field_name} 不能为空"
+        try:
+            parsed = json.loads(text)
+            items = list(parsed) if isinstance(parsed, list) else [text]
+        except (ValueError, TypeError):
+            items = [text]
+    elif isinstance(raw, (list, tuple)):
+        items = list(raw)
+    else:
+        items = [raw]
+
+    out: list[str] = []
+    seen: set[str] = set()
+    for item in items:
+        text = str(item).strip() if item is not None else ""
+        if not text or text in seen:
+            continue
+        seen.add(text)
+        out.append(text)
+
+    if not out:
+        return [], f"{field_name} 不能为空"
+    return out, None
+
+
+def parse_int_list(raw: Any) -> list[int]:
+    """将 JSON 文本/列表统一解析为 int 列表，解析失败返回空列表。"""
+    if raw is None:
+        return []
+    if isinstance(raw, str):
+        try:
+            raw = json.loads(raw)
+        except (ValueError, TypeError):
+            return []
+    if not isinstance(raw, list):
+        return []
+    out: list[int] = []
+    for item in raw:
+        try:
+            out.append(int(item))
+        except (TypeError, ValueError):
+            continue
+    return out
+
+
+def dump_int_list(values: list[int] | None) -> str | None:
+    """将 int 列表序列化为 JSON 文本，空列表/None 返回 None。"""
+    if not values:
+        return None
+    return json.dumps(values, ensure_ascii=False)
+
+
+def _parse_video_ids(raw: Any) -> list[str]:
+    """解析单个 multi_demand_pool_di.video_list（JSON数组/逗号分隔文本）为 vid 字符串列表。"""
+    if raw is None:
+        return []
+    items: list[Any]
+    if isinstance(raw, str):
+        text = raw.strip()
+        if not text:
+            return []
+        try:
+            parsed = json.loads(text)
+            items = list(parsed) if isinstance(parsed, list) else [text]
+        except (ValueError, TypeError):
+            items = [part.strip() for part in text.split(",") if part.strip()]
+    elif isinstance(raw, (list, tuple)):
+        items = list(raw)
+    else:
+        return []
+    return [str(v).strip() for v in items if v is not None and str(v).strip()]
+
+
+def merge_video_ids(pool_rows: list[MultiDemandPoolDi], limit: int = _VIDEO_LIST_LIMIT) -> str | None:
+    """合并多条原始需求行的 video_list，去重保序，最多取前 limit 个，返回 JSON 文本或 None。"""
+    merged: list[str] = []
+    seen: set[str] = set()
+    for row in pool_rows:
+        for vid in _parse_video_ids(row.video_list):
+            if vid not in seen:
+                seen.add(vid)
+                merged.append(vid)
+    if not merged:
+        return None
+    return json.dumps(merged[:limit], ensure_ascii=False)
+
+
+def collect_strategies(pool_rows: list[MultiDemandPoolDi]) -> str | None:
+    """收集多条原始需求行的策略名，去重排序，返回 JSON 文本或 None。"""
+    strategies = sorted({row.strategy.strip() for row in pool_rows if row.strategy and row.strategy.strip()})
+    if not strategies:
+        return None
+    return json.dumps(strategies, ensure_ascii=False)

api/app.py

@@ -9,6 +9,8 @@ from fastapi.staticfiles import StaticFiles
 
 from api.services.category_tree import build_category_tree
 from api.services.demand_belong_category import list_demand_belong_categories
+from api.services.demand_grade import list_demand_grades
+from api.services.demand_grade_videos import list_videos_for_demand_grade
 from api.services.demand_videos import list_videos_for_demand_belong
 from api.services.oss_logs import list_demand_belong_oss_logs
 
@@ -60,6 +62,27 @@ def demand_belong_videos(belong_id: int) -> dict:
     return result
 
 
+@app.get("/api/demand-grade")
+def demand_grade(
+    biz_dt: str | None = Query(
+        default=None,
+        description="业务日 YYYYMMDD；省略则取 demand_grade 最新一日",
+    ),
+) -> dict:
+    """Return demand_grade rows (one per category_id) for the given/latest biz_dt."""
+    items = list_demand_grades(biz_dt=biz_dt)
+    return {"items": items}
+
+
+@app.get("/api/demand-grade/{demand_grade_id}/videos")
+def demand_grade_videos(demand_grade_id: int) -> dict:
+    """Return videos linked to a demand_grade row (vid + title + points JSON)."""
+    result = list_videos_for_demand_grade(demand_grade_id)
+    if result is None:
+        raise HTTPException(status_code=404, detail="demand_grade 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."""

api/services/demand_grade.py

@@ -0,0 +1,25 @@
+"""Load demand_grade rows (expanded one row per category) for the category tree UI."""
+from __future__ import annotations
+
+from typing import Any
+
+from supply_infra.db.repositories.demand_grade_category_rel_repo import (
+    DemandGradeCategoryRelRepository,
+)
+from supply_infra.db.repositories.demand_grade_repo import DemandGradeRepository
+from supply_infra.db.session import get_session
+
+
+def list_demand_grades(biz_dt: str | None = None) -> list[dict[str, Any]]:
+    """
+    Return demand_grade rows joined with demand_grade_category_rel, one row per
+    (demand_grade, category_id) pair — mirrors the shape the frontend already
+    groups by category_id.
+
+    biz_dt defaults to the latest business date present in demand_grade.
+    """
+    with get_session() as session:
+        resolved_biz_dt = biz_dt or DemandGradeRepository(session).get_latest_biz_dt()
+        if not resolved_biz_dt:
+            return []
+        return DemandGradeCategoryRelRepository(session).list_items_with_category(resolved_biz_dt)

api/services/demand_grade_videos.py

@@ -0,0 +1,63 @@
+"""Resolve demand_grade.video_list → multi_demand_video_detail."""
+from __future__ import annotations
+
+import json
+from typing import Any
+
+from supply_infra.db.repositories.demand_grade_repo import DemandGradeRepository
+from supply_infra.db.repositories.multi_demand_video_detail_repo import (
+    MultiDemandVideoDetailRepository,
+)
+from supply_infra.db.session import get_session
+
+
+def _parse_json_list(raw: str | None) -> list[Any]:
+    if not raw:
+        return []
+    try:
+        parsed = json.loads(raw)
+    except json.JSONDecodeError:
+        return []
+    return parsed if isinstance(parsed, list) else []
+
+
+def _parse_video_ids(raw: str | None) -> list[str]:
+    return [str(v).strip() for v in _parse_json_list(raw) if v is not None and str(v).strip()]
+
+
+def list_videos_for_demand_grade(demand_grade_id: int) -> dict[str, Any] | None:
+    """
+    按 demand_grade.id 返回关联视频详情。
+
+    顺序与 video_list 一致；详情表缺失的 vid 仍返回，title/三点 JSON 为空。
+    category_ids 返回该分级需求归属的全部全局树节点 id（可能不止一个）。
+    """
+    with get_session() as session:
+        grade = DemandGradeRepository(session).get_by_id(demand_grade_id)
+        if grade is None:
+            return None
+
+        vids = _parse_video_ids(grade.video_list)
+        details = MultiDemandVideoDetailRepository(session).list_by_vids(vids)
+
+        videos: list[dict[str, Any]] = []
+        for vid in vids:
+            row = details.get(vid)
+            videos.append(
+                {
+                    "vid": vid,
+                    "title": row.title if row else None,
+                    "inspiration_points_json": row.inspiration_points_json if row else None,
+                    "purpose_points_json": row.purpose_points_json if row else None,
+                    "key_points_json": row.key_points_json if row else None,
+                }
+            )
+
+        return {
+            "demand_grade_id": grade.id,
+            "demand_name": grade.demand_name,
+            "category_ids": [int(c) for c in _parse_json_list(grade.category_ids)],
+            "grade": grade.grade,
+            "strategies": _parse_json_list(grade.strategies),
+            "videos": videos,
+        }

jobs/grade_demand_pool.py

@@ -0,0 +1,46 @@
+#!/usr/bin/env python3
+"""手动执行需求分级：循环取批次调用 demand_grade_agent，直到需求池分级完毕。
+
+用法:
+    python jobs/grade_demand_pool.py                      # 默认业务日、批次20、最多200批
+    python jobs/grade_demand_pool.py 20260716              # 指定业务日
+    python jobs/grade_demand_pool.py 20260716 20           # 指定业务日 + 批次大小
+    python jobs/grade_demand_pool.py 20260716 20 5          # 再指定本次最多跑 5 批（用于试跑）
+"""
+from __future__ import annotations
+
+import logging
+import sys
+
+from supply_infra.scheduler.jobs.grade_demand_pool import grade_demand_pool
+
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
+)
+
+
+def main(
+    biz_dt: str | None = None,
+    batch_size_arg: str | None = None,
+    max_batches_arg: str | None = None,
+) -> dict:
+    batch_size = int(batch_size_arg) if batch_size_arg else 20
+    if max_batches_arg is None:
+        max_batches: int | None = 200
+    elif max_batches_arg.lower() in {"all", "0", "-1"}:
+        max_batches = None
+    else:
+        max_batches = int(max_batches_arg)
+
+    result = grade_demand_pool(biz_dt, batch_size=batch_size, max_batches=max_batches)
+    print(result)
+    return result
+
+
+if __name__ == "__main__":
+    main(
+        sys.argv[1] if len(sys.argv) > 1 else None,
+        sys.argv[2] if len(sys.argv) > 2 else None,
+        sys.argv[3] if len(sys.argv) > 3 else None,
+    )

supply_infra/db/models/__init__.py

@@ -3,6 +3,8 @@
 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_grade import DemandGrade
+from supply_infra.db.models.demand_grade_category_rel import DemandGradeCategoryRel
 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
@@ -15,6 +17,8 @@ __all__ = [
     "CategoryTreeWeight",
     "DemandBelongCategory",
     "DemandBelongPoolRel",
+    "DemandGrade",
+    "DemandGradeCategoryRel",
     "DemandPopularityStats",
     "GeneratedDemand",
     "GlobalTreeCategory",

supply_infra/db/models/demand_grade.py

@@ -0,0 +1,68 @@
+from __future__ import annotations
+
+from datetime import datetime
+from decimal import Decimal
+
+from sqlalchemy import BigInteger, Integer, Numeric, String, Text, UniqueConstraint, func
+from sqlalchemy.orm import Mapped, mapped_column
+
+from supply_infra.db.base import Base
+
+
+class DemandGrade(Base):
+    """需求分级结果 — 对 multi_demand_pool_di 中的需求按先验/后验热度评级。"""
+
+    __tablename__ = "demand_grade"
+    __table_args__ = (UniqueConstraint("biz_dt", "demand_name", name="uk_demand_grade"),)
+
+    id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
+    biz_dt: Mapped[str] = mapped_column(String(32), nullable=False, comment="业务日期YYYYMMDD")
+    demand_name: Mapped[str] = mapped_column(
+        String(256), nullable=False, comment="需求名称（去重后的代表词）"
+    )
+    category_ids: Mapped[str | None] = mapped_column(
+        Text, nullable=True, comment="归属的全局树节点id列表JSON数组（展示快照，权威关系见demand_grade_category_rel）"
+    )
+    grade: Mapped[str] = mapped_column(String(4), nullable=False, comment="等级 S/A/B/C/D")
+    score: Mapped[Decimal | None] = mapped_column(
+        Numeric(6, 2), nullable=True, comment="数值分（可选，辅助同级排序）"
+    )
+    prior_total_score: Mapped[Decimal | None] = mapped_column(
+        Numeric(16, 8), nullable=True, comment="落库时的先验 total_score 快照"
+    )
+    posterior_rov_avg: Mapped[Decimal | None] = mapped_column(
+        Numeric(16, 8), nullable=True, comment="落库时的后验 real_rov_7d_avg 快照"
+    )
+    posterior_rov_count: Mapped[int] = mapped_column(
+        Integer, nullable=False, default=0, comment="落库时的后验样本数快照"
+    )
+    has_posterior: Mapped[int] = mapped_column(
+        Integer, nullable=False, default=0, comment="是否有后验验证数据 0-无 1-有"
+    )
+    related_pool_ids: Mapped[str] = mapped_column(
+        Text,
+        nullable=False,
+        comment="关联的原始 multi_demand_pool_di.id 列表JSON数组（必填，用于回溯原始需求）",
+    )
+    video_list: Mapped[str | None] = mapped_column(
+        Text,
+        nullable=True,
+        comment="关联视频列表JSON（最多10个，由related_pool_ids对应原始行的video_list合并去重得到）",
+    )
+    strategies: Mapped[str | None] = mapped_column(
+        Text,
+        nullable=True,
+        comment="来源策略列表JSON数组（由related_pool_ids对应原始行的strategy去重得到）",
+    )
+    reason: Mapped[str] = mapped_column(Text, nullable=False, comment="判断依据")
+    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/models/demand_grade_category_rel.py

@@ -0,0 +1,42 @@
+from __future__ import annotations
+
+from datetime import datetime
+
+from sqlalchemy import BigInteger, Index, UniqueConstraint, func
+from sqlalchemy.orm import Mapped, mapped_column
+
+from supply_infra.db.base import Base
+
+
+class DemandGradeCategoryRel(Base):
+    """需求分级结果与全局树节点的归属关系，支撑按分类高效查询已分级需求。"""
+
+    __tablename__ = "demand_grade_category_rel"
+    __table_args__ = (
+        UniqueConstraint(
+            "demand_grade_id",
+            "category_id",
+            name="uk_demand_grade_category",
+        ),
+        Index("idx_demand_grade_id", "demand_grade_id"),
+        Index("idx_category_id", "category_id"),
+    )
+
+    id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
+    demand_grade_id: Mapped[int] = mapped_column(
+        BigInteger, nullable=False, comment="demand_grade.id"
+    )
+    category_id: Mapped[int] = mapped_column(
+        BigInteger, nullable=False, comment="global_tree_category.id"
+    )
+    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

@@ -10,6 +10,10 @@ from supply_infra.db.repositories.demand_belong_category_repo import (
 from supply_infra.db.repositories.demand_belong_pool_rel_repo import (
     DemandBelongPoolRelRepository,
 )
+from supply_infra.db.repositories.demand_grade_category_rel_repo import (
+    DemandGradeCategoryRelRepository,
+)
+from supply_infra.db.repositories.demand_grade_repo import DemandGradeRepository
 from supply_infra.db.repositories.demand_popularity_stats_repo import (
     DemandPopularityStatsRepository,
 )
@@ -27,6 +31,8 @@ __all__ = [
     "CategoryTreeWeightRepository",
     "DemandBelongCategoryRepository",
     "DemandBelongPoolRelRepository",
+    "DemandGradeCategoryRelRepository",
+    "DemandGradeRepository",
     "DemandPopularityStatsRepository",
     "GeneratedDemandRepository",
     "GlobalTreeCategoryRepository",

supply_infra/db/repositories/category_tree_weight_repo.py

@@ -52,6 +52,38 @@ class CategoryTreeWeightRepository(BaseRepository[CategoryTreeWeight]):
         stmt = select(CategoryTreeWeight).where(CategoryTreeWeight.biz_dt == biz_dt)
         return list(self.session.scalars(stmt).all())
 
+    def get_by_category_ids(
+        self,
+        category_ids: list[int],
+        biz_dt: str | None = None,
+    ) -> list[CategoryTreeWeight]:
+        """按 category_id 列表查询权重行；未指定 biz_dt 时取每个节点各自的最新业务日。"""
+        if not category_ids:
+            return []
+
+        if biz_dt:
+            stmt = select(CategoryTreeWeight).where(
+                CategoryTreeWeight.category_id.in_(category_ids),
+                CategoryTreeWeight.biz_dt == biz_dt,
+            )
+            return list(self.session.scalars(stmt).all())
+
+        latest_dt_subq = (
+            select(
+                CategoryTreeWeight.category_id,
+                func.max(CategoryTreeWeight.biz_dt).label("max_biz_dt"),
+            )
+            .where(CategoryTreeWeight.category_id.in_(category_ids))
+            .group_by(CategoryTreeWeight.category_id)
+            .subquery()
+        )
+        stmt = select(CategoryTreeWeight).join(
+            latest_dt_subq,
+            (CategoryTreeWeight.category_id == latest_dt_subq.c.category_id)
+            & (CategoryTreeWeight.biz_dt == latest_dt_subq.c.max_biz_dt),
+        )
+        return list(self.session.scalars(stmt).all())
+
     def get_by_category_biz_dt(
         self, category_id: int, biz_dt: str
     ) -> CategoryTreeWeight | None:

supply_infra/db/repositories/demand_belong_category_repo.py

@@ -16,6 +16,28 @@ class DemandBelongCategoryRepository(BaseRepository[DemandBelongCategory]):
 
     model = DemandBelongCategory
 
+    def get_by_ids(self, ids: Iterable[int]) -> list[DemandBelongCategory]:
+        """按 id 批量查询，跳过已软删除记录。"""
+        id_list = [int(i) for i in ids]
+        if not id_list:
+            return []
+        stmt = select(DemandBelongCategory).where(
+            DemandBelongCategory.id.in_(id_list),
+            DemandBelongCategory.is_delete == 0,
+        )
+        return list(self.session.scalars(stmt).all())
+
+    def search_by_name_like(self, keyword: str) -> list[DemandBelongCategory]:
+        """按名称模糊匹配（双向包含关系不在此处理，仅 LIKE %keyword%），跳过已软删除记录。"""
+        keyword = (keyword or "").strip()
+        if not keyword:
+            return []
+        stmt = select(DemandBelongCategory).where(
+            DemandBelongCategory.name.like(f"%{keyword}%"),
+            DemandBelongCategory.is_delete == 0,
+        )
+        return list(self.session.scalars(stmt).all())
+
     def get_existing_names(self, names: Iterable[str]) -> set[str]:
         """返回 names 中已存在于表内的名称（含软删除）。"""
         name_list = [n for n in names if n]

supply_infra/db/repositories/demand_belong_pool_rel_repo.py

@@ -18,6 +18,23 @@ class DemandBelongPoolRelRepository(BaseRepository[DemandBelongPoolRel]):
 
     model = DemandBelongPoolRel
 
+    def get_belong_ids_by_pool_ids(self, pool_ids: Iterable[int]) -> dict[int, list[int]]:
+        """反查池表行归属的需求归属分类 id：pool_id -> [belong_id, ...]。"""
+        id_list = [int(p) for p in pool_ids]
+        if not id_list:
+            return {}
+
+        result: dict[int, list[int]] = {}
+        for i in range(0, len(id_list), _BATCH_SIZE):
+            batch = id_list[i : i + _BATCH_SIZE]
+            stmt = select(
+                DemandBelongPoolRel.multi_demand_pool_di_id,
+                DemandBelongPoolRel.demand_belong_category_id,
+            ).where(DemandBelongPoolRel.multi_demand_pool_di_id.in_(batch))
+            for pool_id, belong_id in self.session.execute(stmt).all():
+                result.setdefault(int(pool_id), []).append(int(belong_id))
+        return result
+
     def get_existing_pairs(self, pairs: Iterable[RelPair]) -> set[RelPair]:
         """返回 pairs 中已存在的 (belong_id, pool_id)。"""
         pair_list = [(int(b), int(p)) for b, p in pairs]

supply_infra/db/repositories/demand_grade_category_rel_repo.py

@@ -0,0 +1,62 @@
+from __future__ import annotations
+
+from sqlalchemy import delete, select
+from sqlalchemy.dialects.mysql import insert
+
+from supply_infra.db.models.demand_grade import DemandGrade
+from supply_infra.db.models.demand_grade_category_rel import DemandGradeCategoryRel
+from supply_infra.db.repositories.base import BaseRepository
+
+
+class DemandGradeCategoryRelRepository(BaseRepository[DemandGradeCategoryRel]):
+    """demand_grade ↔ global_tree_category 归属关系 — 支撑按分类高效查询已分级需求。"""
+
+    model = DemandGradeCategoryRel
+
+    def replace_for_demand_grade(self, demand_grade_id: int, category_ids: list[int]) -> None:
+        """覆盖某个 demand_grade 的分类映射：先删除旧的，再写入新的（空列表则只删）。"""
+        self.session.execute(
+            delete(DemandGradeCategoryRel).where(
+                DemandGradeCategoryRel.demand_grade_id == demand_grade_id
+            )
+        )
+        unique_ids = sorted({int(c) for c in category_ids if c is not None})
+        if not unique_ids:
+            return
+        rows = [
+            {"demand_grade_id": demand_grade_id, "category_id": category_id}
+            for category_id in unique_ids
+        ]
+        stmt = insert(DemandGradeCategoryRel).values(rows).prefix_with("IGNORE")
+        self.session.execute(stmt)
+
+    def list_items_with_category(self, biz_dt: str) -> list[dict]:
+        """按 biz_dt 展开为一行一个 category_id，供前端按分类展示需求列表使用。"""
+        stmt = (
+            select(
+                DemandGrade.id,
+                DemandGrade.demand_name,
+                DemandGradeCategoryRel.category_id,
+                DemandGrade.grade,
+                DemandGrade.score,
+                DemandGrade.reason,
+                DemandGrade.strategies,
+                DemandGrade.biz_dt,
+            )
+            .join(DemandGradeCategoryRel, DemandGradeCategoryRel.demand_grade_id == DemandGrade.id)
+            .where(DemandGrade.biz_dt == biz_dt)
+            .order_by(DemandGrade.grade, DemandGrade.demand_name)
+        )
+        return [
+            {
+                "id": int(row.id),
+                "demand_name": row.demand_name,
+                "category_id": int(row.category_id),
+                "grade": row.grade,
+                "score": float(row.score) if row.score is not None else None,
+                "reason": row.reason,
+                "strategies": row.strategies,
+                "biz_dt": row.biz_dt,
+            }
+            for row in self.session.execute(stmt).all()
+        ]

supply_infra/db/repositories/demand_grade_repo.py

@@ -0,0 +1,87 @@
+from __future__ import annotations
+
+from collections.abc import Iterable
+
+from sqlalchemy import func, select
+from sqlalchemy.dialects.mysql import insert
+
+from supply_infra.db.models.demand_grade import DemandGrade
+from supply_infra.db.repositories.base import BaseRepository
+
+_BATCH_SIZE = 500
+
+_UPSERT_COLUMNS = (
+    "category_ids",
+    "grade",
+    "score",
+    "prior_total_score",
+    "posterior_rov_avg",
+    "posterior_rov_count",
+    "has_posterior",
+    "related_pool_ids",
+    "video_list",
+    "strategies",
+    "reason",
+)
+
+
+class DemandGradeRepository(BaseRepository[DemandGrade]):
+    """需求分级结果表 — 按 (biz_dt, demand_name) 增量/更新写入。"""
+
+    model = DemandGrade
+
+    def get_existing_demand_names(self, biz_dt: str, names: Iterable[str] | None = None) -> set[str]:
+        """返回指定业务日已分级的需求名集合；传入 names 时只在其中查交集。"""
+        stmt = select(DemandGrade.demand_name).where(DemandGrade.biz_dt == biz_dt)
+        if names is not None:
+            name_list = [n for n in names if n]
+            if not name_list:
+                return set()
+            stmt = stmt.where(DemandGrade.demand_name.in_(name_list))
+        return {n for n in self.session.scalars(stmt).all() if n}
+
+    def count_by_biz_dt(self, biz_dt: str) -> int:
+        """统计指定业务日已分级的需求数。"""
+        stmt = select(func.count()).select_from(DemandGrade).where(DemandGrade.biz_dt == biz_dt)
+        return int(self.session.scalar(stmt) or 0)
+
+    def list_by_biz_dt(self, biz_dt: str) -> list[DemandGrade]:
+        """返回指定业务日的全部分级结果，按等级、需求名排序。"""
+        stmt = (
+            select(DemandGrade)
+            .where(DemandGrade.biz_dt == biz_dt)
+            .order_by(DemandGrade.grade, DemandGrade.demand_name)
+        )
+        return list(self.session.scalars(stmt).all())
+
+    def get_latest_biz_dt(self) -> str | None:
+        """返回 demand_grade 中最新的业务日期。"""
+        stmt = select(func.max(DemandGrade.biz_dt))
+        return self.session.scalar(stmt)
+
+    def get_ids_by_names(self, biz_dt: str, names: Iterable[str]) -> dict[str, int]:
+        """按 (biz_dt, demand_name) 反查 id，供 upsert 后写关联表使用。"""
+        name_list = [n for n in names if n]
+        if not name_list:
+            return {}
+        stmt = select(DemandGrade.demand_name, DemandGrade.id).where(
+            DemandGrade.biz_dt == biz_dt,
+            DemandGrade.demand_name.in_(name_list),
+        )
+        return {name: int(id_) for name, id_ in self.session.execute(stmt).all()}
+
+    def bulk_upsert(self, rows: list[dict]) -> int:
+        """按 (biz_dt, demand_name) 批量 upsert。"""
+        if not rows:
+            return 0
+
+        affected = 0
+        for i in range(0, len(rows), _BATCH_SIZE):
+            batch = rows[i : i + _BATCH_SIZE]
+            stmt = insert(DemandGrade).values(batch)
+            stmt = stmt.on_duplicate_key_update(
+                **{col: stmt.inserted[col] for col in _UPSERT_COLUMNS}
+            )
+            result = self.session.execute(stmt)
+            affected += result.rowcount or 0
+        return affected

supply_infra/db/repositories/demand_popularity_stats_repo.py

@@ -47,6 +47,24 @@ class DemandPopularityStatsRepository(BaseRepository[DemandPopularityStats]):
         stmt = select(DemandPopularityStats).where(DemandPopularityStats.biz_dt == biz_dt)
         return list(self.session.scalars(stmt).all())
 
+    def search_by_word_name(
+        self,
+        keyword: str,
+        biz_dt: str | None = None,
+    ) -> list[DemandPopularityStats]:
+        """按 demand_word_name 精确+模糊搜索；未指定 biz_dt 时不限日期，按 biz_dt 降序返回。"""
+        keyword = (keyword or "").strip()
+        if not keyword:
+            return []
+
+        stmt = select(DemandPopularityStats).where(
+            DemandPopularityStats.demand_word_name.like(f"%{keyword}%")
+        )
+        if biz_dt:
+            stmt = stmt.where(DemandPopularityStats.biz_dt == biz_dt)
+        stmt = stmt.order_by(DemandPopularityStats.biz_dt.desc())
+        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]:

supply_infra/db/repositories/multi_demand_pool_di_repo.py

@@ -17,6 +17,121 @@ class MultiDemandPoolDiRepository(BaseRepository[MultiDemandPoolDi]):
 
     model = MultiDemandPoolDi
 
+    def get_latest_biz_dt(self) -> str | None:
+        """返回需求池中最新业务日；无数据时返回 None。"""
+        stmt = select(func.max(MultiDemandPoolDi.biz_dt))
+        return self.session.scalar(stmt)
+
+    def list_distinct_demand_name_summaries(
+        self,
+        biz_dt: str,
+        *,
+        limit: int = 50,
+        offset: int = 0,
+        exclude_names: list[str] | None = None,
+    ) -> list[dict]:
+        """
+        按业务日分页列出去重需求词及聚合统计。
+
+        返回每项包含：demand_name、row_count（出现行数）、strategies（策略列表）、
+        total_weight、total_video_count、max_real_rov_7d、max_real_vov_7d。
+        按 max_real_rov_7d 降序、total_weight 降序排列，优先暴露有后验数据/高权重的需求词。
+        """
+        stmt = (
+            select(
+                MultiDemandPoolDi.demand_name,
+                func.count().label("row_count"),
+                func.group_concat(MultiDemandPoolDi.strategy.distinct()).label("strategies"),
+                func.sum(MultiDemandPoolDi.weight).label("total_weight"),
+                func.sum(MultiDemandPoolDi.video_count).label("total_video_count"),
+                func.max(MultiDemandPoolDi.real_rov_7d).label("max_real_rov_7d"),
+                func.max(MultiDemandPoolDi.real_vov_7d).label("max_real_vov_7d"),
+            )
+            .where(MultiDemandPoolDi.biz_dt == biz_dt)
+        )
+        if exclude_names:
+            stmt = stmt.where(MultiDemandPoolDi.demand_name.notin_(exclude_names))
+        stmt = (
+            stmt.group_by(MultiDemandPoolDi.demand_name)
+            .order_by(
+                func.max(MultiDemandPoolDi.real_rov_7d).desc(),
+                func.sum(MultiDemandPoolDi.weight).desc(),
+            )
+            .limit(limit)
+            .offset(offset)
+        )
+
+        rows = self.session.execute(stmt).all()
+        return [
+            {
+                "demand_name": name,
+                "row_count": int(row_count or 0),
+                "strategies": (strategies or "").split(",") if strategies else [],
+                "total_weight": float(total_weight) if total_weight is not None else None,
+                "total_video_count": int(total_video_count) if total_video_count is not None else None,
+                "max_real_rov_7d": float(max_rov) if max_rov is not None else None,
+                "max_real_vov_7d": float(max_vov) if max_vov is not None else None,
+            }
+            for name, row_count, strategies, total_weight, total_video_count, max_rov, max_vov in rows
+        ]
+
+    def count_distinct_demand_names(self, biz_dt: str) -> int:
+        """统计指定业务日去重需求词总数。"""
+        stmt = select(func.count(func.distinct(MultiDemandPoolDi.demand_name))).where(
+            MultiDemandPoolDi.biz_dt == biz_dt
+        )
+        return int(self.session.scalar(stmt) or 0)
+
+    def search_rows_by_name_fragment(self, biz_dt: str, keyword: str) -> list[dict]:
+        """
+        按业务日 + 需求名双向包含关系搜索明细行。
+
+        匹配 demand_name LIKE %keyword% 或 keyword LIKE %demand_name%（互相包含），
+        用于把同语义、措辞不同的需求词合并到一起判断。返回按 demand_name 去重后的明细。
+        """
+        keyword = (keyword or "").strip()
+        if not keyword:
+            return []
+
+        stmt = select(MultiDemandPoolDi).where(
+            MultiDemandPoolDi.biz_dt == biz_dt,
+            MultiDemandPoolDi.demand_name.like(f"%{keyword}%"),
+        )
+        rows = list(self.session.scalars(stmt).all())
+
+        if len(keyword) >= 2:
+            broad_stmt = select(MultiDemandPoolDi).where(
+                MultiDemandPoolDi.biz_dt == biz_dt,
+            )
+            seen_ids = {int(r.id) for r in rows}
+            for row in self.session.scalars(broad_stmt).all():
+                if int(row.id) in seen_ids:
+                    continue
+                name = row.demand_name or ""
+                if name and name in keyword:
+                    rows.append(row)
+                    seen_ids.add(int(row.id))
+
+        return [
+            {
+                "id": int(row.id),
+                "demand_name": row.demand_name,
+                "strategy": row.strategy,
+                "weight": float(row.weight) if row.weight is not None else None,
+                "video_count": int(row.video_count) if row.video_count is not None else None,
+                "real_rov_7d": float(row.real_rov_7d) if row.real_rov_7d is not None else None,
+                "real_vov_7d": float(row.real_vov_7d) if row.real_vov_7d is not None else None,
+            }
+            for row in rows
+        ]
+
+    def get_by_ids(self, ids: list[int]) -> list[MultiDemandPoolDi]:
+        """按 id 批量查询完整行。"""
+        if not ids:
+            return []
+        stmt = select(MultiDemandPoolDi).where(MultiDemandPoolDi.id.in_(ids))
+        return list(self.session.scalars(stmt).all())
+
     def count_by_biz_dt(self, biz_dt: str) -> int:
         """统计指定业务日期去重行数（strategy + demand_id）。"""
         stmt = (

supply_infra/scheduler/app.py

@@ -6,6 +6,7 @@ from apscheduler.schedulers.blocking import BlockingScheduler
 from apscheduler.triggers.cron import CronTrigger
 
 from supply_infra.config import get_infra_settings
+from supply_infra.scheduler.jobs.grade_demand_pool import grade_demand_pool
 from supply_infra.scheduler.jobs.sync_global_tree_odps_to_mysql import sync_global_tree_odps_to_mysql
 from supply_infra.scheduler.jobs.sync_multi_demand_pool_odps_to_mysql import (
     sync_multi_demand_pool_odps_to_mysql,
@@ -37,6 +38,17 @@ def create_scheduler() -> BlockingScheduler:
         replace_existing=True,
     )
 
+    # 每天 13:30 对需求池循环打分级（在 12:00 同步/热度计算完成后运行）。
+    # 循环控制在 job 内部：每批取少量未分级需求调用一次 agent、等待完成后再取下一批，
+    # 避免单次 agent 会话内自行分页导致上下文无限增长；max_batches 作为单次运行的安全阀。
+    scheduler.add_job(
+        grade_demand_pool,
+        trigger=CronTrigger(hour=13, minute=30),
+        id="grade_demand_pool",
+        name="需求池分级评估",
+        replace_existing=True,
+    )
+
     logger.info("Scheduler configured with %d job(s)", len(scheduler.get_jobs()))
     return scheduler
 

supply_infra/scheduler/jobs/grade_demand_pool.py

@@ -0,0 +1,142 @@
+"""
+定时任务：对 multi_demand_pool_di 需求池中的需求循环打分级。
+
+循环控制在本文件（调度任务侧），而不是让 agent 在一次运行内自行分页遍历——
+每一批只挑选少量（默认 20 个）尚未分级的需求词，调用一次 demand_grade_agent
+并等待其运行结束（agent 内部会调用 batch_save_demand_grades 落库），再重新
+查询"已分级"集合、找出下一批新词，如此循环直到没有更多待分级需求或达到本次
+运行的批次上限，从而避免单次 agent 会话上下文无限增长。
+"""
+from __future__ import annotations
+
+import logging
+from datetime import datetime
+
+from agents.demand_grade_agent.run import main as grade_demand_words
+from supply_infra.db.repositories.demand_grade_repo import DemandGradeRepository
+from supply_infra.db.repositories.multi_demand_pool_di_repo import MultiDemandPoolDiRepository
+from supply_infra.db.session import get_session
+
+logger = logging.getLogger(__name__)
+
+_DEFAULT_BATCH_SIZE = 20
+_DEFAULT_MAX_BATCHES = 200
+
+
+def _resolve_biz_dt(biz_dt: str | None) -> str | None:
+    if biz_dt:
+        return biz_dt
+    with get_session() as session:
+        return MultiDemandPoolDiRepository(session).get_latest_biz_dt()
+
+
+def _fetch_next_batch(biz_dt: str, batch_size: int, graded_names: set[str]) -> list[str]:
+    """按 exclude_names 过滤已分级词，取下一批（不用 offset，避免和落库进度错位）。"""
+    with get_session() as session:
+        summaries = MultiDemandPoolDiRepository(session).list_distinct_demand_name_summaries(
+            biz_dt,
+            limit=batch_size,
+            offset=0,
+            exclude_names=list(graded_names) if graded_names else None,
+        )
+    return [item["demand_name"] for item in summaries]
+
+
+def _fetch_graded_names(biz_dt: str) -> set[str]:
+    with get_session() as session:
+        return DemandGradeRepository(session).get_existing_demand_names(biz_dt)
+
+
+def grade_demand_pool(
+    biz_dt: str | None = None,
+    *,
+    batch_size: int = _DEFAULT_BATCH_SIZE,
+    max_batches: int | None = _DEFAULT_MAX_BATCHES,
+) -> dict:
+    """
+    循环对需求池中尚未分级的需求打分级，每批调用一次 agent 并等待其完成。
+
+    Args:
+        biz_dt: 业务日期 YYYYMMDD；不传则取需求池最新业务日。
+        batch_size: 每批交给 agent 的需求词数量。
+        max_batches: 本次运行最多执行多少批（安全阀，避免单次运行时间过长/无限循环）；
+                     None 表示不限制，跑到没有待分级需求为止。
+
+    Returns:
+        运行统计：{"biz_dt", "total", "graded_before", "graded_after", "batches_run", "stopped_reason"}
+    """
+    resolved_biz_dt = _resolve_biz_dt(biz_dt)
+    if not resolved_biz_dt:
+        result = {"stopped_reason": "no_biz_dt", "batches_run": 0}
+        logger.warning("Grade demand pool: no biz_dt available, skip. result=%s", result)
+        return result
+
+    with get_session() as session:
+        total = MultiDemandPoolDiRepository(session).count_distinct_demand_names(resolved_biz_dt)
+
+    graded_names = _fetch_graded_names(resolved_biz_dt)
+    graded_before = len(graded_names)
+
+    logger.info(
+        "Grade demand pool start: biz_dt=%s total=%d graded=%d batch_size=%d max_batches=%s",
+        resolved_biz_dt,
+        total,
+        graded_before,
+        batch_size,
+        max_batches,
+    )
+
+    batches_run = 0
+    stopped_reason = "no_more_pending"
+
+    while True:
+        if max_batches is not None and batches_run >= max_batches:
+            stopped_reason = "max_batches_reached"
+            break
+
+        batch = _fetch_next_batch(resolved_biz_dt, batch_size, graded_names)
+        if not batch:
+            stopped_reason = "no_more_pending"
+            break
+
+        batches_run += 1
+        logger.info(
+            "Grade demand pool batch %d (size=%d): %s",
+            batches_run,
+            len(batch),
+            batch,
+        )
+
+        try:
+            grade_demand_words(batch, biz_dt=resolved_biz_dt)
+        except Exception as e:
+            logger.error(
+                "Grade demand pool batch %d failed: %s", batches_run, e, exc_info=True
+            )
+            stopped_reason = "batch_failed"
+            break
+
+        new_graded_names = _fetch_graded_names(resolved_biz_dt)
+        if len(new_graded_names) <= len(graded_names):
+            # agent 没有对这批产生任何新的落库结果，避免死循环重复拿到同一批
+            logger.warning(
+                "Grade demand pool batch %d made no progress (graded still %d), stop",
+                batches_run,
+                len(graded_names),
+            )
+            stopped_reason = "stalled"
+            graded_names = new_graded_names
+            break
+        graded_names = new_graded_names
+
+    result = {
+        "biz_dt": resolved_biz_dt,
+        "total": total,
+        "graded_before": graded_before,
+        "graded_after": len(graded_names),
+        "batches_run": batches_run,
+        "stopped_reason": stopped_reason,
+        "run_at": datetime.now().isoformat(),
+    }
+    logger.info("Grade demand pool completed: %s", result)
+    return result

web/src/api/demand.ts

@@ -1,17 +1,18 @@
-import type { DemandBelongResponse, DemandVideosResponse } from '../types/demand'
+import type { DemandGradeResponse, DemandGradeVideosResponse } from '../types/demand'
 
-export async function fetchDemandBelongCategory(): Promise<DemandBelongResponse> {
-  const res = await fetch('/api/demand-belong-category')
+export async function fetchDemandGrade(bizDt?: string | null): Promise<DemandGradeResponse> {
+  const query = bizDt ? `?biz_dt=${encodeURIComponent(bizDt)}` : ''
+  const res = await fetch(`/api/demand-grade${query}`)
   if (!res.ok) {
-    throw new Error(`加载需求归属失败: ${res.status} ${res.statusText}`)
+    throw new Error(`加载需求分级失败: ${res.status} ${res.statusText}`)
   }
   return res.json()
 }
 
-export async function fetchDemandBelongVideos(
-  belongId: number,
-): Promise<DemandVideosResponse> {
-  const res = await fetch(`/api/demand-belong-category/${belongId}/videos`)
+export async function fetchDemandGradeVideos(
+  demandGradeId: number,
+): Promise<DemandGradeVideosResponse> {
+  const res = await fetch(`/api/demand-grade/${demandGradeId}/videos`)
   if (!res.ok) {
     throw new Error(`加载关联视频失败: ${res.status} ${res.statusText}`)
   }

web/src/components/CategoryTree.vue

@@ -9,7 +9,7 @@ import {
   filterTreeByDim,
   maxTreeDepth,
 } from '../types/category'
-import type { DemandBelongItem, DemandsByCategory } from '../types/demand'
+import type { DemandGradeItem, DemandsByCategory } from '../types/demand'
 import { downloadCategoryTreeHtml } from '../utils/exportCategoryTreeHtml'
 
 const FULL_TREE_KEY = 'full' as const
@@ -65,7 +65,7 @@ const collapsedMap = ref<Record<number, true>>({})
 const drawerOpen = ref(false)
 const drawerCategoryId = ref<number | null>(null)
 const drawerCategoryName = ref('')
-const drawerItems = ref<DemandBelongItem[]>([])
+const drawerItems = ref<DemandGradeItem[]>([])
 
 const treePanelRef = ref<HTMLElement | null>(null)
 const isPanning = ref(false)
@@ -148,7 +148,7 @@ function expandAll() {
 function onInspect(payload: {
   categoryId: number
   categoryName: string
-  items: DemandBelongItem[]
+  items: DemandGradeItem[]
 }) {
   if (drawerOpen.value && drawerCategoryId.value === payload.categoryId) {
     closeDrawer()

web/src/components/DemandPathPanel.vue

@@ -1,12 +1,13 @@
 <script setup lang="ts">
 import { computed, nextTick, ref, watch } from 'vue'
-import { fetchDemandBelongVideos } from '../api/demand'
-import type { DemandBelongItem, DemandVideoItem } from '../types/demand'
+import { fetchDemandGradeVideos } from '../api/demand'
+import type { DemandGradeItem, DemandVideoItem } from '../types/demand'
+import { gradeRank, parseStrategies } from '../types/demand'
 
 const props = defineProps<{
   open: boolean
   categoryName: string
-  items: DemandBelongItem[]
+  items: DemandGradeItem[]
   /** When set, prepend a "当前节点" column. */
   nodeName?: string | null
   nodePath?: string | null
@@ -27,6 +28,12 @@ const selectedDemand = computed(
   () => props.items.find((item) => item.id === selectedDemandId.value) ?? null,
 )
 
+const selectedStrategies = computed(() => parseStrategies(selectedDemand.value?.strategies))
+
+const sortedItems = computed(() =>
+  [...props.items].sort((a, b) => gradeRank(a.grade) - gradeRank(b.grade)),
+)
+
 const selectedVideo = computed(
   () => videos.value.find((v) => v.vid === selectedVid.value) ?? null,
 )
@@ -103,7 +110,12 @@ watch(
   },
 )
 
-async function selectDemand(item: DemandBelongItem) {
+function gradeClass(grade: string | null | undefined): string {
+  const g = (grade || '').toUpperCase()
+  return g ? `grade-${g}` : 'grade-none'
+}
+
+async function selectDemand(item: DemandGradeItem) {
   if (selectedDemandId.value === item.id) return
   selectedDemandId.value = item.id
   selectedVid.value = null
@@ -111,7 +123,7 @@ async function selectDemand(item: DemandBelongItem) {
   videosError.value = null
   videosLoading.value = true
   try {
-    const res = await fetchDemandBelongVideos(item.id)
+    const res = await fetchDemandGradeVideos(item.id)
     videos.value = res.videos ?? []
   } catch (e) {
     videosError.value = e instanceof Error ? e.message : String(e)
@@ -170,17 +182,27 @@ function selectVideo(video: DemandVideoItem) {
           ×
         </button>
       </div>
-      <div v-if="items.length" class="card-list">
+      <div v-if="sortedItems.length" class="card-list">
         <button
-          v-for="item in items"
+          v-for="item in sortedItems"
           :key="item.id"
           type="button"
           class="card demand-card"
           :class="{ active: selectedDemandId === item.id }"
           @click="selectDemand(item)"
         >
-          <span class="card-label">需求词</span>
-          <span class="card-value">{{ item.name || '—' }}</span>
+          <div class="demand-card-head">
+            <span class="card-label">需求词</span>
+            <span class="grade-badge" :class="gradeClass(item.grade)">{{ item.grade || '—' }}</span>
+          </div>
+          <span class="card-value">{{ item.demand_name || '—' }}</span>
+          <div v-if="parseStrategies(item.strategies).length" class="strategy-tags">
+            <span
+              v-for="s in parseStrategies(item.strategies)"
+              :key="s"
+              class="strategy-tag"
+            >{{ s }}</span>
+          </div>
         </button>
       </div>
       <div v-else class="empty">暂无需求词</div>
@@ -197,6 +219,14 @@ function selectVideo(video: DemandVideoItem) {
     <section class="col col-video">
       <h3 class="col-title">真实视频实例</h3>
       <template v-if="selectedDemand">
+        <div class="detail-meta">
+          <span class="grade-badge" :class="gradeClass(selectedDemand.grade)">
+            {{ selectedDemand.grade || '—' }}
+          </span>
+          <div v-if="selectedStrategies.length" class="strategy-tags">
+            <span v-for="s in selectedStrategies" :key="s" class="strategy-tag">{{ s }}</span>
+          </div>
+        </div>
         <div v-if="videosLoading" class="empty">加载中…</div>
         <div v-else-if="videosError" class="empty error">{{ videosError }}</div>
         <div v-else-if="videos.length" class="card-list">
@@ -399,6 +429,86 @@ function selectVideo(video: DemandVideoItem) {
   background: linear-gradient(180deg, #eff6ff 0%, #fff 100%);
 }
 
+.demand-card-head {
+  display: flex;
+  align-items: center;
+  justify-content: space-between;
+  width: 100%;
+  gap: 8px;
+}
+
+.grade-badge {
+  display: inline-flex;
+  align-items: center;
+  justify-content: center;
+  min-width: 22px;
+  height: 20px;
+  padding: 0 6px;
+  border-radius: 6px;
+  font-size: 11px;
+  font-weight: 700;
+  letter-spacing: 0.02em;
+  flex-shrink: 0;
+}
+
+.grade-badge.grade-S {
+  background: #fee2e2;
+  color: #b91c1c;
+}
+
+.grade-badge.grade-A {
+  background: #ffedd5;
+  color: #c2410c;
+}
+
+.grade-badge.grade-B {
+  background: #fef9c3;
+  color: #a16207;
+}
+
+.grade-badge.grade-C {
+  background: #e0f2fe;
+  color: #0369a1;
+}
+
+.grade-badge.grade-D {
+  background: #f1f5f9;
+  color: #64748b;
+}
+
+.grade-badge.grade-none {
+  background: #f1f5f9;
+  color: #94a3b8;
+}
+
+.strategy-tags {
+  display: flex;
+  flex-wrap: wrap;
+  gap: 4px;
+  margin-top: 2px;
+}
+
+.strategy-tag {
+  display: inline-flex;
+  align-items: center;
+  height: 18px;
+  padding: 0 6px;
+  border-radius: 999px;
+  background: rgba(99, 102, 241, 0.12);
+  color: #4338ca;
+  font-size: 10px;
+  font-weight: 600;
+  white-space: nowrap;
+}
+
+.detail-meta {
+  display: flex;
+  flex-wrap: wrap;
+  align-items: center;
+  gap: 8px;
+  margin-bottom: 6px;
+}
+
 .demand-card .card-label {
   color: #1d4ed8;
 }

web/src/components/IcicleHeatTree.vue

@@ -29,7 +29,7 @@ import {
   type IcicleRect,
   type PreparedNode,
 } from '../types/heatTree'
-import type { DemandBelongItem, DemandsByCategory } from '../types/demand'
+import type { DemandGradeItem, DemandsByCategory } from '../types/demand'
 
 /** Min cell height in leftmost two columns so labels stay readable. */
 const MIN_READABLE_HEIGHT = 18
@@ -88,7 +88,7 @@ const tooltipNode = ref<PreparedNode | null>(null)
 const inspectOpen = ref(false)
 const inspectNode = ref<PreparedNode | null>(null)
 
-const inspectItems = computed<DemandBelongItem[]>(() => {
+const inspectItems = computed<DemandGradeItem[]>(() => {
   if (!inspectNode.value) return []
   return props.demandsByCategory?.[inspectNode.value.id] ?? []
 })

web/src/components/TreeNode.vue

@@ -8,7 +8,7 @@ import {
   heatTextColor,
   scoreHeatT,
 } from '../types/category'
-import type { DemandBelongItem, DemandsByCategory } from '../types/demand'
+import type { DemandGradeItem, DemandsByCategory } from '../types/demand'
 
 const props = defineProps<{
   node: CategoryNode
@@ -21,12 +21,12 @@ const props = defineProps<{
   weightScale: number[]
   inspectCategoryId: number | null
   inspectCategoryName: string
-  inspectItems: DemandBelongItem[]
+  inspectItems: DemandGradeItem[]
 }>()
 
 const emit = defineEmits<{
   toggle: [id: number]
-  inspect: [payload: { categoryId: number; categoryName: string; items: DemandBelongItem[] }]
+  inspect: [payload: { categoryId: number; categoryName: string; items: DemandGradeItem[] }]
   closeInspect: []
 }>()
 
@@ -39,7 +39,7 @@ function isExpanded(): boolean {
   return props.depth < props.expandDepth
 }
 
-function demandsForNode(): DemandBelongItem[] {
+function demandsForNode(): DemandGradeItem[] {
   return props.demandsByCategory[props.node.id] ?? []
 }
 

web/src/types/demand.ts

@@ -1,12 +1,18 @@
-export interface DemandBelongItem {
+/** A demand_grade row, expanded one row per category_id it belongs to. */
+export interface DemandGradeItem {
   id: number
-  name: string | null
+  demand_name: string
   category_id: number
+  grade: string
+  score: number | null
   reason: string | null
+  /** JSON-encoded string array of source strategies, e.g. '["新热事件","逐月"]'. */
+  strategies: string | null
+  biz_dt: string
 }
 
-export interface DemandBelongResponse {
-  items: DemandBelongItem[]
+export interface DemandGradeResponse {
+  items: DemandGradeItem[]
 }
 
 export interface DemandVideoItem {
@@ -17,17 +23,19 @@ export interface DemandVideoItem {
   key_points_json: string | null
 }
 
-export interface DemandVideosResponse {
-  demand_belong_id: number
-  name: string | null
-  category_id: number
+export interface DemandGradeVideosResponse {
+  demand_grade_id: number
+  demand_name: string
+  category_ids: number[]
+  grade: string
+  strategies: string[]
   videos: DemandVideoItem[]
 }
 
 /** category_id → demands belonging to that category */
-export type DemandsByCategory = Record<number, DemandBelongItem[]>
+export type DemandsByCategory = Record<number, DemandGradeItem[]>
 
-export function groupDemandsByCategory(items: DemandBelongItem[]): DemandsByCategory {
+export function groupDemandsByCategory(items: DemandGradeItem[]): DemandsByCategory {
   const map: DemandsByCategory = {}
   for (const item of items) {
     const key = item.category_id
@@ -36,3 +44,22 @@ export function groupDemandsByCategory(items: DemandBelongItem[]): DemandsByCate
   }
   return map
 }
+
+/** Parse the JSON-encoded strategies string into a plain string array. */
+export function parseStrategies(raw: string | null | undefined): string[] {
+  if (!raw) return []
+  try {
+    const parsed = JSON.parse(raw)
+    return Array.isArray(parsed) ? parsed.map((s) => String(s)) : []
+  } catch {
+    return []
+  }
+}
+
+const GRADE_ORDER = ['S', 'A', 'B', 'C', 'D']
+
+export function gradeRank(grade: string | null | undefined): number {
+  if (!grade) return GRADE_ORDER.length
+  const idx = GRADE_ORDER.indexOf(grade.toUpperCase())
+  return idx === -1 ? GRADE_ORDER.length : idx
+}

web/src/utils/exportCategoryTreeHtml.ts

@@ -610,8 +610,9 @@ button, select { font-family: inherit; }
   color: #64748b;
   letter-spacing: 0.02em;
 }
-.drawer-body .col-name { width: 36%; }
-.drawer-body .col-reason { width: 64%; }
+.drawer-body .col-name { width: 30%; }
+.drawer-body .col-grade { width: 12%; }
+.drawer-body .col-reason { width: 58%; }
 .drawer-body tbody tr:hover td { background: #f8fafc; }
 .drawer-empty {
   padding: 48px 20px;
@@ -951,9 +952,9 @@ const EXPORT_JS = `
       els.drawerBody.innerHTML = '<div class="drawer-empty">该节点暂无需求词</div>';
     } else {
       els.drawerBody.innerHTML =
-        '<table><thead><tr><th class="col-name">需求词</th><th class="col-reason">原因</th></tr></thead><tbody>' +
+        '<table><thead><tr><th class="col-name">需求词</th><th class="col-grade">等级</th><th class="col-reason">原因</th></tr></thead><tbody>' +
         items.map(function (item) {
-          return '<tr><td>' + esc(item.name || '—') + '</td><td>' + esc(item.reason || '—') + '</td></tr>';
+          return '<tr><td>' + esc(item.demand_name || '—') + '</td><td>' + esc(item.grade || '—') + '</td><td>' + esc(item.reason || '—') + '</td></tr>';
         }).join('') +
         '</tbody></table>';
     }

web/src/views/CategoryTreeView.vue

@@ -2,7 +2,7 @@
 import { onMounted, ref } from 'vue'
 import CategoryTree from '../components/CategoryTree.vue'
 import { fetchCategoryTree } from '../api/category'
-import { fetchDemandBelongCategory } from '../api/demand'
+import { fetchDemandGrade } from '../api/demand'
 import type { CategoryNode, WeightDimMeta } from '../types/category'
 import type { DemandsByCategory } from '../types/demand'
 import { groupDemandsByCategory } from '../types/demand'
@@ -18,7 +18,7 @@ onMounted(async () => {
   try {
     const [tree, demands] = await Promise.all([
       fetchCategoryTree(),
-      fetchDemandBelongCategory(),
+      fetchDemandGrade(),
     ])
     nodes.value = tree.nodes ?? []
     dims.value = tree.dims ?? []

web/src/views/GlobalDemandMapView.vue

@@ -2,7 +2,7 @@
 import { onMounted, ref } from 'vue'
 import IcicleHeatTree from '../components/IcicleHeatTree.vue'
 import { fetchCategoryTree } from '../api/category'
-import { fetchDemandBelongCategory } from '../api/demand'
+import { fetchDemandGrade } from '../api/demand'
 import type { CategoryNode } from '../types/category'
 import type { DemandsByCategory } from '../types/demand'
 import { groupDemandsByCategory } from '../types/demand'
@@ -17,7 +17,7 @@ onMounted(async () => {
   try {
     const [tree, demands] = await Promise.all([
       fetchCategoryTree(),
-      fetchDemandBelongCategory(),
+      fetchDemandGrade(),
     ])
     nodes.value = tree.nodes ?? []
     bizDt.value = tree.biz_dt ?? null