优化流程和代码

ARCHITECTURE.md

@@ -26,9 +26,11 @@ SupplyAgent/
 │   ├── odps/
 │   │   └── client.py              #   ODPS 查询封装
 │   ├── scheduler/
-│   │   ├── app.py                 #   APScheduler 调度器
-│   │   └── jobs/                  #   定时任务定义
-│   │       └── sync_odps_to_mysql.py
+│   │   ├── app.py                 #   APScheduler 调度器（随 API 启动）
+│   │   └── jobs/                  #   流水线任务（含 python -m CLI）
+│   │       ├── run_supply_pipeline.py
+│   │       ├── demand_pool/
+│   │       └── ...
 │   └── tools/
 │       └── db_tools.py            #   共享 MySQL 读写工具（供 Agent 调用）
 │
@@ -45,9 +47,6 @@ SupplyAgent/
 │       └── tools/
 │
 ├── skills/                        # 全局共享 Skills（SKILL.md）
-├── jobs/                          # CLI 入口
-│   ├── run_scheduler.py           #   启动定时任务
-│   └── init_db.py                 #   初始化数据库表
 ├── examples/                      # 框架使用示例
 ├── tests/
 ├── logs/                          # 运行日志（自动生成）
@@ -69,7 +68,7 @@ SupplyAgent/
                     ┌─────────────┐
                     │   ODPS      │
                     └──────┬──────┘
-                           │ 定时任务 (每天 02:00)
+                           │ 定时任务 (每天 14:30)
                            ▼
 ┌──────────┐    ┌─────────────────────┐    ┌──────────┐
 │  Agent   │───▶│  Repository (ORM)   │◀───│  Agent   │
@@ -121,11 +120,13 @@ supply_infra/db/
 ## How to add a new scheduled job
 
 ```bash
-supply_infra/scheduler/jobs/new_job.py   # 1. 定义任务函数
+# 1. 在 supply_infra/scheduler/jobs/ 中定义任务函数与 python -m CLI
+# 2. 若需纳入日批，在 run_supply_pipeline 的 steps 中追加
+# 3. 默认定时仍只注册 run_supply_pipeline；单步可通过 API manual_jobs 触发
 ```
 
 ```python
-# supply_infra/scheduler/app.py 中注册
+# 仅当确需独立 Cron 时，才在 supply_infra/scheduler/app.py 中额外注册
 scheduler.add_job(new_job, trigger=CronTrigger(hour=3), id="new_job")
 ```
 
