需求分配完整流程修改

api/app.py

@@ -15,7 +15,7 @@ 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
 from supply_infra.db import init_db
-from supply_infra.scheduler import get_scheduler_status, start_scheduler, stop_scheduler
+from supply_infra.scheduler.app import get_scheduler_status, start_scheduler, stop_scheduler
 
 
 @asynccontextmanager

api/services/demand_grade_videos.py

@@ -8,6 +8,9 @@ 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.repositories.multi_demand_video_point_repo import (
+    MultiDemandVideoPointRepository,
+)
 from supply_infra.db.session import get_session
 
 
@@ -39,17 +42,23 @@ def list_videos_for_demand_grade(demand_grade_id: int) -> dict[str, Any] | None:
 
         vids = _parse_video_ids(grade.video_list)
         details = MultiDemandVideoDetailRepository(session).list_by_vids(vids)
+        points_by_vid = MultiDemandVideoPointRepository(session).json_fields_by_video_ids(
+            vids
+        )
 
         videos: list[dict[str, Any]] = []
         for vid in vids:
             row = details.get(vid)
+            point_fields = points_by_vid.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,
+                    "inspiration_points_json": point_fields.get(
+                        "inspiration_points_json"
+                    ),
+                    "purpose_points_json": point_fields.get("purpose_points_json"),
+                    "key_points_json": point_fields.get("key_points_json"),
                 }
             )
 

api/services/demand_videos.py

