修改定时任务执行逻辑

api/app.py

@@ -1,6 +1,7 @@
 """FastAPI application — category tree API on port 8080."""
 from __future__ import annotations
 
+from contextlib import asynccontextmanager
 from pathlib import Path
 
 from fastapi import FastAPI, HTTPException, Query
@@ -13,8 +14,19 @@ 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
+from supply_infra.db import init_db
+from supply_infra.scheduler import get_scheduler_status, start_scheduler, stop_scheduler
 
-app = FastAPI(title="SupplyAgent API", version="0.1.0")
+
+@asynccontextmanager
+async def lifespan(_app: FastAPI):
+    init_db()
+    start_scheduler()
+    yield
+    stop_scheduler()
+
+
+app = FastAPI(title="SupplyAgent API", version="0.1.0", lifespan=lifespan)
 
 app.add_middleware(
     CORSMiddleware,
@@ -35,6 +47,12 @@ def health() -> dict[str, str]:
     return {"status": "ok"}
 
 
+@app.get("/api/scheduler/status")
+def scheduler_status() -> dict:
+    """Return scheduler enabled/running state and next run times."""
+    return get_scheduler_status()
+
+
 @app.get("/api/category-tree")
 def category_tree(
     biz_dt: str | None = Query(

api/run.py

@@ -1,10 +1,16 @@
 """Start the SupplyAgent API on port 8080."""
 from __future__ import annotations
 
+import logging
+
 import uvicorn
 
 
 def main() -> None:
+    logging.basicConfig(
+        level=logging.INFO,
+        format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
+    )
     uvicorn.run(
         "api.app:app",
         host="0.0.0.0",

jobs/grade_demand_pool.py

@@ -1,6 +1,8 @@
 #!/usr/bin/env python3
 """手动执行需求分级：循环取批次调用 demand_grade_agent，直到需求池分级完毕。
 
+默认每轮 5 个线程并行，各处理一批互不重叠的需求词。
+
 用法:
     python jobs/grade_demand_pool.py                      # 默认业务日、批次20、最多200批
     python jobs/grade_demand_pool.py 20260716              # 指定业务日
@@ -27,7 +29,8 @@ def main(
 ) -> dict:
     batch_size = int(batch_size_arg) if batch_size_arg else 20
     if max_batches_arg is None:
-        max_batches: int | None = 200
+        # 交给下游 job 根据数据库数据自动计算动态上限。
+        max_batches = None
     elif max_batches_arg.lower() in {"all", "0", "-1"}:
         max_batches = None
     else:

jobs/run_supply_pipeline.py

@@ -0,0 +1,22 @@
+#!/usr/bin/env python3
+"""手动执行供应链数据流水线（全局树 → 需求池 → 分级）。"""
+
+import logging
+import sys
+
+from supply_infra.scheduler.jobs.run_supply_pipeline import run_supply_pipeline
+
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
+)
+
+
+def main() -> None:
+    biz_dt = sys.argv[1] if len(sys.argv) > 1 else None
+    result = run_supply_pipeline(biz_dt)
+    print(result)
+
+
+if __name__ == "__main__":
+    main()

supply_infra/db/models/__init__.py

@@ -12,6 +12,7 @@ from supply_infra.db.models.global_tree_element import GlobalTreeElement
 from supply_infra.db.models.multi_demand_pool_di import MultiDemandPoolDi
 from supply_infra.db.models.multi_demand_video_detail import MultiDemandVideoDetail
 from supply_infra.db.models.oss_log import OssLog
+from supply_infra.db.models.scheduler_job_execution import SchedulerJobExecution
 
 __all__ = [
     "CategoryTreeWeight",
@@ -26,4 +27,5 @@ __all__ = [
     "MultiDemandPoolDi",
     "MultiDemandVideoDetail",
     "OssLog",
+    "SchedulerJobExecution",
 ]

supply_infra/db/models/scheduler_job_execution.py

@@ -0,0 +1,51 @@
+from __future__ import annotations
+
+from datetime import datetime
+
+from sqlalchemy import BigInteger, Numeric, String, Text, func
+from sqlalchemy.orm import Mapped, mapped_column
+
+from supply_infra.db.base import Base
+
+
+class SchedulerJobExecution(Base):
+    """定时任务执行记录（开始/结束各记一条，同一 run_id 关联）。"""
+
+    __tablename__ = "scheduler_job_execution"
+
+    id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
+    run_id: Mapped[str] = mapped_column(
+        String(36),
+        nullable=False,
+        index=True,
+        comment="同一次执行的唯一标识",
+    )
+    job_name: Mapped[str] = mapped_column(String(128), nullable=False, comment="定时任务名称")
+    job_id: Mapped[str | None] = mapped_column(String(64), nullable=True, comment="调度器 job id")
+    status: Mapped[str] = mapped_column(
+        String(32),
+        nullable=False,
+        comment="状态: started/finished/failed/skipped",
+    )
+    event_time: Mapped[datetime] = mapped_column(nullable=False, comment="事件发生时间")
+    biz_dt: Mapped[str | None] = mapped_column(String(8), nullable=True, comment="业务日 YYYYMMDD")
+    started_at: Mapped[datetime | None] = mapped_column(nullable=True, comment="任务开始时间")
+    finished_at: Mapped[datetime | None] = mapped_column(nullable=True, comment="任务结束时间")
+    duration_seconds: Mapped[float | None] = mapped_column(
+        Numeric(10, 2),
+        nullable=True,
+        comment="执行耗时（秒）",
+    )
+    error_message: Mapped[str | None] = mapped_column(Text, nullable=True, comment="失败原因")
+    detail: Mapped[str | None] = mapped_column(Text, nullable=True, comment="执行结果摘要 JSON")
+    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_video_detail_repo import (
     MultiDemandVideoDetailRepository,
 )
 from supply_infra.db.repositories.oss_log_repo import OssLogRepository
+from supply_infra.db.repositories.scheduler_job_execution_repo import (
+    SchedulerJobExecutionRepository,
+)
 
 __all__ = [
     "BaseRepository",
@@ -40,4 +43,5 @@ __all__ = [
     "MultiDemandPoolDiRepository",
     "MultiDemandVideoDetailRepository",
     "OssLogRepository",
+    "SchedulerJobExecutionRepository",
 ]

supply_infra/db/repositories/scheduler_job_execution_repo.py

@@ -0,0 +1,50 @@
+from __future__ import annotations
+
+from datetime import datetime
+
+from sqlalchemy import select
+
+from supply_infra.db.models.scheduler_job_execution import SchedulerJobExecution
+from supply_infra.db.repositories.base import BaseRepository
+
+
+class SchedulerJobExecutionRepository(BaseRepository[SchedulerJobExecution]):
+    model = SchedulerJobExecution
+
+    def log_event(
+        self,
+        *,
+        run_id: str,
+        job_name: str,
+        status: str,
+        event_time: datetime,
+        job_id: str | None = None,
+        biz_dt: str | None = None,
+        started_at: datetime | None = None,
+        finished_at: datetime | None = None,
+        duration_seconds: float | None = None,
+        error_message: str | None = None,
+        detail: str | None = None,
+    ) -> SchedulerJobExecution:
+        entity = SchedulerJobExecution(
+            run_id=run_id,
+            job_name=job_name,
+            job_id=job_id,
+            status=status,
+            event_time=event_time,
+            biz_dt=biz_dt,
+            started_at=started_at,
+            finished_at=finished_at,
+            duration_seconds=duration_seconds,
+            error_message=error_message,
+            detail=detail,
+        )
+        return self.add(entity)
+
+    def list_recent(self, *, limit: int = 50) -> list[SchedulerJobExecution]:
+        stmt = (
+            select(SchedulerJobExecution)
+            .order_by(SchedulerJobExecution.event_time.desc(), SchedulerJobExecution.id.desc())
+            .limit(limit)
+        )
+        return list(self.session.scalars(stmt).all())

supply_infra/scheduler/__init__.py

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

supply_infra/scheduler/app.py

@@ -1,71 +1,140 @@
 from __future__ import annotations
 
 import logging
+import signal
+import time
+from typing import TYPE_CHECKING
 
-from apscheduler.schedulers.blocking import BlockingScheduler
+from apscheduler.schedulers.background import BackgroundScheduler
 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,
-)
+from supply_infra.scheduler.constants import SUPPLY_PIPELINE_JOB_ID, SUPPLY_PIPELINE_JOB_NAME
+from supply_infra.scheduler.jobs.run_supply_pipeline import run_supply_pipeline
+
+if TYPE_CHECKING:
+    from apscheduler.schedulers.base import BaseScheduler
 
 logger = logging.getLogger(__name__)
 
+_scheduler: BackgroundScheduler | None = None
 
-def create_scheduler() -> BlockingScheduler:
-    """Create and configure the scheduler with all registered jobs."""
-    settings = get_infra_settings()
-    scheduler = BlockingScheduler(timezone=settings.scheduler_timezone)
+_PIPELINE_CRON_HOURS = "9,15,21"
 
-    # 每天凌晨 2:30 从 ODPS 同步全局树元素与分类到 MySQL
-    scheduler.add_job(
-        sync_global_tree_odps_to_mysql,
-        trigger=CronTrigger(hour=2, minute=30),
-        id="sync_global_tree_odps_to_mysql",
-        name="ODPS → MySQL 全局树同步",
-        replace_existing=True,
-    )
 
-    # 每天 12:00 从 ODPS 同步策略需求天级表到 MySQL（当天 dt）
-    scheduler.add_job(
-        sync_multi_demand_pool_odps_to_mysql,
-        trigger=CronTrigger(hour=12, minute=0),
-        id="sync_multi_demand_pool_odps_to_mysql",
-        name="ODPS → MySQL 策略需求池同步",
-        replace_existing=True,
-    )
+def create_scheduler() -> BackgroundScheduler:
+    """Create and configure the scheduler with the chained supply pipeline job."""
+    settings = get_infra_settings()
+    scheduler = BackgroundScheduler(timezone=settings.scheduler_timezone)
 
-    # 每天 13:30 对需求池循环打分级（在 12:00 同步/热度计算完成后运行）。
-    # 循环控制在 job 内部：每批取少量未分级需求调用一次 agent、等待完成后再取下一批，
-    # 避免单次 agent 会话内自行分页导致上下文无限增长；max_batches 作为单次运行的安全阀。
+    # 每天 9:00 / 15:00 / 21:00 串行执行：全局树 → 需求池 → 分级
     scheduler.add_job(
-        grade_demand_pool,
-        trigger=CronTrigger(hour=13, minute=30),
-        id="grade_demand_pool",
-        name="需求池分级评估",
+        run_supply_pipeline,
+        trigger=CronTrigger(hour=_PIPELINE_CRON_HOURS, minute=0),
+        id=SUPPLY_PIPELINE_JOB_ID,
+        name=SUPPLY_PIPELINE_JOB_NAME,
         replace_existing=True,
+        max_instances=1,
+        coalesce=True,
+        misfire_grace_time=3600,
     )
 
     logger.info("Scheduler configured with %d job(s)", len(scheduler.get_jobs()))
     return scheduler
 
 
-def run_scheduler() -> None:
-    """Start the blocking scheduler (CLI entry point)."""
+def _log_next_runs(scheduler: BaseScheduler) -> None:
+    for job in scheduler.get_jobs():
+        logger.info("  - %s | next run: %s", job.name, job.next_run_time)
+
+
+def start_scheduler() -> BackgroundScheduler | None:
+    """Start the background scheduler (idempotent). Used by API lifespan."""
+    global _scheduler
+
     settings = get_infra_settings()
     if not settings.scheduler_enabled:
         logger.warning("Scheduler is disabled (SCHEDULER_ENABLED=false)")
-        return
+        return None
+
+    if _scheduler is not None and _scheduler.running:
+        return _scheduler
 
-    scheduler = create_scheduler()
+    _scheduler = create_scheduler()
     logger.info("Starting scheduler...")
-    for job in scheduler.get_jobs():
-        logger.info("  - %s | next run: %s", job.name, job.next_run_time)
+    _log_next_runs(_scheduler)
+    _scheduler.start()
+    return _scheduler
+
+
+def get_scheduler_status() -> dict:
+    """Return scheduler runtime status for health checks."""
+    settings = get_infra_settings()
+    if not settings.scheduler_enabled:
+        return {
+            "enabled": False,
+            "running": False,
+            "timezone": settings.scheduler_timezone,
+            "jobs": [],
+        }
+
+    if _scheduler is None or not _scheduler.running:
+        return {
+            "enabled": True,
+            "running": False,
+            "timezone": settings.scheduler_timezone,
+            "jobs": [],
+        }
+
+    jobs = []
+    for job in _scheduler.get_jobs():
+        next_run = job.next_run_time
+        jobs.append(
+            {
+                "id": job.id,
+                "name": job.name,
+                "next_run_time": next_run.isoformat() if next_run else None,
+            }
+        )
+
+    return {
+        "enabled": True,
+        "running": True,
+        "timezone": settings.scheduler_timezone,
+        "jobs": jobs,
+    }
+
+
+def stop_scheduler() -> None:
+    """Shut down the background scheduler if running."""
+    global _scheduler
+
+    if _scheduler is None or not _scheduler.running:
+        return
+
+    logger.info("Stopping scheduler...")
+    _scheduler.shutdown(wait=False)
+    _scheduler = None
+    logger.info("Scheduler stopped.")
+
+
+def run_scheduler() -> None:
+    """Start the scheduler and block until interrupted (CLI entry point)."""
+    scheduler = start_scheduler()
+    if scheduler is None:
+        return
+
+    def _handle_exit(signum: int, _frame: object) -> None:
+        logger.info("Received signal %s, shutting down scheduler...", signum)
+        stop_scheduler()
+        raise SystemExit(0)
+
+    signal.signal(signal.SIGINT, _handle_exit)
+    signal.signal(signal.SIGTERM, _handle_exit)
 
     try:
-        scheduler.start()
+        while scheduler.running:
+            time.sleep(3600)
     except (KeyboardInterrupt, SystemExit):
+        stop_scheduler()
         logger.info("Scheduler stopped.")

supply_infra/scheduler/constants.py

@@ -0,0 +1,4 @@
+"""Scheduler job identifiers shared by app and job implementations."""
+
+SUPPLY_PIPELINE_JOB_ID = "run_supply_pipeline"
+SUPPLY_PIPELINE_JOB_NAME = "供应链数据流水线"

supply_infra/scheduler/job_execution.py

@@ -0,0 +1,120 @@
+"""定时任务执行记录写入辅助。"""
+from __future__ import annotations
+
+import json
+import logging
+import uuid
+from datetime import datetime
+from typing import Any
+
+from supply_infra.db.repositories.scheduler_job_execution_repo import (
+    SchedulerJobExecutionRepository,
+)
+from supply_infra.db.session import get_session
+
+logger = logging.getLogger(__name__)
+
+STATUS_STARTED = "started"
+STATUS_FINISHED = "finished"
+STATUS_FAILED = "failed"
+STATUS_SKIPPED = "skipped"
+
+
+def _serialize_detail(payload: Any) -> str | None:
+    if payload is None:
+        return None
+    return json.dumps(payload, ensure_ascii=False, default=str)
+
+
+class JobExecutionRecorder:
+    """在任务开始/结束时各写一条执行记录。"""
+
+    def __init__(
+        self,
+        *,
+        job_name: str,
+        job_id: str | None = None,
+        biz_dt: str | None = None,
+    ) -> None:
+        self.run_id = str(uuid.uuid4())
+        self.job_name = job_name
+        self.job_id = job_id
+        self.biz_dt = biz_dt
+        self.started_at = datetime.now()
+
+    def record_started(self) -> None:
+        try:
+            with get_session() as session:
+                SchedulerJobExecutionRepository(session).log_event(
+                    run_id=self.run_id,
+                    job_name=self.job_name,
+                    job_id=self.job_id,
+                    status=STATUS_STARTED,
+                    event_time=self.started_at,
+                    biz_dt=self.biz_dt,
+                    started_at=self.started_at,
+                )
+        except Exception:
+            logger.exception("Failed to record job start: job_name=%s run_id=%s", self.job_name, self.run_id)
+
+    def record_finished(
+        self,
+        *,
+        success: bool,
+        result: dict[str, Any] | None = None,
+        error_message: str | None = None,
+    ) -> None:
+        finished_at = datetime.now()
+        duration_seconds = round((finished_at - self.started_at).total_seconds(), 2)
+        status = STATUS_FINISHED if success else STATUS_FAILED
+
+        try:
+            with get_session() as session:
+                SchedulerJobExecutionRepository(session).log_event(
+                    run_id=self.run_id,
+                    job_name=self.job_name,
+                    job_id=self.job_id,
+                    status=status,
+                    event_time=finished_at,
+                    biz_dt=self.biz_dt,
+                    started_at=self.started_at,
+                    finished_at=finished_at,
+                    duration_seconds=duration_seconds,
+                    error_message=error_message,
+                    detail=_serialize_detail(result),
+                )
+        except Exception:
+            logger.exception(
+                "Failed to record job finish: job_name=%s run_id=%s status=%s",
+                self.job_name,
+                self.run_id,
+                status,
+            )
+
+
+def record_skipped(
+    *,
+    job_name: str,
+    job_id: str | None = None,
+    biz_dt: str | None = None,
+    reason: str,
+) -> None:
+    """记录因并发等原因跳过的执行。"""
+    now = datetime.now()
+    run_id = str(uuid.uuid4())
+    try:
+        with get_session() as session:
+            SchedulerJobExecutionRepository(session).log_event(
+                run_id=run_id,
+                job_name=job_name,
+                job_id=job_id,
+                status=STATUS_SKIPPED,
+                event_time=now,
+                biz_dt=biz_dt,
+                started_at=now,
+                finished_at=now,
+                duration_seconds=0,
+                error_message=reason,
+            )
+    except Exception:
+        logger.exception("Failed to record skipped job: job_name=%s reason=%s", job_name, reason)

supply_infra/scheduler/jobs/grade_demand_pool.py

@@ -2,17 +2,22 @@
 定时任务：对 multi_demand_pool_di 需求池中的需求循环打分级。
 
 循环控制在本文件（调度任务侧），而不是让 agent 在一次运行内自行分页遍历——
-每一批只挑选少量（默认 20 个）尚未分级的需求词，调用一次 demand_grade_agent
-并等待其运行结束（agent 内部会调用 batch_save_demand_grades 落库），再重新
-查询"已分级"集合、找出下一批新词，如此循环直到没有更多待分级需求或达到本次
-运行的批次上限，从而避免单次 agent 会话上下文无限增长。
+每一轮并行启动多个 worker（默认 5 个线程），每个 worker 处理一批互不重叠的
+需求词（默认每批 20 个），调用一次 demand_grade_agent 并等待其运行结束
+（agent 内部会调用 batch_save_demand_grades 落库），再重新查询"已分级"集合、
+分配下一批新词，如此循环直到没有更多待分级需求或达到本次运行的批次上限，
+从而避免单次 agent 会话上下文无限增长，同时提升吞吐。
 """
 from __future__ import annotations
 
 import logging
+import math
+from concurrent.futures import ThreadPoolExecutor, as_completed
 from datetime import datetime
+from zoneinfo import ZoneInfo
 
 from agents.demand_grade_agent.run import main as grade_demand_words
+from supply_infra.config import get_infra_settings
 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
@@ -20,69 +25,107 @@ from supply_infra.db.session import get_session
 logger = logging.getLogger(__name__)
 
 _DEFAULT_BATCH_SIZE = 20
-_DEFAULT_MAX_BATCHES = 200
+_DEFAULT_WORKERS = 5
 
 
 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()
+    # 定时任务默认按“当天业务日”执行；使用 scheduler_timezone 以免机器时区不同导致 biz_dt 偏移。
+    settings = get_infra_settings()
+    tz = ZoneInfo(settings.scheduler_timezone)
+    return datetime.now(tz).strftime("%Y%m%d")
 
 
-def _fetch_next_batch(biz_dt: str, batch_size: int, graded_names: set[str]) -> list[str]:
-    """按 exclude_names 过滤已分级词，取下一批（不用 offset，避免和落库进度错位）。"""
+def _fetch_next_batch(biz_dt: str, batch_size: int, exclude_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,
+            exclude_names=list(exclude_names) if exclude_names else None,
         )
     return [item["demand_name"] for item in summaries]
 
 
+def _allocate_parallel_batches(
+    biz_dt: str,
+    batch_size: int,
+    exclude_names: set[str],
+    *,
+    num_workers: int,
+) -> list[list[str]]:
+    """
+    为同一轮并行 worker 分配互不重叠的批次。
+
+    每取出一批后立即加入 reserved，下一批查询时排除，保证单轮内不会重复分配。
+    """
+    batches: list[list[str]] = []
+    reserved = set(exclude_names)
+    for _ in range(num_workers):
+        batch = _fetch_next_batch(biz_dt, batch_size, reserved)
+        if not batch:
+            break
+        batches.append(batch)
+        reserved.update(batch)
+    return batches
+
+
 def _fetch_graded_names(biz_dt: str) -> set[str]:
     with get_session() as session:
         return DemandGradeRepository(session).get_existing_demand_names(biz_dt)
 
 
+def _run_batch_in_thread(batch: list[str], biz_dt: str, batch_label: str) -> None:
+    logger.info("Grade demand pool %s (size=%d): %s", batch_label, len(batch), batch)
+    grade_demand_words(batch, biz_dt=biz_dt)
+
+
 def grade_demand_pool(
     biz_dt: str | None = None,
     *,
     batch_size: int = _DEFAULT_BATCH_SIZE,
-    max_batches: int | None = _DEFAULT_MAX_BATCHES,
+    max_batches: int | None = None,
+    workers: int = _DEFAULT_WORKERS,
 ) -> dict:
     """
-    循环对需求池中尚未分级的需求打分级，每批调用一次 agent 并等待其完成。
+    循环对需求池中尚未分级的需求打分级，每轮并行调用多个 agent 并等待其完成。
 
     Args:
         biz_dt: 业务日期 YYYYMMDD；不传则取需求池最新业务日。
         batch_size: 每批交给 agent 的需求词数量。
-        max_batches: 本次运行最多执行多少批（安全阀，避免单次运行时间过长/无限循环）；
-                     None 表示不限制，跑到没有待分级需求为止。
+        max_batches: 本次运行最多执行多少批（安全阀，避免单次运行时间过长/无限循环）。
+                     不传/为 None 时，将根据当天数据库待分级需求词数量动态计算上限：
+                     `ceil(total / batch_size) + 10`。
+        workers: 并行线程数；每轮最多同时跑 workers 个互不重叠的批次。
 
     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)
 
+    # max_batches 默认为动态上限：ceil(total / batch_size) + 10
+    # 其中 total 使用“当天需求词去重数”，与本 job 的分页维度一致。
+    if max_batches is None:
+        batches_needed = math.ceil(total / batch_size) if batch_size > 0 else 0
+        max_batches = batches_needed + 10
+
+    worker_count = max(1, workers)
+
     logger.info(
-        "Grade demand pool start: biz_dt=%s total=%d graded=%d batch_size=%d max_batches=%s",
+        "Grade demand pool start: biz_dt=%s total=%d graded=%d batch_size=%d "
+        "workers=%d max_batches=%s",
         resolved_biz_dt,
         total,
         graded_before,
         batch_size,
+        worker_count,
         max_batches,
     )
 
@@ -94,34 +137,63 @@ def grade_demand_pool(
             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"
+        if max_batches is not None:
+            slots = max_batches - batches_run
+            num_workers = min(worker_count, slots)
+        else:
+            num_workers = worker_count
+
+        if num_workers <= 0:
+            stopped_reason = "max_batches_reached"
             break
 
-        batches_run += 1
-        logger.info(
-            "Grade demand pool batch %d (size=%d): %s",
-            batches_run,
-            len(batch),
-            batch,
+        batches = _allocate_parallel_batches(
+            resolved_biz_dt,
+            batch_size,
+            graded_names,
+            num_workers=num_workers,
         )
+        if not batches:
+            stopped_reason = "no_more_pending"
+            break
 
-        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
-            )
+        round_start_batch = batches_run + 1
+        round_failed = False
+
+        with ThreadPoolExecutor(max_workers=len(batches)) as executor:
+            futures = {
+                executor.submit(
+                    _run_batch_in_thread,
+                    batch,
+                    resolved_biz_dt,
+                    f"batch {round_start_batch + idx}",
+                ): batch
+                for idx, batch in enumerate(batches)
+            }
+            for future in as_completed(futures):
+                batch = futures[future]
+                try:
+                    future.result()
+                except Exception as e:
+                    logger.error(
+                        "Grade demand pool batch failed: %s (batch=%s)",
+                        e,
+                        batch,
+                        exc_info=True,
+                    )
+                    round_failed = True
+
+        batches_run += len(batches)
+
+        if round_failed:
             stopped_reason = "batch_failed"
             break
 
         new_graded_names = _fetch_graded_names(resolved_biz_dt)
         if len(new_graded_names) <= len(graded_names):
-            # agent 没有对这批产生任何新的落库结果，避免死循环重复拿到同一批
+            # agent 没有对本轮产生任何新的落库结果，避免死循环重复拿到同一批
             logger.warning(
-                "Grade demand pool batch %d made no progress (graded still %d), stop",
-                batches_run,
+                "Grade demand pool round made no progress (graded still %d), stop",
                 len(graded_names),
             )
             stopped_reason = "stalled"
@@ -135,6 +207,7 @@ def grade_demand_pool(
         "graded_before": graded_before,
         "graded_after": len(graded_names),
         "batches_run": batches_run,
+        "workers": worker_count,
         "stopped_reason": stopped_reason,
         "run_at": datetime.now().isoformat(),
     }

supply_infra/scheduler/jobs/run_supply_pipeline.py

@@ -0,0 +1,122 @@
+"""
+串联执行的供应链数据流水线（可重复执行、幂等）：
+
+1. ODPS → MySQL 全局树同步（T-1 分区）
+2. ODPS → MySQL 策略需求池同步（当天 biz_dt）
+3. 需求池分级评估（同上 biz_dt）
+
+各子步骤内部已做去重（INSERT IGNORE、diff 同步、跳过已分级词等）；
+本文件额外用进程内锁防止同一轮次并发重入。
+"""
+from __future__ import annotations
+
+import logging
+import threading
+from datetime import datetime, timedelta
+from typing import Any
+
+from supply_infra.scheduler.constants import (
+    SUPPLY_PIPELINE_JOB_ID,
+    SUPPLY_PIPELINE_JOB_NAME,
+)
+from supply_infra.scheduler.job_execution import JobExecutionRecorder, record_skipped
+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,
+)
+
+logger = logging.getLogger(__name__)
+
+_pipeline_lock = threading.Lock()
+
+
+def _resolve_dates(biz_dt: str | None) -> tuple[str, str]:
+    """返回 (biz_dt, global_tree_partition_date)，global_tree 使用 biz_dt 前一日。"""
+    resolved_biz_dt = biz_dt or datetime.now().strftime("%Y%m%d")
+    tree_partition = (
+        datetime.strptime(resolved_biz_dt, "%Y%m%d") - timedelta(days=1)
+    ).strftime("%Y%m%d")
+    return resolved_biz_dt, tree_partition
+
+
+def run_supply_pipeline(biz_dt: str | None = None) -> dict[str, Any]:
+    """
+    按顺序执行全局树同步 → 需求池同步 → 需求分级。
+
+    Args:
+        biz_dt: 业务日 YYYYMMDD；省略则取当天。
+
+    Returns:
+        各步骤统计；若上一轮仍在执行则返回 skipped。
+    """
+    resolved_biz_dt, tree_partition = _resolve_dates(biz_dt)
+
+    if not _pipeline_lock.acquire(blocking=False):
+        logger.warning("Supply pipeline already running, skip this round")
+        record_skipped(
+            job_name=SUPPLY_PIPELINE_JOB_NAME,
+            job_id=SUPPLY_PIPELINE_JOB_ID,
+            biz_dt=resolved_biz_dt,
+            reason="already_running",
+        )
+        return {
+            "skipped": True,
+            "reason": "already_running",
+            "run_at": datetime.now().isoformat(),
+        }
+
+    started_at = datetime.now()
+    recorder = JobExecutionRecorder(
+        job_name=SUPPLY_PIPELINE_JOB_NAME,
+        job_id=SUPPLY_PIPELINE_JOB_ID,
+        biz_dt=resolved_biz_dt,
+    )
+    recorder.record_started()
+
+    logger.info(
+        "Supply pipeline start: biz_dt=%s global_tree_partition=%s run_id=%s",
+        resolved_biz_dt,
+        tree_partition,
+        recorder.run_id,
+    )
+
+    result: dict[str, Any] = {
+        "run_id": recorder.run_id,
+        "biz_dt": resolved_biz_dt,
+        "global_tree_partition": tree_partition,
+        "started_at": started_at.isoformat(),
+    }
+    error_message: str | None = None
+    success = False
+
+    try:
+        result["global_tree"] = sync_global_tree_odps_to_mysql(partition_date=tree_partition)
+        result["demand_pool"] = sync_multi_demand_pool_odps_to_mysql(
+            partition_date=resolved_biz_dt,
+        )
+        result["grade"] = grade_demand_pool(biz_dt=resolved_biz_dt)
+        success = True
+        result["success"] = True
+    except Exception as exc:
+        result["success"] = False
+        error_message = str(exc)
+        logger.exception(
+            "Supply pipeline failed: biz_dt=%s global_tree_partition=%s",
+            resolved_biz_dt,
+            tree_partition,
+        )
+        raise
+    finally:
+        finished_at = datetime.now()
+        result["finished_at"] = finished_at.isoformat()
+        result["duration_seconds"] = round((finished_at - started_at).total_seconds(), 2)
+        recorder.record_finished(
+            success=success,
+            result=result,
+            error_message=error_message,
+        )
+        _pipeline_lock.release()
+        logger.info("Supply pipeline finished: %s", result)
+
+    return result

web/src/components/DemandPathPanel.vue

@@ -1,5 +1,5 @@
 <script setup lang="ts">
-import { computed, nextTick, ref, watch } from 'vue'
+import { computed, ref, watch } from 'vue'
 import { fetchDemandGradeVideos } from '../api/demand'
 import type { DemandGradeItem, DemandVideoItem } from '../types/demand'
 import { gradeRank, parseStrategies } from '../types/demand'
@@ -11,13 +11,14 @@ const props = defineProps<{
   /** When set, prepend a "当前节点" column. */
   nodeName?: string | null
   nodePath?: string | null
+  /** Hide the small inline × close button (when parent provides its own). */
+  hideInlineClose?: boolean
 }>()
 
 const emit = defineEmits<{
   close: []
 }>()
 
-const rootRef = ref<HTMLElement | null>(null)
 const selectedDemandId = ref<number | null>(null)
 const videos = ref<DemandVideoItem[]>([])
 const videosLoading = ref(false)
@@ -98,15 +99,11 @@ const topicSections = computed((): TopicSection[] | null => {
 
 watch(
   () => [props.open, props.categoryName] as const,
-  async ([open]) => {
+  () => {
     selectedDemandId.value = null
     videos.value = []
     videosError.value = null
     selectedVid.value = null
-    if (open) {
-      await nextTick()
-      rootRef.value?.scrollIntoView({ behavior: 'smooth', inline: 'nearest', block: 'nearest' })
-    }
   },
 )
 
@@ -138,13 +135,21 @@ function selectVideo(video: DemandVideoItem) {
 </script>
 
 <template>
-  <div v-if="open" ref="rootRef" class="path-rail" role="region" :aria-label="`${categoryName} 路径展开`">
+  <div v-if="open" class="path-rail" role="region" :aria-label="`${categoryName} 路径展开`">
     <!-- 当前节点 -->
     <template v-if="nodeName">
       <section class="col col-node">
         <div class="col-head">
           <h3 class="col-title">当前节点</h3>
-          <button type="button" class="close-btn" title="收起路径" @click="emit('close')">×</button>
+          <button
+            v-if="!hideInlineClose"
+            type="button"
+            class="close-btn"
+            title="收起路径"
+            @click="emit('close')"
+          >
+            ×
+          </button>
         </div>
         <div class="card node-card">
           <span class="card-label">分类节点</span>
@@ -173,7 +178,7 @@ function selectVideo(video: DemandVideoItem) {
       <div class="col-head">
         <h3 class="col-title">细节元素 · 需求词</h3>
         <button
-          v-if="!nodeName"
+          v-if="!nodeName && !hideInlineClose"
           type="button"
           class="close-btn"
           title="收起路径"
@@ -298,7 +303,10 @@ function selectVideo(video: DemandVideoItem) {
   align-items: flex-start;
   gap: 0;
   padding-left: 0;
-  min-height: 80px;
+  min-height: 0;
+  width: max-content;
+  min-width: 100%;
+  max-width: 100%;
 }
 
 .col {
@@ -307,6 +315,8 @@ function selectVideo(video: DemandVideoItem) {
   display: flex;
   flex-direction: column;
   gap: 10px;
+  min-height: 0;
+  max-height: 100%;
 }
 
 .col-topic {
@@ -367,8 +377,9 @@ function selectVideo(video: DemandVideoItem) {
   display: flex;
   flex-direction: column;
   gap: 8px;
-  max-height: 420px;
+  max-height: min(200px, 24vh);
   overflow: auto;
+  overscroll-behavior: contain;
   padding-right: 2px;
 }
 
@@ -593,7 +604,7 @@ function selectVideo(video: DemandVideoItem) {
   border-radius: 12px;
   background: linear-gradient(180deg, #faf5ff 0%, #fff 45%);
   overflow: hidden;
-  max-height: 520px;
+  max-height: min(220px, 26vh);
   display: flex;
   flex-direction: column;
 }
@@ -763,4 +774,18 @@ function selectVideo(video: DemandVideoItem) {
   border-color: #fecaca;
   background: #fef2f2;
 }
+
+@media (max-height: 800px) {
+  .card-list {
+    max-height: min(150px, 20vh);
+  }
+
+  .topic-card {
+    max-height: min(170px, 22vh);
+  }
+
+  .arrow-col {
+    padding-top: 32px;
+  }
+}
 </style>

web/src/components/IcicleHeatTree.vue

@@ -816,14 +816,31 @@ onUnmounted(() => {
     </section>
 
     <div v-if="inspectOpen && inspectNode" class="path-dock">
-      <DemandPathPanel
-        :open="true"
-        :category-name="inspectNode.name"
-        :node-name="inspectNode.name"
-        :node-path="inspectNode.path.join(' / ')"
-        :items="inspectItems"
-        @close="closeInspect"
-      />
+      <div class="path-dock-head">
+        <div class="path-dock-meta">
+          <span class="path-dock-label">节点详情</span>
+          <span class="path-dock-name">{{ inspectNode.name }}</span>
+        </div>
+        <button
+          type="button"
+          class="path-dock-close"
+          aria-label="关闭节点详情"
+          @click="closeInspect"
+        >
+          关闭
+        </button>
+      </div>
+      <div class="path-dock-body">
+        <DemandPathPanel
+          :open="true"
+          :category-name="inspectNode.name"
+          :node-name="inspectNode.name"
+          :node-path="inspectNode.path.join(' / ')"
+          :items="inspectItems"
+          hide-inline-close
+          @close="closeInspect"
+        />
+      </div>
     </div>
   </div>
 </template>
@@ -874,7 +891,7 @@ onUnmounted(() => {
   display: grid;
   grid-template-rows: auto auto auto auto minmax(0, 1fr);
   flex: 1;
-  min-height: 0;
+  min-height: min(240px, 32vh);
   overflow: hidden;
   border: 1px solid #e2e8f0;
   border-radius: 10px;
@@ -1010,15 +1027,87 @@ onUnmounted(() => {
 }
 
 .path-dock {
-  flex-shrink: 0;
-  max-height: 340px;
-  overflow: auto;
+  display: flex;
+  flex-direction: column;
+  flex: 0 1 auto;
+  min-height: 0;
+  max-height: min(32vh, 280px);
+  overflow: hidden;
+  overscroll-behavior: contain;
   border: 1px solid #e2e8f0;
   border-radius: 10px;
   background: #fff;
+  isolation: isolate;
+}
+
+.path-dock-head {
+  display: flex;
+  align-items: center;
+  justify-content: space-between;
+  gap: 12px;
+  flex-shrink: 0;
+  padding: 10px 14px;
+  border-bottom: 1px solid #e2e8f0;
+  background: #f8fafc;
+}
+
+.path-dock-meta {
+  display: flex;
+  align-items: baseline;
+  gap: 8px;
+  min-width: 0;
+}
+
+.path-dock-label {
+  flex-shrink: 0;
+  font-size: 12px;
+  font-weight: 700;
+  color: #64748b;
+  letter-spacing: 0.03em;
+}
+
+.path-dock-name {
+  font-size: 14px;
+  font-weight: 600;
+  color: #0f172a;
+  overflow: hidden;
+  text-overflow: ellipsis;
+  white-space: nowrap;
+}
+
+.path-dock-close {
+  flex-shrink: 0;
+  height: 30px;
+  padding: 0 12px;
+  border: 1px solid #cbd5e1;
+  border-radius: 6px;
+  background: #fff;
+  color: #475569;
+  font-size: 13px;
+  font-weight: 500;
+  cursor: pointer;
+}
+
+.path-dock-close:hover {
+  background: #f1f5f9;
+  border-color: #94a3b8;
+  color: #0f172a;
+}
+
+.path-dock-body {
+  flex: 1;
+  min-height: 0;
+  overflow: auto;
+  overscroll-behavior: contain;
   padding: 12px 14px;
 }
 
+@media (max-height: 800px) {
+  .path-dock {
+    max-height: min(28vh, 220px);
+  }
+}
+
 .depth-axis {
   display: grid;
   padding: 8px 12px;