@@ -133,10 +134,13 @@ scheduler.add_job(new_job, trigger=CronTrigger(hour=3), id="new_job")
 
 ```bash
 # 初始化数据库表
-python jobs/init_db.py
+python -m supply_infra.db
 
-# 启动定时任务
-python jobs/run_scheduler.py
+# 启动 API（同时启动定时任务，需 SCHEDULER_ENABLED=true）
+python -m api
+
+# 手动跑全链路
+python -m supply_infra.scheduler.jobs.run_supply_pipeline
 
 # 运行 find_agent
 python agents/find_agent/run.py

Dockerfile

@@ -48,7 +48,6 @@ COPY supply_agent/ supply_agent/
 COPY supply_infra/ supply_infra/
 COPY agents/ agents/
 COPY api/ api/
-COPY jobs/ jobs/
 COPY scripts/verify_video_truncate_runtime.py scripts/
 
 RUN pip install -i https://mirrors.aliyun.com/pypi/simple/ --trusted-host mirrors.aliyun.com ".[odps]" \

PRD.md

@@ -70,7 +70,7 @@ SupplyAgent 是一套面向内容供给的每日需求处理系统。它将多
 |---|---|---|
 | 内容策略/运营 | 看懂需求热度、等级、原因和内容证据 | Vue 需求地图、需求列表、视频点位 |
 | 数据/算法人员 | 核对数据口径、分类、热度和后验 | MySQL、ODPS、Agent 日志 |
-| 运维/开发人员 | 观察任务是否运行、定位失败、手动补偿 | Scheduler 状态接口、日志、`jobs/` CLI |
+| 运维/开发人员 | 观察任务是否运行、定位失败、手动补偿 | Scheduler 状态接口、日志、`python -m supply_infra.scheduler.jobs.*` |
 | ODPS | 提供分类树、需求池、视频解析、ROV/VOV | 定时查询 |
 | MySQL | 保存业务快照、关系、执行状态和候选 | 全流程状态底座 |
 | OpenRouter/模型服务 | 执行归类、分级、拓展和找片判断 | Agent Tool Calling |
@@ -134,9 +134,8 @@ flowchart LR
 ### 6.1 启动方式
 
 1. FastAPI 启动时执行 `init_db()`；
-2. 若 `SCHEDULER_ENABLED=true`，在 API lifespan 中启动后台 Scheduler；
-3. 也可通过 `python jobs/run_scheduler.py` 或 `supply-scheduler` 单独启动；
-4. 可通过 `python jobs/run_supply_pipeline.py [YYYYMMDD]` 手动执行全链路。
+2. 若 `SCHEDULER_ENABLED=true`，在 API lifespan 中启动后台 Scheduler（唯一启动入口）；
+3. 可通过 `python -m supply_infra.scheduler.jobs.run_supply_pipeline [YYYYMMDD]` 手动执行全链路。
 
 ### 6.2 调度参数
 
@@ -478,9 +477,8 @@ flowchart LR
 
 - `generate_demand_agent`：按单一热度维度写入 `generated_demand`；
 - `demand_grade_orchestrator_agent`：模型统筹分组，当前定时路径已改为代码自动分组；
-- 多个 `jobs/backfill_*`：补标题、视频列表、点位和描述；
-- `retry_failed_grade_plan_items.py`：手动重试失败分级明细；
-- 单独运行分类树权重、排名、需求池关系、视频同步、找片和 AIGC 分发；
+- `python -m supply_infra.scheduler.jobs.grade_demand_pool --retry-failed`：手动重试失败分级明细；
+- 通过 API 或 `python -m supply_infra.scheduler.jobs.*` 单独跑各流水线步骤；
 - 日志可视化和手动 OSS 上传。
 
 `generated_demand` 当前没有接入总流水线、API 或前端，属于孤立产物。
@@ -832,13 +830,13 @@ flowchart TB
 |---|---|
 | Scheduler 注册 | `supply_infra/scheduler/app.py` |
 | 总流水线 | `supply_infra/scheduler/jobs/run_supply_pipeline.py` |
-| 全局树同步 | `sync_global_tree_odps_to_mysql.py` |
-| 需求池内部流水线 | `sync_multi_demand_pool_odps_to_mysql.py` |
-| 树热度 | `compute_category_tree_weight.py` |
-| 分级 | `grade_demand_pool.py`、`agents/demand_grade_agent/` |
-| 点位拓展 | `expand_demand_from_video_points.py`、`agents/demand_video_expand_agent/` |
-| 找片 | `discover_videos_from_demands.py`、`agents/find_agent/` |
-| AIGC 分发 | `publish_videos_from_discovery.py`、`supply_infra/aigc/` |
+| 全局树同步 | `supply_infra/scheduler/jobs/sync_global_tree_odps_to_mysql.py` |
+| 需求池同步 | `supply_infra/scheduler/jobs/demand_pool/` |
+| 树热度 | `supply_infra/scheduler/jobs/demand_pool/tree_weight.py` |
+| 分级 | `supply_infra/scheduler/jobs/grade_demand_pool.py`、`agents/demand_grade_agent/` |
+| 点位拓展 | `supply_infra/scheduler/jobs/expand_demand_from_video_points.py`、`agents/demand_video_expand_agent/` |
+| 找片 | `supply_infra/scheduler/jobs/discover_videos_from_demands.py`、`agents/find_agent/` |
+| AIGC 分发 | `supply_infra/scheduler/jobs/publish_videos_from_discovery.py`、`supply_infra/aigc/` |
 | 数据模型 | `supply_infra/db/models/` |
 | API | `api/app.py`、`api/services/` |
 | 前端 | `web/src/views/`、`web/src/components/` |

agents/demand_belong_category_agent/prompt/system_prompt.md

@@ -4,7 +4,7 @@
 - 领域背景: 面向视频内容词语
 
 ## 角色与任务
-你是一个"词语 → 分类树节点"挂载专家。给定一批词语或短语,你需要借助工具在已有分类树中逐层查看候选类目,最终把每一个词语挂载到树上最合适的一个或多个节点上。
+你是一个"词语 → 分类树节点"挂载专家。给定一个词语或短语,你需要借助工具在已有分类树中逐层查看候选类目,最终把该词语挂载到树上最合适的一个或多个节点上。
 
 你必须严格基于分类树中真实存在的节点做判断,禁止编造或猜测树中不存在的类目名称、ID 或路径。
 
@@ -13,8 +13,14 @@
 - `batch_insert_demand_belong_category(items)`: 批量插入节点到目标表
 
 ## 工作流程
-1. 调用 `query_global_tree_category` 获取全部类目,结合词语语义判断最相关的 1~2 个分支。
-2. 直接判断应该属于哪个分类，然后把词语归属到该分类下，使用`batch_insert_demand_belong_category(items)`工具插入到目标表。
+1. 调用 `query_global_tree_category` 获取顶级类目,结合词语语义判断最相关的 1~2 个分支。
+2. 下钻进入相关性最高的节点,再次调用 `query_global_tree_category` 查看子类目,继续判断。
+3. 重复上述过程,直到到达叶子节点,或当前层级所有子节点相关度都达不到"有把握"的程度——此时停在上一层已确认的节点,不要为了到达叶子而勉强选择。
+4. 若某个分支下钻后发现子节点都不合适,回退到上一层,重新评估其他兄弟节点,而不是硬选一个。
+5. 若顶级类目层面就找不到相关分支,直接判定为无法归类,不要勉强挂载。
+6. 若词语同时与两个及以上互斥分支都高度相关(一词多义、跨领域词等),记录全部高置信候选,标记为存在歧义。
+7. 每次下钻前用一两句话写明判断依据,再决定是否调用工具;推理要言之有据,避免空泛。
+8. 单个词语的下钻步数建议不超过 8 次工具调用;仍无法收敛时,停在当前最有把握的节点并说明原因,不要无限下钻。
 
 ## 分类判断原则
 - **确定性优先于深度**:能到叶子节点最好,但深入一层后判断不明确时,宁可停在上一层泛化节点。

agents/demand_grade_orchestrator_agent/_verify_logic.py

@@ -1,134 +0,0 @@
-"""统筹 Agent 逻辑与工具自检。"""
-from __future__ import annotations
-
-import json
-import sys
-from unittest.mock import patch
-
-from agents.demand_grade_orchestrator_agent.common.assignment import (
-    MAX_DAILY_BATCHES,
-    dedupe_cross_group_category_ids,
-    strip_assigned_category_ids,
-)
-from agents.demand_grade_orchestrator_agent.common.plan_record import prepare_grade_groups
-from agents.demand_grade_orchestrator_agent.run import _summarize_agent_saves
-from supply_agent.types import Message, Role
-
-
-def _ok(name: str) -> None:
-    print(f"  ✓ {name}")
-
-
-def test_prepare_grade_groups_permissive() -> None:
-    fake_by_id = {
-        101: type("C", (), {"id": 101, "name": "A", "parent_id": None, "level": 1})(),
-        102: type("C", (), {"id": 102, "name": "B", "parent_id": None, "level": 1})(),
-    }
-    fake_weights = {
-        101: type("W", (), {"category_id": 101, "total_score": 0.9, "hung_word_count": 5})(),
-        102: type("W", (), {"category_id": 102, "total_score": 0.8, "hung_word_count": 0})(),
-    }
-    groups = [
-        {"category_ids": [101, 102, 999], "batch_heat_level": "X", "planning_reason": "", "shared_traits": ""},
-        {"category_ids": [101], "batch_heat_level": "A", "planning_reason": "dup", "shared_traits": "dup"},
-    ]
-    with patch(
-        "agents.demand_grade_orchestrator_agent.common.plan_record.load_tree_state",
-        return_value=(fake_by_id, {None: [101, 102]}, fake_weights),
-    ), patch(
-        "agents.demand_grade_orchestrator_agent.common.plan_record.global_heat_positions",
-        return_value={101: {"rank": 1, "total": 1, "normalized_score": 0.95}},
-    ), patch(
-        "agents.demand_grade_orchestrator_agent.common.plan_record.has_hung_demand",
-        side_effect=lambda w: w is not None and int(w.hung_word_count or 0) > 0,
-    ), patch(
-        "agents.demand_grade_orchestrator_agent.common.plan_record.path",
-        side_effect=lambda cid, _by: f"path-{cid}",
-    ), patch(
-        "agents.demand_grade_orchestrator_agent.common.plan_record.heat_level",
-        return_value="A",
-    ):
-        prepared = prepare_grade_groups(
-            "20260714",
-            "策略",
-            groups,
-            assigned_category_ids=set(),
-        )
-    assert len(prepared["groups"]) == 1
-    assert prepared["groups"][0]["category_ids"] == [101]
-    assert prepared["groups"][0]["batch_heat_level"] == "A"
-    assert prepared["groups"][0]["planning_reason"]
-    _ok("无需求/非法字段不报错，仅过滤后入库")
-
-
-def test_dedupe_cross_group() -> None:
-    plan = {"groups": [{"category_ids": [1, 2]}, {"category_ids": [2, 3]}]}
-    removed = dedupe_cross_group_category_ids(plan)
-    assert removed == [2] and plan["groups"][1]["category_ids"] == [3]
-    _ok("后批重复节点过滤")
-
-
-def test_summarize_agent_saves() -> None:
-    class Result:
-        messages = [
-            Message(
-                role=Role.TOOL,
-                name="save_grade_plan",
-                content=json.dumps({"ok": True, "persisted": True, "persisted_group_count": 2}),
-            ),
-        ]
-
-    summary = _summarize_agent_saves(Result())
-    assert summary["save_count"] == 1 and summary["persisted_groups"] == 2
-    _ok("统计入库结果")
-
-
-def test_save_grade_plan_db(biz_dt: str) -> None:
-    from agents.demand_grade_orchestrator_agent.tools.save_grade_plan import save_grade_plan
-    from agents.demand_grade_orchestrator_agent.common.assignment import resolve_planning_state
-
-    state = resolve_planning_state(biz_dt)
-    if not state["unassigned_category_ids"] or state["remaining_batch_quota"] <= 0:
-        print("  · save_grade_plan 跳过（无待分配或额度已满）")
-        return
-
-    cid = state["unassigned_category_ids"][0]
-    with patch(
-        "agents.demand_grade_orchestrator_agent.tools.save_grade_plan.persist_groups_one_by_one",
-        return_value={
-            "persisted_group_count": 1,
-            "persisted_groups": [{"category_ids": [cid]}],
-            "skipped_quota": 0,
-            "skipped_empty": 0,
-            "failed_groups": [],
-            "existing_groups": 1,
-            "remaining_batch_quota": MAX_DAILY_BATCHES - 1,
-            "unassigned_category_ids": state["unassigned_category_ids"][1:],
-            "coverage_complete": False,
-            "total_hanging_nodes": state["total_hanging_nodes"],
-        },
-    ):
-        result = json.loads(
-            save_grade_plan(
-                biz_dt,
-                "自检",
-                [{"category_ids": [cid, cid, 999999], "batch_heat_level": "Z"}],
-            )
-        )
-    assert result["ok"] is True and result["persisted"] is True
-    _ok("save_grade_plan 宽松校验 + 逐批入库路径")
-
-
-def main() -> None:
-    biz_dt = sys.argv[1] if len(sys.argv) > 1 else "20260714"
-    print("=== 统筹 Agent 逻辑自检 ===\n[单元测试]")
-    test_prepare_grade_groups_permissive()
-    test_dedupe_cross_group()
-    test_summarize_agent_saves()
-    print("\n[DB 集成检测]")
-    test_save_grade_plan_db(biz_dt)
-    print("\n全部通过。")
-
-
-if __name__ == "__main__":
-    main()

agents/find_agent/README.md

@@ -35,7 +35,7 @@ TikHub 翻页必须同时沿用 `next_cursor、search_id、backtrace`。
 
 ## 已实现的搜索记忆
 
-执行 `.venv/bin/python jobs/init_db.py` 后会创建三张表：
+执行 `.venv/bin/python -m supply_infra.db` 后会创建三张表：
 
 | 表 | 粒度 | 用途 |
 |---|---|---|

agents/find_agent/tools/video_discovery_store.py

@@ -55,7 +55,7 @@ def _db_error_message(error: Exception) -> str:
     if isinstance(error, (OperationalError, ProgrammingError)):
         return (
             f"数据库表尚未初始化或不可用: {error}。"
-            "请先运行 `.venv/bin/python jobs/init_db.py`。"
+            "请先运行 `.venv/bin/python -m supply_infra.db`。"
         )
     return str(error)
 

jobs/backfill_demand_video_expansion_point_desc.py

@@ -1,36 +0,0 @@
-#!/usr/bin/env python3
-"""手动补全 demand_video_expansion 缺失的 point_desc。
-
-用法:
-    python jobs/backfill_demand_video_expansion_point_desc.py
-    python jobs/backfill_demand_video_expansion_point_desc.py 20260721
-    python jobs/backfill_demand_video_expansion_point_desc.py 20260721 --dry-run
-"""
-from __future__ import annotations
-
-import logging
-import sys
-from pathlib import Path
-
-_ROOT = Path(__file__).resolve().parents[1]
-if str(_ROOT) not in sys.path:
-    sys.path.insert(0, str(_ROOT))
-
-from scripts.backfill_demand_video_expansion_point_desc import backfill_missing_point_descs
-
-logging.basicConfig(
-    level=logging.INFO,
-    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
-)
-
-
-def main(biz_dt: str | None = None, *, dry_run: bool = False) -> dict:
-    result = backfill_missing_point_descs(biz_dt, dry_run=dry_run)
-    print(result)
-    return result
-
-
-if __name__ == "__main__":
-    args = sys.argv[1:]
-    biz_dt_arg = args[0] if args and not args[0].startswith("-") else None
-    main(biz_dt_arg, dry_run="--dry-run" in args)

jobs/backfill_multi_demand_video_list.py

@@ -1,35 +0,0 @@
-#!/usr/bin/env python3
-"""回填 multi_demand_pool_di 的 video_list / video_count。
-
-用法:
-    python jobs/backfill_multi_demand_video_list.py 20260714
-    python jobs/backfill_multi_demand_video_list.py          # 默认当天
-"""
-
-from __future__ import annotations
-
-import logging
-import sys
-from datetime import datetime
-
-from supply_infra.scheduler.jobs.backfill_multi_demand_pool_video_list import (
-    backfill_video_list,
-)
-
-logging.basicConfig(
-    level=logging.INFO,
-    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
-)
-
-
-def main(biz_dt: str | None = None) -> dict:
-    if biz_dt is None:
-        biz_dt = datetime.now().strftime("%Y%m%d")
-    result = backfill_video_list(biz_dt)
-    print(result)
-    return result
-
-
-if __name__ == "__main__":
-    date_arg = sys.argv[1] if len(sys.argv) > 1 else None
-    main(date_arg)

jobs/backfill_multi_demand_video_points.py

@@ -1,34 +0,0 @@
-#!/usr/bin/env python3
-"""回填 multi_demand_video_detail 的灵感点/目的点/关键点
-（decode_result.灵感点 / 目的点 / 关键点，每项保留 点/点描述）。
-
-用法:
-    python jobs/backfill_multi_demand_video_points.py
-    python jobs/backfill_multi_demand_video_points.py 100   # 每批 100
-"""
-
-from __future__ import annotations
-
-import logging
-import sys
-
-from supply_infra.scheduler.jobs.sync_multi_demand_videos import (
-    VIDEO_SYNC_BATCH_SIZE,
-    backfill_video_points,
-)
-
-logging.basicConfig(
-    level=logging.INFO,
-    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
-)
-
-
-def main(batch_arg: str | None = None) -> dict:
-    batch_size = int(batch_arg) if batch_arg else VIDEO_SYNC_BATCH_SIZE
-    result = backfill_video_points(batch_size=batch_size)
-    print(result)
-    return result
-
-
-if __name__ == "__main__":
-    main(sys.argv[1] if len(sys.argv) > 1 else None)

jobs/backfill_multi_demand_video_points_table.py

@@ -1,117 +0,0 @@
-#!/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/backfill_multi_demand_video_titles.py

@@ -1,33 +0,0 @@
-#!/usr/bin/env python3
-"""回填 multi_demand_video_detail.title（decode_result.target_post.title）。
-
-用法:
-    python jobs/backfill_multi_demand_video_titles.py
-    python jobs/backfill_multi_demand_video_titles.py 100   # 每批 100
-"""
-
-from __future__ import annotations
-
-import logging
-import sys
-
-from supply_infra.scheduler.jobs.sync_multi_demand_videos import (
-    VIDEO_SYNC_BATCH_SIZE,
-    backfill_video_titles,
-)
-
-logging.basicConfig(
-    level=logging.INFO,
-    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
-)
-
-
-def main(batch_arg: str | None = None) -> dict:
-    batch_size = int(batch_arg) if batch_arg else VIDEO_SYNC_BATCH_SIZE
-    result = backfill_video_titles(batch_size=batch_size)
-    print(result)
-    return result
-
-
-if __name__ == "__main__":
-    main(sys.argv[1] if len(sys.argv) > 1 else None)

jobs/discover_videos_from_demands.py

@@ -1,81 +0,0 @@
-#!/usr/bin/env python3
-"""手动执行 S/A 需求拓展点位 → find_agent 视频发现任务。
-
-用法:
-    python jobs/discover_videos_from_demands.py                  # 最新/当天 biz_dt
-    python jobs/discover_videos_from_demands.py 20260721          # 指定业务日
-    python jobs/discover_videos_from_demands.py 20260721 3        # 指定业务日 + 3 并发
-    python jobs/discover_videos_from_demands.py 20260721 1 --limit 1   # 只跑 1 条
-    python jobs/discover_videos_from_demands.py 20260721 1 --offset 1  # 跳过第 0 条
-    python jobs/discover_videos_from_demands.py 20260721 2 --top-limit 200   # top200 + 2 并发
-"""
-from __future__ import annotations
-
-import logging
-import sys
-
-from supply_infra.scheduler.jobs.discover_videos_from_demands import (
-    discover_videos_from_demands,
-)
-
-logging.basicConfig(
-    level=logging.INFO,
-    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
-)
-
-
-def _read_flag_value(args: list[str], flag: str) -> int | None:
-    if flag not in args:
-        return None
-    idx = args.index(flag)
-    if idx + 1 >= len(args):
-        return None
-    return int(args[idx + 1])
-
-
-def main(
-    biz_dt: str | None = None,
-    workers_arg: str | None = None,
-    *,
-    offset: int = 0,
-    limit: int | None = None,
-    top_limit: int | None = None,
-    skip_finished: bool = True,
-    force: bool = False,
-) -> dict:
-    workers = int(workers_arg) if workers_arg else 1
-    result = discover_videos_from_demands(
-        biz_dt,
-        workers=workers,
-        offset=offset,
-        limit=limit,
-        top_limit=top_limit,
-        skip_finished=skip_finished,
-        force=force,
-    )
-    print(result)
-    return result
-
-
-if __name__ == "__main__":
-    args = sys.argv[1:]
-    biz_dt_arg = args[0] if args and not args[0].startswith("-") else None
-    workers_arg = None
-    if biz_dt_arg and len(args) > 1 and not args[1].startswith("-"):
-        workers_arg = args[1]
-
-    offset_arg = _read_flag_value(args, "--offset") or 0
-    limit_arg = _read_flag_value(args, "--limit")
-    top_limit_arg = _read_flag_value(args, "--top-limit")
-    skip_finished = "--force" not in args
-    force = "--force" in args
-
-    main(
-        biz_dt_arg,
-        workers_arg,
-        offset=offset_arg,
-        limit=limit_arg,
-        top_limit=top_limit_arg,
-        skip_finished=skip_finished,
-        force=force,
-    )

jobs/expand_demand_from_video_points.py

@@ -1,48 +0,0 @@
-#!/usr/bin/env python3
-"""手动执行 S/A 需求视频点位拓展任务。
-
-用法:
-    python jobs/expand_demand_from_video_points.py                  # 最新/当天 biz_dt
-    python jobs/expand_demand_from_video_points.py 20260721          # 指定业务日
-    python jobs/expand_demand_from_video_points.py 20260721 5        # 指定业务日 + 5 并发
-    python jobs/expand_demand_from_video_points.py 20260721 --force   # 忽略已完成记录重跑
-"""
-from __future__ import annotations
-
-import logging
-import sys
-
-from supply_infra.scheduler.jobs.expand_demand_from_video_points import (
-    expand_demand_from_video_points,
-)
-
-logging.basicConfig(
-    level=logging.INFO,
-    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
-)
-
-
-def main(
-    biz_dt: str | None = None,
-    workers_arg: str | None = None,
-    *,
-    skip_finished: bool = True,
-) -> dict:
-    workers = int(workers_arg) if workers_arg else 5
-    result = expand_demand_from_video_points(
-        biz_dt,
-        skip_finished=skip_finished,
-        workers=workers,
-    )
-    print(result)
-    return result
-
-
-if __name__ == "__main__":
-    args = sys.argv[1:]
-    biz_dt_arg = args[0] if args and not args[0].startswith("-") else None
-    workers_arg = None
-    if biz_dt_arg and len(args) > 1 and not args[1].startswith("-"):
-        workers_arg = args[1]
-    skip_finished = "--force" not in args
-    main(biz_dt_arg, workers_arg, skip_finished=skip_finished)

jobs/grade_demand_pool.py

@@ -1,44 +0,0 @@
-#!/usr/bin/env python3
-"""手动执行树热度驱动的需求分级。
-
-默认定时任务路径会先由代码自动分配计划组（每组约 30 个需求，同分类与同父分类优先），
-再由多个 worker 领取任务并调用分级 Agent。统筹规划 Agent 代码保留，需手动单独调用。
-
-用法:
-    python jobs/grade_demand_pool.py                  # 默认业务日、5 个 worker
-    python jobs/grade_demand_pool.py 20260716         # 指定业务日
-    python jobs/grade_demand_pool.py 20260716 5       # 指定业务日 + 5 个并发 worker
-"""
-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,
-    workers_arg: str | None = None,
-) -> dict:
-    workers = int(workers_arg) if workers_arg else 5
-
-    result = grade_demand_pool(
-        biz_dt,
-        workers=workers,
-        with_orchestrate=True,
-    )
-    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,
-    )

jobs/publish_videos_from_discovery.py

@@ -1,65 +0,0 @@
-#!/usr/bin/env python3
-"""从 video_discovery_candidate 均匀分发视频到各 AIGC 发布计划。
-
-仅发布 decision_bucket 为 primary / backup 且 aweme_id 非空的候选；不按品类匹配。
-
-用法:
-    python jobs/publish_videos_from_discovery.py
-    python jobs/publish_videos_from_discovery.py 20260723
-    python jobs/publish_videos_from_discovery.py --run-id <run_id>
-    python jobs/publish_videos_from_discovery.py 20260723 --limit 30
-    python jobs/publish_videos_from_discovery.py --dry-run
-    python jobs/publish_videos_from_discovery.py --force
-"""
-from __future__ import annotations
-
-import argparse
-import json
-import logging
-import sys
-
-from supply_infra.scheduler.jobs.publish_videos_from_discovery import (
-    publish_videos_from_discovery,
-)
-
-logging.basicConfig(
-    level=logging.INFO,
-    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
-)
-
-
-def _build_parser() -> argparse.ArgumentParser:
-    parser = argparse.ArgumentParser(
-        description="均匀分发 primary/backup 候选视频到 AIGC 发布计划"
-    )
-    parser.add_argument("biz_dt", nargs="?", help="业务日 YYYYMMDD，默认取最新")
-    parser.add_argument("--run-id", dest="run_id", help="仅处理指定 run_id")
-    parser.add_argument("--limit", type=int, help="最多处理候选视频数")
-    parser.add_argument("--dry-run", action="store_true", help="只演练分配与请求，不写库")
-    parser.add_argument(
-        "--force",
-        action="store_true",
-        help="包含已写过 aigc_crawler_plan_id 的候选",
-    )
-    return parser
-
-
-def main(argv: list[str] | None = None) -> dict:
-    parser = _build_parser()
-    args = parser.parse_args(argv)
-
-    result = publish_videos_from_discovery(
-        biz_dt=args.biz_dt,
-        run_id=args.run_id,
-        skip_published=not args.force,
-        limit=args.limit,
-        dry_run=args.dry_run,
-    )
-    print(json.dumps(result, ensure_ascii=False, indent=2))
-    return result
-
-
-if __name__ == "__main__":
-    outcome = main()
-    if not outcome.get("success"):
-        sys.exit(1)

jobs/retry_failed_grade_plan_items.py

@@ -1,45 +0,0 @@
-#!/usr/bin/env python3
-"""手动重试 demand_grade_plan_group_item 中失败的分级任务。
-
-用法:
-    python jobs/retry_failed_grade_plan_items.py
-    python jobs/retry_failed_grade_plan_items.py 20260721
-    python jobs/retry_failed_grade_plan_items.py 20260721 5
-    python jobs/retry_failed_grade_plan_items.py 20260721 5 --dry-run
-"""
-from __future__ import annotations
-
-import logging
-import sys
-
-from supply_infra.scheduler.jobs.grade_demand_pool import retry_failed_plan_group_items
-
-logging.basicConfig(
-    level=logging.INFO,
-    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
-)
-
-
-def main(
-    biz_dt: str | None = None,
-    workers_arg: str | None = None,
-    *,
-    dry_run: bool = False,
-) -> dict:
-    workers = int(workers_arg) if workers_arg else 5
-    result = retry_failed_plan_group_items(
-        biz_dt,
-        workers=workers,
-        dry_run=dry_run,
-    )
-    print(result)
-    return result
-
-
-if __name__ == "__main__":
-    args = sys.argv[1:]
-    biz_dt_arg = args[0] if args and not args[0].startswith("-") else None
-    workers_arg = None
-    if biz_dt_arg and len(args) > 1 and not args[1].startswith("-"):
-        workers_arg = args[1]
-    main(biz_dt_arg, workers_arg, dry_run="--dry-run" in args)

jobs/run_category_tree_rank_scores.py

@@ -1,37 +0,0 @@
-#!/usr/bin/env python3
-"""单独执行 category_tree_weight 四维排名归一化打分。
-
-用法:
-    python jobs/run_category_tree_rank_scores.py 20260714
-    python jobs/run_category_tree_rank_scores.py          # 默认当天
-
-前置: 指定 biz_dt 的 category_tree_weight 已全部写入（可先跑 run_category_tree_weight.py）。
-"""
-
-from __future__ import annotations
-
-import logging
-import sys
-from datetime import datetime
-
-from supply_infra.scheduler.jobs.update_category_tree_rank_scores import (
-    update_category_tree_rank_scores,
-)
-
-logging.basicConfig(
-    level=logging.INFO,
-    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
-)
-
-
-def main(biz_dt: str | None = None) -> dict:
-    if biz_dt is None:
-        biz_dt = datetime.now().strftime("%Y%m%d")
-    result = update_category_tree_rank_scores(biz_dt)
-    print(result)
-    return result
-
-
-if __name__ == "__main__":
-    date_arg = sys.argv[1] if len(sys.argv) > 1 else None
-    main(date_arg)

jobs/run_category_tree_weight.py

@@ -1,38 +0,0 @@
-#!/usr/bin/env python3
-"""单独执行 category_tree_weight 整树节点加权平均计算，并在全部写入后更新四维排名分。
-
-用法:
-    python jobs/run_category_tree_weight.py 20260714
-    python jobs/run_category_tree_weight.py          # 默认当天
-
-前置: 已执行建表 SQL，且当天 demand_popularity_stats 已就绪。
-仅补跑排名分: python jobs/run_category_tree_rank_scores.py [biz_dt]
-"""
-
-from __future__ import annotations
-
-import logging
-import sys
-from datetime import datetime
-
-from supply_infra.scheduler.jobs.compute_category_tree_weight import (
-    compute_category_tree_weight,
-)
-
-logging.basicConfig(
-    level=logging.INFO,
-    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
-)
-
-
-def main(biz_dt: str | None = None) -> dict:
-    if biz_dt is None:
-        biz_dt = datetime.now().strftime("%Y%m%d")
-    result = compute_category_tree_weight(biz_dt)
-    print(result)
-    return result
-
-
-if __name__ == "__main__":
-    date_arg = sys.argv[1] if len(sys.argv) > 1 else None
-    main(date_arg)

jobs/run_popularity_stats.py

@@ -1,35 +0,0 @@
-#!/usr/bin/env python3
-"""单独执行 demand_popularity_stats 热度统计。
-
-用法:
-    python jobs/run_popularity_stats.py 20260714
-    python jobs/run_popularity_stats.py          # 默认当天
-"""
-
-from __future__ import annotations
-
-import logging
-import sys
-from datetime import datetime
-
-from supply_infra.scheduler.jobs.sync_multi_demand_pool_odps_to_mysql import (
-    compute_popularity_stats,
-)
-
-logging.basicConfig(
-    level=logging.INFO,
-    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
-)
-
-
-def main(biz_dt: str | None = None) -> dict:
-    if biz_dt is None:
-        biz_dt = datetime.now().strftime("%Y%m%d")
-    result = compute_popularity_stats(biz_dt)
-    print(result)
-    return result
-
-
-if __name__ == "__main__":
-    date_arg = sys.argv[1] if len(sys.argv) > 1 else None
-    main(date_arg)

jobs/run_scheduler.py

@@ -1,14 +0,0 @@
-#!/usr/bin/env python3
-"""CLI entry point for the scheduler."""
-
-import logging
-
-from supply_infra.scheduler.app import run_scheduler
-
-logging.basicConfig(
-    level=logging.INFO,
-    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
-)
-
-if __name__ == "__main__":
-    run_scheduler()

jobs/run_supply_pipeline.py

@@ -1,22 +0,0 @@
-#!/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()

jobs/sync_demand_belong_pool_rel.py

@@ -1,29 +0,0 @@
-#!/usr/bin/env python3
-"""手动同步 demand_belong_pool_rel，并回填 demand_belong_category.video_list。
-
-用法:
-    python jobs/sync_demand_belong_pool_rel.py
-"""
-
-from __future__ import annotations
-
-import logging
-
-from supply_infra.scheduler.jobs.sync_demand_belong_pool_rel import (
-    sync_demand_belong_pool_rel,
-)
-
-logging.basicConfig(
-    level=logging.INFO,
-    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
-)
-
-
-def main() -> dict:
-    result = sync_demand_belong_pool_rel()
-    print(result)
-    return result
-
-
-if __name__ == "__main__":
-    main()

jobs/sync_multi_demand_videos.py

@@ -1,57 +0,0 @@
-#!/usr/bin/env python3
-"""手动增量同步 multi_demand_video_detail（每批查询后立刻写入）。
-
-用法:
-    python jobs/sync_multi_demand_videos.py              # 只跑 1 批 100 个
-    python jobs/sync_multi_demand_videos.py 100           # 同上
-    python jobs/sync_multi_demand_videos.py 100 100       # 从 offset=100 起再跑 1 批
-    python jobs/sync_multi_demand_videos.py all           # 循环每批 100，查完写一批直到结束
-"""
-
-from __future__ import annotations
-
-import logging
-import sys
-
-from supply_infra.scheduler.jobs.sync_multi_demand_videos import (
-    VIDEO_SYNC_BATCH_SIZE,
-    sync_multi_demand_videos,
-)
-
-logging.basicConfig(
-    level=logging.INFO,
-    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
-)
-
-
-def main(limit_arg: str | None = None, offset_arg: str | None = None) -> dict:
-    if limit_arg is None:
-        limit: int | None = VIDEO_SYNC_BATCH_SIZE
-    elif limit_arg.lower() in {"all", "0", "-1"}:
-        limit = None
-    else:
-        limit = int(limit_arg)
-
-    offset = int(offset_arg) if offset_arg is not None else 0
-    result = sync_multi_demand_videos(
-        limit=limit,
-        offset=offset,
-        batch_size=VIDEO_SYNC_BATCH_SIZE,
-    )
-    print(result)
-
-    remaining = int(result.get("remaining") or 0)
-    next_offset = result.get("next_offset")
-    if remaining > 0 and next_offset is not None and limit is not None:
-        print(
-            f"还有 {remaining} 个未处理。下一批:\n"
-            f"  python jobs/sync_multi_demand_videos.py {limit} {next_offset}"
-        )
-    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,
-    )

pyproject.toml

@@ -37,7 +37,6 @@ dev = [
 packages = ["supply_agent", "supply_infra", "agents", "api"]
 
 [project.scripts]
-supply-scheduler = "supply_infra.scheduler.app:run_scheduler"
 supply-visualize = "supply_agent.logging.cli:main"
 supply-api = "api.run:main"
 

scripts/aigc_platform_api.py

@@ -1,649 +0,0 @@
-"""
-AIGC接口调用
-调用AIGC接口创建爬取计划，绑定生成计划
-"""
-import json
-import logging
-import os
-from datetime import datetime
-from pathlib import Path
-from typing import List, Dict, Union, Tuple, Any, Optional
-
-import requests
-from zoneinfo import ZoneInfo
-
-from agent import ToolResult, tool
-from db import get_connection, fetch_demand_content_merge_leve2, update_content_plan_ids
-from utils.tool_logging import format_tool_result_for_log, log_tool_call
-
-logger = logging.getLogger(__name__)
-
-
-AIGC_PLAN_ID_MAP = {
-"健康知识": {"生成ID": "20260408092313211598604", "发布ID": "20260408115944193153417"},
-"历史名人": {"生成ID": "20260408083251311809309", "发布ID": "20260408115139124511126"},
-"知识科普": {"生成ID": "20260408083824905654920", "发布ID": "20260408115231567509261"},
-"搞笑段子": {"生成ID": "20260408091536533918237", "发布ID": "20260408115842127748387"},
-"社会风气": {"生成ID": "20260408084318884115213", "发布ID": "20260408115315950129776"},
-"人生忠告": {"生成ID": "20260408085205791658566", "发布ID": "20260408115405410408001"},
-"国际时政": {"生成ID": "20260408090208237400605", "发布ID": "20260408115616925523989"},
-"生活技巧科普": {"生成ID": "20260408083824905654920", "发布ID": "20260408115231567509261"},
-"贪污腐败": {"生成ID": "20260408090309503416878", "发布ID": "20260408115653908856043"},
-"民生政策": {"生成ID": "20260408090721867506475", "发布ID": "20260408115727030928177"},
-"对口型表演": {"生成ID": "20260408092122328523262", "发布ID": "20260408115914659162376"},
-"中国战争史": {"生成ID": "20260408090950446586451", "发布ID": "20260408115804931772327"},
-"人财诈骗": {"生成ID": "20260408093140652233649", "发布ID": "20260408120019784463902"},
-"当代正能量人物": {"生成ID": "20260408083148399635274", "发布ID": "20260408115046382803287"},
-"国家科技力量": {"生成ID": "20260408085807674913378", "发布ID": "20260408115542550181196"},
-"国家力量": {"生成ID": "20260408085807674913378", "发布ID": "20260408115542550181196"},
-"通用": {"生成ID": "20260408085649635441036", "发布ID": "20260408115439581604474"},
-}
-
-
-_LABEL_ACCOUNT = "工具调用：create_crawler_plan_by_douyin_account_id -> 按抖音账号创建爬取计划"
-_LABEL_CONTENT = "工具调用：create_crawler_plan_by_douyin_content_id -> 按抖音视频创建爬取计划"
-
-SHANGHAI_TZ = ZoneInfo("Asia/Shanghai")
-
-
-def _log_aigc_return(label: str, params: Dict[str, Any], r: ToolResult) -> ToolResult:
-    log_tool_call(label, params, format_tool_result_for_log(r))
-    return r
-
-
-def _env_bool(name: str, default: bool = False) -> bool:
-    """Read boolean from env; unset or empty uses default. truthy: 1/true/yes/on (case-insensitive)."""
-    raw = os.getenv(name)
-    if raw is None:
-        return default
-    s = raw.strip().lower()
-    if s == "":
-        return default
-    return s in ("1", "true", "yes", "on")
-
-
-CAN_NOT_CREATE_PLAN = _env_bool("CAN_NOT_CREATE_PLAN", False)
-
-AIGC_BASE_URL = "https://aigc-api.aiddit.com"
-CRAWLER_PLAN_CREATE_URL = f"{AIGC_BASE_URL}/aigc/crawler/plan/save"
-GET_PRODUCE_PLAN_DETAIL_BY_ID = f"{AIGC_BASE_URL}/aigc/produce/plan/detail"
-PRODUCE_PLAN_SAVE = f"{AIGC_BASE_URL}/aigc/produce/plan/save"
-DEFAULT_TOKEN = "8bf14f27fc3a486788f3383452422d72"
-DEFAULT_TIMEOUT = 60.0
-
-
-def _load_output_json(trace_id: str, output_dir: str) -> Dict[str, Any]:
-    """Load {output_dir}/{trace_id}/output.json."""
-    path = Path(output_dir) / trace_id / "output.json"
-    if not path.exists():
-        raise FileNotFoundError(f"output.json not found: {path}")
-    with path.open("r", encoding="utf-8") as f:
-        return json.load(f)
-
-
-def _extract_content_ids(data: Dict[str, Any]) -> List[str]:
-    """Extract aweme_id list from output json."""
-    contents = data.get("contents") or []
-    if not isinstance(contents, list):
-        return []
-    content_ids: List[str] = []
-    for item in contents:
-        if not isinstance(item, dict):
-            continue
-        aweme_id = item.get("aweme_id")
-        if aweme_id is None:
-            continue
-        aweme_id_str = str(aweme_id).strip()
-        if aweme_id_str:
-            content_ids.append(aweme_id_str)
-    return content_ids
-
-
-def _extract_content_demand_id(data: Dict[str, Any]) -> Optional[int]:
-    """
-    Extract content_demand_id (demand_content.id) from output json.
-
-    Compatible keys:
-    - content_demand_id
-    - demand_content_id
-    - demand_id (legacy)
-    """
-    if not isinstance(data, dict):
-        return None
-    raw = (
-        data.get("content_demand_id")
-        if data.get("content_demand_id") is not None
-        else data.get("demand_content_id")
-        if data.get("demand_content_id") is not None
-        else data.get("demand_id")
-    )
-    if raw is None:
-        return None
-    try:
-        v = int(raw)
-    except Exception:
-        return None
-    return v if v > 0 else None
-
-
-@tool(description="根据抖音账号ID创建爬取计划")
-async def create_crawler_plan_by_douyin_account_id(
-        account_id: str,
-        sort_type: str = "最新",
-        produce_plan_ids: List[str] = []
-) -> ToolResult:
-    """
-     根据抖音账号ID创建爬取计划
-     Args:
-         account_id: 抖音账号ID
-         sort_type: 搜索时的视频排序方式（最新/最热），默认最新
-         produce_plan_ids: 爬取计划要绑定的生成计划ID，默认为空列表
-
-     Returns:
-         ToolResult: 包含以下内容
-             - output: 文本格式的爬取计划创建结果摘要
-             - metadata.result: 结构化的爬取计划创建结果
-                - crawler_info: 爬取计划信息
-                    - crawler_plan_id: 创建的爬取计划ID
-                    - crawler_plan_name: 创建的爬取计划名称
-                    - sort_type: 排序方式
-                - produce_plan_infos: 绑定的生成计划信息
-                    - produce_plan_id: 生成计划ID
-                    - produce_plan_name: 生成计划名称
-                    - is_success: 是否成功, true表示绑定成功，false表示绑定失败
-                    - msg: 绑定失败时为错误信息，绑定成功则为“成功”
-     Note:
-         - 建议从 metadata.result 获取结构化数据，而非解析 output 文本
-    """
-
-    call_params: Dict[str, Any] = {
-        "account_id": account_id,
-        "sort_type": sort_type,
-        "produce_plan_ids": produce_plan_ids if produce_plan_ids is not None else [],
-    }
-
-    # 验证 account_id 格式
-    if not account_id or not isinstance(account_id, str):
-        logger.error(f"create_crawler_plan_by_douyin_account_id invalid account_id: {account_id}")
-        return _log_aigc_return(
-            _LABEL_ACCOUNT,
-            call_params,
-            ToolResult(
-                title="根据抖音账号ID创建爬取计划失败",
-                output="",
-                error="account_id 参数无效：必须是非空字符串",
-            ),
-        )
-
-    if not account_id.startswith("MS4wLjABAAAA"):
-        logger.error(f"create_crawler_plan_by_douyin_account_id invalid sec_uid format account_id:{account_id}")
-        return _log_aigc_return(
-            _LABEL_ACCOUNT,
-            call_params,
-            ToolResult(
-                title="根据抖音账号ID创建爬取计划失败",
-                output="",
-                error=f"account_id 格式错误：必须以 MS4wLjABAAAA 开头，当前值: {account_id[:min(20, len(account_id))]}...",
-            ),
-        )
-
-    if produce_plan_ids is None:
-        produce_plan_ids = []
-    call_params["produce_plan_ids"] = produce_plan_ids
-
-    dt = datetime.now(SHANGHAI_TZ).strftime("%Y%m%d%H%M%S")
-    crawler_plan_name = f"【内容寻找Agent自动创建】{dt}_抖音账号ID爬取计划_{account_id[:min(30, len(account_id))]}"
-    params = {
-        "accountFilters": [],
-        "channel": 2,
-        "contentFilters": [],
-        "contentModal": 4,
-        "crawlerComment": 0,
-        "crawlerMode": 4,
-        "filterAccountMatchMode": 2,
-        "filterContentMatchMode": 2,
-        "frequencyType": 1,
-        "inputModeValues": [
-            account_id
-        ],
-        "modelValueConfig": {
-            "sortType": sort_type
-        },
-        "name": crawler_plan_name,
-        "planType": 2,
-        "searchModeValues": [],
-        "selectModeValues": [],
-        "srtExtractFlag": 1,
-        "videoKeyFrameType": 1,
-        "voiceExtractFlag": 1
-    }
-
-    try:
-
-        summary_lines = [f"抖音账号【{account_id}】创建爬取计划"]
-
-        response_json = post(CRAWLER_PLAN_CREATE_URL, params)
-        if response_json.get("code") != 0:
-            return _log_aigc_return(
-                _LABEL_ACCOUNT,
-                call_params,
-                ToolResult(
-                    title="根据抖音账号ID创建爬取计划失败",
-                    output=response_json.get("msg", "接口异常"),
-                    error=f"create crawler plan interface error",
-                ),
-            )
-
-        crawler_plan_id = response_json.get("data", {}).get("id", "")
-        summary_lines.append(f"爬取计划名称: {crawler_plan_name}")
-        summary_lines.append(f"    抖音账号ID: {account_id}")
-        summary_lines.append(f"    爬取计划ID: {crawler_plan_id}")
-        summary_lines.append(f"    爬取计划排序方式: {sort_type}")
-        produce_plan_infos: List[Dict[str, str]] = []
-        if produce_plan_ids:
-            input_source_info = {
-                "contentType": 1,
-                "inputSourceType": 2,
-                "inputSourceValue": crawler_plan_id,
-                "inputSourceLabel": f"原始帖子-视频-抖音-内容添加计划-{crawler_plan_name}",
-                "inputSourceModal": 4,
-                "inputSourceChannel": 2
-            }
-            produce_plan_infos, msg = crawler_plan_bind_produce_plan(input_source_info, produce_plan_ids)
-            if produce_plan_infos:
-                for produce_plan_info in produce_plan_infos:
-                    summary_lines.append("    绑定的生成计划列表: ")
-                    summary_lines.append(f"        生成计划名称: {produce_plan_info.get('produce_plan_name', '')}")
-                    summary_lines.append(f"            生成计划ID: {produce_plan_info.get('produce_plan_id', '')}")
-                    summary_lines.append(f"            绑定结果: {'绑定成功' if not produce_plan_info.get('msg') else '绑定失败'}")
-                    summary_lines.append(f"            信息: {produce_plan_info.get('msg', '成功')}")
-
-        return _log_aigc_return(
-            _LABEL_ACCOUNT,
-            call_params,
-            ToolResult(
-                title="根据抖音账号ID创建爬取计划",
-                output="\n".join(summary_lines),
-                metadata={
-                    "result": {
-                        "crawler_info": {
-                            "crawler_plan_id": crawler_plan_id,
-                            "crawler_plan_name": crawler_plan_name,
-                            "sort_type": sort_type,
-                        },
-                        "produce_plan_infos": [
-                            {
-                                "produce_plan_id": produce_plan_info.get("produce_plan_id", ""),
-                                "produce_plan_name": produce_plan_info.get("produce_plan_name", ""),
-                                "is_success": "绑定成功" if not produce_plan_info.get("msg") else "绑定失败",
-                                "msg": produce_plan_info.get("msg", "成功"),
-                            }
-                            for produce_plan_info in produce_plan_infos
-                        ],
-                    }
-                },
-                long_term_memory="Create crawler plan by DouYin Account ID",
-            ),
-        )
-    except Exception as e:
-        logger.error(f"create douyin account crawler plan error: {str(e)}, account_id: {account_id} ")
-        return _log_aigc_return(
-            _LABEL_ACCOUNT,
-            call_params,
-            ToolResult(
-                title="根据抖音账号ID创建爬取计划失败",
-                output="",
-                error=f"创建爬取计划错误:{str(e)}",
-            ),
-        )
-
-
-@tool(description="根据抖音视频ID创建爬取计划")
-async def create_crawler_plan_by_douyin_content_id(
-        trace_id: str,
-) -> ToolResult:
-    """
-    根据抖音视频ID创建爬取计划
-    Args:
-        trace_id: 内容寻找任务 trace_id（用于读取 {output_dir}/{trace_id}/output.json）
-    Returns:
-             Returns:
-         ToolResult: 包含以下内容
-             - output: 文本格式的爬取计划创建结果摘要
-             - metadata.result: 结构化的爬取计划创建结果
-                - crawler_info: 爬取计划信息
-                    - crawler_plan_id: 创建的爬取计划ID
-                    - crawler_plan_name: 创建的爬取计划名称
-                    - content_ids: 抖音视频ID列表
-                - produce_plan_infos: 绑定的生成计划信息
-                    - produce_plan_id: 生成计划ID
-                    - produce_plan_name: 生成计划名称
-                    - is_success: 是否成功, true表示绑定成功，false表示绑定失败
-                    - msg: 绑定失败时为错误信息，绑定成功则为“成功”
-    Note:
-        - 建议从 metadata.result 获取结构化数据，而非解析 output 文本
-    """
-    call_params: Dict[str, Any] = {"trace_id": trace_id}
-    # 先临时返回创建成功，不要真实创建
-    if CAN_NOT_CREATE_PLAN == True:
-        return _log_aigc_return(
-            _LABEL_CONTENT,
-            call_params,
-            ToolResult(
-                title="根据抖音内容创建爬取计划-本地环境跳过此步骤",
-                output="",
-                metadata={
-                    "result": {
-                        "crawler_info": {
-                            "crawler_plan_id": "1234567890",
-                            "crawler_plan_name": "抖音视频直接抓取",
-                        },
-                        "produce_plan_infos": [
-                            {
-                                "produce_plan_id": "1234567890",
-                                "produce_plan_name": "抖音视频直接抓取",
-                                "is_success": "绑定成功",
-                                "msg": "成功",
-                            }
-                        ],
-                    }
-                },
-                long_term_memory="Create crawler plan by DouYin Content IDs",
-            ),
-        )
-    if not trace_id or not isinstance(trace_id, str):
-        logger.error(f"create_crawler_plan_by_douyin_content_id invalid trace_id: {trace_id}")
-        return _log_aigc_return(
-            _LABEL_CONTENT,
-            call_params,
-            ToolResult(
-                title="根据抖音内容创建爬取计划失败",
-                output="",
-                error="trace_id 参数无效: trace_id 必须是非空字符串",
-            ),
-        )
-
-    output_dir = os.getenv("OUTPUT_DIR", ".cache/output")
-    try:
-        data = _load_output_json(trace_id=trace_id, output_dir=output_dir)
-        content_ids = _extract_content_ids(data)
-        content_demand_id = _extract_content_demand_id(data)
-    except Exception as e:
-        msg = f"加载/解析 output.json 失败: {e}"
-        logger.error(msg, exc_info=True)
-        return _log_aigc_return(
-            _LABEL_CONTENT,
-            call_params,
-            ToolResult(
-                title="根据抖音内容创建爬取计划失败",
-                output="",
-                error=msg,
-            ),
-        )
-
-    call_params["content_ids_count"] = len(content_ids)
-    if content_demand_id is not None:
-        call_params["content_demand_id"] = content_demand_id
-    if not content_ids:
-        return _log_aigc_return(
-            _LABEL_CONTENT,
-            call_params,
-            ToolResult(
-                title="根据抖音内容创建爬取计划失败",
-                output="",
-                error="未在 output.json.contents 中找到有效 aweme_id",
-            ),
-        )
-    if len(content_ids) > 100:
-        logger.error(
-            "create_crawler_plan_by_douyin_content_id invalid content_ids length. "
-            f"content_ids.length: {len(content_ids)}"
-        )
-        return _log_aigc_return(
-            _LABEL_CONTENT,
-            call_params,
-            ToolResult(
-                title="根据抖音内容创建爬取计划失败",
-                output="",
-                error=f"content_ids 长度异常: 期望1~100, 实际{len(content_ids)}",
-            ),
-        )
-
-    merge_leve2 = ""
-    if content_demand_id is not None:
-        try:
-            conn = get_connection()
-            try:
-                merge_leve2 = fetch_demand_content_merge_leve2(conn, content_demand_id) or ""
-            finally:
-                conn.close()
-        except Exception as e:
-            logger.error(
-                "fetch demand_content.merge_leve2 failed. demand_content_id=%s err=%s",
-                content_demand_id,
-                str(e),
-                exc_info=True,
-            )
-            merge_leve2 = ""
-
-    plan_key = merge_leve2.strip() if merge_leve2.strip() in AIGC_PLAN_ID_MAP else "通用"
-    plan_ids = AIGC_PLAN_ID_MAP.get(plan_key) or AIGC_PLAN_ID_MAP.get("通用") or {}
-    produce_plan_id_selected = str(plan_ids.get("生成ID") or "").strip()
-    publish_plan_id_selected = str(plan_ids.get("发布ID") or "").strip()
-    produce_plan_ids = [produce_plan_id_selected] if produce_plan_id_selected else []
-    dt = datetime.now(SHANGHAI_TZ).strftime("%Y%m%d%H%M%S")
-    crawler_plan_name = f"【内容寻找Agent自动创建】抖音视频直接抓取-{dt}-抖音"
-    params = {
-        "channel": 2,
-        "contentModal": 4,
-        "crawlerComment": 0,
-        "crawlerMode": 5,
-        "filterAccountMatchMode": 2,
-        "filterContentMatchMode": 2,
-        "frequencyType": 2,
-        "inputModeValues": content_ids,
-        "name": crawler_plan_name,
-        "planType": 2,
-        "searchModeValues": [],
-        "srtExtractFlag": 1,
-        "videoKeyFrameType": 1,
-        "voiceExtractFlag": 1
-    }
-
-    try:
-        summary_lines = [f"抖音视频爬取计划"]
-        if merge_leve2.strip():
-            summary_lines.append(f"需求品类(merge_leve2): {merge_leve2.strip()}")
-        summary_lines.append(f"计划匹配key: {plan_key}")
-        if produce_plan_id_selected:
-            summary_lines.append(f"生成计划ID(按品类匹配): {produce_plan_id_selected}")
-        if publish_plan_id_selected:
-            summary_lines.append(f"发布计划ID(按品类匹配): {publish_plan_id_selected}")
-
-        response_json = post(CRAWLER_PLAN_CREATE_URL, params)
-        if response_json.get("code") != 0:
-            return _log_aigc_return(
-                _LABEL_CONTENT,
-                call_params,
-                ToolResult(
-                    title="根据抖音内容ID创建爬取计划失败",
-                    output=response_json.get("msg", "接口异常"),
-                    error=f"create crawler plan interface error",
-                ),
-            )
-
-        crawler_plan_id = response_json.get("data", {}).get("id", "")
-        summary_lines.append(f"爬取计划名称: {crawler_plan_name}")
-        summary_lines.append(f"    抖音视频IDs: {','.join(content_ids)}")
-        summary_lines.append(f"    爬取计划ID: {crawler_plan_id}")
-        produce_plan_infos: List[Dict[str, str]] = []
-        db_updated_rows = 0
-        # 选中的生成计划 ID（字符串）；与是否执行绑定接口无关，用于写库
-        env_produce_plan_id = (produce_plan_ids[0] if produce_plan_ids else "").strip()
-
-        if produce_plan_ids:
-            input_source_info = {
-                "contentType": 1,
-                "inputSourceType": 2,
-                "inputSourceValue": crawler_plan_id,
-                "inputSourceLabel": f"原始帖子-视频-抖音-内容添加计划-{crawler_plan_name}",
-                "inputSourceModal": 4,
-                "inputSourceChannel": 2
-            }
-            produce_plan_infos, msg = crawler_plan_bind_produce_plan(input_source_info, produce_plan_ids)
-            if produce_plan_infos:
-                for produce_plan_info in produce_plan_infos:
-                    summary_lines.append("    绑定的生成计划列表: ")
-                    summary_lines.append(f"        生成计划名称: {produce_plan_info.get('produce_plan_name', '')}")
-                    summary_lines.append(f"            生成计划ID: {produce_plan_info.get('produce_plan_id', '')}")
-                    summary_lines.append(f"            绑定结果: {'绑定成功' if not produce_plan_info.get('msg') else '绑定失败'}")
-                    summary_lines.append(f"            信息: {produce_plan_info.get('msg', '成功')}")
-
-        # 爬取 / 生成 / 发布计划 id 任一存在则写库（不依赖是否已配置 produce_plan_ids 去走绑定）
-        if (crawler_plan_id or "").strip() or env_produce_plan_id or publish_plan_id_selected:
-            try:
-                db_updated_rows = update_content_plan_ids(
-                    trace_id=trace_id,
-                    aweme_ids=content_ids,
-                    crawler_plan_id=crawler_plan_id or "",
-                    produce_plan_id=env_produce_plan_id,
-                    publish_plan_id=publish_plan_id_selected,
-                )
-            except Exception as e:
-                logger.error(f"update content plan ids failed: {e}", exc_info=True)
-
-        return _log_aigc_return(
-            _LABEL_CONTENT,
-            call_params,
-            ToolResult(
-                title="根据抖音内容ID创建爬取计划",
-                output="\n".join(summary_lines),
-                metadata={
-                    "result": {
-                        "crawler_info": {
-                            "crawler_plan_id": crawler_plan_id,
-                            "crawler_plan_name": crawler_plan_name,
-                        },
-                        "produce_plan_infos": [
-                            {
-                                "produce_plan_id": produce_plan_info.get("produce_plan_id", ""),
-                                "produce_plan_name": produce_plan_info.get("produce_plan_name", ""),
-                                "is_success": "绑定成功" if not produce_plan_info.get("msg") else "绑定失败",
-                                "msg": produce_plan_info.get("msg", "成功"),
-                            }
-                            for produce_plan_info in produce_plan_infos
-                        ],
-                    },
-                    "db": {"updated_rows": db_updated_rows},
-                },
-                long_term_memory="Create crawler plan by DouYin Content IDs",
-            ),
-        )
-    except Exception as e:
-        logger.error(f"create douyin content crawler plan error. content_ids: {content_ids}, error: {str(e)}")
-        return _log_aigc_return(
-            _LABEL_CONTENT,
-            call_params,
-            ToolResult(
-                title="根据抖音内容ID创建爬取计划失败",
-                output="",
-                error=f"创建爬取计划错误:{str(e)}",
-            ),
-        )
-
-
-def crawler_plan_bind_produce_plan(
-        input_source_info: Dict[str, Any],
-        produce_plan_ids: List[str],
-) -> Tuple[Union[List[Dict[str, str]], None], str]:
-    if not input_source_info or not produce_plan_ids:
-        return None, f"input_source_info or produce_plan_ids is invalid"
-    input_source_check_key = ["inputSourceModal", "inputSourceChannel", "contentType"]
-    try:
-        if not isinstance(produce_plan_ids, list):
-            return None, f"produce_plan_ids is not list"
-        result: List[Dict[str, str]] = []
-        for produce_plan_id in produce_plan_ids:
-            produce_plan_info = {
-                "produce_plan_id": produce_plan_id,
-            }
-            result.append(produce_plan_info)
-            # 获取生成计划详情，msg不为空表示获取失败
-            produce_plan_detail_info, msg = find_produce_plan_info_by_id(produce_plan_id)
-            if msg:
-                produce_plan_info["msg"] = msg
-                continue
-
-            produce_plan_info["produce_plan_name"] = produce_plan_detail_info.get("name", "")
-
-            input_source_groups = produce_plan_detail_info.get("inputSourceGroups", [])
-            if not input_source_groups:
-                produce_plan_info["msg"] = "生成计划没有输入源组"
-                continue
-            # 查询当前爬取计划要添加到的输入源组下标
-            input_source_index = 0
-            for i in range(len(input_source_groups)):
-                input_source_group = input_source_groups[i]
-                if not input_source_group.get("inputSources", []):
-                    continue
-                first_input_source = input_source_group.get("inputSources")[0]
-                if all(input_source_info.get(k, 0) == first_input_source.get(k, -1) for k in input_source_check_key):
-                    input_source_index = i
-                    break
-
-            # 对应的输入源组添加输入源
-            input_source_group = input_source_groups[input_source_index]
-            input_source_group.get("inputSources", []).append(input_source_info)
-
-            response_json = post(PRODUCE_PLAN_SAVE, produce_plan_detail_info)
-            if response_json.get("code") != 0 or not response_json.get("data", {}):
-                produce_plan_info["msg"] = response_json.get("msg", "爬取计划绑定生成计划异常")
-
-        return result, ""
-    except Exception as e:
-        logger.error(f"crawler_plan_bind_produce_plan error. input_source_info: {json.dumps(input_source_info)}, produce_plan_ids: {produce_plan_ids}, error: {str(e)},")
-        return None, str(e)
-
-
-def find_produce_plan_info_by_id(
-        produce_plan_id: str,
-) -> Tuple[Union[Dict[str, str], None], str]:
-    try:
-        if not produce_plan_id or not isinstance(produce_plan_id, str):
-            return None, f"非法的produce_plan_id: {produce_plan_id}"
-
-        params = {
-            "id": produce_plan_id,
-        }
-        response_json = post(GET_PRODUCE_PLAN_DETAIL_BY_ID, params)
-
-        if response_json.get("code") != 0 or not response_json.get("data", {}):
-            return None, response_json.get("msg", "获取生成计划详情异常")
-
-        return response_json.get("data", {}), ""
-    except Exception as e:
-        logger.error(f"find_produce_plan_info_by_id error. produce_plan_id: {produce_plan_id}, error: {str(e)},")
-        return None, str(e)
-
-
-def post(url: str, params: Any) -> Dict[str, Any]:
-    request = {
-        "baseInfo": {
-            "token": DEFAULT_TOKEN,
-        },
-        "params": params
-    }
-    try:
-        response = requests.post(
-            url=url,
-            json=request,
-            headers={"Content-Type": "application/json"},
-            timeout=DEFAULT_TIMEOUT
-        )
-        response.raise_for_status()
-        response_json = response.json()
-
-        return response_json
-    except Exception as e:
-        logger.error(f"invoke aigc platform error. url: {url}, request: {json.dumps(request)}, error: {str(e)}")
-    return {}

scripts/backfill_demand_video_expansion_point_desc.py

@@ -1,162 +0,0 @@
-#!/usr/bin/env python3
-"""补全 demand_video_expansion 表中缺失的 point_desc。
-
-从 multi_demand_video_point 按 (video_id, point_type, expanded_text=point_data) 匹配；
-查不到则保持空值。
-
-用法:
-  .venv/bin/python scripts/backfill_demand_video_expansion_point_desc.py
-  .venv/bin/python scripts/backfill_demand_video_expansion_point_desc.py --biz-dt 20260721
-  .venv/bin/python scripts/backfill_demand_video_expansion_point_desc.py --biz-dt 20260721 --dry-run
-"""
-from __future__ import annotations
-
-import argparse
-import json
-import logging
-import sys
-from pathlib import Path
-from typing import Any
-
-from sqlalchemy import or_, select, update
-
-_ROOT = Path(__file__).resolve().parents[1]
-if str(_ROOT) not in sys.path:
-    sys.path.insert(0, str(_ROOT))
-
-from agents.demand_video_expand_agent.tools.batch_save_demand_expansions import (
-    _fill_missing_point_descs,
-)
-from supply_infra.db.models.demand_video_expansion import DemandVideoExpansion
-from supply_infra.db.session import get_session
-
-logger = logging.getLogger(__name__)
-_BATCH_SIZE = 500
-
-
-def _list_rows_missing_point_desc(biz_dt: str | None) -> list[dict[str, Any]]:
-    stmt = select(DemandVideoExpansion).where(
-        DemandVideoExpansion.is_delete == 0,
-        or_(
-            DemandVideoExpansion.point_desc.is_(None),
-            DemandVideoExpansion.point_desc == "",
-        ),
-    )
-    if biz_dt:
-        stmt = stmt.where(DemandVideoExpansion.biz_dt == biz_dt)
-    stmt = stmt.order_by(DemandVideoExpansion.id)
-
-    with get_session() as session:
-        rows = session.scalars(stmt).all()
-        return [
-            {
-                "id": int(row.id),
-                "biz_dt": str(row.biz_dt),
-                "video_id": str(row.video_id),
-                "point_type": str(row.point_type),
-                "expanded_text": str(row.expanded_text),
-                "point_desc": row.point_desc,
-            }
-            for row in rows
-        ]
-
-
-def backfill_missing_point_descs(
-    biz_dt: str | None = None,
-    *,
-    dry_run: bool = False,
-) -> dict[str, Any]:
-    rows = _list_rows_missing_point_desc(biz_dt)
-    result: dict[str, Any] = {
-        "biz_dt": biz_dt,
-        "dry_run": dry_run,
-        "missing_total": len(rows),
-        "filled": 0,
-        "still_empty": 0,
-        "updated": 0,
-        "samples": [],
-    }
-    if not rows:
-        return result
-
-    with get_session() as session:
-        _fill_missing_point_descs(rows, session)
-
-    to_update: list[dict[str, Any]] = []
-    for row in rows:
-        if row.get("point_desc"):
-            to_update.append(row)
-            result["filled"] += 1
-            if len(result["samples"]) < 10:
-                result["samples"].append(
-                    {
-                        "id": row["id"],
-                        "video_id": row["video_id"],
-                        "point_type": row["point_type"],
-                        "expanded_text": row["expanded_text"],
-                        "point_desc": row["point_desc"][:80]
-                        if len(str(row["point_desc"])) > 80
-                        else row["point_desc"],
-                    }
-                )
-        else:
-            result["still_empty"] += 1
-
-    if dry_run or not to_update:
-        result["updated"] = 0
-        return result
-
-    with get_session() as session:
-        for i in range(0, len(to_update), _BATCH_SIZE):
-            batch = to_update[i : i + _BATCH_SIZE]
-            for row in batch:
-                session.execute(
-                    update(DemandVideoExpansion)
-                    .where(DemandVideoExpansion.id == int(row["id"]))
-                    .values(point_desc=row["point_desc"])
-                )
-            result["updated"] += len(batch)
-
-    return result
-
-
-def main(argv: list[str] | None = None) -> int:
-    parser = argparse.ArgumentParser(
-        description="补全 demand_video_expansion 缺失的 point_desc",
-    )
-    parser.add_argument("--biz-dt", default=None, help="业务日期 YYYYMMDD，默认全表")
-    parser.add_argument("--dry-run", action="store_true", help="仅统计，不写库")
-    parser.add_argument("--json", action="store_true", help="以 JSON 输出结果")
-    args = parser.parse_args(argv)
-
-    logging.basicConfig(
-        level=logging.INFO,
-        format="%(asctime)s %(levelname)s %(name)s: %(message)s",
-    )
-
-    result = backfill_missing_point_descs(args.biz_dt, dry_run=bool(args.dry_run))
-
-    if args.json:
-        print(json.dumps(result, ensure_ascii=False, indent=2, default=str))
-    else:
-        print("\n=== point_desc 补全 ===")
-        print(f"biz_dt={result.get('biz_dt') or '全部'}")
-        print(f"dry_run={result.get('dry_run')}")
-        print(f"缺失记录={result.get('missing_total')}")
-        print(f"可补全={result.get('filled')}")
-        print(f"仍为空={result.get('still_empty')}")
-        print(f"已更新={result.get('updated')}")
-        if result.get("samples"):
-            print("\n示例:")
-            for item in result["samples"]:
-                print(
-                    f"  id={item['id']} video={item['video_id']} "
-                    f"type={item['point_type']} text={item['expanded_text']!r} "
-                    f"desc={item['point_desc']!r}"
-                )
-
-    return 0
-
-
-if __name__ == "__main__":
-    raise SystemExit(main())

scripts/retry_failed_grade_plan_items.py

@@ -1,100 +0,0 @@
-#!/usr/bin/env python3
-"""重试 demand_grade_plan_group_item 中 status=failed 的分级任务。
-
-用法:
-  .venv/bin/python scripts/retry_failed_grade_plan_items.py
-  .venv/bin/python scripts/retry_failed_grade_plan_items.py --biz-dt 20260721
-  .venv/bin/python scripts/retry_failed_grade_plan_items.py --biz-dt 20260721 --workers 5
-  .venv/bin/python scripts/retry_failed_grade_plan_items.py --biz-dt 20260721 --dry-run
-  .venv/bin/python scripts/retry_failed_grade_plan_items.py --group-id 12 --group-id 15
-"""
-from __future__ import annotations
-
-import argparse
-import json
-import logging
-import sys
-from pathlib import Path
-
-_ROOT = Path(__file__).resolve().parents[1]
-if str(_ROOT) not in sys.path:
-    sys.path.insert(0, str(_ROOT))
-
-from supply_infra.scheduler.jobs.grade_demand_pool import retry_failed_plan_group_items
-from supply_infra.scheduler.plan_group_batch import MAX_DEMANDS_PER_BATCH
-
-logger = logging.getLogger(__name__)
-
-
-def main(argv: list[str] | None = None) -> int:
-    parser = argparse.ArgumentParser(
-        description="重试 demand_grade_plan_group_item 中失败的分级任务",
-    )
-    parser.add_argument("--biz-dt", default=None, help="业务日期 YYYYMMDD，默认当天")
-    parser.add_argument("--workers", type=int, default=5, help="并发执行的 plan_group 数")
-    parser.add_argument(
-        "--max-demands-per-batch",
-        type=int,
-        default=MAX_DEMANDS_PER_BATCH,
-        help=f"每个 Agent 子批次最多处理的需求条数，默认 {MAX_DEMANDS_PER_BATCH}",
-    )
-    parser.add_argument(
-        "--group-id",
-        type=int,
-        action="append",
-        dest="group_ids",
-        help="仅重试指定 group_id，可重复传入",
-    )
-    parser.add_argument(
-        "--dry-run",
-        action="store_true",
-        help="仅列出将要重试的 failed 记录，不实际执行",
-    )
-    parser.add_argument("--json", action="store_true", help="以 JSON 打印结果")
-    args = parser.parse_args(argv)
-
-    logging.basicConfig(
-        level=logging.INFO,
-        format="%(asctime)s %(levelname)s %(name)s: %(message)s",
-    )
-
-    result = retry_failed_plan_group_items(
-        args.biz_dt,
-        workers=max(1, int(args.workers)),
-        max_demands_per_batch=max(1, min(int(args.max_demands_per_batch), MAX_DEMANDS_PER_BATCH)),
-        group_ids=args.group_ids,
-        dry_run=bool(args.dry_run),
-    )
-
-    if args.json:
-        print(json.dumps(result, ensure_ascii=False, indent=2, default=str))
-    else:
-        reset = result.get("reset") or {}
-        print("\n=== 失败任务重试 ===")
-        print(f"biz_dt={result.get('biz_dt')}")
-        print(f"dry_run={result.get('dry_run')}")
-        print(f"failed_items={result.get('failed_items', 0)}")
-        print(f"reset_items={reset.get('reset_items', 0)}")
-        print(f"reset_groups={reset.get('reset_groups', 0)}")
-        if reset.get("group_ids"):
-            print(f"group_ids={reset.get('group_ids')}")
-        if not result.get("dry_run"):
-            print(f"graded: {result.get('graded_before')} -> {result.get('graded_after')}")
-            print(f"remaining_failed={result.get('remaining_failed', 0)}")
-            print(f"group_status={result.get('group_status')}")
-            print(f"success={result.get('success')}")
-        elif reset.get("items"):
-            print("\n待重试明细:")
-            for item in reset["items"][:20]:
-                print(
-                    f"  item_id={item['item_id']} group_id={item['group_id']} "
-                    f"demand={item['demand_name']!r}"
-                )
-            if len(reset["items"]) > 20:
-                print(f"  ... 另有 {len(reset['items']) - 20} 条")
-
-    return 0 if result.get("success") else 1
-
-
-if __name__ == "__main__":
-    raise SystemExit(main())

scripts/run_grade_plan_groups.py

@@ -1,85 +0,0 @@
-#!/usr/bin/env python3
-"""批量执行 demand_grade_plan_group 分级任务。
-
-与定时任务共用 supply_infra.scheduler.jobs.grade_demand_pool.grade_demand_pool。
-
-Usage:
-  .venv/bin/python scripts/run_grade_plan_groups.py
-  .venv/bin/python scripts/run_grade_plan_groups.py --biz-dt 20260721
-  .venv/bin/python scripts/run_grade_plan_groups.py --biz-dt 20260721 --workers 5
-  .venv/bin/python scripts/run_grade_plan_groups.py --biz-dt 20260721 --with-orchestrate
-"""
-from __future__ import annotations
-
-import argparse
-import json
-import logging
-import sys
-from pathlib import Path
-
-_ROOT = Path(__file__).resolve().parents[1]
-if str(_ROOT) not in sys.path:
-    sys.path.insert(0, str(_ROOT))
-
-from supply_infra.scheduler.plan_group_batch import MAX_DEMANDS_PER_BATCH
-from supply_infra.scheduler.jobs.grade_demand_pool import grade_demand_pool
-
-logger = logging.getLogger(__name__)
-
-
-def main(argv: list[str] | None = None) -> int:
-    parser = argparse.ArgumentParser(description="批量执行 demand_grade_plan_group 分级任务")
-    parser.add_argument("--biz-dt", default="20260721", help="业务日期 YYYYMMDD，默认 20260721")
-    parser.add_argument(
-        "--max-demands-per-batch",
-        type=int,
-        default=MAX_DEMANDS_PER_BATCH,
-        help=f"每个 Agent 子批次最多处理的需求条数，默认 {MAX_DEMANDS_PER_BATCH}",
-    )
-    parser.add_argument("--workers", type=int, default=5, help="并发执行的 plan_group 数")
-    parser.add_argument(
-        "--max-rounds",
-        type=int,
-        default=0,
-        help="最多执行轮数，0 表示直到没有 pending 任务",
-    )
-    parser.add_argument(
-        "--with-orchestrate",
-        action="store_true",
-        help="执行前先自动分配计划组（每组约 30 个需求；非统筹 Agent）",
-    )
-    parser.add_argument(
-        "--json",
-        action="store_true",
-        help="最终以 JSON 打印摘要",
-    )
-    args = parser.parse_args(argv)
-
-    logging.basicConfig(
-        level=logging.INFO,
-        format="%(asctime)s %(levelname)s %(name)s: %(message)s",
-    )
-
-    result = grade_demand_pool(
-        str(args.biz_dt).strip(),
-        workers=max(1, int(args.workers)),
-        max_demands_per_batch=max(1, min(int(args.max_demands_per_batch), MAX_DEMANDS_PER_BATCH)),
-        with_orchestrate=bool(args.with_orchestrate),
-        max_rounds=max(0, int(args.max_rounds)),
-    )
-
-    if args.json:
-        print(json.dumps(result, ensure_ascii=False, indent=2, default=str))
-    else:
-        print("\n=== 批量分级完成 ===")
-        print(f"biz_dt={result.get('biz_dt')}")
-        print(f"完成任务组={result.get('groups_run')}")
-        print(f"已分级: {result.get('graded_before')} -> {result.get('graded_after')}")
-        print(f"任务状态: {(result.get('group_status') or result.get('plan_execution', {}).get('final_snapshot', {}).get('group_status'))}")
-        print(f"是否全部完成: {result.get('success')}")
-
-    return 0 if result.get("success") else 1
-
-
-if __name__ == "__main__":
-    raise SystemExit(main())

scripts/verify_find_agent_flow.py

@@ -1,59 +0,0 @@
-"""Smoke-check find_agent demand flow without calling the LLM."""
-from __future__ import annotations
-
-import json
-import sys
-
-from agents.find_agent.demand_run import (
-    build_find_agent_user_input,
-    build_run_input_payload,
-    filter_pending_contexts,
-    list_find_demand_contexts,
-    pick_find_demand_context,
-    prepare_video_discovery_run,
-)
-from agents.find_agent.async_runner import _run_coroutine
-
-
-def main() -> int:
-    biz_dt, contexts = list_find_demand_contexts()
-    pending, stats = filter_pending_contexts(contexts, biz_dt)
-    ctx = pick_find_demand_context(biz_dt, index=0)
-    if ctx is None:
-        print(json.dumps({"ok": False, "error": "no context"}, ensure_ascii=False))
-        return 1
-
-    run_id, skip_reason = prepare_video_discovery_run(ctx, force=True)
-    if skip_reason or not run_id:
-        print(json.dumps({"ok": False, "skip_reason": skip_reason}, ensure_ascii=False))
-        return 1
-
-    prompt = build_find_agent_user_input(ctx, run_id)
-    payload = build_run_input_payload(ctx)
-
-    async def _noop():
-        return None
-
-    # 验证主线程 asyncio 运行器可用
-    _run_coroutine(_noop())
-
-    print(
-        json.dumps(
-            {
-                "ok": True,
-                "biz_dt": biz_dt,
-                "total": len(contexts),
-                "pending": len(pending),
-                "stats": stats,
-                "run_id": run_id,
-                "prompt_lines": len(prompt.splitlines()),
-                "payload_keys": list(payload.keys()),
-            },
-            ensure_ascii=False,
-        )
-    )
-    return 0
-
-
-if __name__ == "__main__":
-    sys.exit(main())

scripts/verify_find_agent_from_demand.py

@@ -1,126 +0,0 @@
-"""从 demand_grade + demand_video_expansion 选取一条记录，调用 find_agent。"""
-from __future__ import annotations
-
-import argparse
-import json
-import logging
-
-from agents.find_agent.demand_run import (
-    build_find_agent_user_input,
-    build_run_input_payload,
-    discover_videos_for_demand,
-    list_find_demand_contexts,
-    pick_find_demand_context,
-    serialize_find_demand_context,
-)
-
-logging.basicConfig(
-    level=logging.INFO,
-    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
-)
-logger = logging.getLogger(__name__)
-
-
-def _load_target_context(args: argparse.Namespace):
-    if args.list_only:
-        biz_dt, contexts = list_find_demand_contexts(args.biz_dt)
-        print(
-            json.dumps(
-                {
-                    "biz_dt": biz_dt,
-                    "total": len(contexts),
-                    "items": [serialize_find_demand_context(ctx) for ctx in contexts],
-                },
-                ensure_ascii=False,
-                indent=2,
-            )
-        )
-        return None
-
-    ctx = pick_find_demand_context(
-        args.biz_dt,
-        index=args.index,
-        demand_grade_id=args.demand_grade_id,
-    )
-    if ctx is None:
-        biz_dt, contexts = list_find_demand_contexts(args.biz_dt)
-        raise SystemExit(
-            "未找到匹配的待执行记录。"
-            f" biz_dt={biz_dt}, total={len(contexts)}, index={args.index},"
-            f" demand_grade_id={args.demand_grade_id}"
-        )
-    return ctx
-
-
-def main() -> None:
-    parser = argparse.ArgumentParser(
-        description="从 S/A 需求拓展点位中选取一条记录，调用 find_agent。",
-    )
-    parser.add_argument("--biz-dt", help="业务日 YYYYMMDD，默认取最新 demand_grade.biz_dt")
-    parser.add_argument(
-        "--index",
-        type=int,
-        default=0,
-        help="在未指定 demand_grade_id 时，选取第几条记录（从 0 开始）",
-    )
-    parser.add_argument("--demand-grade-id", type=int, help="指定 demand_grade.id")
-    parser.add_argument(
-        "--list-only",
-        action="store_true",
-        help="只列出可执行记录，不调用 Agent",
-    )
-    parser.add_argument(
-        "--dry-run",
-        action="store_true",
-        help="只打印组装后的上下文与 prompt，不调用 Agent",
-    )
-    parser.add_argument(
-        "--force",
-        action="store_true",
-        help="忽略当天已执行记录，强制重跑",
-    )
-    args = parser.parse_args()
-
-    ctx = _load_target_context(args)
-    if ctx is None:
-        return
-
-    summary = serialize_find_demand_context(ctx)
-    run_payload = build_run_input_payload(ctx)
-    print("CONTEXT " + json.dumps(summary, ensure_ascii=False), flush=True)
-    print("RUN_PAYLOAD " + json.dumps(run_payload, ensure_ascii=False), flush=True)
-
-    if args.dry_run:
-        user_input = build_find_agent_user_input(ctx, "<pre-create-on-run>")
-        print("PROMPT_BEGIN", flush=True)
-        print(user_input, flush=True)
-        print("PROMPT_END", flush=True)
-        return
-
-    execution = discover_videos_for_demand(ctx, force=args.force)
-    if execution.skipped:
-        print("SKIPPED " + json.dumps({"skip_reason": execution.skip_reason}, ensure_ascii=False), flush=True)
-        return
-
-    agent_result = execution.agent_result
-    if agent_result is None:
-        raise SystemExit("Agent 未返回结果")
-    print(
-        "AGENT_REPORT "
-        + json.dumps(
-            {
-                "run_id": execution.run_id,
-                "iterations": agent_result.iterations,
-                "content_chars": len(agent_result.content or ""),
-            },
-            ensure_ascii=False,
-        ),
-        flush=True,
-    )
-    print("FINAL_CONTENT_BEGIN", flush=True)
-    print(agent_result.content or "", flush=True)
-    print("FINAL_CONTENT_END", flush=True)
-
-
-if __name__ == "__main__":
-    main()

scripts/verify_find_agent_live.py

@@ -1,226 +0,0 @@
-"""Run a real find_agent acceptance case and clean its database records by default."""
-
-from __future__ import annotations
-
-import argparse
-import asyncio
-import json
-from typing import Any
-
-from sqlalchemy import delete, func, select
-
-from agents.find_agent import create_find_agent
-from agents.find_agent.agent import _validate_report_buckets
-from supply_agent.config import Settings
-from supply_agent.types import AgentEventType
-from supply_infra.db.models.video_discovery import (
-    VideoDiscoveryCandidate,
-    VideoDiscoveryRun,
-    VideoDiscoverySearch,
-)
-from supply_infra.db.session import get_session
-
-
-def _parse_json(raw: str) -> dict[str, Any]:
-    try:
-        value = json.loads(raw)
-    except (TypeError, json.JSONDecodeError):
-        return {}
-    return value if isinstance(value, dict) else {}
-
-
-def _cleanup_run(run_id: str) -> int:
-    with get_session() as session:
-        session.execute(
-            delete(VideoDiscoveryCandidate).where(
-                VideoDiscoveryCandidate.run_id == run_id
-            )
-        )
-        session.execute(
-            delete(VideoDiscoverySearch).where(
-                VideoDiscoverySearch.run_id == run_id
-            )
-        )
-        session.execute(
-            delete(VideoDiscoveryRun).where(VideoDiscoveryRun.run_id == run_id)
-        )
-    with get_session() as session:
-        remaining = session.scalar(
-            select(func.count(VideoDiscoveryRun.id)).where(
-                VideoDiscoveryRun.run_id == run_id
-            )
-        )
-    return int(remaining or 0)
-
-
-async def _run(args: argparse.Namespace) -> None:
-    agent = create_find_agent(settings=Settings.from_env(LOG_ENABLED=False))
-    relevant_points = [{"point": point} for point in args.relevant_point]
-    prompt = (
-        "这是一次真实验收测试，请完整执行并严格遵守完成条件。\n"
-        f"demand_word：{args.demand_word}\n"
-        f"seed_video_title：{args.seed_video_title}\n"
-        f"relevant_points：{json.dumps(relevant_points, ensure_ascii=False)}\n"
-        "请寻找老年人或临近退休人群可能喜欢观看和分享的视频。"
-    )
-
-    tool_calls: list[str] = []
-    tool_payloads: list[dict[str, Any]] = []
-    run_ids: list[str] = []
-    domain_errors: list[dict[str, Any]] = []
-    audit_results: list[dict[str, Any]] = []
-    states: list[dict[str, Any]] = []
-    final_content = ""
-    done_data: dict[str, Any] = {}
-
-    try:
-        async for event in agent.astream(prompt):
-            if event.type == AgentEventType.TOOL_CALL:
-                name = str(event.data.get("name") or "")
-                call_args = _parse_json(str(event.data.get("arguments") or "{}"))
-                tool_calls.append(name)
-                tool_payloads.append({"name": name, "args": call_args})
-                print(f"TOOL_CALL {len(tool_calls)} {name}", flush=True)
-            elif event.type == AgentEventType.TOOL_RESULT:
-                name = str(event.data.get("name") or "")
-                data = _parse_json(str(event.data.get("content") or ""))
-                if name == "create_video_discovery_run" and data.get("run_id"):
-                    run_ids.append(str(data["run_id"]))
-                if name in {
-                    "audit_video_discovery_process",
-                    "audit_video_discovery_run",
-                } and not data.get("error"):
-                    audit_results.append(data)
-                if name == "query_video_discovery_state" and not data.get("error"):
-                    states.append(data)
-                if data.get("error"):
-                    domain_errors.append(
-                        {
-                            "tool": name,
-                            "budget_exhausted": bool(data.get("budget_exhausted")),
-                            "error": str(data["error"])[:300],
-                        }
-                    )
-                status = "DOMAIN_ERROR" if data.get("error") else "OK"
-                if event.data.get("is_error") and not data.get("error"):
-                    status = "ERROR"
-                extra = ""
-                if data.get("error"):
-                    extra = " error=" + json.dumps(
-                        str(data["error"])[:240],
-                        ensure_ascii=False,
-                    )
-                if name in {
-                    "audit_video_discovery_process",
-                    "audit_video_discovery_run",
-                } and not data.get("error"):
-                    audit_flag = data.get("can_finish")
-                    extra = f" can_finish={audit_flag}"
-                    if audit_flag is False:
-                        extra += " violations=" + json.dumps(
-                            data.get("critical_violations"),
-                            ensure_ascii=False,
-                        )
-                print(f"TOOL_RESULT {name} {status}{extra}", flush=True)
-            elif event.type == AgentEventType.MESSAGE:
-                final_content = str(event.data.get("content") or "")
-            elif event.type == AgentEventType.DONE:
-                done_data = event.data
-                if not final_content:
-                    final_content = str(event.data.get("content") or "")
-
-        unique_keywords: list[str] = []
-        recorded_sources: list[Any] = []
-        pagination_calls = 0
-        for entry in tool_payloads:
-            name = entry["name"]
-            call_args = entry["args"]
-            if name in {"douyin_search", "douyin_search_tikhub"}:
-                keyword = str(call_args.get("keyword") or "").strip()
-                if keyword and keyword not in unique_keywords:
-                    unique_keywords.append(keyword)
-                if (
-                    str(call_args.get("cursor") or "0") not in {"", "0"}
-                    or call_args.get("search_id")
-                ):
-                    pagination_calls += 1
-            if name == "record_video_search_page":
-                recorded_sources.append(call_args.get("source_type"))
-
-        last_state = states[-1] if states else {}
-        final_bucket_error = (
-            _validate_report_buckets(final_content, last_state)
-            if final_content and last_state
-            else "缺少最终文本或最终状态"
-        )
-        report = {
-            "model": agent.model,
-            "iterations": done_data.get("iterations"),
-            "tool_calls": len(tool_calls),
-            "tool_names": tool_calls,
-            "unique_search_keywords": unique_keywords,
-            "pagination_calls": pagination_calls,
-            "recorded_source_types": recorded_sources,
-            "qwen_calls": tool_calls.count("qwen_video_analyze"),
-            "audit_calls": len(audit_results),
-            "last_audit_can_finish": (
-                audit_results[-1].get("can_finish") if audit_results else None
-            ),
-            "last_audit_violations": (
-                audit_results[-1].get("critical_violations")
-                if audit_results
-                else None
-            ),
-            "domain_errors": domain_errors,
-            "state_search_count": len(last_state.get("searches", [])),
-            "state_candidate_count": len(last_state.get("candidates", [])),
-            "state_run": last_state.get("run"),
-            "final_content_chars": len(final_content),
-            "final_has_primary_section": "主推荐" in final_content,
-            "final_has_backup_section": "补充推荐" in final_content,
-            "final_bucket_consistent": final_bucket_error is None,
-            "final_bucket_error": final_bucket_error,
-            "max_iterations_failure": final_content.startswith(
-                "Max iterations reached"
-            ),
-            "run_ids": run_ids,
-        }
-        print("AGENT_REPORT " + json.dumps(report, ensure_ascii=False), flush=True)
-        print("FINAL_CONTENT_BEGIN", flush=True)
-        print(final_content, flush=True)
-        print("FINAL_CONTENT_END", flush=True)
-    finally:
-        if not args.keep_data:
-            for run_id in set(run_ids):
-                remaining = _cleanup_run(run_id)
-                print(f"CLEANUP {run_id} remaining={remaining}", flush=True)
-
-
-def _arguments() -> argparse.Namespace:
-    parser = argparse.ArgumentParser()
-    parser.add_argument(
-        "--demand-word",
-        default="个人养老金税收优惠",
-    )
-    parser.add_argument(
-        "--seed-video-title",
-        default="个人养老金制度全面实施，退休前这样缴纳可以享受税收优惠",
-    )
-    parser.add_argument(
-        "--relevant-point",
-        action="append",
-        default=[
-            "个人养老金缴费如何抵扣个税",
-            "适合临近退休人群转发给家人了解",
-        ],
-    )
-    parser.add_argument(
-        "--keep-data",
-        action="store_true",
-        help="保留本次运行的数据库记录；默认清理。",
-    )
-    return parser.parse_args()
-
-
-if __name__ == "__main__":
-    asyncio.run(_run(_arguments()))

supply_infra/aigc/plan_map.py

@@ -5,7 +5,7 @@ from typing import Any, TypeVar
 
 T = TypeVar("T")
 
-# 与 scripts/aigc_platform_api.py 保持一致；按 (生成ID, 发布ID) 去重后做均匀分发。
+# 品类 → (生成计划ID, 发布计划ID)；按 pair 去重后做均匀分发。
 AIGC_PLAN_ID_MAP: dict[str, dict[str, str]] = {
     "健康知识": {"生成ID": "20260408092313211598604", "发布ID": "20260408115944193153417"},
     "历史名人": {"生成ID": "20260408083251311809309", "发布ID": "20260408115139124511126"},

jobs/init_db.py → supply_infra/db/__main__.py

@@ -1,5 +1,4 @@
-#!/usr/bin/env python3
-"""CLI entry point to initialize database tables."""
+"""CLI: ``python -m supply_infra.db`` 初始化数据库表。"""
 
 from supply_infra.config import get_infra_settings
 from supply_infra.db import init_db

supply_infra/scheduler/_verify_auto_assign.py

@@ -1,47 +0,0 @@
-"""自动分配计划组逻辑自检。"""
-from __future__ import annotations
-
-from supply_infra.scheduler.plan_group_batch import pack_category_units
-
-
-def _demands(n: int) -> list[dict]:
-    return [{"pool_id": i, "demand_name": f"d{i}"} for i in range(n)]
-
-
-def test_pack_same_parent_prefers_single_group() -> None:
-  units = [
-      (101, 10, _demands(12)),
-      (102, 10, _demands(15)),
-      (201, 20, _demands(10)),
-  ]
-  groups = pack_category_units(units, max_demands_per_group=30)
-  assert groups == [[101, 102], [201]]
-
-
-def test_pack_large_category_alone() -> None:
-  units = [
-      (101, 10, _demands(35)),
-      (102, 10, _demands(5)),
-  ]
-  groups = pack_category_units(units, max_demands_per_group=30)
-  assert groups == [[101], [102]]
-
-
-def test_pack_exact_capacity() -> None:
-  units = [
-      (101, 10, _demands(15)),
-      (102, 10, _demands(15)),
-  ]
-  groups = pack_category_units(units, max_demands_per_group=30)
-  assert groups == [[101, 102]]
-
-
-def main() -> None:
-  test_pack_same_parent_prefers_single_group()
-  test_pack_large_category_alone()
-  test_pack_exact_capacity()
-  print("auto_assign_grade_plan: all tests passed")
-
-
-if __name__ == "__main__":
-  main()

supply_infra/scheduler/app.py

@@ -1,8 +1,6 @@
 from __future__ import annotations
 
 import logging
-import signal
-import time
 from typing import TYPE_CHECKING
 
 from apscheduler.schedulers.background import BackgroundScheduler
@@ -133,25 +131,3 @@ def stop_scheduler() -> None:
     _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:
-        while scheduler.running:
-            time.sleep(3600)
-    except (KeyboardInterrupt, SystemExit):
-        stop_scheduler()
-        logger.info("Scheduler stopped.")

supply_infra/scheduler/cli_result.py

@@ -0,0 +1,33 @@
+"""CLI job result helpers: logging, JSON stdout, exit code contract."""
+from __future__ import annotations
+
+import json
+import logging
+import sys
+from collections.abc import Callable
+from typing import Any
+
+
+def configure_cli_logging() -> None:
+    logging.basicConfig(
+        level=logging.INFO,
+        format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
+    )
+
+
+def emit_and_exit(result: Any) -> None:
+    """Print one JSON line to stdout; exit 1 when result.success is False."""
+    print(json.dumps(result, ensure_ascii=False, default=str))
+    if isinstance(result, dict) and result.get("success") is False:
+        sys.exit(1)
+    sys.exit(0)
+
+
+def run_cli(action: Callable[[], Any], *, label: str) -> None:
+    """Configure logging, run action, emit JSON result, and exit."""
+    configure_cli_logging()
+    try:
+        emit_and_exit(action())
+    except Exception as exc:
+        logging.exception("%s failed", label)
+        emit_and_exit({"success": False, "error": str(exc)})

supply_infra/scheduler/jobs/backfill_multi_demand_pool_video_list.py

@@ -1,35 +0,0 @@
-"""手动维护任务：回填需求池记录的 video_list / video_count。"""
-from __future__ import annotations
-
-import logging
-from typing import Any
-
-from supply_infra.db.repositories.multi_demand_pool_di_repo import MultiDemandPoolDiRepository
-from supply_infra.db.session import get_session
-from supply_infra.odps.client import get_odps_client
-from supply_infra.scheduler.jobs.sync_multi_demand_pool_odps_to_mysql import _to_mysql_rows
-
-logger = logging.getLogger(__name__)
-
-
-def backfill_video_list(partition_date: str) -> dict[str, Any]:
-    """从 ODPS 回填指定分区的 video_list / video_count（每条最多前 10 个 video_id）。"""
-    logger.info("Backfill video_list for partition: %s", partition_date)
-    raw_rows = get_odps_client().fetch_multi_demand_pool(partition_date)
-    mysql_rows = _to_mysql_rows(raw_rows, partition_date)
-
-    with get_session() as session:
-        updated = MultiDemandPoolDiRepository(session).update_video_fields(
-            partition_date,
-            mysql_rows,
-        )
-
-    result = {
-        "partition_date": partition_date,
-        "fetched": len(raw_rows),
-        "unique_rows": len(mysql_rows),
-        "updated": updated,
-        "with_video": sum(1 for row in mysql_rows if row.get("video_list")),
-    }
-    logger.info("Backfill video_list completed: %s", result)
-    return result

supply_infra/scheduler/jobs/demand_pool/__init__.py

@@ -0,0 +1,5 @@
+"""策略需求池同步任务（ODPS → MySQL + 归属/热度/树权重/视频）。"""
+
+from supply_infra.scheduler.jobs.demand_pool.sync import sync_multi_demand_pool_odps_to_mysql
+
+__all__ = ["sync_multi_demand_pool_odps_to_mysql"]

supply_infra/scheduler/jobs/demand_pool/__main__.py

@@ -0,0 +1,15 @@
+"""策略需求池同步任务 CLI：``python -m supply_infra.scheduler.jobs.demand_pool``。"""
+from __future__ import annotations
+
+import sys
+
+from supply_infra.scheduler.cli_result import run_cli
+from supply_infra.scheduler.jobs.demand_pool.sync import sync_multi_demand_pool_odps_to_mysql
+
+if __name__ == "__main__":
+    run_cli(
+        lambda: sync_multi_demand_pool_odps_to_mysql(
+            partition_date=sys.argv[1] if len(sys.argv) > 1 else None,
+        ),
+        label="sync_multi_demand_pool_odps_to_mysql",
+    )

supply_infra/scheduler/jobs/sync_multi_demand_pool_odps_to_mysql.py → supply_infra/scheduler/jobs/demand_pool/sync.py

@@ -33,13 +33,11 @@ from supply_infra.db.repositories.demand_popularity_stats_repo import (
 from supply_infra.db.repositories.multi_demand_pool_di_repo import MultiDemandPoolDiRepository
 from supply_infra.db.session import get_session
 from supply_infra.odps.client import get_odps_client
-from supply_infra.scheduler.jobs.compute_category_tree_weight import (
+from supply_infra.scheduler.jobs.demand_pool.belong_rel import sync_demand_belong_pool_rel
+from supply_infra.scheduler.jobs.demand_pool.tree_weight import (
     compute_category_tree_weight,
 )
-from supply_infra.scheduler.jobs.sync_demand_belong_pool_rel import (
-    sync_demand_belong_pool_rel,
-)
-from supply_infra.scheduler.jobs.sync_multi_demand_videos import sync_multi_demand_videos
+from supply_infra.scheduler.jobs.demand_pool.videos import sync_multi_demand_videos
 
 logger = logging.getLogger(__name__)
 

supply_infra/scheduler/jobs/compute_category_tree_weight.py → supply_infra/scheduler/jobs/demand_pool/tree_weight.py

@@ -28,10 +28,8 @@ from supply_infra.db.repositories.demand_popularity_stats_repo import (
 from supply_infra.db.repositories.global_tree_category_repo import (
     GlobalTreeCategoryRepository,
 )
+from supply_agent.ranking import rank_to_scores
 from supply_infra.db.session import get_session
-from supply_infra.scheduler.jobs.update_category_tree_rank_scores import (
-    update_category_tree_rank_scores,
-)
 
 logger = logging.getLogger(__name__)
 
@@ -44,7 +42,15 @@ METRIC_KEYS: tuple[str, ...] = (
     "real_vov_7d",
 )
 
+POP_DIM_KEYS: tuple[str, ...] = (
+    "ext_pop",
+    "plat_sust_pop",
+    "plat_ly_pop",
+    "recent_pop",
+)
+
 _UPSERT_BATCH = 500
+_RANK_UPDATE_BATCH = 500
 
 
 @dataclass
@@ -222,6 +228,99 @@ def _materialize_stats_row(stats: Any) -> SimpleNamespace:
     return SimpleNamespace(**payload)
 
 
+def _materialize_weight_row(row: Any) -> dict[str, Any]:
+    """在 session 内抽出标量，避免 DetachedInstanceError。"""
+    payload: dict[str, Any] = {
+        "category_id": int(row.category_id),
+        "biz_dt": str(row.biz_dt),
+    }
+    for dim in POP_DIM_KEYS:
+        count = int(getattr(row, f"{dim}_count", 0) or 0)
+        payload[f"{dim}_count"] = count
+        if count > 0:
+            avg_raw = getattr(row, f"{dim}_avg", None)
+            payload[f"{dim}_avg"] = float(avg_raw) if avg_raw is not None else None
+        else:
+            payload[f"{dim}_avg"] = None
+    return payload
+
+
+def _build_rank_score_rows(
+    rows: list[dict[str, Any]],
+) -> list[dict[str, Any]]:
+    """根据已写入的权重行计算各节点排名分。"""
+    dim_scores: dict[str, dict[int, float]] = {}
+    for dim in POP_DIM_KEYS:
+        candidates: list[tuple[int, float]] = []
+        for row in rows:
+            count = int(row[f"{dim}_count"])
+            if count <= 0:
+                continue
+            avg_raw = row[f"{dim}_avg"]
+            if avg_raw is None:
+                continue
+            avg = float(avg_raw)
+            candidates.append((int(row["category_id"]), avg))
+        dim_scores[dim] = rank_to_scores(candidates)
+
+    updates: list[dict[str, Any]] = []
+    for row in rows:
+        category_id = int(row["category_id"])
+        biz_dt = str(row["biz_dt"])
+        score_values = {
+            f"{dim}_score": dim_scores[dim].get(category_id)
+            for dim in POP_DIM_KEYS
+        }
+        non_null_scores = [v for v in score_values.values() if v is not None]
+        total = sum(non_null_scores) if non_null_scores else None
+        updates.append(
+            {
+                "category_id": category_id,
+                "biz_dt": biz_dt,
+                **{
+                    col: _optional_dec(score_values[col])
+                    for col in (
+                        "ext_pop_score",
+                        "plat_sust_pop_score",
+                        "plat_ly_pop_score",
+                        "recent_pop_score",
+                    )
+                },
+                "total_score": _optional_dec(total),
+            }
+        )
+    return updates
+
+
+def _update_category_tree_rank_scores(biz_dt: str) -> dict[str, Any]:
+    """在 category_tree_weight 全部写入后，按 biz_dt 更新四维排名分与 total_score。"""
+    with get_session() as session:
+        repo = CategoryTreeWeightRepository(session)
+        rows = [
+            _materialize_weight_row(row)
+            for row in repo.list_by_biz_dt(biz_dt)
+        ]
+
+    if not rows:
+        logger.info("Category tree rank scores: no rows, skip biz_dt=%s", biz_dt)
+        return {"biz_dt": biz_dt, "nodes": 0, "updated": 0}
+
+    updates = _build_rank_score_rows(rows)
+    updated = 0
+    with get_session() as session:
+        repo = CategoryTreeWeightRepository(session)
+        for i in range(0, len(updates), _RANK_UPDATE_BATCH):
+            updated += repo.update_rank_scores(updates[i : i + _RANK_UPDATE_BATCH])
+
+    result = {
+        "biz_dt": biz_dt,
+        "nodes": len(updates),
+        "updated": updated,
+    }
+    logger.info("Category tree rank scores completed: %s", result)
+    return result
+
+
 def compute_category_tree_weight(biz_dt: str) -> dict[str, Any]:
     """按 biz_dt 计算整棵树各维度加权平均分并写入 category_tree_weight。"""
     with get_session() as session:
@@ -279,6 +378,6 @@ def compute_category_tree_weight(biz_dt: str) -> dict[str, Any]:
     }
     logger.info("Category tree weight completed: %s", result)
 
-    rank_stats = update_category_tree_rank_scores(biz_dt)
+    rank_stats = _update_category_tree_rank_scores(biz_dt)
     result["rank_scores"] = rank_stats
     return result

supply_infra/scheduler/jobs/sync_multi_demand_videos.py → supply_infra/scheduler/jobs/demand_pool/videos.py

@@ -276,182 +276,3 @@ def sync_multi_demand_videos(
     }
     logger.info("Multi demand video sync completed: %s", result)
     return result
-
-
-def backfill_video_titles(
-    batch_size: int = VIDEO_SYNC_BATCH_SIZE,
-) -> dict[str, Any]:
-    """为 title 为空的已有行，从 ODPS 昨天分区回填 target_post.title。"""
-    decode_dt = (datetime.now() - timedelta(days=1)).strftime("%Y%m%d")
-    chunk = max(1, int(batch_size))
-    logger.info("Backfill video titles: decode_dt=%s batch_size=%d", decode_dt, chunk)
-
-    with get_session() as session:
-        pending = MultiDemandVideoDetailRepository(session).list_vids_missing_title()
-
-    if not pending:
-        result = {
-            "decode_dt": decode_dt,
-            "pending": 0,
-            "batches": 0,
-            "updated": 0,
-            "missing_in_odps": 0,
-            "skipped_no_title": 0,
-        }
-        logger.info("No vids missing title: %s", result)
-        return result
-
-    odps = get_odps_client()
-    batches = [pending[i : i + chunk] for i in range(0, len(pending), chunk)]
-    total_updated = 0
-    total_missing = 0
-    total_skipped = 0
-
-    for idx, batch_vids in enumerate(batches, start=1):
-        logger.info(
-            "Title backfill batch %d/%d: %d vids",
-            idx,
-            len(batches),
-            len(batch_vids),
-        )
-        odps_rows = odps.fetch_topic_decode_results(
-            decode_dt, batch_vids, batch_size=len(batch_vids)
-        )
-        titles_by_vid: dict[str, str] = {}
-        for row in odps_rows:
-            raw_vid = row.get("vid")
-            if raw_vid is None:
-                continue
-            vid = str(raw_vid).strip()
-            if not vid or vid in titles_by_vid:
-                continue
-            payload = _parse_decode_result(row.get("decode_result"))
-            if payload is None:
-                total_skipped += 1
-                continue
-            title = _extract_title(payload)
-            if title is None:
-                total_skipped += 1
-                continue
-            titles_by_vid[vid] = title
-
-        with get_session() as session:
-            updated = MultiDemandVideoDetailRepository(session).update_titles(
-                titles_by_vid
-            )
-        total_updated += updated
-
-        odps_vids = {
-            str(r.get("vid")).strip()
-            for r in odps_rows
-            if r.get("vid") is not None and str(r.get("vid")).strip()
-        }
-        total_missing += len(set(batch_vids) - odps_vids)
-        logger.info(
-            "Title backfill batch %d/%d done: updated=%d",
-            idx,
-            len(batches),
-            updated,
-        )
-
-    result = {
-        "decode_dt": decode_dt,
-        "pending": len(pending),
-        "batches": len(batches),
-        "updated": total_updated,
-        "missing_in_odps": total_missing,
-        "skipped_no_title": total_skipped,
-    }
-    logger.info("Backfill video titles completed: %s", result)
-    return result
-
-
-def backfill_video_points(
-    batch_size: int = VIDEO_SYNC_BATCH_SIZE,
-) -> dict[str, Any]:
-    """为灵感点/目的点/关键点字段缺失的已有行，从 ODPS 昨天分区回填。"""
-    decode_dt = (datetime.now() - timedelta(days=1)).strftime("%Y%m%d")
-    chunk = max(1, int(batch_size))
-    logger.info("Backfill video points: decode_dt=%s batch_size=%d", decode_dt, chunk)
-
-    with get_session() as session:
-        pending = MultiDemandVideoDetailRepository(session).list_vids_missing_points()
-
-    if not pending:
-        result = {
-            "decode_dt": decode_dt,
-            "pending": 0,
-            "batches": 0,
-            "updated": 0,
-            "missing_in_odps": 0,
-            "skipped_no_decode": 0,
-        }
-        logger.info("No vids missing points: %s", result)
-        return result
-
-    odps = get_odps_client()
-    batches = [pending[i : i + chunk] for i in range(0, len(pending), chunk)]
-    total_updated = 0
-    total_missing = 0
-    total_skipped = 0
-
-    for idx, batch_vids in enumerate(batches, start=1):
-        logger.info(
-            "Points backfill batch %d/%d: %d vids",
-            idx,
-            len(batches),
-            len(batch_vids),
-        )
-        odps_rows = odps.fetch_topic_decode_results(
-            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:
-                continue
-            vid = str(raw_vid).strip()
-            if not vid or vid in points_by_vid:
-                continue
-            payload = _parse_decode_result(row.get("decode_result"))
-            if payload is None:
-                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:
-            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 = {
-            str(r.get("vid")).strip()
-            for r in odps_rows
-            if r.get("vid") is not None and str(r.get("vid")).strip()
-        }
-        total_missing += len(set(batch_vids) - odps_vids)
-        logger.info(
-            "Points backfill batch %d/%d done: updated=%d",
-            idx,
-            len(batches),
-            updated,
-        )
-
-    result = {
-        "decode_dt": decode_dt,
-        "pending": len(pending),
-        "batches": len(batches),
-        "updated": total_updated,
-        "missing_in_odps": total_missing,
-        "skipped_no_decode": total_skipped,
-    }
-    logger.info("Backfill video points completed: %s", result)
-    return result

supply_infra/scheduler/jobs/discover_videos_from_demands.py

@@ -226,3 +226,36 @@ def discover_videos_from_demands(
 
     logger.info("discover_videos_from_demands finished: %s", result)
     return result
+
+
+if __name__ == "__main__":
+    import sys
+
+    from supply_infra.scheduler.cli_result import run_cli
+
+    def _read_flag_value(args: list[str], flag: str) -> int | None:
+        if flag not in args:
+            return None
+        idx = args.index(flag)
+        if idx + 1 >= len(args):
+            return None
+        return int(args[idx + 1])
+
+    _args = sys.argv[1:]
+    _biz_dt = _args[0] if _args and not _args[0].startswith("-") else None
+    _workers_arg = None
+    if _biz_dt and len(_args) > 1 and not _args[1].startswith("-"):
+        _workers_arg = _args[1]
+
+    run_cli(
+        lambda: discover_videos_from_demands(
+            _biz_dt,
+            workers=int(_workers_arg) if _workers_arg else 1,
+            offset=_read_flag_value(_args, "--offset") or 0,
+            limit=_read_flag_value(_args, "--limit"),
+            top_limit=_read_flag_value(_args, "--top-limit"),
+            skip_finished="--force" not in _args,
+            force="--force" in _args,
+        ),
+        label="discover_videos_from_demands",
+    )

supply_infra/scheduler/jobs/expand_demand_from_video_points.py

@@ -333,3 +333,25 @@ def expand_demand_from_video_points(
 
     logger.info("expand_demand_from_video_points finished: %s", result)
     return result
+
+
+if __name__ == "__main__":
+    import sys
+
+    from supply_infra.scheduler.cli_result import run_cli
+
+    _args = sys.argv[1:]
+    _biz_dt = _args[0] if _args and not _args[0].startswith("-") else None
+    _workers_arg = None
+    if _biz_dt and len(_args) > 1 and not _args[1].startswith("-"):
+        _workers_arg = _args[1]
+    _skip_finished = "--force" not in _args
+
+    run_cli(
+        lambda: expand_demand_from_video_points(
+            _biz_dt,
+            skip_finished=_skip_finished,
+            workers=int(_workers_arg) if _workers_arg else 5,
+        ),
+        label="expand_demand_from_video_points",
+    )

supply_infra/scheduler/jobs/grade_demand_pool.py

@@ -387,3 +387,25 @@ def grade_demand_pool(
             "error": str(exc),
             "run_at": datetime.now().isoformat(),
         }
+
+
+if __name__ == "__main__":
+    import sys
+
+    from supply_infra.scheduler.cli_result import run_cli
+
+    _argv = [a for a in sys.argv[1:] if a != "--retry-failed"]
+    _retry_failed = "--retry-failed" in sys.argv[1:]
+    _biz_dt = _argv[0] if len(_argv) > 0 else None
+    _workers = int(_argv[1]) if len(_argv) > 1 else 5
+
+    def _main() -> dict[str, Any]:
+        if _retry_failed:
+            return retry_failed_plan_group_items(_biz_dt, workers=_workers)
+        return grade_demand_pool(
+            _biz_dt,
+            workers=_workers,
+            with_orchestrate=True,
+        )
+
+    run_cli(_main, label="grade_demand_pool")

supply_infra/scheduler/jobs/publish_videos_from_discovery.py

@@ -216,3 +216,40 @@ def publish_videos_from_discovery(
         "failed_batch_count": len(failed_batches),
         "batches": [item.to_dict() for item in batch_results],
     }
+
+
+if __name__ == "__main__":
+    import argparse
+
+    from supply_infra.scheduler.cli_result import run_cli
+
+    def _build_parser() -> argparse.ArgumentParser:
+        parser = argparse.ArgumentParser(
+            description="均匀分发 primary/backup 候选视频到 AIGC 发布计划"
+        )
+        parser.add_argument("biz_dt", nargs="?", help="业务日 YYYYMMDD，默认取最新")
+        parser.add_argument("--run-id", dest="run_id", help="仅处理指定 run_id")
+        parser.add_argument("--limit", type=int, help="最多处理候选视频数")
+        parser.add_argument(
+            "--dry-run",
+            action="store_true",
+            help="只演练分配与请求，不写库",
+        )
+        parser.add_argument(
+            "--force",
+            action="store_true",
+            help="包含已写过 aigc_crawler_plan_id 的候选",
+        )
+        return parser
+
+    _args = _build_parser().parse_args()
+    run_cli(
+        lambda: publish_videos_from_discovery(
+            biz_dt=_args.biz_dt,
+            run_id=_args.run_id,
+            skip_published=not _args.force,
+            limit=_args.limit,
+            dry_run=_args.dry_run,
+        ),
+        label="publish_videos_from_discovery",
+    )

supply_infra/scheduler/jobs/run_supply_pipeline.py

@@ -8,19 +8,22 @@
 5. find_agent 视频发现（top 200 需求，2 线程并行）
 6. AIGC 发布（find_agent 完成后发布全部符合条件的视频）
 
-各子步骤内部已做去重（INSERT IGNORE、diff 同步、跳过已分级词等）；
+各子步骤以独立 CLI 子进程执行（``python -m supply_infra.scheduler.jobs.*``）；
+内部已做去重（INSERT IGNORE、diff 同步、跳过已分级词等）。
 本文件额外用进程内锁防止同一轮次并发重入，并隔离各步骤异常：前一步失败时记录
 error 后继续后续步骤，最终返回失败结果而不向 APScheduler 抛异常。
 """
 from __future__ import annotations
 
 import logging
+import sys
 import threading
 from datetime import datetime, timedelta
-from typing import Any, Callable
+from typing import Any
 from zoneinfo import ZoneInfo
 
 from supply_infra.config import get_infra_settings
+from supply_infra.scheduler.cli_result import run_cli
 from supply_infra.scheduler.constants import (
     PIPELINE_FIND_AGENT_TOP_DEMANDS,
     PIPELINE_FIND_AGENT_WORKERS,
@@ -28,20 +31,7 @@ from supply_infra.scheduler.constants import (
     SUPPLY_PIPELINE_JOB_NAME,
 )
 from supply_infra.scheduler.job_execution import JobExecutionRecorder, record_skipped
-from supply_infra.scheduler.jobs.discover_videos_from_demands import (
-    discover_videos_from_demands,
-)
-from supply_infra.scheduler.jobs.expand_demand_from_video_points import (
-    expand_demand_from_video_points,
-)
-from supply_infra.scheduler.jobs.grade_demand_pool import grade_demand_pool
-from supply_infra.scheduler.jobs.publish_videos_from_discovery import (
-    publish_videos_from_discovery,
-)
-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.step_runner import run_step_module
 
 logger = logging.getLogger(__name__)
 
@@ -61,24 +51,6 @@ def _resolve_dates(biz_dt: str | None) -> tuple[str, str]:
     return resolved_biz_dt, tree_partition
 
 
-def _run_step(
-    step_name: str,
-    action: Callable[[], Any],
-) -> tuple[Any, bool, str | None]:
-    """执行单个流水线步骤；失败只转成结果，不允许异常越过定时任务边界。"""
-    try:
-        payload = action()
-    except Exception as exc:
-        logger.exception("Supply pipeline step failed: step=%s", step_name)
-        return {"success": False, "error": str(exc)}, False, str(exc)
-
-    if isinstance(payload, dict) and payload.get("success") is False:
-        error = str(payload.get("error") or f"{step_name} returned success=False")
-        logger.error("Supply pipeline step reported failure: step=%s error=%s", step_name, error)
-        return payload, False, error
-    return payload, True, None
-
-
 def _preflight_failure_result(biz_dt: str | None, exc: Exception) -> dict[str, Any]:
     """日期/配置解析失败时也记录结果并正常返回，避免调度线程出现未捕获异常。"""
     started_at = datetime.now()
@@ -104,7 +76,7 @@ def _preflight_failure_result(biz_dt: str | None, exc: Exception) -> dict[str, A
 
 def run_supply_pipeline(biz_dt: str | None = None) -> dict[str, Any]:
     """
-    按顺序执行全局树同步 → 需求池同步 → 需求分级 → 视频点位拓展
+    按顺序以子进程执行：全局树同步 → 需求池同步 → 需求分级 → 视频点位拓展
     → find_agent 找片 → AIGC 发布。
 
     Args:
@@ -159,45 +131,48 @@ def run_supply_pipeline(biz_dt: str | None = None) -> dict[str, Any]:
     success = False
 
     try:
-        steps: list[tuple[str, Callable[[], Any]]] = [
+        steps: list[tuple[str, str, list[str]]] = [
             (
                 "global_tree",
-                lambda: sync_global_tree_odps_to_mysql(partition_date=tree_partition),
+                "supply_infra.scheduler.jobs.sync_global_tree_odps_to_mysql",
+                [tree_partition],
             ),
             (
                 "demand_pool",
-                lambda: sync_multi_demand_pool_odps_to_mysql(
-                    partition_date=resolved_biz_dt,
-                ),
+                "supply_infra.scheduler.jobs.demand_pool",
+                [resolved_biz_dt],
             ),
             (
                 "grade",
-                lambda: grade_demand_pool(biz_dt=resolved_biz_dt),
+                "supply_infra.scheduler.jobs.grade_demand_pool",
+                [resolved_biz_dt],
             ),
             (
                 "expand_video_points",
-                lambda: expand_demand_from_video_points(
-                    biz_dt=resolved_biz_dt,
-                    workers=5,
-                ),
+                "supply_infra.scheduler.jobs.expand_demand_from_video_points",
+                [resolved_biz_dt, "5"],
             ),
             (
                 "discover_videos",
-                lambda: discover_videos_from_demands(
-                    biz_dt=resolved_biz_dt,
-                    workers=PIPELINE_FIND_AGENT_WORKERS,
-                    top_limit=PIPELINE_FIND_AGENT_TOP_DEMANDS,
-                ),
+                "supply_infra.scheduler.jobs.discover_videos_from_demands",
+                [
+                    resolved_biz_dt,
+                    str(PIPELINE_FIND_AGENT_WORKERS),
+                    "--top-limit",
+                    str(PIPELINE_FIND_AGENT_TOP_DEMANDS),
+                ],
             ),
             (
                 "publish_videos",
-                lambda: publish_videos_from_discovery(
-                    biz_dt=resolved_biz_dt,
-                ),
+                "supply_infra.scheduler.jobs.publish_videos_from_discovery",
+                [resolved_biz_dt],
             ),
         ]
-        for step_name, action in steps:
-            payload, step_success, step_error = _run_step(step_name, action)
+        for step_name, module, module_args in steps:
+            payload, step_success, step_error = run_step_module(
+                module,
+                module_args,
+            )
             result[step_name] = payload
             step_status[step_name] = {
                 "success": step_success,
@@ -209,7 +184,7 @@ def run_supply_pipeline(biz_dt: str | None = None) -> dict[str, Any]:
         success = all(item["success"] for item in step_status.values())
         result["success"] = success
     except Exception as exc:
-        # 保护流水线编排本身；正常子步骤异常应已由 _run_step 消化。
+        # 保护流水线编排本身；正常子步骤异常应已由 run_step_module 消化。
         result["success"] = False
         errors.append(f"pipeline: {exc}")
         logger.exception(
@@ -239,3 +214,10 @@ def run_supply_pipeline(biz_dt: str | None = None) -> dict[str, Any]:
         logger.info("Supply pipeline finished: %s", result)
 
     return result
+
+
+if __name__ == "__main__":
+    run_cli(
+        lambda: run_supply_pipeline(sys.argv[1] if len(sys.argv) > 1 else None),
+        label="run_supply_pipeline",
+    )

supply_infra/scheduler/jobs/sync_global_tree_odps_to_mysql.py

@@ -260,3 +260,16 @@ def sync_global_tree_odps_to_mysql(partition_date: str | None = None) -> dict:
     }
     logger.info("Global tree sync completed: %s", result)
     return result
+
+
+if __name__ == "__main__":
+    import sys
+
+    from supply_infra.scheduler.cli_result import run_cli
+
+    run_cli(
+        lambda: sync_global_tree_odps_to_mysql(
+            partition_date=sys.argv[1] if len(sys.argv) > 1 else None,
+        ),
+        label="sync_global_tree_odps_to_mysql",
+    )

supply_infra/scheduler/jobs/update_category_tree_rank_scores.py

@@ -1,133 +0,0 @@
-"""
-category_tree_weight 四维热度全局排名归一化打分。
-
-在 category_tree_weight 全部节点写入完成后执行：
-1. 对 ext_pop / plat_sust_pop / plat_ly_pop / recent_pop 各自独立做全局排名
-2. 将排名映射为 [1/n, 1] 区间的归一化分（count>0 的节点参与排名，其余为 null）
-3. 四维分相加写入 total_score（无有效维度时为 null）
-"""
-from __future__ import annotations
-
-import logging
-from decimal import Decimal
-from typing import Any
-
-from supply_agent.ranking import rank_to_scores
-from supply_infra.db.repositories.category_tree_weight_repo import (
-    CategoryTreeWeightRepository,
-)
-from supply_infra.db.session import get_session
-
-logger = logging.getLogger(__name__)
-
-POP_DIM_KEYS: tuple[str, ...] = (
-    "ext_pop",
-    "plat_sust_pop",
-    "plat_ly_pop",
-    "recent_pop",
-)
-
-_UPDATE_BATCH = 500
-
-
-def _dec(value: float, places: int = 8) -> Decimal:
-    return Decimal(str(round(float(value), places)))
-
-
-def _optional_dec(value: float | None, places: int = 8) -> Decimal | None:
-    if value is None:
-        return None
-    return _dec(value, places)
-
-
-def _materialize_weight_row(row: Any) -> dict[str, Any]:
-    """在 session 内抽出标量，避免 DetachedInstanceError。"""
-    payload: dict[str, Any] = {
-        "category_id": int(row.category_id),
-        "biz_dt": str(row.biz_dt),
-    }
-    for dim in POP_DIM_KEYS:
-        count = int(getattr(row, f"{dim}_count", 0) or 0)
-        payload[f"{dim}_count"] = count
-        if count > 0:
-            avg_raw = getattr(row, f"{dim}_avg", None)
-            payload[f"{dim}_avg"] = float(avg_raw) if avg_raw is not None else None
-        else:
-            payload[f"{dim}_avg"] = None
-    return payload
-
-
-def _build_rank_score_rows(
-    rows: list[dict[str, Any]],
-) -> list[dict[str, Any]]:
-    """根据已写入的权重行计算各节点排名分。"""
-    dim_scores: dict[str, dict[int, float]] = {}
-    for dim in POP_DIM_KEYS:
-        candidates: list[tuple[int, float]] = []
-        for row in rows:
-            count = int(row[f"{dim}_count"])
-            if count <= 0:
-                continue
-            avg_raw = row[f"{dim}_avg"]
-            if avg_raw is None:
-                continue
-            avg = float(avg_raw)
-            candidates.append((int(row["category_id"]), avg))
-        dim_scores[dim] = rank_to_scores(candidates)
-
-    updates: list[dict[str, Any]] = []
-    for row in rows:
-        category_id = int(row["category_id"])
-        biz_dt = str(row["biz_dt"])
-        score_values = {
-            f"{dim}_score": dim_scores[dim].get(category_id)
-            for dim in POP_DIM_KEYS
-        }
-        non_null_scores = [v for v in score_values.values() if v is not None]
-        total = sum(non_null_scores) if non_null_scores else None
-        updates.append(
-            {
-                "category_id": category_id,
-                "biz_dt": biz_dt,
-                **{
-                    col: _optional_dec(score_values[col])
-                    for col in (
-                        "ext_pop_score",
-                        "plat_sust_pop_score",
-                        "plat_ly_pop_score",
-                        "recent_pop_score",
-                    )
-                },
-                "total_score": _optional_dec(total),
-            }
-        )
-    return updates
-
-
-def update_category_tree_rank_scores(biz_dt: str) -> dict[str, Any]:
-    """在 category_tree_weight 全部写入后，按 biz_dt 更新四维排名分与 total_score。"""
-    with get_session() as session:
-        repo = CategoryTreeWeightRepository(session)
-        rows = [
-            _materialize_weight_row(row)
-            for row in repo.list_by_biz_dt(biz_dt)
-        ]
-
-    if not rows:
-        logger.info("Category tree rank scores: no rows, skip biz_dt=%s", biz_dt)
-        return {"biz_dt": biz_dt, "nodes": 0, "updated": 0}
-
-    updates = _build_rank_score_rows(rows)
-    updated = 0
-    with get_session() as session:
-        repo = CategoryTreeWeightRepository(session)
-        for i in range(0, len(updates), _UPDATE_BATCH):
-            updated += repo.update_rank_scores(updates[i : i + _UPDATE_BATCH])
-
-    result = {
-        "biz_dt": biz_dt,
-        "nodes": len(updates),
-        "updated": updated,
-    }
-    logger.info("Category tree rank scores completed: %s", result)
-    return result

supply_infra/scheduler/manual_jobs.py

@@ -78,7 +78,7 @@ def _run_demand_pool_job(
     biz_dt: str | None,
     partition_date: str | None,
 ) -> dict[str, Any]:
-    from supply_infra.scheduler.jobs.sync_multi_demand_pool_odps_to_mysql import (
+    from supply_infra.scheduler.jobs.demand_pool import (
         sync_multi_demand_pool_odps_to_mysql,
     )
 
@@ -100,19 +100,6 @@ def _run_biz_dt_job(
     return fn(biz_dt)
 
 
-def _run_no_arg_job(
-    import_path: str,
-    fn_name: str,
-    *,
-    biz_dt: str | None,
-    partition_date: str | None,
-) -> dict[str, Any]:
-    del biz_dt, partition_date
-    module = __import__(import_path, fromlist=[fn_name])
-    fn = getattr(module, fn_name)
-    return fn()
-
-
 MANUAL_JOBS: dict[str, ManualJobSpec] = {
     SUPPLY_PIPELINE_JOB_ID: ManualJobSpec(
         job_id=SUPPLY_PIPELINE_JOB_ID,
@@ -137,16 +124,6 @@ MANUAL_JOBS: dict[str, ManualJobSpec] = {
         accepts_biz_dt=True,
         accepts_partition_date=True,
     ),
-    "sync_demand_belong_pool_rel": ManualJobSpec(
-        job_id="sync_demand_belong_pool_rel",
-        name="需求词与需求池匹配",
-        description="补充 demand_belong_pool_rel 并回填 demand_belong_category.video_list",
-        runner=lambda **kwargs: _run_no_arg_job(
-            "supply_infra.scheduler.jobs.sync_demand_belong_pool_rel",
-            "sync_demand_belong_pool_rel",
-            **kwargs,
-        ),
-    ),
     "grade_demand_pool": ManualJobSpec(
         job_id="grade_demand_pool",
         name="需求池分级",
@@ -191,16 +168,6 @@ MANUAL_JOBS: dict[str, ManualJobSpec] = {
         ),
         accepts_biz_dt=True,
     ),
-    "sync_multi_demand_videos": ManualJobSpec(
-        job_id="sync_multi_demand_videos",
-        name="需求池视频详情同步",
-        description="从 TikHub 拉取 multi_demand_pool_di 关联视频标题与详情",
-        runner=lambda **kwargs: _run_no_arg_job(
-            "supply_infra.scheduler.jobs.sync_multi_demand_videos",
-            "sync_multi_demand_videos",
-            **kwargs,
-        ),
-    ),
 }
 
 

supply_infra/scheduler/step_runner.py

@@ -0,0 +1,122 @@
+"""Run pipeline step modules as isolated subprocesses via ``python -m``."""
+from __future__ import annotations
+
+import json
+import logging
+import subprocess
+import sys
+from pathlib import Path
+from typing import Any
+
+logger = logging.getLogger(__name__)
+
+_REPO_ROOT = Path(__file__).resolve().parents[2]
+
+
+def run_step_module(
+    module: str,
+    args: list[str] | None = None,
+    *,
+    timeout: float | None = None,
+) -> tuple[Any, bool, str | None]:
+    """
+    Execute ``python -m <module>`` with the current interpreter.
+
+    Returns:
+        (payload, success, error_message)
+    """
+    cmd = [sys.executable, "-m", module, *(args or [])]
+    logger.info("Pipeline step subprocess start: cmd=%s", cmd)
+    try:
+        completed = subprocess.run(
+            cmd,
+            cwd=str(_REPO_ROOT),
+            capture_output=True,
+            text=True,
+            timeout=timeout,
+            check=False,
+        )
+    except subprocess.TimeoutExpired as exc:
+        error = f"step timed out after {timeout}s: {module}"
+        logger.exception(error)
+        stderr_tail = (exc.stderr or "")[-2000:] if isinstance(exc.stderr, str) else ""
+        return (
+            {
+                "success": False,
+                "error": error,
+                "stderr": stderr_tail,
+            },
+            False,
+            error,
+        )
+    except Exception as exc:
+        error = f"failed to launch step module {module}: {exc}"
+        logger.exception(error)
+        return {"success": False, "error": error}, False, error
+
+    stdout = (completed.stdout or "").strip()
+    stderr = (completed.stderr or "").strip()
+    if stderr:
+        logger.info(
+            "Pipeline step stderr: module=%s exit=%s stderr_tail=%s",
+            module,
+            completed.returncode,
+            stderr[-2000:],
+        )
+
+    payload: Any
+    parse_error: str | None = None
+    if not stdout:
+        payload = {
+            "success": False,
+            "error": "empty stdout from step module",
+            "exit_code": completed.returncode,
+            "stderr": stderr[-2000:] if stderr else None,
+        }
+        parse_error = "empty stdout from step module"
+    else:
+        # Prefer the last non-empty line in case logging leaked to stdout.
+        last_line = stdout.splitlines()[-1]
+        try:
+            payload = json.loads(last_line)
+        except json.JSONDecodeError:
+            try:
+                payload = json.loads(stdout)
+            except json.JSONDecodeError as exc:
+                parse_error = f"invalid JSON from step module: {exc}"
+                payload = {
+                    "success": False,
+                    "error": parse_error,
+                    "exit_code": completed.returncode,
+                    "stdout_tail": stdout[-2000:],
+                    "stderr": stderr[-2000:] if stderr else None,
+                }
+
+    success = completed.returncode == 0
+    if isinstance(payload, dict) and payload.get("success") is False:
+        success = False
+
+    error: str | None = None
+    if not success:
+        if isinstance(payload, dict):
+            error = str(
+                payload.get("error")
+                or parse_error
+                or f"{module} failed with exit_code={completed.returncode}"
+            )
+        else:
+            error = parse_error or f"{module} failed with exit_code={completed.returncode}"
+        logger.error(
+            "Pipeline step failed: module=%s exit=%s error=%s",
+            module,
+            completed.returncode,
+            error,
+        )
+    else:
+        logger.info(
+            "Pipeline step ok: module=%s exit=%s",
+            module,
+            completed.returncode,
+        )
+
+    return payload, success, error

zhangbo.md

@@ -68,7 +68,6 @@ SupplyAgent/
 ├── supply_infra/       MySQL、ODPS、OSS、定时任务
 ├── api/                FastAPI 查询接口和静态前端托管
 ├── web/                Vue 3 业务前端
-├── jobs/               数据任务的手动 CLI 入口
 ├── scripts/            部署、日志上传、日志可视化脚本
 ├── visualization/      早期/独立的业务设计可视化材料
 ├── Dockerfile          前端构建 + Python 运行镜像
@@ -76,6 +75,8 @@ SupplyAgent/
 └── requirements.txt    另一份运行依赖清单
 ```
 