@@ -10,6 +10,9 @@ from supply_infra.db.repositories.demand_belong_category_repo import (
 from supply_infra.db.repositories.multi_demand_video_detail_repo import (
     MultiDemandVideoDetailRepository,
 )
+from supply_infra.db.repositories.multi_demand_video_point_repo import (
+    MultiDemandVideoPointRepository,
+)
 from supply_infra.db.session import get_session
 
 
@@ -38,19 +41,23 @@ def list_videos_for_demand_belong(belong_id: int) -> dict[str, Any] | None:
 
         vids = _parse_video_ids(belong.video_list)
         details = MultiDemandVideoDetailRepository(session).list_by_vids(vids)
+        points_by_vid = MultiDemandVideoPointRepository(session).json_fields_by_video_ids(
+            vids
+        )
 
         videos: list[dict[str, Any]] = []
         for vid in vids:
             row = details.get(vid)
+            point_fields = points_by_vid.get(vid, {})
             videos.append(
                 {
                     "vid": vid,
                     "title": row.title if row else None,
-                    "inspiration_points_json": (
-                        row.inspiration_points_json if row else None
+                    "inspiration_points_json": point_fields.get(
+                        "inspiration_points_json"
                     ),
-                    "purpose_points_json": row.purpose_points_json if row else None,
-                    "key_points_json": row.key_points_json if row else None,
+                    "purpose_points_json": point_fields.get("purpose_points_json"),
+                    "key_points_json": point_fields.get("key_points_json"),
                 }
             )
 

jobs/backfill_multi_demand_video_points_table.py

@@ -0,0 +1,117 @@
+#!/usr/bin/env python3
+"""将 multi_demand_video_detail 三个 JSON 点位列迁移到 multi_demand_video_point 表。
+
+用法:
+    python jobs/backfill_multi_demand_video_points_table.py
+    python jobs/backfill_multi_demand_video_points_table.py 500   # 每批 500 条视频
+"""
+
+from __future__ import annotations
+
+import logging
+import sys
+
+from sqlalchemy import select
+
+from supply_infra.db.models.multi_demand_video_detail import MultiDemandVideoDetail
+from supply_infra.db.repositories.multi_demand_video_point_repo import (
+    MultiDemandVideoPointRepository,
+)
+from supply_infra.db.session import get_session
+from supply_infra.video_points import points_from_json_fields
+
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
+)
+logger = logging.getLogger(__name__)
+
+_DEFAULT_BATCH_SIZE = 200
+
+
+def backfill_video_points_table(batch_size: int = _DEFAULT_BATCH_SIZE) -> dict:
+    """从 detail 表 JSON 列回填点位表，跳过已迁移的 video_id。"""
+    chunk = max(1, int(batch_size))
+    total_rows = 0
+    total_points = 0
+    batches = 0
+    offset = 0
+
+    while True:
+        with get_session() as session:
+            stmt = (
+                select(MultiDemandVideoDetail)
+                .where(
+                    MultiDemandVideoDetail.inspiration_points_json.is_not(None)
+                    | MultiDemandVideoDetail.purpose_points_json.is_not(None)
+                    | MultiDemandVideoDetail.key_points_json.is_not(None)
+                )
+                .order_by(MultiDemandVideoDetail.id)
+                .offset(offset)
+                .limit(chunk)
+            )
+            rows = list(session.scalars(stmt).all())
+            if not rows:
+                break
+
+            vids = [str(row.vid) for row in rows if row.vid]
+            existing = MultiDemandVideoPointRepository(session).list_video_ids_with_points(
+                vids
+            )
+
+            points_by_vid: dict[str, list] = {}
+            for row in rows:
+                vid = str(row.vid).strip() if row.vid else ""
+                if not vid or vid in existing:
+                    continue
+                point_rows = points_from_json_fields(
+                    vid,
+                    inspiration_points_json=row.inspiration_points_json,
+                    purpose_points_json=row.purpose_points_json,
+                    key_points_json=row.key_points_json,
+                )
+                if point_rows:
+                    points_by_vid[vid] = point_rows
+
+            inserted = 0
+            if points_by_vid:
+                inserted = MultiDemandVideoPointRepository(session).replace_for_video_ids(
+                    points_by_vid
+                )
+
+        batch_count = len(rows)
+        migrated = len(points_by_vid)
+        total_rows += batch_count
+        total_points += inserted
+        batches += 1
+        offset += batch_count
+        logger.info(
+            "Batch %d: scanned=%d migrated_videos=%d inserted_points=%d offset=%d",
+            batches,
+            batch_count,
+            migrated,
+            inserted,
+            offset,
+        )
+
+        if batch_count < chunk:
+            break
+
+    result = {
+        "batches": batches,
+        "scanned_rows": total_rows,
+        "inserted_points": total_points,
+    }
+    logger.info("Backfill multi_demand_video_point completed: %s", result)
+    return result
+
+
+def main(batch_arg: str | None = None) -> dict:
+    batch_size = int(batch_arg) if batch_arg else _DEFAULT_BATCH_SIZE
+    result = backfill_video_points_table(batch_size=batch_size)
+    print(result)
+    return result
+
+
+if __name__ == "__main__":
+    main(sys.argv[1] if len(sys.argv) > 1 else None)

jobs/init_db.py

@@ -1,8 +1,16 @@
 #!/usr/bin/env python3
 """CLI entry point to initialize database tables."""
 
+from supply_infra.config import get_infra_settings
 from supply_infra.db import init_db
 
 if __name__ == "__main__":
-    init_db()
-    print("Database tables created.")
+    settings = get_infra_settings()
+    result = init_db()
+    print(
+        f"Database ready: {settings.mysql_host}:{settings.mysql_port}/{settings.mysql_database}"
+    )
+    if result["created"]:
+        print("Created tables:", ", ".join(result["created"]))
+    else:
+        print("No new tables created (all ORM tables already exist).")

jobs/run_scheduler.py

@@ -3,7 +3,7 @@
 
 import logging
 
-from supply_infra.scheduler import run_scheduler
+from supply_infra.scheduler.app import run_scheduler
 
 logging.basicConfig(
     level=logging.INFO,

supply_infra/db/models/__init__.py

@@ -5,13 +5,18 @@ 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_grade_plan import DemandGradePlan, DemandGradePlanGroup
+from supply_infra.db.models.demand_grade_plan import (
+    DemandGradePlan,
+    DemandGradePlanGroup,
+    DemandGradePlanGroupItem,
+)
 from supply_infra.db.models.demand_popularity_stats import DemandPopularityStats
 from supply_infra.db.models.generated_demand import GeneratedDemand
 from supply_infra.db.models.global_tree_category import GlobalTreeCategory
 from supply_infra.db.models.global_tree_element import GlobalTreeElement
 from supply_infra.db.models.multi_demand_pool_di import MultiDemandPoolDi
 from supply_infra.db.models.multi_demand_video_detail import MultiDemandVideoDetail
+from supply_infra.db.models.multi_demand_video_point import MultiDemandVideoPoint
 from supply_infra.db.models.oss_log import OssLog
 from supply_infra.db.models.scheduler_job_execution import SchedulerJobExecution
 
@@ -23,12 +28,14 @@ __all__ = [
     "DemandGradeCategoryRel",
     "DemandGradePlan",
     "DemandGradePlanGroup",
+    "DemandGradePlanGroupItem",
     "DemandPopularityStats",
     "GeneratedDemand",
     "GlobalTreeCategory",
     "GlobalTreeElement",
     "MultiDemandPoolDi",
     "MultiDemandVideoDetail",
+    "MultiDemandVideoPoint",
     "OssLog",
     "SchedulerJobExecution",
 ]

supply_infra/db/models/demand_grade_plan.py

@@ -51,3 +51,26 @@ class DemandGradePlanGroup(Base):
     finished_at: Mapped[datetime | None] = mapped_column(nullable=True)
     create_time: Mapped[datetime] = mapped_column(nullable=False, server_default=func.now())
     update_time: Mapped[datetime] = mapped_column(nullable=False, server_default=func.now(), onupdate=func.now())
+
+
+class DemandGradePlanGroupItem(Base):
+    """计划组内待执行分级的具体需求池记录。"""
+
+    __tablename__ = "demand_grade_plan_group_item"
+    __table_args__ = (
+        UniqueConstraint("group_id", "pool_id", name="uk_demand_grade_plan_group_item"),
+        Index("idx_demand_grade_plan_group_item_group_status", "group_id", "status", "sort_order"),
+        Index("idx_demand_grade_plan_group_item_biz_dt", "biz_dt"),
+    )
+
+    id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
+    group_id: Mapped[int] = mapped_column(BigInteger, nullable=False)
+    biz_dt: Mapped[str] = mapped_column(String(8), nullable=False)
+    pool_id: Mapped[int] = mapped_column(BigInteger, nullable=False)
+    demand_name: Mapped[str] = mapped_column(String(512), nullable=False)
+    sort_order: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
+    status: Mapped[str] = mapped_column(String(32), nullable=False, default="pending")
+    error_message: Mapped[str | None] = mapped_column(Text, nullable=True)
+    finished_at: Mapped[datetime | None] = mapped_column(nullable=True)
+    create_time: Mapped[datetime] = mapped_column(nullable=False, server_default=func.now())
+    update_time: Mapped[datetime] = mapped_column(nullable=False, server_default=func.now(), onupdate=func.now())

supply_infra/db/models/multi_demand_video_point.py

@@ -0,0 +1,48 @@
+from __future__ import annotations
+
+from datetime import datetime
+
+from sqlalchemy import BigInteger, Index, String, Text, func
+from sqlalchemy.orm import Mapped, mapped_column
+
+from supply_infra.db.base import Base
+
+POINT_TYPE_INSPIRATION = "inspiration"
+POINT_TYPE_PURPOSE = "purpose"
+POINT_TYPE_KEY = "key"
+
+POINT_TYPES = (POINT_TYPE_INSPIRATION, POINT_TYPE_PURPOSE, POINT_TYPE_KEY)
+
+
+class MultiDemandVideoPoint(Base):
+    """需求池视频点位表 — 灵感点/目的点/关键点按行存储。"""
+
+    __tablename__ = "multi_demand_video_point"
+    __table_args__ = (
+        Index("idx_multi_demand_video_point_video_type", "video_id", "point_type"),
+    )
+
+    id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
+    video_id: Mapped[str] = mapped_column(String(64), nullable=False, comment="视频id")
+    point_type: Mapped[str] = mapped_column(
+        String(32),
+        nullable=False,
+        comment="点类型：inspiration / purpose / key",
+    )
+    point_data: Mapped[str | None] = mapped_column(
+        Text, nullable=True, comment="点"
+    )
+    point_desc: Mapped[str | None] = mapped_column(
+        Text, nullable=True, 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/repositories/__init__.py

@@ -25,6 +25,9 @@ from supply_infra.db.repositories.multi_demand_pool_di_repo import MultiDemandPo
 from supply_infra.db.repositories.multi_demand_video_detail_repo import (
     MultiDemandVideoDetailRepository,
 )
+from supply_infra.db.repositories.multi_demand_video_point_repo import (
+    MultiDemandVideoPointRepository,
+)
 from supply_infra.db.repositories.oss_log_repo import OssLogRepository
 from supply_infra.db.repositories.scheduler_job_execution_repo import (
     SchedulerJobExecutionRepository,
@@ -44,6 +47,7 @@ __all__ = [
     "GlobalTreeElementRepository",
     "MultiDemandPoolDiRepository",
     "MultiDemandVideoDetailRepository",
+    "MultiDemandVideoPointRepository",
     "OssLogRepository",
     "SchedulerJobExecutionRepository",
 ]

supply_infra/db/repositories/demand_grade_plan_repo.py

@@ -5,10 +5,16 @@ import uuid
 from datetime import datetime
 from typing import Any
 
-from sqlalchemy import select, update
+from sqlalchemy import func, select, update
 
-from supply_infra.db.models.demand_grade_plan import DemandGradePlan, DemandGradePlanGroup
+from supply_infra.db.models.demand_grade_plan import (
+    DemandGradePlan,
+    DemandGradePlanGroup,
+    DemandGradePlanGroupItem,
+)
 from supply_infra.db.repositories.base import BaseRepository
+from supply_infra.db.repositories.demand_grade_repo import DemandGradeRepository
+from supply_infra.scheduler.plan_group_batch import resolve_demands_for_category_ids
 
 
 class DemandGradePlanRepository(BaseRepository[DemandGradePlan]):
@@ -92,6 +98,19 @@ class DemandGradePlanRepository(BaseRepository[DemandGradePlan]):
                 planning_reason=str(group["planning_reason"]), shared_traits=shared_traits,
                 status="pending",
             ))
+            group_row = self.session.scalar(
+                select(DemandGradePlanGroup)
+                .where(DemandGradePlanGroup.plan_id == plan_id, DemandGradePlanGroup.group_no == group_no)
+                .limit(1)
+            )
+            if group_row is not None:
+                graded_names = DemandGradeRepository(self.session).get_existing_demand_names(biz_dt)
+                self.materialize_group_items(
+                    int(group_row.id),
+                    biz_dt,
+                    list(group.get("category_ids") or []),
+                    graded_names=graded_names,
+                )
 
     def list_pending_group_ids(self, biz_dt: str) -> list[int]:
         """返回当天待执行的 pending 任务 id。"""
@@ -150,3 +169,108 @@ class DemandGradePlanRepository(BaseRepository[DemandGradePlan]):
             select(DemandGradePlanGroup.status).where(DemandGradePlanGroup.biz_dt == biz_dt)
         ).scalars().all()
         return {status: sum(value == status for value in rows) for status in ("pending", "running", "finished", "failed")}
+
+    def group_item_count(self, group_id: int) -> int:
+        return int(self.session.scalar(
+            select(func.count())
+            .select_from(DemandGradePlanGroupItem)
+            .where(DemandGradePlanGroupItem.group_id == int(group_id))
+        ) or 0)
+
+    def materialize_group_items(
+        self,
+        group_id: int,
+        biz_dt: str,
+        category_ids: list[int],
+        *,
+        graded_names: set[str] | None = None,
+    ) -> int:
+        """将 category_ids 展开为组内需求明细；已存在明细时跳过。"""
+        if self.group_item_count(group_id) > 0:
+            return 0
+
+        graded = graded_names or set()
+        demands = resolve_demands_for_category_ids(biz_dt, category_ids)
+        created = 0
+        for sort_order, demand in enumerate(demands, start=1):
+            demand_name = str(demand["demand_name"])
+            status = "skipped" if demand_name in graded else "pending"
+            self.add(DemandGradePlanGroupItem(
+                group_id=int(group_id),
+                biz_dt=biz_dt,
+                pool_id=int(demand["pool_id"]),
+                demand_name=demand_name,
+                sort_order=sort_order,
+                status=status,
+            ))
+            created += 1
+        return created
+
+    def materialize_pending_groups(self, biz_dt: str, *, graded_names: set[str] | None = None) -> int:
+        """为当天尚未物化明细的 pending 计划组补写需求列表。"""
+        created = 0
+        for group in self.list_groups_by_biz_dt(biz_dt):
+            if group.status != "pending":
+                continue
+            if self.group_item_count(int(group.id)) > 0:
+                continue
+            category_ids = json.loads(group.category_ids)
+            created += self.materialize_group_items(
+                int(group.id),
+                biz_dt,
+                category_ids,
+                graded_names=graded_names,
+            )
+        return created
+
+    def list_pending_group_items(
+        self,
+        group_id: int,
+        *,
+        limit: int | None = None,
+    ) -> list[dict[str, Any]]:
+        stmt = (
+            select(DemandGradePlanGroupItem)
+            .where(
+                DemandGradePlanGroupItem.group_id == int(group_id),
+                DemandGradePlanGroupItem.status == "pending",
+            )
+            .order_by(DemandGradePlanGroupItem.sort_order, DemandGradePlanGroupItem.id)
+        )
+        if limit is not None:
+            stmt = stmt.limit(max(1, int(limit)))
+        rows = self.session.scalars(stmt).all()
+        return [
+            {
+                "item_id": int(row.id),
+                "pool_id": int(row.pool_id),
+                "demand_name": str(row.demand_name),
+            }
+            for row in rows
+        ]
+
+    def summarize_group_items(self, group_id: int) -> dict[str, int]:
+        rows = self.session.execute(
+            select(DemandGradePlanGroupItem.status)
+            .where(DemandGradePlanGroupItem.group_id == int(group_id))
+        ).scalars().all()
+        statuses = ("pending", "finished", "failed", "skipped")
+        return {status: sum(value == status for value in rows) for status in statuses}
+
+    def mark_group_items_status(
+        self,
+        item_ids: list[int],
+        *,
+        status: str,
+        error_message: str | None = None,
+    ) -> None:
+        if not item_ids:
+            return
+        values: dict[str, Any] = {"status": status, "error_message": error_message}
+        if status in {"finished", "failed", "skipped"}:
+            values["finished_at"] = datetime.now()
+        self.session.execute(
+            update(DemandGradePlanGroupItem)
+            .where(DemandGradePlanGroupItem.id.in_([int(item_id) for item_id in item_ids]))
+            .values(**values)
+        )

supply_infra/db/repositories/multi_demand_video_point_repo.py

@@ -0,0 +1,117 @@
+from __future__ import annotations
+
+from collections.abc import Iterable
+from typing import Any
+
+from sqlalchemy import delete, select
+
+from supply_infra.db.models.multi_demand_video_point import MultiDemandVideoPoint
+from supply_infra.db.repositories.base import BaseRepository
+from supply_infra.video_points import json_fields_from_point_rows
+
+_BATCH_SIZE = 1000
+
+
+class MultiDemandVideoPointRepository(BaseRepository[MultiDemandVideoPoint]):
+    """需求池视频点位表 — 按 video_id 批量替换与查询。"""
+
+    model = MultiDemandVideoPoint
+
+    def list_video_ids_with_points(self, video_ids: Iterable[str]) -> set[str]:
+        """返回 video_ids 中在点位表已有记录的视频 id。"""
+        vid_list = [v for v in video_ids if v]
+        if not vid_list:
+            return set()
+
+        existing: set[str] = set()
+        for i in range(0, len(vid_list), _BATCH_SIZE):
+            batch = vid_list[i : i + _BATCH_SIZE]
+            stmt = (
+                select(MultiDemandVideoPoint.video_id)
+                .where(MultiDemandVideoPoint.video_id.in_(batch))
+                .distinct()
+            )
+            existing.update(
+                str(v) for v in self.session.scalars(stmt).all() if v
+            )
+        return existing
+
+    def list_by_video_ids(
+        self, video_ids: Iterable[str]
+    ) -> dict[str, list[dict[str, Any]]]:
+        """按 video_id 批量查询点位行，返回 video_id → 行列表。"""
+        vid_list = [v for v in video_ids if v]
+        if not vid_list:
+            return {}
+
+        result: dict[str, list[dict[str, Any]]] = {}
+        for i in range(0, len(vid_list), _BATCH_SIZE):
+            batch = vid_list[i : i + _BATCH_SIZE]
+            stmt = (
+                select(MultiDemandVideoPoint)
+                .where(MultiDemandVideoPoint.video_id.in_(batch))
+                .order_by(
+                    MultiDemandVideoPoint.video_id,
+                    MultiDemandVideoPoint.point_type,
+                    MultiDemandVideoPoint.id,
+                )
+            )
+            for row in self.session.scalars(stmt).all():
+                result.setdefault(str(row.video_id), []).append(
+                    {
+                        "point_type": row.point_type,
+                        "point_data": row.point_data,
+                        "point_desc": row.point_desc,
+                    }
+                )
+        return result
+
+    def json_fields_by_video_ids(
+        self, video_ids: Iterable[str]
+    ) -> dict[str, dict[str, str | None]]:
+        """按 video_id 返回三个 JSON 列（API 兼容）。"""
+        rows_by_vid = self.list_by_video_ids(video_ids)
+        return {
+            vid: json_fields_from_point_rows(rows)
+            for vid, rows in rows_by_vid.items()
+        }
+
+    def replace_for_video_ids(
+        self, points_by_video_id: dict[str, list[dict[str, Any]]]
+    ) -> int:
+        """按 video_id 全量替换点位：先删后插。"""
+        if not points_by_video_id:
+            return 0
+
+        video_ids = [v for v in points_by_video_id if v]
+        if not video_ids:
+            return 0
+
+        for i in range(0, len(video_ids), _BATCH_SIZE):
+            batch = video_ids[i : i + _BATCH_SIZE]
+            self.session.execute(
+                delete(MultiDemandVideoPoint).where(
+                    MultiDemandVideoPoint.video_id.in_(batch)
+                )
+            )
+
+        insert_rows: list[dict[str, Any]] = []
+        for video_id, rows in points_by_video_id.items():
+            if not video_id or not rows:
+                continue
+            for row in rows:
+                insert_rows.append(
+                    {
+                        "video_id": video_id,
+                        "point_type": row["point_type"],
+                        "point_data": row.get("point_data"),
+                        "point_desc": row.get("point_desc"),
+                    }
+                )
+
+        inserted = 0
+        for i in range(0, len(insert_rows), _BATCH_SIZE):
+            batch = insert_rows[i : i + _BATCH_SIZE]
+            self.session.bulk_insert_mappings(MultiDemandVideoPoint, batch)
+            inserted += len(batch)
+        return inserted

supply_infra/db/session.py

@@ -4,7 +4,7 @@ from collections.abc import Generator
 from contextlib import contextmanager
 from typing import Any
 
-from sqlalchemy import create_engine
+from sqlalchemy import create_engine, inspect
 from sqlalchemy.orm import Session, sessionmaker
 
 from supply_infra.config import get_infra_settings
@@ -29,11 +29,17 @@ def get_engine():
     return _engine
 
 
-def init_db() -> None:
+def init_db() -> dict[str, list[str]]:
     """Create all tables (dev / first-run). Import models before calling."""
     import supply_infra.db.models  # noqa: F401 — register all models
 
-    Base.metadata.create_all(bind=get_engine())
+    engine = get_engine()
+    inspector = inspect(engine)
+    before = set(inspector.get_table_names())
+    Base.metadata.create_all(bind=engine)
+    after = set(inspect(engine).get_table_names())
+    created = sorted(after - before)
+    return {"created": created}
 
 
 @contextmanager

supply_infra/scheduler/__init__.py

@@ -1,17 +1 @@
 """Scheduler for periodic jobs."""
-
-from supply_infra.scheduler.app import (
-    create_scheduler,
-    get_scheduler_status,
-    run_scheduler,
-    start_scheduler,
-    stop_scheduler,
-)
-
-__all__ = [
-    "create_scheduler",
-    "get_scheduler_status",
-    "run_scheduler",
-    "start_scheduler",
-    "stop_scheduler",
-]

supply_infra/scheduler/jobs/grade_demand_pool.py

@@ -14,10 +14,7 @@ from supply_infra.db.repositories.demand_grade_plan_repo import DemandGradePlanR
 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
-from supply_infra.scheduler.plan_group_batch import (
-    MAX_DEMANDS_PER_BATCH,
-    list_pending_demands_by_category_ids,
-)
+from supply_infra.scheduler.plan_group_batch import MAX_DEMANDS_PER_BATCH, split_even_batches
 
 logger = logging.getLogger(__name__)
 
@@ -30,36 +27,57 @@ def _resolve_biz_dt(biz_dt: str | None) -> str:
     return datetime.now(ZoneInfo(get_infra_settings().scheduler_timezone)).strftime("%Y%m%d")
 
 
-def _graded_names(biz_dt: str) -> set[str]:
+def _materialize_pending_group_items(biz_dt: str) -> int:
+    """执行前为 pending 计划组物化待分级需求明细。"""
     with get_session() as session:
-        return DemandGradeRepository(session).get_existing_demand_names(biz_dt)
+        graded_names = DemandGradeRepository(session).get_existing_demand_names(biz_dt)
+        return DemandGradePlanRepository(session).materialize_pending_groups(
+            biz_dt,
+            graded_names=graded_names,
+        )
+
+
+def _group_success(item_summary: dict[str, int], processed_batches: int) -> bool:
+    if item_summary.get("pending", 0) > 0:
+        return False
+    total = sum(item_summary.values())
+    if total == 0:
+        return True
+    if processed_batches > 0:
+        return True
+    return item_summary.get("failed", 0) == 0 and (
+        item_summary.get("finished", 0) > 0 or item_summary.get("skipped", 0) > 0
+    )
 
 
 def _run_group(biz_dt: str, group_id: int, *, max_demands_per_batch: int) -> int:
-    """领取并执行一个固定任务，组内按批处理，失败子批跳过并继续。"""
+    """领取并执行一个固定任务，从组内需求明细表按批读取。"""
     with get_session() as session:
         group = DemandGradePlanRepository(session).claim_group(biz_dt, group_id)
     if group is None:
         logger.warning("分级任务已被其他 worker 领取或状态已变化: biz_dt=%s group_id=%s", biz_dt, group_id)
         return 0
 
+    gid = int(group["id"])
     processed_batches = 0
     batch_errors: list[str] = []
-    skipped_names: set[str] = set()
     try:
-        while True:
-            graded_before = _graded_names(biz_dt) | skipped_names
-            demands = list_pending_demands_by_category_ids(
-                biz_dt,
-                group["category_ids"],
-                max_demands=max_demands_per_batch,
-                excluded_demand_names=sorted(graded_before),
-            )
-            if not demands:
-                break
+        with get_session() as session:
+            items = DemandGradePlanRepository(session).list_pending_group_items(gid)
 
+        for batch_items in split_even_batches(items, max_per_batch=max_demands_per_batch):
+            demands = [
+                {"pool_id": item["pool_id"], "demand_name": item["demand_name"]}
+                for item in batch_items
+            ]
+            item_ids = [int(item["item_id"]) for item in batch_items]
             try:
                 grade_demand_words(demands, biz_dt=biz_dt)
+                with get_session() as session:
+                    DemandGradePlanRepository(session).mark_group_items_status(
+                        item_ids,
+                        status="finished",
+                    )
                 processed_batches += 1
             except Exception as exc:
                 logger.exception(
@@ -69,12 +87,19 @@ def _run_group(biz_dt: str, group_id: int, *, max_demands_per_batch: int) -> int
                     len(demands),
                 )
                 batch_errors.append(str(exc))
-                skipped_names.update(item["demand_name"] for item in demands)
+                with get_session() as session:
+                    DemandGradePlanRepository(session).mark_group_items_status(
+                        item_ids,
+                        status="failed",
+                        error_message=str(exc),
+                    )
 
         with get_session() as session:
-            DemandGradePlanRepository(session).finish_group(
-                group["id"],
-                success=processed_batches > 0,
+            repo = DemandGradePlanRepository(session)
+            item_summary = repo.summarize_group_items(gid)
+            repo.finish_group(
+                gid,
+                success=_group_success(item_summary, processed_batches),
                 error_message="; ".join(batch_errors) if batch_errors else None,
             )
     except Exception as exc:
@@ -86,7 +111,7 @@ def _run_group(biz_dt: str, group_id: int, *, max_demands_per_batch: int) -> int
         try:
             with get_session() as session:
                 DemandGradePlanRepository(session).finish_group(
-                    group["id"],
+                    gid,
                     success=False,
                     error_message=str(exc),
                 )
@@ -202,6 +227,9 @@ def _grade_demand_pool_impl(
         except Exception:
             logger.exception("统筹 Agent 执行失败，继续处理数据库中已有任务: biz_dt=%s", resolved_biz_dt)
 
+    materialized = _materialize_pending_group_items(resolved_biz_dt)
+    logger.info("物化计划组需求明细: biz_dt=%s items=%s", resolved_biz_dt, materialized)
+
     plan_execution = execute_plan_tasks_until_complete(
         resolved_biz_dt,
         workers=max(1, int(workers)),
@@ -218,6 +246,7 @@ def _grade_demand_pool_impl(
         "total": total,
         "graded_before": graded_before,
         "graded_after": graded_after,
+        "materialized_items": materialized,
         "planned_category_count": len(final_snapshot["assigned_category_ids"]),
         "planned_groups": final_snapshot["planned_groups"],
         "group_status": final_snapshot["group_status"],

supply_infra/scheduler/jobs/sync_multi_demand_videos.py

@@ -13,22 +13,21 @@ from supply_infra.db.repositories.multi_demand_pool_di_repo import MultiDemandPo
 from supply_infra.db.repositories.multi_demand_video_detail_repo import (
     MultiDemandVideoDetailRepository,
 )
+from supply_infra.db.repositories.multi_demand_video_point_repo import (
+    MultiDemandVideoPointRepository,
+)
 from supply_infra.db.session import get_session
 from supply_infra.odps.client import ODPSClient, get_odps_client
+from supply_infra.video_points import (
+    extract_points_json_from_decode,
+    points_from_decode_payload,
+)
 
 logger = logging.getLogger(__name__)
 
 _FINAL_TOPIC_KEY = "最终选题"
 _TARGET_POST_KEY = "target_post"
 _TITLE_KEY = "title"
-_POINT_KEY = "点"
-_POINT_DESC_KEY = "点描述"
-# decode_result 中的点位 key → 落库字段名
-_POINT_FIELD_MAP = {
-    "灵感点": "inspiration_points_json",
-    "目的点": "purpose_points_json",
-    "关键点": "key_points_json",
-}
 VIDEO_SYNC_BATCH_SIZE = 100
 
 
@@ -83,33 +82,9 @@ def _extract_final_topic_json(payload: dict[str, Any]) -> str | None:
     return json.dumps(final_topic, ensure_ascii=False)
 
 
-def _extract_points(payload: dict[str, Any], key: str) -> str | None:
-    """从 decode_result[key]（灵感点/目的点/关键点）取每项的 点/点描述，重组为 JSON 文本。"""
-    items = payload.get(key)
-    if not isinstance(items, list):
-        return None
-
-    points: list[dict[str, Any]] = []
-    for item in items:
-        if not isinstance(item, dict):
-            continue
-        point = item.get(_POINT_KEY)
-        point_desc = item.get(_POINT_DESC_KEY)
-        if point is None and point_desc is None:
-            continue
-        points.append({_POINT_KEY: point, _POINT_DESC_KEY: point_desc})
-
-    if not points:
-        return None
-    return json.dumps(points, ensure_ascii=False)
-
-
 def _extract_all_points(payload: dict[str, Any]) -> dict[str, str | None]:
-    """从 decode_result 提取灵感点/目的点/关键点三个字段，返回 落库字段名 → JSON 文本。"""
-    return {
-        field: _extract_points(payload, key)
-        for key, field in _POINT_FIELD_MAP.items()
-    }
+    """从 decode_result 提取灵感点/目的点/关键点三个 JSON 列。"""
+    return extract_points_json_from_decode(payload)
 
 
 def _extract_title(payload: dict[str, Any]) -> str | None:
@@ -144,6 +119,7 @@ def _sync_one_batch(
     )
 
     insert_rows: list[dict[str, Any]] = []
+    point_rows_by_vid: dict[str, list[dict[str, Any]]] = {}
     skipped_no_topic = 0
     seen_vids: set[str] = set()
     for row in odps_rows:
@@ -170,11 +146,17 @@ def _sync_one_batch(
                 **_extract_all_points(payload),
             }
         )
+        point_rows = points_from_decode_payload(vid, payload)
+        if point_rows:
+            point_rows_by_vid[vid] = point_rows
 
     with get_session() as session:
-        inserted = MultiDemandVideoDetailRepository(session).bulk_insert_ignore(
-            insert_rows
-        )
+        detail_repo = MultiDemandVideoDetailRepository(session)
+        inserted = detail_repo.bulk_insert_ignore(insert_rows)
+        if point_rows_by_vid:
+            MultiDemandVideoPointRepository(session).replace_for_video_ids(
+                point_rows_by_vid
+            )
 
     odps_vids = {
         str(r.get("vid")).strip()
@@ -424,6 +406,7 @@ def backfill_video_points(
             decode_dt, batch_vids, batch_size=len(batch_vids)
         )
         points_by_vid: dict[str, dict[str, str | None]] = {}
+        point_rows_by_vid: dict[str, list[dict[str, Any]]] = {}
         for row in odps_rows:
             raw_vid = row.get("vid")
             if raw_vid is None:
@@ -436,11 +419,17 @@ def backfill_video_points(
                 total_skipped += 1
                 continue
             points_by_vid[vid] = _extract_all_points(payload)
+            point_rows = points_from_decode_payload(vid, payload)
+            if point_rows:
+                point_rows_by_vid[vid] = point_rows
 
         with get_session() as session:
-            updated = MultiDemandVideoDetailRepository(session).update_points(
-                points_by_vid
-            )
+            detail_repo = MultiDemandVideoDetailRepository(session)
+            updated = detail_repo.update_points(points_by_vid)
+            if point_rows_by_vid:
+                MultiDemandVideoPointRepository(session).replace_for_video_ids(
+                    point_rows_by_vid
+                )
         total_updated += updated
 
         odps_vids = {

supply_infra/scheduler/plan_group_batch.py

@@ -1,4 +1,4 @@
-"""从 plan_group 的 category_ids 查询待分级需求词。"""
+"""计划组需求明细：物化与读取。"""
 from __future__ import annotations
 
 from typing import Any
@@ -12,6 +12,33 @@ from supply_infra.db.session import get_session
 MAX_DEMANDS_PER_BATCH = 30
 
 
+def split_even_batches(items: list[Any], *, max_per_batch: int = MAX_DEMANDS_PER_BATCH) -> list[list[Any]]:
+    """按总数均分子批次：总数不超过上限则一批；否则递增组数直到每组不超过上限，余数从前组分配。"""
+    n = len(items)
+    if n == 0:
+        return []
+    cap = max(1, int(max_per_batch))
+    if n <= cap:
+        return [items]
+
+    k = 2
+    while True:
+        base = n // k
+        rem = n % k
+        max_size = base + 1 if rem > 0 else base
+        if max_size <= cap:
+            break
+        k += 1
+
+    batches: list[list[Any]] = []
+    idx = 0
+    for i in range(k):
+        size = base + (1 if i < rem else 0)
+        batches.append(items[idx : idx + size])
+        idx += size
+    return batches
+
+
 def _priority_sort_key(name: str, priority_index: dict) -> tuple:
     return (
         priority_index.get(name, {}).get("source_rank_score") is None,
@@ -21,16 +48,12 @@ def _priority_sort_key(name: str, priority_index: dict) -> tuple:
     )
 
 
-def list_pending_demands_by_category_ids(
+def resolve_demands_for_category_ids(
     biz_dt: str,
     category_ids: list[int],
-    *,
-    max_demands: int = MAX_DEMANDS_PER_BATCH,
-    excluded_demand_names: list[str] | None = None,
 ) -> list[dict[str, Any]]:
-    """按分类节点取待分级需求池记录（pool_id + demand_name），每批最多 max_demands 条。"""
+    """按分类节点解析全部需求池记录（pool_id + demand_name）。"""
     selected_ids = list(dict.fromkeys(int(value) for value in category_ids))
-    excluded = set(excluded_demand_names or [])
     with get_session() as session:
         belongs = DemandBelongCategoryRepository(session).list_by_category_ids(selected_ids)
         pool_ids_by_belong = DemandBelongPoolRelRepository(session).get_pool_ids_by_belong_ids(
@@ -44,10 +67,10 @@ def list_pending_demands_by_category_ids(
         for row in pool_rows:
             if row.biz_dt != biz_dt or not row.demand_name:
                 continue
-            demand_name = str(row.demand_name)
-            if demand_name in excluded:
-                continue
-            candidates.append({"pool_id": int(row.id), "demand_name": demand_name})
+            candidates.append({
+                "pool_id": int(row.id),
+                "demand_name": str(row.demand_name),
+            })
 
     candidates.sort(
         key=lambda item: (
@@ -55,5 +78,4 @@ def list_pending_demands_by_category_ids(
             item["pool_id"],
         )
     )
-    limit = max(1, min(int(max_demands), MAX_DEMANDS_PER_BATCH))
-    return candidates[:limit]
+    return candidates

supply_infra/video_points.py

@@ -0,0 +1,150 @@
+"""视频点位解析与序列化 — decode_result / JSON 字段 ↔ 行记录。"""
+from __future__ import annotations
+
+import json
+from typing import Any
+
+from supply_infra.db.models.multi_demand_video_point import (
+    POINT_TYPE_INSPIRATION,
+    POINT_TYPE_KEY,
+    POINT_TYPE_PURPOSE,
+)
+
+_POINT_KEY = "点"
+_POINT_DESC_KEY = "点描述"
+
+# decode_result 中的 key → point_type
+DECODE_RESULT_KEY_TO_POINT_TYPE = {
+    "灵感点": POINT_TYPE_INSPIRATION,
+    "目的点": POINT_TYPE_PURPOSE,
+    "关键点": POINT_TYPE_KEY,
+}
+
+# 原 JSON 列名 → point_type
+JSON_FIELD_TO_POINT_TYPE = {
+    "inspiration_points_json": POINT_TYPE_INSPIRATION,
+    "purpose_points_json": POINT_TYPE_PURPOSE,
+    "key_points_json": POINT_TYPE_KEY,
+}
+
+# point_type → 原 JSON 列名（API 兼容）
+POINT_TYPE_TO_JSON_FIELD = {v: k for k, v in JSON_FIELD_TO_POINT_TYPE.items()}
+
+
+def _item_to_row(
+    video_id: str, point_type: str, item: dict[str, Any]
+) -> dict[str, str | None] | None:
+    point = item.get(_POINT_KEY)
+    point_desc = item.get(_POINT_DESC_KEY)
+    if point is None and point_desc is None:
+        return None
+    return {
+        "video_id": video_id,
+        "point_type": point_type,
+        "point_data": str(point) if point is not None else None,
+        "point_desc": str(point_desc) if point_desc is not None else None,
+    }
+
+
+def points_from_decode_payload(
+    video_id: str, payload: dict[str, Any]
+) -> list[dict[str, str | None]]:
+    """从 decode_result 提取全部点位行。"""
+    rows: list[dict[str, str | None]] = []
+    for decode_key, point_type in DECODE_RESULT_KEY_TO_POINT_TYPE.items():
+        items = payload.get(decode_key)
+        if not isinstance(items, list):
+            continue
+        for item in items:
+            if not isinstance(item, dict):
+                continue
+            row = _item_to_row(video_id, point_type, item)
+            if row:
+                rows.append(row)
+    return rows
+
+
+def points_from_json_fields(
+    video_id: str,
+    *,
+    inspiration_points_json: str | None = None,
+    purpose_points_json: str | None = None,
+    key_points_json: str | None = None,
+) -> list[dict[str, str | None]]:
+    """从 multi_demand_video_detail 的三个 JSON 列提取点位行。"""
+    field_values = {
+        "inspiration_points_json": inspiration_points_json,
+        "purpose_points_json": purpose_points_json,
+        "key_points_json": key_points_json,
+    }
+    rows: list[dict[str, str | None]] = []
+    for field, point_type in JSON_FIELD_TO_POINT_TYPE.items():
+        text = field_values.get(field)
+        if not text:
+            continue
+        try:
+            items = json.loads(text)
+        except json.JSONDecodeError:
+            continue
+        if not isinstance(items, list):
+            continue
+        for item in items:
+            if not isinstance(item, dict):
+                continue
+            row = _item_to_row(video_id, point_type, item)
+            if row:
+                rows.append(row)
+    return rows
+
+
+def json_fields_from_point_rows(
+    rows: list[dict[str, Any]],
+) -> dict[str, str | None]:
+    """将点位行重组为三个 JSON 列（保持 API 兼容）。"""
+    grouped: dict[str, list[dict[str, str | None]]] = {
+        POINT_TYPE_INSPIRATION: [],
+        POINT_TYPE_PURPOSE: [],
+        POINT_TYPE_KEY: [],
+    }
+    for row in rows:
+        point_type = row.get("point_type")
+        if point_type not in grouped:
+            continue
+        grouped[point_type].append(
+            {
+                _POINT_KEY: row.get("point_data"),
+                _POINT_DESC_KEY: row.get("point_desc"),
+            }
+        )
+
+    result: dict[str, str | None] = {}
+    for point_type, field in POINT_TYPE_TO_JSON_FIELD.items():
+        items = grouped[point_type]
+        result[field] = (
+            json.dumps(items, ensure_ascii=False) if items else None
+        )
+    return result
+
+
+def extract_points_json_from_decode(payload: dict[str, Any]) -> dict[str, str | None]:
+    """从 decode_result 提取灵感点/目的点/关键点 JSON 列（原逻辑）。"""
+    result: dict[str, str | None] = {}
+    for decode_key, field in {
+        k: POINT_TYPE_TO_JSON_FIELD[v]
+        for k, v in DECODE_RESULT_KEY_TO_POINT_TYPE.items()
+    }.items():
+        items = payload.get(decode_key)
+        if not isinstance(items, list):
+            result[field] = None
+            continue
+        points: list[dict[str, Any]] = []
+        for item in items:
+            if not isinstance(item, dict):
+                continue
+            point = item.get(_POINT_KEY)
+            point_desc = item.get(_POINT_DESC_KEY)
+            if point is None and point_desc is None:
+                continue
+            points.append({_POINT_KEY: point, _POINT_DESC_KEY: point_desc})
+        result[field] = json.dumps(points, ensure_ascii=False) if points else None
+    return result

web/src/components/DemandPathPanel.vue

@@ -100,8 +100,9 @@ const topicSections = computed((): TopicSection[] | null => {
 })
 
 watch(
-  () => [props.open, props.categoryName] as const,
-  () => {
+  () => props.open,
+  (open) => {
+    if (open) return
     selectedDemandId.value = null
     videos.value = []
     videosError.value = null
@@ -110,13 +111,18 @@ watch(
 )
 
 watch(
-  () => props.highlightDemandId,
-  (nextId) => {
+  () => [props.highlightDemandId, props.items] as const,
+  ([nextId, items]) => {
     if (!props.open) return
-    if (!nextId) return
-    const item = props.items.find((i) => i.id === nextId)
+    if (!nextId) {
+      selectedDemandId.value = null
+      videos.value = []
+      videosError.value = null
+      selectedVid.value = null
+      return
+    }
+    const item = items.find((i) => i.id === nextId)
     if (!item) return
-    if (selectedDemandId.value === nextId) return
     void selectDemand(item)
   },
   { immediate: true },
@@ -315,10 +321,11 @@ function selectVideo(video: DemandVideoItem) {
 .path-rail {
   display: flex;
   flex-direction: row;
-  align-items: flex-start;
+  align-items: stretch;
   gap: 0;
   padding-left: 0;
   min-height: 0;
+  height: 100%;
   width: max-content;
   min-width: 100%;
   max-width: 100%;
@@ -331,7 +338,7 @@ function selectVideo(video: DemandVideoItem) {
   flex-direction: column;
   gap: 10px;
   min-height: 0;
-  max-height: 100%;
+  height: 100%;
 }
 
 .col-topic {
@@ -392,7 +399,8 @@ function selectVideo(video: DemandVideoItem) {
   display: flex;
   flex-direction: column;
   gap: 8px;
-  max-height: min(200px, 24vh);
+  flex: 1;
+  min-height: 80px;
   overflow: auto;
   overscroll-behavior: contain;
   padding-right: 2px;
@@ -619,7 +627,8 @@ function selectVideo(video: DemandVideoItem) {
   border-radius: 12px;
   background: linear-gradient(180deg, #faf5ff 0%, #fff 45%);
   overflow: hidden;
-  max-height: min(220px, 26vh);
+  flex: 1;
+  min-height: 80px;
   display: flex;
   flex-direction: column;
 }
@@ -791,14 +800,6 @@ function selectVideo(video: DemandVideoItem) {
 }
 
 @media (max-height: 800px) {
-  .card-list {
-    max-height: min(150px, 20vh);
-  }
-
-  .topic-card {
-    max-height: min(170px, 22vh);
-  }
-
   .arrow-col {
     padding-top: 32px;
   }

web/src/components/IcicleHeatTree.vue

@@ -90,9 +90,69 @@ const tooltipNode = ref<PreparedNode | null>(null)
 const inspectOpen = ref(false)
 const inspectNode = ref<PreparedNode | null>(null)
 
+const PATH_DOCK_HEIGHT_KEY = 'icicle-path-dock-height'
+const PATH_DOCK_MIN = 120
+const PATH_DOCK_MAX_RATIO = 0.75
+
+function loadPathDockHeight(): number {
+  const raw = localStorage.getItem(PATH_DOCK_HEIGHT_KEY)
+  const n = raw ? Number(raw) : NaN
+  return Number.isFinite(n) && n >= PATH_DOCK_MIN ? n : 240
+}
+
+const pathDockHeight = ref(loadPathDockHeight())
+const pathDockResizing = ref(false)
+
+function pathDockMaxHeight(): number {
+  return Math.floor(window.innerHeight * PATH_DOCK_MAX_RATIO)
+}
+
+function onPathDockResizeStart(event: PointerEvent) {
+  event.preventDefault()
+  pathDockResizing.value = true
+  const startY = event.clientY
+  const startHeight = pathDockHeight.value
+
+  const onMove = (e: PointerEvent) => {
+    const delta = startY - e.clientY
+    pathDockHeight.value = Math.max(
+      PATH_DOCK_MIN,
+      Math.min(pathDockMaxHeight(), startHeight + delta),
+    )
+  }
+
+  const onUp = () => {
+    pathDockResizing.value = false
+    localStorage.setItem(PATH_DOCK_HEIGHT_KEY, String(pathDockHeight.value))
+    window.removeEventListener('pointermove', onMove)
+    window.removeEventListener('pointerup', onUp)
+    window.removeEventListener('pointercancel', onUp)
+  }
+
+  window.addEventListener('pointermove', onMove)
+  window.addEventListener('pointerup', onUp)
+  window.addEventListener('pointercancel', onUp)
+}
+
 const inspectItems = computed<DemandGradeItem[]>(() => {
   if (!inspectNode.value) return []
-  return props.demandsByCategory?.[inspectNode.value.id] ?? []
+  const categoryId = inspectNode.value.id
+  const all = props.demandsByCategory?.[categoryId] ?? []
+
+  if (
+    listSelectedDemandId.value != null &&
+    listSelectedCategoryId.value === categoryId
+  ) {
+    const picked = all.find((item) => item.id === listSelectedDemandId.value)
+    if (picked) return [picked]
+  }
+  return all
+})
+
+const inspectDemandName = computed(() => {
+  if (listSelectedDemandId.value == null) return null
+  const card = demandCards.value.find((c) => c.id === listSelectedDemandId.value)
+  return card?.demand_name ?? null
 })
 
 const listSelectedDemandId = ref<number | null>(null)
@@ -932,10 +992,24 @@ onUnmounted(() => {
       </aside>
     </div>
 
-    <div v-if="inspectOpen && inspectNode" class="path-dock">
+    <div
+      v-if="inspectOpen && inspectNode"
+      class="path-dock"
+      :class="{ resizing: pathDockResizing }"
+      :style="{ height: `${pathDockHeight}px` }"
+    >
+      <div
+        class="path-dock-resize"
+        role="separator"
+        aria-orientation="horizontal"
+        aria-label="调整详情区域高度"
+        title="上下拖拽调整高度"
+        @pointerdown="onPathDockResizeStart"
+      />
       <div class="path-dock-head">
         <div class="path-dock-meta">
           <span class="path-dock-label">节点详情</span>
+          <span v-if="inspectDemandName" class="path-dock-demand">{{ inspectDemandName }}</span>
           <span class="path-dock-name">{{ inspectNode.name }}</span>
         </div>
         <button
@@ -1180,9 +1254,8 @@ onUnmounted(() => {
 .path-dock {
   display: flex;
   flex-direction: column;
-  flex: 0 1 auto;
+  flex: 0 0 auto;
   min-height: 0;
-  max-height: min(32vh, 280px);
   overflow: hidden;
   overscroll-behavior: contain;
   border: 1px solid #e2e8f0;
@@ -1191,6 +1264,38 @@ onUnmounted(() => {
   isolation: isolate;
 }
 
+.path-dock-resize {
+  flex-shrink: 0;
+  height: 10px;
+  cursor: ns-resize;
+  touch-action: none;
+  position: relative;
+  background: #f8fafc;
+  border-bottom: 1px solid #e2e8f0;
+}
+
+.path-dock-resize::before {
+  content: '';
+  position: absolute;
+  left: 50%;
+  top: 50%;
+  transform: translate(-50%, -50%);
+  width: 40px;
+  height: 4px;
+  border-radius: 999px;
+  background: #cbd5e1;
+  transition: background 0.15s ease;
+}
+
+.path-dock-resize:hover::before,
+.path-dock.resizing .path-dock-resize::before {
+  background: #64748b;
+}
+
+.path-dock.resizing {
+  user-select: none;
+}
+
 .path-dock-head {
   display: flex;
   align-items: center;
@@ -1226,6 +1331,15 @@ onUnmounted(() => {
   white-space: nowrap;
 }
 
+.path-dock-demand {
+  font-size: 14px;
+  font-weight: 700;
+  color: #1d4ed8;
+  overflow: hidden;
+  text-overflow: ellipsis;
+  white-space: nowrap;
+}
+
 .path-dock-close {
   flex-shrink: 0;
   height: 30px;
@@ -1251,12 +1365,8 @@ onUnmounted(() => {
   overflow: auto;
   overscroll-behavior: contain;
   padding: 12px 14px;
-}
-
-@media (max-height: 800px) {
-  .path-dock {
-    max-height: min(28vh, 220px);
-  }
+  display: flex;
+  flex-direction: column;
 }
 
 .depth-axis {