+任务 CLI 已并入 `supply_infra/scheduler/jobs/`（`python -m ...`）。
+
 模块之间的依赖方向：
 
 ```text
@@ -479,7 +480,7 @@ multi_demand_pool_di.id
 
 ### 8.5 四维排名分和全局热度
 
-`update_category_tree_rank_scores.py` 只对四项先验排名：
+`demand_pool/tree_weight.py` 在写入权重后会对四项先验排名：
 
 - 各维只让 `count > 0` 的节点参与；
 - 按 avg 降序排名；
@@ -498,14 +499,12 @@ multi_demand_pool_di.id
 
 ## 9. 定时任务和数据流水线
 
-### 9.1 已注册的两个定时任务
-
-`supply_infra/scheduler/app.py` 当前只注册：
+### 9.1 已注册的定时任务
 
-1. 每天 `02:30`：同步昨天的全局树和元素；
-2. 每天 `12:00`：同步当天的策略需求池并执行后续完整流水线。
+`supply_infra/scheduler/app.py` 当前只注册一条任务：
 
-时区来自 `SCHEDULER_TIMEZONE`。
+1. 每天 `14:30`（`SCHEDULER_TIMEZONE`）：串行执行全链路  
+   `run_supply_pipeline`（全局树 → 需求池 → 分级 → 拓展 → 找片 → AIGC 发布）。
 
 ### 9.2 全局树同步
 
@@ -523,7 +522,7 @@ multi_demand_pool_di.id
 
 ### 9.3 策略需求池完整流水线
 
-`sync_multi_demand_pool_odps_to_mysql` 实际执行顺序：
+`supply_infra.scheduler.jobs.demand_pool`（`sync_multi_demand_pool_odps_to_mysql`）实际执行顺序：
 
 1. 比较 ODPS 与 MySQL 当日去重行数；
 2. 行数不同时执行差异同步；
@@ -540,16 +539,15 @@ multi_demand_pool_di.id
 
 ### 9.4 手动任务入口
 
-`jobs/` 提供：
+用 `python -m` 调用（与定时流水线同源）：
+
+- `python -m supply_infra.db`：初始化数据库；
+- `python -m supply_infra.scheduler.jobs.run_supply_pipeline`：全链路；
+- `python -m supply_infra.scheduler.jobs.demand_pool`：需求池同步（含内部阶段）；
+- `python -m supply_infra.scheduler.jobs.grade_demand_pool`：分级（可加 `--retry-failed`）；
+- 以及 global_tree / expand / discover / publish 各步骤模块。
 
-- 初始化数据库；
-- 启动 Scheduler；
-- 单独同步需求池视频；
-- 回填视频列表和标题；
-- 单独计算词级热度；
-- 单独计算树节点权重；
-- 单独计算四维排名；
-- 单独同步需求词与需求池关系。
+也可通过 API `POST /api/scheduler/jobs/{job_id}/run` 手动触发。
 
 ---
 
@@ -655,7 +653,6 @@ API 虽然返回真实 ROV/VOV，但当前冰柱图标签没有提供真实 ROV/
 `pyproject.toml` 注册：
 
 - `supply-api`；
-- `supply-scheduler`；
 - `supply-visualize`。
 
 ### 12.2 本地开发
@@ -810,9 +807,9 @@ Docker 使用两阶段构建：
 
 如果继续开发，建议按以下顺序进入代码：
 
-1. `supply_infra/scheduler/jobs/sync_multi_demand_pool_odps_to_mysql.py`：理解主业务流水线；
+1. `supply_infra/scheduler/jobs/demand_pool/`：理解需求池同步主业务；
 2. `supply_infra/db/models/`：理解真实数据对象；
-3. `supply_infra/scheduler/jobs/compute_category_tree_weight.py`：理解树上热度；
+3. `supply_infra/scheduler/jobs/demand_pool/tree_weight.py`：理解树上热度；
 4. `agents/demand_belong_category_agent/`：理解需求词挂树；
 5. `agents/generate_demand_agent/`：理解平台需求生成；
 6. `api/services/category_tree.py`：理解后端对前端的数据形态；