docs: 更新 AGENTS.md（批量处理/断点存档/环境变量）

- 右眼 MarginR 误用 width-50 改为 50（左右眼对称内边距）
- 字幕颜色改为半透明白（&H80FFFFFF），避免遮挡 VR 画面

AGENTS.md

@@ -20,8 +20,8 @@ ASS）在**同一个进程**内运行，不再启动子进程、不再走节点
 vrsub/
 ├── src/wov_sdk/       # 协议数据模型（NodeManifest/InvokeRequest/InvokeResponse/
 │                      #   WorkflowDefinition 等），与分布式版保持一致
-├── src/wov_app/       # 应用层：main/config/db/registry/scheduler/seed/routers
+├── src/wov_app/       # 应用层：main/config/db/registry/scheduler/batch/seed/routers
-│   └── routers/       #   apps.py（用户端）、workflows.py（管理端）
+│   └── routers/       #   apps.py（用户端）、workflows.py（管理端）、batch.py（批量处理）
 ├── nodes/             # 进程内节点实现：echo/ffmpeg/whisper/llm/ass
 ├── manifests/         # 各节点清单 JSON（echo.json/ffmpeg.json/...）
 ├── workflows/         # 默认工作流定义 JSON（模型/链路均为数据，改模型不改代码）
@@ -40,7 +40,7 @@ vrsub/
   任务，按工作流 DAG 拓扑顺序调用节点，产物按
   `data/storage/runs/<run_id>/steps/<node_id>/` 落盘并登记到 artifacts 表。
 - **前端**：由 FastAPI 静态挂载 `web/`，节点注册/实例管理页面已移除，
-  仅保留应用中心、任务管理、管理后台（工作流）与工作流编排。
+  仅保留应用中心、任务管理、批量处理、管理后台（工作流）与工作流编排。
 - **工作流编排页（web/workflow.html）**：支持新建工作流（空表单预填演示模板），
   从列表"编辑"加载任一工作流的最新定义（ID 锁定，保存即追加新版本）；"版本"
   查看全部历史版本并可"加载到编辑器"（对比/回滚后另存新版本）；管理后台
@@ -58,9 +58,9 @@ vrsub/
 | `faster-whisper` | `audio_uri`（16kHz 单声道） | `srt_uri` | 参数：`language`、`task`、`model_path`、`device`、`compute_type`、`beam_size`、`vad_filter`（默认开）、`condition_on_previous_text`、`chunk_seconds` |
 | `llm-translate` | `srt_uri` | `cn_srt_uri` | 参数：`target_language`、`model` |
 | `vlm-ocr` | `image_uri` | `text`、`text_uri` | 直接调本地 Ollama 多模态模型（glm-ocr）的 `/api/chat` 做视频帧 OCR（流式 + 5s 上限），参数：`model`、`ollama_host`、`prompt`、`timeout_seconds`、`keep_alive`、`num_predict`、`temperature`、`repeat_penalty` |
-| `frame-extract` | `video_uri` | `frames_manifest`、`frame_count` | 按**帧间隔**抽帧（解析 fps → step=round(间隔秒×fps)，ffmpeg select 按帧号精确取帧，帧时间=帧号/fps 无累计偏差）并 crop 裁切字幕区域，参数：`interval_seconds`（默认 0.5）、`crop`（[x,y,w,h] 0~1）。**帧文件必须按帧号数值排序读取**（`_sorted_frame_files`）：ffmpeg `%04d` 编号超过 9999 帧后扩为 5 位，字典序 `sorted()` 会把 5 位编号排在 4 位之前导致时间与图像错位（真实发生于 run_339ec7ee437f 的 14236 帧任务，回归测试见 `test_frame_files_read_order_matches_frame_number`） |
+| `frame-extract` | `video_uri` | `frames_manifest`、`frame_count` | 按**帧间隔**抽帧（解析 fps → step=round(间隔秒×fps)，ffmpeg select 按帧号精确取帧，帧时间=帧号/fps 无累计偏差）并 crop 裁切字幕区域，参数：`interval_seconds`（默认 0.5）、`crop`（[x,y,w,h] 0~1，**默认画面底部 1/4** `[0,0.75,1,0.25]`——字幕很少出现在画面上半部分，2026-08 调整）。**帧文件必须按帧号数值排序读取**（`_sorted_frame_files`）：ffmpeg `%04d` 编号超过 9999 帧后扩为 5 位，字典序 `sorted()` 会把 5 位编号排在 4 位之前导致时间与图像错位（真实发生于 run_339ec7ee437f 的 14236 帧任务，回归测试见 `test_frame_files_read_order_matches_frame_number`） |
 | `subtitle-ocr` | `frames_manifest` | `srt_uri`、`count` | 自适应线程池并发逐帧调 vlm-ocr → 垃圾过滤（无文字帧）→ 相同字幕合并（记录最后可见帧）→ 组装 SRT，消失时间=最后可见帧+采样间隔（间隔从帧清单推导），参数：`min_chars`、`min_alnum_ratio`、`garbage_tokens`、`pool_min_workers`/`pool_max_workers`/`pool_window_seconds`/`pool_fast_threshold`/`pool_slow_threshold` |
-| `llm-filter` | `srt_uri` | `srt_uri`、`kept`、`removed` | 两级过滤：①**规则层**（不调 LLM）正则确定性删除——横线装饰、URL/邮箱/**裸网址域名**（含中文夹杂的注册地址）、**HTML/水印模式**（html code/标签/javascript 等）、overlay token（html/marketing 等）、单双 ASCII 字符；②**LLM 五类分类**（garbage/overlay/noise 删，repeat/dialogue 留，未识别回退保留）每条连同前后各 `context_size`（默认 10）条纯文本分批判断——**上下文净化**：喂给 LLM 的是**过滤后的字幕**，规则层确定性垃圾从上下文中剔除（原文不进 LLM），避免覆盖层垃圾污染场景判断误删真实对话（回归：run_011d01f19999 曾 190 条含 ≥4 汉字对话被误删）；**长文本保护**：≥`min_keep_len`（默认 12）时 noise 不构成删除依据——LLM 判定不稳定，长度是必要兜底（实测移除保护后新增误删 124 条真实长对话）。**限流自适应**：LLM 调用 429/5xx 指数退避重试（最多 3 次，1s/2s/4s），worker 捕获限流错误时调用线程池 `report_failure()` **内存中临时降低最大线程数并缩容**（连续无错误窗口后逐步回升），失败条目在收紧后的并发下**重试一轮**，二次仍失败才整体失败——20 并发一拥而上触发 429 时自动收敛到配额内而不打挂任务。**按文本去重**（忽略空白/大小写，相同文本只调一次 LLM，上下文取首次出现）保证判定一致并省调用。参数：`context_size`、`min_keep_len`、`overlay_tokens`（JSON 数组）、`dedupe`（默认开）、`model`、`pool_min_workers`/`pool_max_workers`/`pool_window_seconds`/`pool_fast_threshold`/`pool_slow_threshold`。回归数据：testdata/ocr_srt_run_ac7f480a3ccb.srt（真实任务 1666 条 OCR 输出） |
+| `llm-filter` | `srt_uri` | `srt_uri`、`kept`、`removed` | 两级过滤：①**规则层**（不调 LLM）正则确定性删除——横线装饰、URL/邮箱/**裸网址域名**（含中文夹杂的注册地址）、**HTML/水印模式**（html code/标签/javascript 等）、overlay token（html/marketing 等）、单双 ASCII 字符；②**LLM 五类分类**（garbage/overlay/noise 删，repeat/dialogue 留，未识别回退保留）每条连同前后各 `context_size`（默认 10）条纯文本分批判断——**上下文净化**：喂给 LLM 的是**过滤后的字幕**，规则层确定性垃圾从上下文中剔除（原文不进 LLM），避免覆盖层垃圾污染场景判断误删真实对话（回归：run_011d01f19999 曾 190 条含 ≥4 汉字对话被误删）；**长文本保护**：≥`min_keep_len`（默认 12）时 noise 不构成删除依据——LLM 判定不稳定，长度是必要兜底（实测移除保护后新增误删 124 条真实长对话）。**限流自适应**：LLM 调用 429/5xx 指数退避重试（最多 3 次，1s/2s/4s），worker 捕获限流错误时调用线程池 `report_failure()` **内存中临时降低最大线程数并缩容**（连续无错误窗口后逐步回升），失败条目在收紧后的并发下**重试一轮**，二次仍失败才整体失败——20 并发一拥而上触发 429 时自动收敛到配额内而不打挂任务。**节点级断点存档**（2026-08）：每条判定成功立即追加 `filter_partial.jsonl`（`{"index","category"}`，多线程加锁串行化），失败/中断后重跑只重判未判定条目，已判定结果复用（与 OCR 存档同机制）。**按文本去重**（忽略空白/大小写，相同文本只调一次 LLM，上下文取首次出现）保证判定一致并省调用。参数：`context_size`、`min_keep_len`、`overlay_tokens`（JSON 数组）、`dedupe`（默认开）、`model`、`pool_min_workers`/`pool_max_workers`/`pool_window_seconds`/`pool_fast_threshold`/`pool_slow_threshold`。回归数据：testdata/ocr_srt_run_ac7f480a3ccb.srt（真实任务 1666 条 OCR 输出） |
 | `srt-to-dual-eye-ass` | `cn_srt_uri` | `ass_uri` | 参数：`resolution`，如 `3840x1920` |
 
 ### 模型权重解析（本地优先）
@@ -190,6 +190,8 @@ subtitle-ocr 逐帧调 vlm-ocr 时使用 `nodes/adaptive_pool.py` 的自适应
 | `WOV_CLEANUP_ENABLED` | `1` | 开启孤儿数据定时清理 |
 | `WOV_CLEANUP_INTERVAL_SECONDS` | `3600` | 孤儿清理扫描周期（秒） |
 | `WOV_CLEANUP_GRACE_SECONDS` | `3600` | 孤儿清理宽限期（秒） |
+| `WOV_BATCH_ENABLED` | `1` | 开启文件夹批量处理引擎（处理 source=batch 任务） |
+| `WOV_BATCH_INTERVAL_SECONDS` | `1.0` | 批量引擎轮询间隔 |
 | `WHISPER_MODEL_PATH` | 见上 | 显式指定 whisper 模型路径 |
 | `WHISPER_DEVICE` | `auto` | 转写设备 |
 | `LLM_API_BASE` | `https://api.siliconflow.cn/v1/chat/completions` | LLM 兼容接口 |
@@ -274,6 +276,53 @@ http://127.0.0.1:8000/docs      API 文档
     （线程池 `on_progress` 回调，每任务完成触发）；
   - whisper：分块转写打印"分块 X/Y 完成 offset=... 耗时 Zs (Nx 实时, 累计 ...s)"；
   - frame-extract：ffmpeg `-progress` 输出解析 `frame=N`，打印"抽帧进度 X/Y 帧 (Z 帧/s)"。
+## 文件夹批量处理（2026-08）
+
+本地版核心能力：**不把视频上传到工作目录**，直接读取用户所选文件夹下的全部
+视频，逐个执行所选流水线。入口为批量处理页（`web/batch.html`，导航"批量处理"），
+后端为 `src/wov_app/batch.py` 的 `BatchWorker`（单线程轮询线程，处理
+`source=batch` 的运行，与主调度器互不抢占）与 `routers/batch.py`。
+- **路径选择（2026-08 起不用手敲路径）**：批量页点击"选择文件夹…"按钮弹出
+  目录树选择器（懒加载），选完回填只读路径框。浏览器拿不到所选文件夹的绝对
+  路径，因此由**本地后端**提供目录浏览：`GET /api/batch/roots`（Windows 盘符 /
+  POSIX 根 + 家目录）、`GET /api/batch/dirs?path=`（列直接子目录，隐藏目录
+  过滤；不存在/不可读返回空列表不报 500）。只暴露目录名，不返回文件内容。
+- **数据落盘**：每个视频的中间态（`runs/<run_id>/steps/...`）与最终产物都放在
+  **视频所在目录的同名文件夹**（`movie.mp4` → `movie/`，去掉扩展名，
+  `work_dir_for` 推导）；最终产物在任务完成后从节点产物目录**复制**到同名
+  文件夹根目录，并写 `batch.done.json` 完成标记（记录 workflow_id 与产物文件名）。
+- **已处理过的不再处理**：同名文件夹已有**同工作流**完成标记且产物文件齐全 →
+  直接 SKIPPED；不同工作流的标记不互相误判（换流水线会重新处理）。
+- **任务参数**：`POST /api/batch/jobs {folder, workflow_id, recursive}` 创建批量
+  任务（校验文件夹/已发布工作流/有版本/至少一个视频，失败 422）；任务入
+  `batch_jobs` 表，每个视频一行 `batch_videos`（PENDING/RUNNING/PAUSED/
+  COMPLETED/FAILED/SKIPPED）。可用任意已发布流水线（demo / zh-direct /
+  ocr-subtitle 等），前端下拉选择。
+- **执行复用**：每个视频创建一个 `source=batch` 的 run（`input_uri` 直接指向
+  本地视频路径），用 per-video 的 `WorkflowScheduler` 实例（storage=同名文件夹）
+  执行——完整复用 DAG 拓扑执行、产物表登记与**断点续跑**逻辑。
+- **暂停/继续**：`POST /api/batch/jobs/{id}/pause` 把任务置 PAUSED 并暂停当前
+  run（写 `paused.flag`；whisper **分块间**检查、OCR 逐帧检查后中止，当前节点
+  执行完才停）；`resume` 恢复 QUEUED，引擎从断点继续——PAUSED 视频的 run 显式
+  resume 后从产物表续跑，未开始的视频接着处理。重启进程后 RUNNING 残留 run 由
+  `recover_interrupted_runs` 恢复，暂停的继续处理。
+- **失败容错**：单个视频失败（节点失败/文件缺失）记为 FAILED，批量任务继续
+  处理后续视频，结束后统计 done/failed；DAG 解析/任务级异常把任务置 FAILED。
+  **重跑保留产物**（2026-08，修复 run_e2b74e89e232 实测）：FAILED 视频重新处理
+  时不再 reset_run 清空产物记录，而是保留已完成节点的 artifacts 恢复 QUEUED，
+  execute_run 从产物表跳过已完成节点、只重跑失败节点——extract/ocr 等长耗时
+  成果不浪费；配合 llm-filter/OCR 的节点级断点存档，失败节点自身也只重判未完成
+  条目。前端对"部分失败"（COMPLETED 且 failed>0）用红色徽章醒目标示。
+- **产物下载**：`GET /api/batch/jobs/{id}/videos/{vid}/download?alias=result`
+  从完成标记解析产物文件并返回（只读同名文件夹根目录）。
+- **孤儿清理保护**：`source=batch` 的运行**跳过**自动清理——产物不在主存储
+  目录下，普通孤儿逻辑会误删记录并连带删除 `input_uri` 的父目录（用户的整个
+  视频文件夹）。
+- **删除任务**：`DELETE /api/batch/jobs/{id}` 只清理数据库记录（含关联 run），
+  磁盘上的同名文件夹与产物属于用户数据，保留不删。
+- **环境变量**：`WOV_BATCH_ENABLED`（默认 1）、`WOV_BATCH_INTERVAL_SECONDS`
+  （默认 1.0）。
+
 
 ## 测试与覆盖率
 

nodes/ass.py

@@ -16,7 +16,7 @@ def _ass_header(resolution: str) -> str:
     width, height = resolution.lower().split("x", 1)
     # 左眼样式占左半边，右眼样式占右半边，各留 50px 内边距。
     left_margin = 50
-    right_margin = int(width) - 50
+    right_margin = 50
     return f"""[Script Info]
 Title: VR Dual-Eye Subtitle
 ScriptType: v4.00+
@@ -28,8 +28,8 @@ ScaledBorderAndShadow: yes
 
 [V4+ Styles]
 Format: Name,Fontname,Fontsize,PrimaryColour,SecondaryColour,OutlineColour,BackColour,Bold,Italic,Underline,StrikeOut,ScaleX,ScaleY,Spacing,Angle,BorderStyle,Outline,Shadow,Alignment,MarginL,MarginR,MarginV,Encoding
-Style: LeftEye,Arial,50,&H00FFFFFF,&H000000FF,&H00000000,&H80000000,0,0,0,0,50,100,0,0,1,4,0,2,{left_margin},{int(width) // 2},{int(height) // 2 + 60},1
+Style: LeftEye,Arial,50,&H80FFFFFF,&H000000FF,&H00000000,&H80000000,0,0,0,0,50,100,0,0,1,4,0,2,{left_margin},{int(width) // 2},{int(height) // 2 + 60},1
-Style: RightEye,Arial,50,&H00FFFFFF,&H000000FF,&H00000000,&H80000000,0,0,0,0,50,100,0,0,1,4,0,2,{int(width) // 2},{right_margin},{int(height) // 2 + 60},1
+Style: RightEye,Arial,50,&H80FFFFFF,&H000000FF,&H00000000,&H80000000,0,0,0,0,50,100,0,0,1,4,0,2,{int(width) // 2},{right_margin},{int(height) // 2 + 60},1
 
 [Events]
 Format: Layer,Start,End,Style,Name,MarginL,MarginR,MarginV,Effect,Text

nodes/frame_extract.py

@@ -21,8 +21,8 @@ from wov_sdk.models import InvokeRequest, InvokeResponse
 
 logger = get_logger("frame-extract")
 
-# 默认裁切：画面底部 18% 区域（常见字幕位置）。
+# 默认裁切：画面底部 1/4 区域（字幕区，字幕很少出现在画面上半部分）。
-DEFAULT_CROP = [0.0, 0.82, 1.0, 0.18]
+DEFAULT_CROP = [0.0, 0.75, 1.0, 0.25]
 
 
 def _parse_crop(raw) -> list[float] | None:

nodes/llm_filter.py

@@ -33,6 +33,7 @@ from __future__ import annotations
 import json
 import os
 import re
+import threading
 import time
 import urllib.error
 import urllib.request
@@ -71,6 +72,39 @@ _ALL_CATEGORIES = (
 )
 DELETE_CATEGORIES = {CATEGORY_GARBAGE, CATEGORY_OVERLAY, CATEGORY_NOISE}
 
+# 判定存档文件名：位于节点 output_dir，每行 {"index": 条目标引, "category": 类别}。
+# 每条 LLM 判定成功即追加一行；进程被杀/节点失败（如 429 限流）后重跑时，
+# 只对未判定的条目重新调用 LLM，已判定结果直接复用（类似 OCR 的断点存档）。
+_PARTIAL_NAME = "filter_partial.jsonl"
+
+# 判定存档追加写锁：多线程判定并发完成时串行化追加，避免行交错。
+_partial_lock = threading.Lock()
+
+
+def _load_partial(output_dir: Path) -> dict[int, str]:
+    """读取判定存档，返回 {条目标引: 类别}；无存档/损坏行跳过。"""
+    path = output_dir / _PARTIAL_NAME
+    if not path.is_file():
+        return {}
+    result: dict[int, str] = {}
+    for line in path.read_text(encoding="utf-8").splitlines():
+        if not line.strip():
+            continue
+        try:
+            item = json.loads(line)
+        except json.JSONDecodeError:
+            # 进程被杀时可能残留半行写入：跳过，对应条目视为未判定。
+            continue
+        result[int(item["index"])] = str(item["category"])
+    return result
+
+
+def _append_partial(output_dir: Path, index: int, category: str) -> None:
+    """线程安全地把一条判定结果追加到存档（成功判定后立即落盘）。"""
+    with _partial_lock:
+        with (output_dir / _PARTIAL_NAME).open("a", encoding="utf-8") as fh:
+            fh.write(json.dumps({"index": index, "category": category}, ensure_ascii=False) + "\n")
+
 # 规则层正则：横线装饰（含全角/半角横线、下划线、中点、句点等符号组合）。
 _DASH_RE = re.compile(r"^[\s\-—_~=•・。．.、]+$")
 # 规则层正则：URL / 邮箱。
@@ -295,7 +329,11 @@ def invoke(request: InvokeRequest) -> InvokeResponse:
     llm_needed = [i for i, verdict in enumerate(rule_verdicts) if verdict is None]
 
     # 阶段 2：LLM 分类层（去重：相同文本只判一次，上下文取首次出现）。
-    cat_by_index: dict[int, str] = {}
+    # 节点级断点存档：output_dir 提前建好，重跑时只重判未判定条目。
+    output_dir = Path(request.output_dir)
+    output_dir.mkdir(parents=True, exist_ok=True)
+    partial = _load_partial(output_dir)
+    cat_by_index: dict[int, str] = dict(partial)
     if llm_needed:
         if dedupe:
             first_of_key: dict[str, int] = {}
@@ -304,20 +342,24 @@ def invoke(request: InvokeRequest) -> InvokeResponse:
                 key = _dedup_key(entries[i]["text"])
                 if key not in first_of_key:
                     first_of_key[key] = i
-                    pool_indices.append(i)
+                    # 断点续跑：该键首次出现已在存档判定过则跳过（结果复用）。
+                    if i not in partial:
+                        pool_indices.append(i)
         else:
-            pool_indices = llm_needed
+            # 断点续跑：只处理未判定的条目。
-
+            pool_indices = [i for i in llm_needed if i not in partial]
         # 单条判断的工作函数：返回类别词；overlay_tokens 用于上下文净化。
         def judge_one(index: int) -> str:
             try:
-                return _judge_category(entries, index, context_size, request.params, overlay_tokens)
+                category = _judge_category(entries, index, context_size, request.params, overlay_tokens)
             except urllib.error.HTTPError as exc:
                 # 限流/服务端错误：通知线程池临时降低最大并发，避免持续超配额。
                 if exc.code == 429 or 500 <= exc.code < 600:
                     pool.report_failure()
                 raise
-
+            # 判定成功立即落盘（断点存档）：失败/中断后重跑不重复调用已判定条目。
+            _append_partial(output_dir, index, category)
+            return category
         # 进度日志：打印已判定条数、总数、平均处理速度（条/s）、最近窗口
         # 平均单条耗时与当前线程数（与 OCR 节点同一回调协议）。
         def log_progress(done: int, total: int, rate: float, avg_time: float, workers: int) -> None:

nodes/subtitle_ocr.py

@@ -59,6 +59,28 @@ def _load_partial(output_dir: Path) -> dict[int, str]:
         result[int(item["frame"])] = str(item["text"])
     return result
 
+
+def _format_eta(seconds: float) -> str:
+    """把剩余秒数格式化为可读的预计完成时间（如 34分13秒 / 2小时05分）。"""
+    total = max(0, int(seconds))
+    hours, remainder = divmod(total, 3600)
+    minutes, secs = divmod(remainder, 60)
+    if hours:
+        return f"{hours}小时{minutes:02d}分"
+    if minutes:
+        return f"{minutes}分{secs:02d}秒"
+    return f"{secs}秒"
+
+
+def _eta_suffix(done: int, total: int, rate: float) -> str:
+    """根据当前处理速度计算剩余时间后缀（供进度日志追加）。
+
+    剩余时间 = 剩余帧数 / 当前速度；速度为 0（刚开始或耗时不可测）时
+    返回空串，进度提示不显示 ETA。
+    """
+    if rate <= 0:
+        return ""
+    return f", 预计剩余 {_format_eta((total - done) / rate)}"
 # 默认垃圾词：无文字帧的模型输出可能反复出现这些词。
 def _sampling_interval(manifest: list[dict], default: float) -> float:
     """从帧清单时间轴推导采样间隔（相邻帧时间差的中位数）。
@@ -206,12 +228,14 @@ def invoke(request: InvokeRequest) -> InvokeResponse:
 
     if pending:
         # 进度日志：打印已识别帧数、总数、平均处理速度（帧/s）、最近窗口平均
-        # 单帧耗时与当前线程数——便于判断多线程是否因单帧处理过慢而未启用
+        # 单帧耗时、当前线程数与**预计剩余完成时间**（剩余帧/当前速度）——
-        # （窗口平均响应 ≥ fast_threshold 时自适应池不会扩容）。
+        # 便于判断多线程是否因单帧处理过慢而未启用（窗口平均响应 ≥
+        # fast_threshold 时自适应池不会扩容）以及整体还需要多久。
         def log_progress(done: int, total: int, rate: float, avg_time: float, workers: int) -> None:
             logger.info(
-                "OCR 进度 %d/%d 帧 (%.1f 帧/s, 平均 %.2fs/帧, 线程 %d/%d)",
+                "OCR 进度 %d/%d 帧 (%.1f 帧/s, 平均 %.2fs/帧, 线程 %d/%d%s)",
                 done, total, rate, avg_time, workers, pool.max_workers,
+                _eta_suffix(done, total, rate),
             )
 
         # 自适应并发调用 vlm-ocr：10s 窗口内平均响应 < 0.3s 则加 1 线程（上限

nodes/whisper.py

@@ -22,6 +22,9 @@ from wov_sdk.models import InvokeRequest, InvokeResponse
 # 转写进度日志：输出到主进程控制台，长音频分块时可见每块进度。
 logger = get_logger("whisper")
 
+# 暂停信号文件名：位于 run 根目录（<storage>/runs/<run_id>/paused.flag），
+# 与 subtitle-ocr 节点约定一致；批量暂停时由暂停接口写入，分块间检查即中止。
+PAUSE_FLAG = "paused.flag"
 
 def _is_windows() -> bool:
     """判断当前是否为 Windows，供测试单独注入覆盖。"""
@@ -216,7 +219,7 @@ def invoke(request: InvokeRequest) -> InvokeResponse:
         model_path = resolve_model_path(request.params)
         device = str(request.params.get("device") or os.getenv("WHISPER_DEVICE", "auto"))
         # auto 让 faster-whisper 根据硬件自动选择 float16/int8 等计算类型。
-        compute_type = str(request.params.get("compute_type") or "auto")
+        compute_type = str(request.params.get("compute_type") or "float16")
         model = WhisperModel(
             model_path,
             device=device,
@@ -241,6 +244,12 @@ def invoke(request: InvokeRequest) -> InvokeResponse:
         srt_number = 1
         transcribe_started = time.monotonic()
         for chunk_index, chunk in enumerate(chunks, start=1):
+            # 暂停检查：批量暂停时在 run 根目录写 paused.flag，whisper 在分块
+            # 之间检查该信号（默认 60s 一块，暂停粒度不超过一块），检测到即抛
+            # 异常，由 invoke 转 failed、调度器保持任务 PAUSED，继续时整个节点
+            # 重新转写（whisper 没有节点级断点存档，产物只在结束时一次性写出）。
+            if (Path(request.output_dir).parent.parent / PAUSE_FLAG).exists():
+                raise RuntimeError(f"whisper 被暂停（run {request.run_id}）")
             chunk_started = time.monotonic()
             segments, _info = model.transcribe(
                 str(chunk),

scripts/rerun_filter_011d01f19999.py

@@ -0,0 +1,26 @@
+"""一次性脚本：用上下文净化后的 llm_filter 重跑 run_011d01f19999 的 filter 节点。
+
+目的：验证"上下文净化"修复对误删真实对话的恢复效果（对比原输出 885 保留 / 805 删除）。
+"""
+from pathlib import Path
+from dotenv import load_dotenv
+
+load_dotenv(Path(".env"))
+
+from nodes.llm_filter import invoke  # noqa: E402
+from wov_sdk.models import InvokeRequest  # noqa: E402
+
+run_root = Path("data/storage/runs/run_011d01f19999/steps")
+out_dir = run_root / "filter_rerun_v2"
+
+resp = invoke(
+    InvokeRequest(
+        run_id="run_011d01f19999",
+        node_instance_id="",
+        inputs={"srt_uri": str(run_root / "ocr/subtitle.srt")},
+        params={"pool_min_workers": 8, "pool_max_workers": 8},
+        output_dir=str(out_dir),
+    )
+)
+print("status:", resp.status, "| error:", resp.error)
+print("outputs:", resp.outputs)

src/wov_app/batch.py

@@ -0,0 +1,480 @@
+"""文件夹批量处理引擎。
+
+本地版的核心能力：**不把视频上传到工作目录**，而是直接读取用户所选文件夹
+下的所有视频，逐个调用现有的工作流流水线（复用 WorkflowScheduler 的 DAG
+执行与断点续跑逻辑）。
+
+数据落盘约定：
+
+- 每个视频的中间态数据（runs/、chunks/、帧图等）与最终产物都存放在**视频
+  所在目录的同名文件夹**里（movie.mp4 → movie/），源视频目录保持干净。
+- 最终产物（SRT/ASS）在任务完成后从节点产物目录复制到同名文件夹根目录，
+  同时写入 `batch.done.json` 完成标记；再次批量处理同一文件夹时，已有完成
+  标记且产物文件齐全的视频直接跳过（已经处理过的不再处理）。
+
+暂停/恢复语义（对应前端"暂停/继续"按钮）：
+
+- 暂停批量任务：把批量任务置为 PAUSED，并暂停当前正在执行的 run（写
+  paused.flag，whisper 按分块、OCR 按帧检查后停止），处理中的视频保持
+  PAUSED，后续视频不再开始。
+- 继续：批量任务恢复 QUEUED，引擎从断点继续——PAUSED 视频的 run 显式
+  resume 后由 execute_run 从产物表断点续跑，已完成节点不重复执行。
+
+主调度器不会抢占批量 run（next_queued_run 排除 source=batch），批量引擎
+使用 per-video 的 WorkflowScheduler 实例，storage 指向同名文件夹。
+"""
+
+from __future__ import annotations
+
+import json
+import shutil
+import threading
+import time
+import uuid
+from datetime import datetime, timezone
+from pathlib import Path
+
+from wov_app import registry
+from wov_app.config import BATCH_INTERVAL_SECONDS
+from wov_app.db import Database
+from wov_app.logging import get_logger
+from wov_app.scheduler import WorkflowScheduler
+from wov_sdk.models import WorkflowDefinition
+
+# 批量引擎运行日志：任务进度、视频逐个处理与暂停/续跑等状态变化。
+logger = get_logger("batch")
+
+# 识别为视频文件的扩展名（大小写不敏感，扫描时统一转小写比较）。
+VIDEO_EXTENSIONS = {
+    ".mp4", ".mkv", ".avi", ".mov", ".webm", ".flv", ".ts",
+    ".m4v", ".wmv", ".mpg", ".mpeg", ".3gp",
+}
+
+# 暂停信号文件名：与节点约定一致，位于 run 根目录（<work_dir>/runs/<run_id>/）。
+PAUSE_FLAG = "paused.flag"
+
+# 视频完成标记文件名：位于同名文件夹根目录，记录该视频已完成的工作流与最终
+# 产物文件名，跨批量任务去重（已经处理过的不再处理）。
+MARKER_NAME = "batch.done.json"
+
+
+def _now_iso() -> str:
+    """返回当前 UTC 时间的 ISO 格式字符串。"""
+    return datetime.now(timezone.utc).isoformat()
+
+
+def scan_videos(folder: Path, recursive: bool = True) -> list[Path]:
+    """扫描文件夹下的全部视频文件，按路径排序保证处理顺序确定。
+
+    recursive=True 时递归扫描子文件夹；recursive=False 只扫描顶层。
+    """
+    if recursive:
+        paths = [
+            p for p in folder.rglob("*")
+            if p.is_file() and p.suffix.lower() in VIDEO_EXTENSIONS
+        ]
+    else:
+        paths = [
+            p for p in folder.glob("*")
+            if p.is_file() and p.suffix.lower() in VIDEO_EXTENSIONS
+        ]
+    return sorted(paths)
+
+
+def work_dir_for(video: Path) -> Path:
+    """返回视频的同名文件夹：去掉扩展名，位于视频所在目录。
+
+    例如 movie.mp4 → 旁边的 movie/ 文件夹，中间态与最终产物都放这里。
+    """
+    return video.parent / video.stem
+
+
+def load_marker(work_dir: Path) -> dict | None:
+    """读取同名文件夹里的完成标记；不存在或损坏时返回 None。"""
+    path = work_dir / MARKER_NAME
+    if not path.is_file():
+        return None
+    try:
+        data = json.loads(path.read_text(encoding="utf-8"))
+        return data if isinstance(data, dict) else None
+    except (json.JSONDecodeError, OSError):
+        # 半行写入或权限异常时保守视为未完成，允许重新处理。
+        return None
+
+
+def ensure_videos(db: Database, job_id: str, videos: list[Path]) -> None:
+    """为扫描到的视频补齐 batch_videos 明细；已存在的记录保持不变。
+
+    同一批量任务反复处理（暂停/续跑）时保留每个视频的状态，已完成的不重置。
+    """
+    existing = {item["video_path"] for item in db.list_batch_videos(job_id)}
+    now = _now_iso()
+    for video in videos:
+        path = str(video)
+        if path in existing:
+            continue
+        db.create_batch_video({
+            "id": f"bv_{uuid.uuid4().hex[:12]}",
+            "job_id": job_id,
+            "video_path": path,
+            "work_dir": str(work_dir_for(video)),
+            "run_id": None,
+            "status": "PENDING",
+            "error": None,
+            "created_at": now,
+            "updated_at": now,
+        })
+
+
+def create_job(
+    db: Database,
+    folder_path: str,
+    workflow_id: str,
+    recursive: bool = True,
+) -> str:
+    """创建批量任务：校验文件夹与工作流、扫描视频、落库明细，返回任务 ID。
+
+    校验失败抛出 ValueError（由路由层转为 422 响应）；扫描到的视频全部
+    登记为 batch_videos 明细，引擎轮询到该任务后逐个处理。
+    """
+    folder = Path(folder_path).expanduser()
+    if not folder.is_dir():
+        raise ValueError("folder not found")
+    workflow = db.get_workflow(workflow_id)
+    if workflow is None or not workflow["published"]:
+        raise ValueError("published workflow not found")
+    if db.get_latest_workflow_version(workflow_id) is None:
+        raise ValueError("workflow has no version")
+    videos = scan_videos(folder, recursive)
+    if not videos:
+        raise ValueError("no videos found in folder")
+
+    job_id = f"batch_{uuid.uuid4().hex[:12]}"
+    now = _now_iso()
+    db.create_batch_job({
+        "id": job_id,
+        "folder_path": str(folder),
+        "workflow_id": workflow_id,
+        "recursive": int(recursive),
+        "status": "QUEUED",
+        "progress": 0,
+        "total": 0,
+        "done": 0,
+        "failed": 0,
+        "current_video": None,
+        "error": None,
+        "created_at": now,
+        "updated_at": now,
+    })
+    ensure_videos(db, job_id, videos)
+    logger.info("创建批量任务 %s: 文件夹 %s, 工作流 %s, 视频 %d 个", job_id, folder, workflow_id, len(videos))
+    return job_id
+
+class BatchWorker:
+    """批量处理引擎：单线程轮询 QUEUED 批量任务，逐视频调用现有调度器执行。"""
+
+    def __init__(
+        self,
+        db: Database,
+        interval_seconds: float | None = None,
+    ) -> None:
+        """保存数据库依赖并初始化轮询线程控制字段。"""
+        self.db = db
+        self.interval_seconds = interval_seconds or BATCH_INTERVAL_SECONDS
+        self._thread: threading.Thread | None = None
+        self._stopping = False
+
+    def start(self) -> None:
+        """启动批量处理线程；重复调用无副作用。"""
+        if self._thread is not None:
+            return
+        # 独立运行时确保节点已注册；重复注册幂等。
+        registry.register_all()
+        self._stopping = False
+        self._thread = threading.Thread(
+            target=self._loop,
+            name="wov-batch-worker",
+            daemon=True,
+        )
+        self._thread.start()
+
+    def stop(self) -> None:
+        """请求停止并等待轮询线程退出。"""
+        self._stopping = True
+        if self._thread is not None:
+            self._thread.join(timeout=5)
+            self._thread = None
+
+    def _loop(self) -> None:
+        """轮询循环：有排队中的批量任务就处理，否则休眠一个间隔。"""
+        while not self._stopping:
+            try:
+                job = self.db.next_queued_batch_job()
+                if job is not None:
+                    self._process_job(job)
+                else:
+                    time.sleep(self.interval_seconds)
+            except Exception:  # noqa: BLE001
+                # 单次轮询异常不杀死线程，记录后跳过本轮（与主调度器一致）。
+                logger.exception("批量引擎轮询异常，跳过本轮")
+                time.sleep(self.interval_seconds)
+
+    # ------------------------------------------------------------------
+    # 任务执行
+    # ------------------------------------------------------------------
+
+    def _process_job(self, job: dict) -> None:
+        """处理一个批量任务：校验、扫描、补齐明细、逐视频执行并复制产物。
+
+        job 以 QUEUED 状态进入，处理期间置 RUNNING；全部视频处理完置
+        COMPLETED；被暂停时保持 PAUSED；校验失败置 FAILED。
+        """
+        job_id = str(job["id"])
+        try:
+            self._run_job(job_id)
+        except Exception as exc:  # noqa: BLE001
+            # 任务级兜底：任何未捕获异常都记录到任务而不是卡死在 RUNNING。
+            logger.exception("批量任务 %s 处理异常", job_id)
+            self.db.update_batch_job(job_id, status="FAILED", error=str(exc), updated_at=_now_iso())
+
+    def _run_job(self, job_id: str) -> None:
+        """批量任务主流程（内部实现，异常由 _process_job 统一处理）。"""
+        job = self.db.get_batch_job(job_id)
+        if job is None:
+            return
+        folder = Path(job["folder_path"])
+        if not folder.is_dir():
+            self.db.update_batch_job(job_id, status="FAILED", error="folder not found", updated_at=_now_iso())
+            return
+
+        workflow = self.db.get_workflow(job["workflow_id"])
+        if workflow is None or not workflow["published"]:
+            self.db.update_batch_job(job_id, status="FAILED", error="workflow not found or unpublished", updated_at=_now_iso())
+            return
+        version = self.db.get_latest_workflow_version(job["workflow_id"])
+        if version is None:
+            self.db.update_batch_job(job_id, status="FAILED", error="workflow has no version", updated_at=_now_iso())
+            return
+        definition = WorkflowDefinition.from_dict(version["definition"])
+        definition.validate()
+
+        # 扫描当前文件夹的视频，为新增视频补齐明细（已有明细保留原状态，
+        # 保证暂停/续跑时已完成与进行中的视频不被重置）。
+        self._ensure_videos(job_id, scan_videos(folder, bool(job["recursive"])))
+
+        items = self.db.list_batch_videos(job_id)
+        total = len(items)
+        self.db.update_batch_job(
+            job_id, status="RUNNING", total=total, progress=0,
+            current_video=None, error=None, updated_at=_now_iso(),
+        )
+
+        for index, item in enumerate(items):
+            # 暂停检查：批量任务被暂停后停止处理后续视频，等待用户继续。
+            current = self.db.get_batch_job(job_id)
+            if current is None or current["status"] == "PAUSED":
+                logger.info("批量任务 %s 已暂停，停止在视频 %s", job_id, item["video_path"])
+                return
+
+            # 已完成/已跳过的视频不再处理。
+            if item["status"] in ("COMPLETED", "SKIPPED"):
+                continue
+
+            video = Path(item["video_path"])
+            if not video.is_file():
+                self.db.update_batch_video(item["id"], status="FAILED", error="video file not found", updated_at=_now_iso())
+                continue
+
+            work_dir = Path(item["work_dir"])
+            # 同名文件夹里已有同工作流的完成标记且产物齐全 → 直接跳过。
+            if self._is_done(work_dir, job["workflow_id"]):
+                self.db.update_batch_video(item["id"], status="SKIPPED", updated_at=_now_iso())
+                continue
+
+            self.db.update_batch_job(
+                job_id, current_video=str(video),
+                progress=index / total if total else 0,
+                updated_at=_now_iso(),
+            )
+            try:
+                self._process_video(job, item, version, definition, work_dir)
+            except Exception as exc:  # noqa: BLE001
+                # 单视频兜底：不中断整个批量任务，记录错误后继续下一个视频。
+                logger.exception("批量任务 %s 视频 %s 处理异常", job_id, video)
+                self.db.update_batch_video(item["id"], status="FAILED", error=str(exc), updated_at=_now_iso())
+
+            # 重新读取视频明细：_process_video 可能刚创建 run（快照里 run_id
+            # 还是 None），必须取最新记录才能拿到 run_id 判断暂停状态。
+            item = self.db.get_batch_video(item["id"])
+            # 视频处理中被暂停：批量任务整体保持 PAUSED，等待用户继续。
+            run = self.db.get_run(item["run_id"]) if item and item.get("run_id") else None
+            if run is not None and run["status"] == "PAUSED":
+                self.db.update_batch_video(item["id"], status="PAUSED", updated_at=_now_iso())
+                self.db.update_batch_job(job_id, status="PAUSED", updated_at=_now_iso())
+                return
+
+        # 全部视频处理完成：汇总已处理与失败数量，任务置为 COMPLETED。
+        items = self.db.list_batch_videos(job_id)
+        done = sum(1 for item in items if item["status"] in ("COMPLETED", "SKIPPED"))
+        failed = sum(1 for item in items if item["status"] == "FAILED")
+        self.db.update_batch_job(
+            job_id, status="COMPLETED", progress=1.0, done=done, failed=failed,
+            current_video=None, error=None, updated_at=_now_iso(),
+        )
+        logger.info(
+            "批量任务 %s 完成: 共 %d 个视频, 完成/跳过 %d, 失败 %d",
+            job_id, total, done, failed,
+        )
+
+    def _process_video(
+        self,
+        job: dict,
+        item: dict,
+        version: dict,
+        definition: WorkflowDefinition,
+        work_dir: Path,
+    ) -> None:
+        """处理单个视频：建 run（复用现有调度器）并执行，完成后复制最终产物。
+
+        per-video 的 WorkflowScheduler 以同名文件夹为 storage，中间态落在
+        <work_dir>/runs/<run_id>/steps/ 下；产物表记录全部节点输出，暂停后
+        续跑从产物表重建已完成节点（断点续跑）。
+        """
+        video = Path(item["video_path"])
+        work_dir.mkdir(parents=True, exist_ok=True)
+        run_id = item.get("run_id")
+        if run_id is None:
+            # 首次处理：创建 source=batch 的运行，input_uri 直接指向本地视频
+            # （不再上传副本），调度器按工作流 DAG 动态组装节点执行。
+            run_id = f"run_{uuid.uuid4().hex[:12]}"
+            now = _now_iso()
+            self.db.create_run({
+                "id": run_id,
+                "workflow_id": job["workflow_id"],
+                "workflow_version": int(version["version"]),
+                "status": "QUEUED",
+                "progress": 0,
+                "input_uri": str(video),
+                "param_overrides": None,
+                "source": "batch",
+                "created_at": now,
+                "updated_at": now,
+            })
+            self.db.update_batch_video(item["id"], run_id=run_id, updated_at=_now_iso())
+
+        run = self.db.get_run(run_id)
+        # 已完成（例如上次运行刚完成但未来得及标记）：直接复制产物后返回。
+        if run["status"] == "COMPLETED":
+            self._copy_finals(run_id, work_dir, definition, job["workflow_id"])
+            self.db.update_batch_video(item["id"], status="COMPLETED", error=None, updated_at=_now_iso())
+            return
+        # 暂停的 run 显式 resume 回 QUEUED，由 execute_run 从产物表断点续跑。
+        if run["status"] == "PAUSED":
+            self.db.resume_run(run_id, _now_iso())
+        elif run["status"] == "FAILED":
+            # 失败重跑：**保留**已完成节点的产物记录，只恢复 QUEUED——
+            # execute_run 从产物表重建已完成节点并跳过，只重跑失败节点。
+            # 不再 reset_run 清空产物：extract/ocr 等长耗时节点的成果（如
+            # ABP-885 的 22222 帧 OCR）会被白白丢弃重做（run_e2b74e89e232 实测）。
+            self.db.update_run(run_id, status="QUEUED", error=None, updated_at=_now_iso())
+        elif run["status"] == "RUNNING":
+            # 上次进程被杀残留：恢复 QUEUED（保留产物）由 execute_run 续跑。
+            self.db.update_run(run_id, status="QUEUED", updated_at=_now_iso())
+        # 清除可能残留的暂停信号（重启/异常中断后），避免本次执行误暂停。
+        (work_dir / "runs" / run_id / PAUSE_FLAG).unlink(missing_ok=True)
+
+        scheduler = WorkflowScheduler(self.db, work_dir)
+        scheduler.execute_run(run_id)
+
+        run = self.db.get_run(run_id)
+        if run["status"] == "COMPLETED":
+            # 复制最终产物到同名文件夹根目录并写完成标记（已处理过不再处理）。
+            self._copy_finals(run_id, work_dir, definition, job["workflow_id"])
+            self.db.update_batch_video(item["id"], status="COMPLETED", error=None, updated_at=_now_iso())
+        else:
+            # FAILED 或 PAUSED：由调用方根据 run 状态更新视频状态与任务状态。
+            self.db.update_batch_video(item["id"], status=run["status"], error=run.get("error"), updated_at=_now_iso())
+
+    # ------------------------------------------------------------------
+    # 明细与产物
+    # ------------------------------------------------------------------
+
+    def _ensure_videos(self, job_id: str, videos: list[Path]) -> None:
+        """为扫描到的视频补齐 batch_videos 明细；已存在的记录保持不变。"""
+        ensure_videos(self.db, job_id, videos)
+
+    def _is_done(self, work_dir: Path, workflow_id: str) -> bool:
+        """判断同名文件夹是否已完成当前工作流的处理。
+
+        完成标记记录 workflow_id 与最终产物文件名；只有工作流一致且产物文件
+        全部存在时才视为已处理（不同工作流的产物不互相误判为完成）。
+        """
+        marker = load_marker(work_dir)
+        if marker is None or marker.get("workflow_id") != workflow_id:
+            return False
+        finals = marker.get("finals") or {}
+        return bool(finals) and all((work_dir / name).is_file() for name in finals.values())
+
+    def _copy_finals(
+        self,
+        run_id: str,
+        work_dir: Path,
+        definition: WorkflowDefinition,
+        workflow_id: str,
+    ) -> None:
+        """把最终产物从节点目录复制到同名文件夹根目录，并写完成标记。
+
+        调度器收尾时已把产物重命名为 上传文件名.标识.时间戳（如
+        movie.zh-CN.20260819120000.srt），这里原样复制，文件名保留辨识度。
+        """
+        work_dir.mkdir(parents=True, exist_ok=True)
+        finals: dict[str, str] = {}
+        for alias in definition.final_outputs:
+            artifact = self.db.get_artifact(run_id, alias)
+            if artifact is None:
+                continue
+            source = Path(artifact["uri"])
+            if not source.is_file():
+                continue
+            target = work_dir / source.name
+            # 已存在的产物直接复用，避免重复复制。
+            if not target.is_file() or target.stat().st_size != source.stat().st_size:
+                shutil.copy2(source, target)
+            finals[alias] = source.name
+        marker = {
+            "workflow_id": workflow_id,
+            "workflow_version": definition.version,
+            "run_id": run_id,
+            "completed_at": _now_iso(),
+            "finals": finals,
+        }
+        (work_dir / MARKER_NAME).write_text(
+            json.dumps(marker, ensure_ascii=False, indent=2),
+            encoding="utf-8",
+        )
+
+    # ------------------------------------------------------------------
+    # 暂停/继续
+    # ------------------------------------------------------------------
+
+    def pause_job(self, job_id: str) -> None:
+        """暂停批量任务：停止当前 run 与后续视频处理。
+
+        先把任务置为 PAUSED（引擎在视频间检查后停下），再暂停所有排队/运行
+        中的 run 并写 paused.flag（whisper 按分块、OCR 按帧检查后中止）。
+        """
+        self.db.update_batch_job(job_id, status="PAUSED", updated_at=_now_iso())
+        for item in self.db.list_batch_videos(job_id):
+            if not item.get("run_id"):
+                continue
+            run = self.db.get_run(item["run_id"])
+            if run is None or run["status"] not in ("QUEUED", "RUNNING"):
+                continue
+            self.db.pause_run(item["run_id"], _now_iso())
+            run_dir = Path(item["work_dir"]) / "runs" / item["run_id"]
+            run_dir.mkdir(parents=True, exist_ok=True)
+            (run_dir / PAUSE_FLAG).write_text("", encoding="utf-8")
+
+    def resume_job(self, job_id: str) -> None:
+        """继续批量任务：置回 QUEUED，引擎从断点续跑（PAUSED 视频逐个 resume）。"""
+        self.db.update_batch_job(job_id, status="QUEUED", updated_at=_now_iso())

src/wov_app/config.py

@@ -20,6 +20,11 @@ STORAGE_DIR = Path(os.getenv("WOV_STORAGE_DIR", str(DATA_DIR / "storage")))
 # 调度器轮询排队任务的间隔（秒）。
 SCHEDULER_INTERVAL_SECONDS = float(os.getenv("WOV_SCHEDULER_INTERVAL_SECONDS", "1.0"))
 
+# 批量处理引擎开关与轮询间隔：引擎处理 source=batch 的任务（文件夹批量处理），
+# 与主调度器互不抢占；测试与不需要该功能的部署可关闭。
+BATCH_ENABLED = os.getenv("WOV_BATCH_ENABLED", "1") == "1"
+BATCH_INTERVAL_SECONDS = float(os.getenv("WOV_BATCH_INTERVAL_SECONDS", "1.0"))
+
 # 孤儿数据清理器配置：定时扫描并清理无对应文件/记录的死数据。
 CLEANUP_ENABLED = os.getenv("WOV_CLEANUP_ENABLED", "1") == "1"
 # 清理扫描周期（秒），默认每小时一次。

src/wov_app/db.py

@@ -73,6 +73,7 @@ class Database:
                     error TEXT,
                     input_uri TEXT,
                     param_overrides TEXT,
+                    source TEXT NOT NULL DEFAULT 'upload',
                     created_at TEXT NOT NULL,
                     updated_at TEXT NOT NULL,
                     FOREIGN KEY(workflow_id) REFERENCES workflows(id)
@@ -91,6 +92,39 @@ class Database:
                     UNIQUE(run_id, name),
                     FOREIGN KEY(run_id) REFERENCES workflow_runs(id)
                 );
+
+                -- 批量处理任务表：一次"文件夹批量处理"对应一条记录，记录目标
+                -- 文件夹、所选工作流与整体状态。批量引擎与 Web 页面共用。
+                CREATE TABLE IF NOT EXISTS batch_jobs (
+                    id TEXT PRIMARY KEY,
+                    folder_path TEXT NOT NULL,
+                    workflow_id TEXT NOT NULL,
+                    recursive INTEGER NOT NULL DEFAULT 1,
+                    status TEXT NOT NULL,
+                    progress REAL NOT NULL DEFAULT 0,
+                    total INTEGER NOT NULL DEFAULT 0,
+                    done INTEGER NOT NULL DEFAULT 0,
+                    failed INTEGER NOT NULL DEFAULT 0,
+                    current_video TEXT,
+                    error TEXT,
+                    created_at TEXT NOT NULL,
+                    updated_at TEXT NOT NULL
+                );
+
+                -- 批量视频明细表：一次批量任务处理的每个视频一条记录，保存其
+                -- 对应的工作流 run（断点续跑复用 workflow_runs 的产物状态）。
+                CREATE TABLE IF NOT EXISTS batch_videos (
+                    id TEXT PRIMARY KEY,
+                    job_id TEXT NOT NULL,
+                    video_path TEXT NOT NULL,
+                    work_dir TEXT NOT NULL,
+                    run_id TEXT,
+                    status TEXT NOT NULL,
+                    error TEXT,
+                    created_at TEXT NOT NULL,
+                    updated_at TEXT NOT NULL,
+                    FOREIGN KEY(job_id) REFERENCES batch_jobs(id)
+                );
                 """
             )
 
@@ -101,7 +135,10 @@ class Database:
             ]
             if "param_overrides" not in columns:
                 conn.execute("ALTER TABLE workflow_runs ADD COLUMN param_overrides TEXT")
-
+            # 旧库迁移：workflow_runs 补充 source 列（upload=网页上传 / batch=批量处理），
+            # 批量引擎与主调度器据此隔离任务，避免互相抢占。
+            if "source" not in columns:
+                conn.execute("ALTER TABLE workflow_runs ADD COLUMN source TEXT NOT NULL DEFAULT 'upload'")
     def upsert_workflow(self, workflow: dict[str, Any]) -> None:
         """插入或更新工作流概要信息。"""
         with self._connect() as conn:
@@ -210,15 +247,20 @@ class Database:
         return versions
 
     def create_run(self, run: dict[str, Any]) -> None:
-        """创建一条排队中的工作流运行记录。"""
+        """创建一条排队中的工作流运行记录。
+
+        source 标识任务来源：upload（网页上传，默认）由主调度器执行；
+        batch（文件夹批量处理）由批量引擎执行，input_uri 直接指向本地视频。
+        """
         with self._connect() as conn:
             conn.execute(
                 """
                 INSERT INTO workflow_runs (
                     id, workflow_id, workflow_version, status, current_node_id,
-                    progress, error, input_uri, param_overrides, created_at, updated_at
+                    progress, error, input_uri, param_overrides, source,
+                    created_at, updated_at
                 )
-                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                 """,
                 (
                     run["id"],
@@ -232,11 +274,11 @@ class Database:
                     json.dumps(run["param_overrides"], ensure_ascii=False)
                     if run.get("param_overrides")
                     else None,
+                    run.get("source", "upload"),
                     run["created_at"],
                     run["updated_at"],
                 ),
             )
-
     def get_run(self, run_id: str) -> dict[str, Any] | None:
         """按 ID 读取任务运行记录。"""
         with self._connect() as conn:
@@ -308,12 +350,14 @@ class Database:
         只取 QUEUED：PAUSED 任务必须由用户显式 resume（转回 QUEUED）后调度器
         才重新执行。修复回归——此前把 PAUSED 也当可执行任务拾起，execute_run
         会先置 RUNNING 再检查暂停，导致"点击暂停反而开始任务"。
+        同时排除 source=batch 的批量运行：批量任务由批量引擎使用视频旁的
+        同名文件夹作为 storage 执行，主调度器拾起会用错存储目录。
         """
         with self._connect() as conn:
             row = conn.execute(
                 """
                 SELECT * FROM workflow_runs
-                WHERE status = 'QUEUED'
+                WHERE status = 'QUEUED' AND source != 'batch'
                 ORDER BY created_at ASC
                 LIMIT 1
                 """
@@ -416,3 +460,152 @@ class Database:
         with self._connect() as conn:
             conn.execute("DELETE FROM artifacts WHERE run_id = ?", (run_id,))
             conn.execute("DELETE FROM workflow_runs WHERE id = ?", (run_id,))
+
+    # ------------------------------------------------------------------
+    # 批量处理任务（batch_jobs / batch_videos）数据访问。
+    # 批量引擎与批量管理页共用这些方法，规则与 workflow_runs 一致。
+    # ------------------------------------------------------------------
+
+    def create_batch_job(self, job: dict[str, Any]) -> None:
+        """插入一条批量处理任务记录。"""
+        with self._connect() as conn:
+            conn.execute(
+                """
+                INSERT INTO batch_jobs (
+                    id, folder_path, workflow_id, recursive, status, progress,
+                    total, done, failed, current_video, error, created_at, updated_at
+                )
+                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+                """,
+                (
+                    job["id"],
+                    job["folder_path"],
+                    job["workflow_id"],
+                    int(job.get("recursive", 1)),
+                    job["status"],
+                    float(job.get("progress", 0)),
+                    int(job.get("total", 0)),
+                    int(job.get("done", 0)),
+                    int(job.get("failed", 0)),
+                    job.get("current_video"),
+                    job.get("error"),
+                    job["created_at"],
+                    job["updated_at"],
+                ),
+            )
+
+    def get_batch_job(self, job_id: str) -> dict[str, Any] | None:
+        """按 ID 读取批量任务记录。"""
+        with self._connect() as conn:
+            row = conn.execute("SELECT * FROM batch_jobs WHERE id = ?", (job_id,)).fetchone()
+        return dict(row) if row else None
+
+    def list_batch_jobs(self, limit: int = 50) -> list[dict[str, Any]]:
+        """按创建时间倒序返回最近的批量任务。"""
+        with self._connect() as conn:
+            rows = conn.execute(
+                "SELECT * FROM batch_jobs ORDER BY created_at DESC LIMIT ?",
+                (limit,),
+            ).fetchall()
+        return [dict(row) for row in rows]
+
+    def list_batch_job_ids(self) -> list[str]:
+        """返回全部批量任务 ID，供孤儿清理区分批量运行使用。"""
+        with self._connect() as conn:
+            rows = conn.execute("SELECT id FROM batch_jobs").fetchall()
+        return [row["id"] for row in rows]
+
+    def next_queued_batch_job(self) -> dict[str, Any] | None:
+        """按创建时间返回最早一条排队（QUEUED）的批量任务。
+
+        批量引擎单线程顺序处理，同一时刻只执行一个批量任务。
+        """
+        with self._connect() as conn:
+            row = conn.execute(
+                """
+                SELECT * FROM batch_jobs
+                WHERE status = 'QUEUED'
+                ORDER BY created_at ASC
+                LIMIT 1
+                """
+            ).fetchone()
+        return dict(row) if row else None
+
+    def update_batch_job(self, job_id: str, **fields: Any) -> None:
+        """更新批量任务字段，同时刷新 updated_at；未知字段会被忽略。"""
+        allowed = {
+            "status",
+            "progress",
+            "total",
+            "done",
+            "failed",
+            "current_video",
+            "error",
+        }
+        updates = {key: value for key, value in fields.items() if key in allowed}
+        if not updates:
+            return
+        updates["updated_at"] = fields.get("updated_at")
+        assignments = ", ".join(f"{key} = ?" for key in updates)
+        values = list(updates.values()) + [job_id]
+        with self._connect() as conn:
+            conn.execute(f"UPDATE batch_jobs SET {assignments} WHERE id = ?", values)
+
+    def delete_batch_job(self, job_id: str) -> None:
+        """删除批量任务记录及其全部视频明细（不含 workflow_runs）。"""
+        with self._connect() as conn:
+            conn.execute("DELETE FROM batch_videos WHERE job_id = ?", (job_id,))
+            conn.execute("DELETE FROM batch_jobs WHERE id = ?", (job_id,))
+
+    def create_batch_video(self, item: dict[str, Any]) -> None:
+        """插入一条批量视频明细记录。"""
+        with self._connect() as conn:
+            conn.execute(
+                """
+                INSERT INTO batch_videos (
+                    id, job_id, video_path, work_dir, run_id, status, error,
+                    created_at, updated_at
+                )
+                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
+                """,
+                (
+                    item["id"],
+                    item["job_id"],
+                    item["video_path"],
+                    item["work_dir"],
+                    item.get("run_id"),
+                    item["status"],
+                    item.get("error"),
+                    item["created_at"],
+                    item["updated_at"],
+                ),
+            )
+
+    def get_batch_video(self, video_id: str) -> dict[str, Any] | None:
+        """按 ID 读取批量视频明细。"""
+        with self._connect() as conn:
+            row = conn.execute(
+                "SELECT * FROM batch_videos WHERE id = ?", (video_id,)
+            ).fetchone()
+        return dict(row) if row else None
+
+    def list_batch_videos(self, job_id: str) -> list[dict[str, Any]]:
+        """按创建时间返回一次批量任务的全部视频明细。"""
+        with self._connect() as conn:
+            rows = conn.execute(
+                "SELECT * FROM batch_videos WHERE job_id = ? ORDER BY created_at",
+                (job_id,),
+            ).fetchall()
+        return [dict(row) for row in rows]
+
+    def update_batch_video(self, video_id: str, **fields: Any) -> None:
+        """更新批量视频字段，同时刷新 updated_at；未知字段会被忽略。"""
+        allowed = {"run_id", "status", "error"}
+        updates = {key: value for key, value in fields.items() if key in allowed}
+        if not updates:
+            return
+        updates["updated_at"] = fields.get("updated_at")
+        assignments = ", ".join(f"{key} = ?" for key in updates)
+        values = list(updates.values()) + [video_id]
+        with self._connect() as conn:
+            conn.execute(f"UPDATE batch_videos SET {assignments} WHERE id = ?", values)

src/wov_app/main.py

@@ -21,10 +21,11 @@ from fastapi.middleware.cors import CORSMiddleware  # noqa: E402
 from fastapi.staticfiles import StaticFiles  # noqa: E402
 
 from wov_app import registry
-from wov_app.config import DB_PATH, STORAGE_DIR, WORKSPACE_ROOT
+from wov_app.batch import BatchWorker
+from wov_app.config import BATCH_ENABLED, DB_PATH, STORAGE_DIR, WORKSPACE_ROOT
 from wov_app.db import Database
 from wov_app.maintenance import OrphanCleaner
-from wov_app.routers import apps, workflows
+from wov_app.routers import apps, batch, workflows
 from wov_app.scheduler import WorkflowScheduler
 from wov_app.seed import seed_default_workflows
 
@@ -50,12 +51,19 @@ async def lifespan(app: FastAPI):
     # 孤儿数据清理默认开启，定时清除死数据；测试可关闭。
     if os.getenv("WOV_CLEANUP_ENABLED", "1") == "1":
         cleaner.start()
+    # 批量处理引擎：处理用户所选文件夹的批量任务（source=batch 的运行），
+    # 与主调度器互不抢占；默认开启，测试可关闭。
+    batch_worker = BatchWorker(db)
+    if os.getenv("WOV_BATCH_ENABLED", "1") == "1":
+        batch_worker.start()
     # 共享对象挂到 app.state，路由通过 Depends 延迟获取。
     app.state.db = db
     app.state.scheduler = scheduler
     app.state.cleaner = cleaner
+    app.state.batch = batch_worker
     yield
-    # 退出时先停调度器与清理器，避免残留后台线程。
+    # 退出时先停批量引擎、调度器与清理器，避免残留后台线程。
+    batch_worker.stop()
     cleaner.stop()
     scheduler.stop()
 
@@ -72,6 +80,7 @@ app.add_middleware(
 
 app.include_router(workflows.router)
 app.include_router(apps.router)
+app.include_router(batch.router)
 
 
 @app.get("/health")

src/wov_app/maintenance.py

@@ -82,6 +82,11 @@ class OrphanCleaner:
             run = self.db.get_run(run_id)
             if run is None:
                 continue
+            # 批量处理运行（source=batch）跳过清理：其产物在用户视频旁的同名
+            # 文件夹里，不在主存储目录下；_has_files 检查不到会误判为孤儿删除，
+            # 且 _remove_run 还会删除 input_uri 的父目录（用户的视频文件夹）。
+            if run.get("source") == "batch":
+                continue
             if run["status"] != "COMPLETED":
                 continue
             if not self._expired(run.get("updated_at")):

src/wov_app/routers/batch.py

@@ -0,0 +1,205 @@
+"""批量处理 API 路由。
+
+面向本地版用户：输入文件夹路径与所选工作流创建批量任务，直接处理所选
+文件夹下的所有视频（**不上传副本**），每个视频的中间态与最终产物都存放在
+视频旁的同名文件夹中。支持暂停/继续（重新开始继续上次未完成的内容）、
+已处理视频自动跳过与最终产物下载。
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+from fastapi import APIRouter, Depends, HTTPException, Query
+from fastapi.responses import FileResponse
+
+from wov_app import batch as batch_engine
+from wov_app.db import Database
+from wov_app.schemas import BatchJobCreate
+
+router = APIRouter(tags=["batch"])
+
+
+def _get_db() -> Database:
+    """从 FastAPI 应用状态中延迟获取数据库实例。"""
+    from wov_app.main import app
+
+    return app.state.db
+
+def _get_worker() -> batch_engine.BatchWorker | None:
+    """从 FastAPI 应用状态中延迟获取批量引擎实例。"""
+    from wov_app.main import app
+
+    return getattr(app.state, "batch", None)
+
+
+def _enrich_videos(db: Database, videos: list[dict]) -> list[dict]:
+    """为每个视频补充最终产物清单（从同名文件夹的完成标记读取）。
+
+    finals 形如 {别名: 文件名}（如 {"cn_srt": "movie.zh-CN.20260819.srt"}），
+    前端据此渲染下载链接；未完成的视频没有产物。
+    """
+    for video in videos:
+        marker = batch_engine.load_marker(Path(video["work_dir"]))
+        video["finals"] = (marker or {}).get("finals") or {}
+    return videos
+
+@router.post("/api/batch/jobs")
+def create_batch_job(
+    payload: BatchJobCreate,
+    db: Database = Depends(_get_db),
+) -> dict:
+    """创建批量处理任务：校验文件夹与工作流，扫描视频并登记明细。"""
+    try:
+        job_id = batch_engine.create_job(
+            db, payload.folder, payload.workflow_id, payload.recursive
+        )
+    except ValueError as exc:
+        raise HTTPException(status_code=422, detail=str(exc)) from exc
+    job = db.get_batch_job(job_id)
+    job["videos"] = _enrich_videos(db, db.list_batch_videos(job_id))
+    return job
+
+
+@router.get("/api/batch/jobs")
+def list_batch_jobs(db: Database = Depends(_get_db)) -> list[dict]:
+    """返回最近的批量任务列表（不含视频明细，明细按需单独查询）。"""
+    return db.list_batch_jobs()
+
+
+@router.get("/api/batch/jobs/{job_id}")
+def get_batch_job(job_id: str, db: Database = Depends(_get_db)) -> dict:
+    """返回批量任务详情，附带每个视频的处理状态与最终产物清单。"""
+    job = db.get_batch_job(job_id)
+    if job is None:
+        raise HTTPException(status_code=404, detail="batch job not found")
+    job["videos"] = _enrich_videos(db, db.list_batch_videos(job_id))
+    return job
+
+
+@router.post("/api/batch/jobs/{job_id}/pause")
+def pause_batch_job(
+    job_id: str,
+    db: Database = Depends(_get_db),
+) -> dict:
+    """暂停批量任务：当前 run 在分块/帧边界停下，后续视频不再开始。"""
+    worker = _get_worker()
+    if db.get_batch_job(job_id) is None:
+        raise HTTPException(status_code=404, detail="batch job not found")
+    if worker is None:
+        raise HTTPException(status_code=503, detail="batch worker not available")
+    worker.pause_job(job_id)
+    return {"id": job_id, "status": "PAUSED"}
+
+
+@router.post("/api/batch/jobs/{job_id}/resume")
+def resume_batch_job(
+    job_id: str,
+    db: Database = Depends(_get_db),
+) -> dict:
+    """继续批量任务：从上次断点（未完成的视频）继续处理。"""
+    worker = _get_worker()
+    if db.get_batch_job(job_id) is None:
+        raise HTTPException(status_code=404, detail="batch job not found")
+    if worker is None:
+        raise HTTPException(status_code=503, detail="batch worker not available")
+    worker.resume_job(job_id)
+    return {"id": job_id, "status": "QUEUED"}
+
+
+@router.delete("/api/batch/jobs/{job_id}")
+def delete_batch_job(job_id: str, db: Database = Depends(_get_db)) -> dict:
+    """删除批量任务：移除任务、明细记录与关联的 run 记录。
+
+    磁盘上的同名文件夹与产物属于用户数据，保留不删（与任务页删除接口的
+    行为不同），只清理数据库记录。
+    """
+    if db.get_batch_job(job_id) is None:
+        raise HTTPException(status_code=404, detail="batch job not found")
+    for item in db.list_batch_videos(job_id):
+        if item.get("run_id"):
+            db.delete_run(item["run_id"])
+    db.delete_batch_job(job_id)
+    return {"deleted": job_id}
+
+
+@router.get("/api/batch/jobs/{job_id}/videos/{video_id}/download")
+def download_batch_video(
+    job_id: str,
+    video_id: str,
+    alias: str = Query(...),
+    db: Database = Depends(_get_db),
+) -> FileResponse:
+    """下载视频的最终产物：从同名文件夹的完成标记解析产物文件名后返回。
+
+    alias 为工作流 final_outputs 的别名（如 cn_srt / ass）；只有完成标记
+    中记录且文件真实存在的产物才可下载。
+    """
+    video = db.get_batch_video(video_id)
+    if video is None or video["job_id"] != job_id:
+        raise HTTPException(status_code=404, detail="video not found")
+    marker = batch_engine.load_marker(Path(video["work_dir"]))
+    if marker is None or alias not in (marker.get("finals") or {}):
+        raise HTTPException(status_code=404, detail="artifact not found")
+    path = Path(video["work_dir"]) / marker["finals"][alias]
+    if not path.is_file():
+        raise HTTPException(status_code=404, detail="artifact file missing")
+    return FileResponse(path, filename=path.name)
+
+
+# ---------------------------------------------------------------------------
+# 本地目录浏览（目录树选择器）
+#
+# 浏览器出于安全限制拿不到所选文件夹的绝对路径，因此由**本地后端**提供目录
+# 浏览能力：roots 返回可浏览的根（Windows 盘符 / POSIX 根 + 家目录），dirs
+# 返回指定目录的直接子目录，前端据此渲染懒加载目录树，点击选择后回填路径。
+# ---------------------------------------------------------------------------
+
+
+@router.get("/api/batch/roots")
+def list_batch_roots() -> list[dict]:
+    """返回本地文件系统的可浏览根目录列表。
+
+    Windows 返回存在的盘符（C:\、D:\…）与家目录；POSIX 返回根目录 / 与
+    家目录。每个条目为 {"path": 绝对路径, "name": 展示名}。
+    """
+    import os
+    import string
+
+    roots: list[dict] = []
+    if os.name == "nt":
+        # Windows：探测 A-Z 盘符，仅返回实际存在的驱动器。
+        for letter in string.ascii_uppercase:
+            drive = Path(f"{letter}:\\")
+            if drive.exists():
+                roots.append({"path": str(drive), "name": f"{letter}:\\"})
+    else:
+        roots.append({"path": "/", "name": "/"})
+    home = Path.home()
+    roots.append({"path": str(home), "name": f"家目录（{home.name}）"})
+    return roots
+
+
+@router.get("/api/batch/dirs")
+def list_batch_dirs(path: str = Query(...)) -> dict:
+    """列出指定目录下的直接子目录，供目录树节点懒加载。
+
+    目录不存在、不可读或个别子项无权限时返回空列表（不报 500），前端保持
+    树可用；隐藏点开头的隐藏目录减少噪音。
+    """
+    target = Path(path).expanduser()
+    dirs: list[dict] = []
+    if target.is_dir():
+        try:
+            for child in sorted(target.iterdir()):
+                try:
+                    # 隐藏目录（.git 等）不展示；只列目录，不暴露文件内容。
+                    if child.is_dir() and not child.name.startswith("."):
+                        dirs.append({"path": str(child), "name": child.name})
+                except OSError:
+                    # 单个子项不可读（权限不足）时跳过，不拖垮整棵树。
+                    continue
+        except OSError:
+            # 整个目录不可读（如无权限的系统目录）时返回空列表。
+            dirs = []
+    return {"path": str(target), "dirs": dirs}

src/wov_app/schemas.py

@@ -20,3 +20,18 @@ class WorkflowCreate(BaseModel):
     description: str = ""
     # DAG 原始字典，后端会解析并校验为 WorkflowDefinition。
     definition: dict[str, Any]
+
+
+class BatchJobCreate(BaseModel):
+    """创建文件夹批量处理任务的请求体。
+
+    批量处理直接读取用户所选文件夹下的视频（不上传副本），每个视频使用其
+    同名文件夹存放中间态与最终产物。
+    """
+
+    # 视频所在文件夹的本地路径（可绝对路径，浏览器输入/粘贴）。
+    folder: str = Field(min_length=1)
+    # 要执行的已发布工作流 ID（可执行任意所选流水线）。
+    workflow_id: str = Field(min_length=1)
+    # 是否递归扫描子文件夹，默认开启。
+    recursive: bool = True

tests/test_apps_api.py

@@ -293,12 +293,12 @@ def test_create_run_with_param_overrides() -> None:
         uploaded = client.post(
             f"/api/apps/{workflow_id}/runs",
             files={"file": ("sample.txt", b"x", "text/plain")},
-            data={"params": '{"step": {"crop": [0, 0.82, 1, 0.18]}}'},
+            data={"params": '{"step": {"crop": [0, 0.75, 1, 0.25]}}'},
         )
         assert uploaded.status_code == 200
         run_id = uploaded.json()["id"]
         run = client.get(f"/api/runs/{run_id}").json()
-        assert run["param_overrides"] == {"step": {"crop": [0, 0.82, 1, 0.18]}}
+        assert run["param_overrides"] == {"step": {"crop": [0, 0.75, 1, 0.25]}}
         # 非法 JSON 返回 422。
         bad = client.post(
             f"/api/apps/{workflow_id}/runs",

tests/test_db.py

@@ -329,3 +329,103 @@ def test_restore_run_outputs(tmp_path) -> None:
         "extract": {"frames_manifest": "frames.json"}
     }
     assert db.restore_run_outputs("run_none") == {}
+
+
+def test_run_source_column_default_and_next_queued(tmp_path) -> None:
+    """source 列默认 upload；主调度器不拾取 batch 来源的运行。"""
+    db = Database(tmp_path / "wov.db")
+    db.upsert_workflow({"id": "demo", "name": "Demo", "published": 1, "latest_version": 1})
+    now = "2026-01-01T00:00:00+00:00"
+    db.create_run(
+        {
+            "id": "run_upload",
+            "workflow_id": "demo",
+            "workflow_version": 1,
+            "status": "QUEUED",
+            "progress": 0,
+            "created_at": now,
+            "updated_at": now,
+        }
+    )
+    db.create_run(
+        {
+            "id": "run_batch",
+            "workflow_id": "demo",
+            "workflow_version": 1,
+            "status": "QUEUED",
+            "progress": 0,
+            "source": "batch",
+            "created_at": now,
+            "updated_at": now,
+        }
+    )
+    # 默认 source 为 upload，可显式指定 batch。
+    assert db.get_run("run_upload")["source"] == "upload"
+    assert db.get_run("run_batch")["source"] == "batch"
+    # 主调度器只取非 batch 运行，批量运行由批量引擎单独拾起。
+    assert db.next_queued_run()["id"] == "run_upload"
+
+
+def test_db_migration_adds_source_column(tmp_path) -> None:
+    """旧库迁移：缺少 source 列的库打开后自动补列并默认 upload。"""
+    import sqlite3
+
+    db_path = tmp_path / "old.db"
+    conn = sqlite3.connect(db_path)
+    conn.execute(
+        "CREATE TABLE workflow_runs (id TEXT PRIMARY KEY, workflow_id TEXT NOT NULL,"
+        " workflow_version INTEGER NOT NULL, status TEXT NOT NULL, current_node_id TEXT,"
+        " progress REAL NOT NULL DEFAULT 0, error TEXT, input_uri TEXT,"
+        " param_overrides TEXT, created_at TEXT NOT NULL, updated_at TEXT NOT NULL)"
+    )
+    conn.commit()
+    conn.close()
+
+    Database(db_path)
+    conn = sqlite3.connect(db_path)
+    columns = [row[1] for row in conn.execute("PRAGMA table_info(workflow_runs)")]
+    conn.close()
+    assert "source" in columns
+
+
+def test_batch_jobs_and_videos_crud(tmp_path) -> None:
+    """批量任务/视频明细的增删改查与排队查询。"""
+    db = Database(tmp_path / "wov.db")
+    now = "2026-01-01T00:00:00+00:00"
+    db.create_batch_job(
+        {
+            "id": "batch_1", "folder_path": "/videos", "workflow_id": "demo",
+            "recursive": 1, "status": "QUEUED", "progress": 0, "total": 2,
+            "done": 0, "failed": 0, "current_video": None, "error": None,
+            "created_at": now, "updated_at": now,
+        }
+    )
+    assert db.next_queued_batch_job()["id"] == "batch_1"
+    assert db.list_batch_job_ids() == ["batch_1"]
+    assert db.list_batch_jobs()[0]["total"] == 2
+
+    db.create_batch_video(
+        {
+            "id": "bv_1", "job_id": "batch_1", "video_path": "/videos/a.mp4",
+            "work_dir": "/videos/a", "run_id": None, "status": "PENDING",
+            "error": None, "created_at": now, "updated_at": now,
+        }
+    )
+    db.update_batch_video("bv_1", status="COMPLETED", updated_at=now)
+    assert db.get_batch_video("bv_1")["status"] == "COMPLETED"
+    assert len(db.list_batch_videos("batch_1")) == 1
+
+    # 未知字段更新被忽略（不会报错也不会改状态）。
+    db.update_batch_video("bv_1", bogus=1, updated_at=now)
+    db.update_batch_job("batch_1", bogus=1, updated_at=now)
+    assert db.get_batch_video("bv_1")["status"] == "COMPLETED"
+
+    db.update_batch_job("batch_1", status="COMPLETED", done=1, failed=0, progress=1.0, updated_at=now)
+    job = db.get_batch_job("batch_1")
+    assert job["status"] == "COMPLETED" and job["done"] == 1
+    # 完成后不再排队。
+    assert db.next_queued_batch_job() is None
+
+    db.delete_batch_job("batch_1")
+    assert db.get_batch_job("batch_1") is None
+    assert db.get_batch_video("bv_1") is None

tests/test_llm_filter.py

@@ -709,3 +709,111 @@ def test_real_run_rules_and_dialogue_regression(monkeypatch, tmp_path) -> None:
     })
     assert len(fake.bodies) == unique_llm
     assert len(fake.bodies) < len(entries)  # 去重确实省调用。
+
+
+class _FailOnTarget429:
+    """模拟持续限流：目标字幕含指定词时恒抛 429，其余正常返回 dialogue。
+
+    用于构造"部分条目成功、个别条目持续 429"的断点重跑场景：第一次 invoke
+    整体失败但成功条目已写存档；解除限流后第二次 invoke 只重判失败条目。
+    """
+
+    def __init__(self, marker: str) -> None:
+        self.marker = marker
+        self.enabled = True
+        self.bodies: list[dict] = []
+
+    def __call__(self, request, timeout=None):
+        body = json.loads(request.data.decode("utf-8"))
+        self.bodies.append(body)
+        target = next(
+            line for line in body["messages"][1]["content"].splitlines()
+            if line.startswith("【目标】")
+        )
+        if self.enabled and self.marker in target:
+            raise urllib.error.HTTPError(request.full_url, 429, "rate limited", {}, None)
+        payload = json.dumps({"choices": [{"message": {"content": "dialogue"}}]}).encode()
+        return FakeResponse(payload)
+
+
+def _llm_invoke(srt_text: str, out: Path) -> tuple[object, Path]:
+    """用给定 SRT 文本构造并执行一次 llm-filter invoke，返回 (响应, 输入文件)。"""
+    srt = out.parent / "in.srt"
+    srt.write_text(srt_text, encoding="utf-8")
+    return (
+        invoke(
+            InvokeRequest(
+                run_id="r", node_instance_id="",
+                inputs={"srt_uri": str(srt)},
+                params={},
+                output_dir=str(out),
+            )
+        ),
+        srt,
+    )
+
+
+def test_load_and_append_partial(tmp_path) -> None:
+    """判定存档读写：无存档/损坏行跳过，追加后可读回。"""
+    from nodes.llm_filter import _PARTIAL_NAME, _append_partial, _load_partial
+
+    out = tmp_path / "out"
+    assert _load_partial(out) == {}  # 目录不存在 → 空。
+    out.mkdir()
+    assert _load_partial(out) == {}  # 无存档 → 空。
+    # 损坏行（半行写入）跳过，正常行读回。
+    (out / _PARTIAL_NAME).write_text(
+        '{"index": 0, "category": "dialogue"}\n\n{broken\n{"index": 3, "category": "garbage"}\n',
+        encoding="utf-8",
+    )
+    assert _load_partial(out) == {0: "dialogue", 3: "garbage"}
+    # 追加一条后读回。
+    _append_partial(out, 5, "overlay")
+    assert _load_partial(out)[5] == "overlay"
+
+
+def test_invoke_resume_skips_archived_judgments(monkeypatch, tmp_path) -> None:
+    """断点存档：已判定条目重跑时不重复调用 LLM（去重后只补判未判定）。"""
+    fake = _patch_llm(monkeypatch, contents=["dialogue"] * 4)
+    out = tmp_path / "out"
+    out.mkdir()
+    # 预置存档：索引 0、1 已判定（模拟上次失败前已完成的部分）。
+    from nodes.llm_filter import _append_partial
+
+    _append_partial(out, 0, "dialogue")
+    _append_partial(out, 1, "dialogue")
+    response, _srt = _llm_invoke(_SRT, out)
+    assert response.status == "completed", response.error
+    # 4 条唯一文本中 2 条已存档，只调用剩余 2 条。
+    assert len(fake.bodies) == 2
+    # 输出与全量判定一致：全部 dialogue → 4 条都保留。
+    assert response.outputs["kept"] == 4
+    assert Path(response.outputs["srt_uri"]).read_text(encoding="utf-8").count("-->") == 4
+
+
+def test_invoke_partial_failure_then_resume(monkeypatch, tmp_path) -> None:
+    """真实断点重跑：个别条目持续 429 → 整体失败但成功判定已写存档，
+    解除限流后重跑只补判失败条目，最终产物与一次跑完一致。"""
+    fake = _FailOnTarget429(marker="无意义")
+    monkeypatch.setattr("nodes.llm_filter.urllib.request.urlopen", fake)
+    monkeypatch.setattr("nodes.llm_filter.time.sleep", lambda s: None)
+    out = tmp_path / "out"
+    first, _srt = _llm_invoke(_SRT, out)
+    assert first.status == "failed"
+    assert "429" in (first.error or "")
+    calls_first = len(fake.bodies)
+    # 成功判定的 3 条已写存档；持续 429 的"答：无意义杂项"（索引 2）不在存档。
+    from nodes.llm_filter import _load_partial
+
+    partial = _load_partial(out)
+    assert 2 not in partial
+    assert len(partial) == 3
+    # 解除限流后重跑：只补判失败条目（1 次调用），其余复用存档。
+    fake.enabled = False
+    second, _srt = _llm_invoke(_SRT, out)
+    assert second.status == "completed", second.error
+    assert len(fake.bodies) == calls_first + 1
+    # 输出：解除限流后全部判定为 dialogue → 4 条都保留（与"一次跑完"一致）。
+    out_text = Path(second.outputs["srt_uri"]).read_text(encoding="utf-8")
+    assert out_text.count("-->") == 4
+    assert "00:00:09,000" in out_text

tests/test_maintenance.py

@@ -120,6 +120,38 @@ def test_cleaner_removes_run_with_empty_steps_dir(tmp_path) -> None:
     assert not steps.exists()
 
 
+def test_cleaner_skips_batch_runs(tmp_path) -> None:
+    """批量处理运行（source=batch）跳过清理：绝不删除用户视频文件夹。
+
+    批量 run 的产物在用户视频旁的同名文件夹里，不在主存储目录下；若按普通
+    孤儿逻辑处理，_has_files 检查不到会误删记录，_remove_run 还会连带删除
+    input_uri 的父目录（用户的整个视频文件夹）。
+    """
+    db = _db(tmp_path)
+    video_folder = tmp_path / "my_videos"
+    video_folder.mkdir()
+    video = video_folder / "a.mp4"
+    video.write_bytes(b"real")
+    # 过期、COMPLETED、主存储无任何产物文件——普通任务会被清理的条件全满足。
+    db.upsert_workflow({"id": "flow", "name": "F", "published": 1, "latest_version": 1})
+    db.create_run(
+        {
+            "id": "run_batch",
+            "workflow_id": "flow",
+            "workflow_version": 1,
+            "status": "COMPLETED",
+            "progress": 1,
+            "input_uri": str(video),
+            "source": "batch",
+            "created_at": "2020-01-01T00:00:00+00:00",
+            "updated_at": "2020-01-01T00:00:00+00:00",
+        }
+    )
+    cleaner = OrphanCleaner(db, tmp_path / "storage", grace_seconds=3600)
+    assert cleaner.clean_once() == 0
+    assert db.get_run("run_batch") is not None
+    assert video_folder.exists()
+    assert video.exists()
 def test_cleaner_default_config_and_defensive_branches(tmp_path, monkeypatch) -> None:
     """验证默认配置构造、缺失/非法时间与缺失任务记录的防御分支。"""
     db = _db(tmp_path)

tests/test_nodes.py

@@ -309,7 +309,7 @@ def test_format_timestamp() -> None:
 
 
 def test_whisper_success(tmp_path, monkeypatch) -> None:
-    """验证成功转写会生成 SRT 并默认使用 auto 设备/计算类型。"""
+    """验证成功转写会生成 SRT 并默认使用 auto 设备、float16 计算类型。"""
     FakeWhisperModel.instances.clear()
     _install_fake_whisper(monkeypatch)
     _make_wav(tmp_path / "audio.wav", 5)
@@ -320,9 +320,39 @@ def test_whisper_success(tmp_path, monkeypatch) -> None:
     assert "01:00:00,500 --> 01:00:02,250" in content
     _, kwargs = FakeWhisperModel.instances[-1]
     assert kwargs["device"] == "auto"
-    assert kwargs["compute_type"] == "auto"
+    assert kwargs["compute_type"] == "float16"
 
 
+def test_whisper_pause_flag_stops_between_chunks(tmp_path, monkeypatch) -> None:
+    """验证 paused.flag 存在时 whisper 在分块边界中止（批量暂停机制）。
+
+    run 根目录（output_dir 的上上级）的 paused.flag 由暂停接口写入，whisper
+    在每个分块转写前检查；检测到即抛异常，invoke 统一转 failed 响应，调度器
+    捕获后保持任务 PAUSED。
+    """
+    _install_fake_whisper(monkeypatch)
+    _make_wav(tmp_path / "audio.wav", 5)
+    # 暂停信号位于 run 根目录：output_dir 为 runs/run_p/steps/whisper，
+    # 其上上级即 runs/run_p（与调度器/OCR 的目录约定一致）。
+    (tmp_path / "runs" / "run_p").mkdir(parents=True)
+    (tmp_path / "runs" / "run_p" / "paused.flag").write_text("", encoding="utf-8")
+    # 注入两个分块路径，确保进入分块循环并执行至少一次暂停检查。
+    monkeypatch.setattr(
+        "nodes.whisper._split_audio",
+        lambda *args, **kwargs: [tmp_path / "chunk_1.wav", tmp_path / "chunk_2.wav"],
+    )
+    response = whisper_invoke(
+        _whisper_request(
+            tmp_path,
+            params={"language": "ja", "chunk_seconds": 60},
+            output_dir=str(tmp_path / "runs" / "run_p" / "steps" / "whisper"),
+        )
+    )
+    assert response.status == "failed"
+    assert "被暂停" in response.error
+    # 暂停时不会写出 SRT 产物。
+    assert not (tmp_path / "runs" / "run_p" / "steps" / "whisper" / "transcript.srt").exists()
+
 def test_load_cuda_libraries_linux(monkeypatch) -> None:
     """验证 Linux 下进程内预加载 nvidia 动态库，含失败跳过分支。"""
     import ctypes

tests/test_ocr_flow.py

@@ -539,3 +539,29 @@ def test_ocr_interrupts_on_pause_flag(monkeypatch, tmp_path) -> None:
     assert "暂停" in (response.error or "")
     assert calls == []  # 一帧都没有真正 OCR。
     assert not (out_dir / "ocr_partial.jsonl").exists()  # 未处理帧不入存档。
+
+
+def test_format_eta() -> None:
+    """预计剩余时间的格式化：秒/分/小时三种量级与边界值。"""
+    from nodes.subtitle_ocr import _format_eta
+
+    assert _format_eta(0) == "0秒"
+    assert _format_eta(59.9) == "59秒"          # 不足 1 分只显示秒。
+    assert _format_eta(61) == "1分01秒"
+    assert _format_eta(2058.4) == "34分18秒"
+    assert _format_eta(3600) == "1小时00分"      # 1 小时整。
+    assert _format_eta(3725) == "1小时02分"      # 超过 1 小时只显示到分钟。
+    assert _format_eta(-5) == "0秒"             # 负数钳制为 0。
+
+
+def test_eta_suffix() -> None:
+    """进度日志的 ETA 后缀：速度为 0 不显示，速度正常时按剩余帧估算。"""
+    from nodes.subtitle_ocr import _eta_suffix
+
+    # 速率未知（0/负）→ 不显示 ETA。
+    assert _eta_suffix(done=100, total=100, rate=0) == ""
+    assert _eta_suffix(done=100, total=100, rate=-1) == ""
+    # 2673/22222 帧、9.5 帧/s：剩余 (22222-2673)/9.5 ≈ 2057.8s ≈ 34分。
+    assert _eta_suffix(done=2673, total=22222, rate=9.5) == ", 预计剩余 34分17秒"
+    # 全部完成时剩余 0 秒。
+    assert _eta_suffix(done=22222, total=22222, rate=9.5) == ", 预计剩余 0秒"

web/admin.html

@@ -13,6 +13,7 @@
       <nav>
         <a href="/">应用中心</a>
         <a href="/tasks.html">任务管理</a>
+        <a href="/batch.html">批量处理</a>
         <a href="/admin.html">管理后台</a>
         <a href="/workflow.html">工作流</a>
       </nav>

web/assets/batch.js

@@ -0,0 +1,345 @@
+// VRSub 批量处理页脚本：创建/列表/暂停/继续/删除批量任务，展开查看每视频明细。
+// 复用 app.js 的 api/escapeHtml/badge/formatTime/formatElapsed/progressBar 辅助函数。
+
+// 已发布工作流缓存（id → 名称），用于任务列表展示工作流名。
+const WORKFLOW_NAMES = new Map();
+
+// 已展开明细的任务 ID 集合（防止轮询刷新时折叠用户展开的行）。
+const expandedJobs = new Set();
+
+// 加载已发布工作流填充下拉框；同时缓存 id → 名称供列表展示。
+async function loadBatchWorkflowOptions() {
+  const select = document.getElementById("batchWorkflow");
+  if (!select) return;
+  const apps = await api("/api/apps");
+  select.innerHTML = apps.length
+    ? apps
+        .map((app) => `<option value="${escapeHtml(app.id)}">${escapeHtml(app.name)}</option>`)
+        .join("")
+    : '<option value="">暂无已发布工作流</option>';
+  apps.forEach((app) => WORKFLOW_NAMES.set(app.id, app.name));
+}
+
+// 创建批量任务：POST 文件夹路径与所选工作流，成功后清空输入并刷新列表。
+async function createBatchJob() {
+  const button = document.getElementById("batchStart");
+  const hint = document.getElementById("batchCreateHint");
+  const folder = document.getElementById("batchFolder").value.trim();
+  const workflowId = document.getElementById("batchWorkflow").value;
+  const recursive = document.getElementById("batchRecursive").checked;
+  if (!folder) {
+    hint.textContent = "请填写视频文件夹路径";
+    return;
+  }
+  if (!workflowId) {
+    hint.textContent = "请先在工作流页发布一个工作流";
+    return;
+  }
+  button.disabled = true;
+  hint.textContent = "正在扫描视频并创建任务…";
+  try {
+    const job = await api("/api/batch/jobs", {
+      method: "POST",
+      body: JSON.stringify({ folder, workflow_id: workflowId, recursive }),
+    });
+    hint.textContent = `已创建任务 ${job.id}，共 ${job.videos.length} 个视频`;
+    document.getElementById("batchFolder").value = "";
+  } catch (error) {
+    hint.textContent = `创建失败：${error.message}`;
+  } finally {
+    button.disabled = false;
+  }
+  await loadBatchJobs();
+}
+
+// 渲染单个任务的进度：已完成/跳过与失败计入不同颜色，总数为任务视频数。
+function batchProgress(job) {
+  const total = job.total || 0;
+  const done = job.done || 0;
+  const failed = job.failed || 0;
+  if (!total) return "待扫描";
+  const percent = Math.round(((done + failed) / total) * 100);
+  // 有失败视频时追加红色失败计数，避免把"部分失败"误看成全部完成。
+  const failedText = failed > 0 ? ` <span class="danger">，失败 ${failed}</span>` : "";
+  return `${progressBar(percent, job.status === "FAILED")} ${done}/${total} 完成 ${percent}%${failedText}`;
+}
+
+// 渲染任务的暂停/继续/删除/详情按钮。
+function batchActions(job) {
+  const id = encodeURIComponent(job.id);
+  const canPause = job.status === "RUNNING" || job.status === "QUEUED";
+  const canResume = job.status === "PAUSED";
+  return `
+    ${canPause ? `<button class="warn" data-batch-pause="${id}">暂停</button>` : ""}
+    ${canResume ? `<button class="warn" data-batch-resume="${id}">继续</button>` : ""}
+    <button class="danger" data-batch-delete="${id}">删除</button>
+    <button data-batch-detail="${id}">详情</button>
+  `;
+}
+
+// 加载批量任务列表并渲染；已展开的任务自动重新拉取明细（展开行内嵌）。
+async function loadBatchJobs() {
+  const tbody = document.getElementById("batchJobList");
+  if (!tbody) return;
+  const jobs = await api("/api/batch/jobs");
+  tbody.innerHTML = jobs.length
+    ? (
+        await Promise.all(
+          jobs.map(async (job) => {
+            const statusHtml =
+              job.status === "COMPLETED" && (job.failed || 0) > 0
+                ? `<span class="badge error">部分失败</span> <span class="muted">${job.failed} 个视频失败，展开详情查看</span>`
+                : job.status === "RUNNING"
+                  ? `${badge(job.status)} <span class="muted">处理中 ${formatElapsed(job.updated_at)}</span>`
+                  : job.status === "QUEUED"
+                    ? `${badge(job.status)} <span class="muted">排队中</span>`
+                    : job.status === "PAUSED"
+                      ? `${badge(job.status)} <span class="muted">已暂停 ${formatElapsed(job.updated_at)}</span>`
+                      : badge(job.status);
+            const current = job.current_video ? PathBase(job.current_video) : "-";
+            const row = `
+              <tr>
+                <td title="${escapeHtml(job.error || "")}">${escapeHtml(job.id)}</td>
+                <td title="${escapeHtml(job.folder_path)}">${escapeHtml(PathBase(job.folder_path))}</td>
+                <td>${escapeHtml(WORKFLOW_NAMES.get(job.workflow_id) || job.workflow_id)}</td>
+                <td>${statusHtml}</td>
+                <td>${batchProgress(job)}</td>
+                <td>${escapeHtml(current)}</td>
+                <td>${escapeHtml(formatTime(job.created_at))}</td>
+                <td>${batchActions(job)}</td>
+              </tr>`;
+            // 展开中的任务在下方追加明细行（每视频状态与产物下载）。
+            const detailRow = expandedJobs.has(job.id)
+              ? `<tr class="batch-detail-row"><td colspan="8">${await videoDetailHtml(job.id)}</td></tr>`
+              : "";
+            return row + detailRow;
+          }),
+        )
+      ).join("")
+    : '<tr><td colspan="8">暂无批量任务，请在上方创建。</td></tr>';
+}
+
+// 取路径的最后一段（兼容 Windows 反斜杠与 Unix 斜杠）。
+function PathBase(path) {
+  return String(path).split(/[\\/]/).pop() || path;
+}
+
+// 拉取任务详情并渲染视频明细表：文件、状态、错误与产物下载链接。
+async function videoDetailHtml(jobId) {
+  const job = await api(`/api/batch/jobs/${encodeURIComponent(jobId)}`);
+  const videos = job.videos || [];
+  if (!videos.length) return "（暂无视频）";
+  const rows = videos
+    .map((video) => {
+      const finals = video.finals || {};
+      const links = Object.keys(finals)
+        .map(
+          (alias) =>
+            `<a class="download-link" href="/api/batch/jobs/${encodeURIComponent(jobId)}/videos/${encodeURIComponent(video.id)}/download?alias=${encodeURIComponent(alias)}">${escapeHtml(alias)}</a>`,
+        )
+        .join(" ");
+      return `
+        <tr>
+          <td>${escapeHtml(PathBase(video.video_path))}</td>
+          <td>${badge(video.status)}</td>
+          <td title="${escapeHtml(video.error || "")}">${escapeHtml(video.error || "-")}</td>
+          <td>${links || "-"}</td>
+        </tr>`;
+    })
+    .join("");
+  return `<table class="inner-table"><thead><tr><th>视频</th><th>状态</th><th>错误</th><th>产物</th></tr></thead><tbody>${rows}</tbody></table>`;
+}
+
+// 暂停批量任务：当前 run 在分块/帧边界停下，后续视频不再开始。
+async function pauseBatchJob(jobId) {
+  try {
+    const result = await api(`/api/batch/jobs/${jobId}/pause`, { method: "POST" });
+    alert(`批量任务 ${result.id} 已暂停，继续后从未完成处接着处理`);
+  } catch (error) {
+    alert(`暂停失败：${error.message}`);
+  }
+  await loadBatchJobs();
+}
+
+// 继续批量任务：恢复排队，从上次断点（未完成视频）继续处理。
+async function resumeBatchJob(jobId) {
+  try {
+    const result = await api(`/api/batch/jobs/${jobId}/resume`, { method: "POST" });
+    alert(`批量任务 ${result.id} 已恢复执行`);
+  } catch (error) {
+    alert(`继续失败：${error.message}`);
+  }
+  await loadBatchJobs();
+}
+
+// 删除批量任务：只清理数据库记录，磁盘上的同名文件夹与产物保留。
+async function deleteBatchJob(jobId) {
+  if (!confirm("删除批量任务？磁盘上的产物文件会保留。")) return;
+  try {
+    await api(`/api/batch/jobs/${jobId}`, { method: "DELETE" });
+  } catch (error) {
+    alert(`删除失败：${error.message}`);
+  }
+  await loadBatchJobs();
+}
+
+// ---------------------------------------------------------------------------
+// 目录树选择器：点击"选择文件夹"按钮弹出，懒加载浏览本地目录后回填路径。
+// 浏览器拿不到文件夹绝对路径，目录列表由本地后端（/api/batch/roots、dirs）提供。
+// ---------------------------------------------------------------------------
+
+// 当前选中的目录绝对路径（未选中时为 null）。
+let dirPickerSelection = null;
+
+// 打开目录树选择器：显示模态框并加载可浏览根目录。
+async function openDirPicker() {
+  document.getElementById("dirPicker").classList.remove("hidden");
+  const tree = document.getElementById("dirTree");
+  tree.innerHTML = '<div class="muted">加载中…</div>';
+  dirPickerSelection = null;
+  document.getElementById("dirPickerUse").disabled = true;
+  document.getElementById("dirPickerSelected").textContent = "";
+  try {
+    const roots = await api("/api/batch/roots");
+    tree.innerHTML = "";
+    roots.forEach((root) => tree.appendChild(dirNode(root, 0)));
+  } catch (error) {
+    tree.innerHTML = `<div class="muted">加载失败：${escapeHtml(error.message)}</div>`;
+  }
+}
+
+// 关闭目录树选择器。
+function closeDirPicker() {
+  document.getElementById("dirPicker").classList.add("hidden");
+}
+
+// 渲染一个目录节点行：缩进 + 展开箭头 + 名称；点击行选中并切换展开。
+function dirNode(dir, depth) {
+  const row = document.createElement("div");
+  row.className = "dir-node";
+  row.dataset.path = dir.path;
+  row.dataset.expanded = "false";
+
+  const indent = document.createElement("span");
+  indent.className = "dir-indent";
+  indent.style.width = `${depth * 18}px`;
+  const arrow = document.createElement("span");
+  arrow.className = "dir-arrow";
+  arrow.textContent = "▸";
+  const name = document.createElement("span");
+  name.className = "dir-name";
+  name.textContent = dir.name;
+  name.title = dir.path;
+
+  row.append(indent, arrow, name);
+  row.addEventListener("click", (event) => {
+    event.stopPropagation();
+    selectDir(row);
+    // 首次点击展开子目录（懒加载），再次点击折叠。
+    if (row.dataset.expanded === "false") {
+      expandDir(row, depth);
+    } else {
+      collapseDir(row);
+    }
+  });
+  return row;
+}
+
+// 选中目录行：取消其它行高亮，更新底部展示与"使用此文件夹"按钮。
+function selectDir(row) {
+  document.querySelectorAll("#dirTree .dir-node.selected").forEach((item) => item.classList.remove("selected"));
+  row.classList.add("selected");
+  dirPickerSelection = row.dataset.path;
+  document.getElementById("dirPickerSelected").textContent = row.dataset.path;
+  document.getElementById("dirPickerUse").disabled = false;
+}
+
+// 展开目录：请求后端列出子目录，把子节点行插到当前行之后。
+async function expandDir(row, depth) {
+  const arrow = row.querySelector(".dir-arrow");
+  arrow.textContent = "…";
+  let children;
+  try {
+    const data = await api(`/api/batch/dirs?path=${encodeURIComponent(row.dataset.path)}`);
+    children = document.createElement("div");
+    children.className = "dir-children";
+    data.dirs.forEach((child) => children.appendChild(dirNode(child, depth + 1)));
+    if (!data.dirs.length) {
+      const empty = document.createElement("div");
+      empty.className = "dir-empty muted";
+      empty.textContent = "（无子目录）";
+      children.appendChild(empty);
+    }
+  } catch (error) {
+    children = document.createElement("div");
+    children.className = "dir-children";
+    const err = document.createElement("div");
+    err.className = "dir-empty muted";
+    err.textContent = `加载失败：${error.message}`;
+    children.appendChild(err);
+  }
+  row.dataset.expanded = "true";
+  arrow.textContent = "▾";
+  row.insertAdjacentElement("afterend", children);
+}
+
+// 折叠目录：移除已展开的子节点行。
+function collapseDir(row) {
+  row.dataset.expanded = "false";
+  row.querySelector(".dir-arrow").textContent = "▸";
+  const children = row.nextElementSibling;
+  if (children && children.classList.contains("dir-children")) {
+    children.remove();
+  }
+}
+
+// 使用当前选中目录：回填路径输入框并关闭选择器。
+function useDirPicker() {
+  if (!dirPickerSelection) return;
+  document.getElementById("batchFolder").value = dirPickerSelection;
+  closeDirPicker();
+}
+
+// 展开/折叠任务明细（事件委托处理列表内的按钮）。
+async function onBatchJobClick(event) {
+  const target = event.target;
+  const pause = target.dataset.batchPause;
+  const resume = target.dataset.batchResume;
+  const del = target.dataset.batchDelete;
+  const detail = target.dataset.batchDetail;
+  if (pause) return pauseBatchJob(pause);
+  if (resume) return resumeBatchJob(resume);
+  if (del) return deleteBatchJob(del);
+  if (detail) {
+    // 切换展开状态后刷新列表（展开行会拉取最新明细）。
+    if (expandedJobs.has(detail)) {
+      expandedJobs.delete(detail);
+    } else {
+      expandedJobs.add(detail);
+    }
+    await loadBatchJobs();
+  }
+}
+
+// 页面初始化：绑定事件并启动列表轮询（每 2 秒刷新一次进度）。
+document.addEventListener("DOMContentLoaded", async () => {
+  const startButton = document.getElementById("batchStart");
+  if (!startButton) return;
+  startButton.addEventListener("click", createBatchJob);
+  // 目录树选择器：打开/关闭/使用/点击遮罩关闭。
+  document.getElementById("batchBrowse").addEventListener("click", openDirPicker);
+  document.getElementById("dirPickerClose").addEventListener("click", closeDirPicker);
+  document.getElementById("dirPickerUse").addEventListener("click", useDirPicker);
+  document.getElementById("dirPicker").addEventListener("click", (event) => {
+    // 点击遮罩（面板外部）关闭选择器。
+    if (event.target.id === "dirPicker") closeDirPicker();
+  });
+  document.addEventListener("keydown", (event) => {
+    // Esc 关闭选择器。
+    if (event.key === "Escape") closeDirPicker();
+  });
+  document.getElementById("batchJobList").addEventListener("click", onBatchJobClick);
+  await loadBatchWorkflowOptions();
+  await loadBatchJobs();
+  setInterval(loadBatchJobs, 2000);
+});

web/assets/styles.css

@@ -346,3 +346,150 @@ th {
   pointer-events: auto;
   cursor: crosshair;
 }
+
+/* 批量处理页：明细行的内嵌表格样式。 */
+.batch-detail-row td {
+  background: #fafbfc;
+  padding: 8px 12px;
+}
+
+/* 内嵌表格：无边框、行距紧凑，嵌入批量任务明细行中。 */
+.inner-table {
+  width: 100%;
+  border-collapse: collapse;
+}
+
+.inner-table th {
+  text-align: left;
+  font-size: 12px;
+  color: #666;
+  padding: 4px 8px;
+  border-bottom: 1px solid #e5e7eb;
+}
+
+.inner-table td {
+  padding: 4px 8px;
+  font-size: 13px;
+  border-bottom: 1px solid #f0f1f3;
+}
+
+/* 批量页创建表单的复选框与标签同行展示。 */
+.form-row .inline {
+  margin-left: 16px;
+  font-weight: 400;
+}
+
+/* 批量页：路径输入框 + "选择文件夹"按钮同行。 */
+.path-picker {
+  display: flex;
+  gap: 8px;
+  flex: 1;
+}
+
+.path-picker input {
+  flex: 1;
+  background: #f7f8fa;
+}
+
+/* 目录树选择器：全屏遮罩 + 居中面板。 */
+.dir-picker {
+  position: fixed;
+  inset: 0;
+  background: rgba(0, 0, 0, 0.35);
+  z-index: 50;
+  display: flex;
+  align-items: center;
+  justify-content: center;
+}
+
+.dir-picker.hidden {
+  display: none;
+}
+
+.dir-picker-panel {
+  width: min(560px, 92vw);
+  max-height: 70vh;
+  display: flex;
+  flex-direction: column;
+  background: #fff;
+  border-radius: 10px;
+  box-shadow: 0 12px 40px rgba(0, 0, 0, 0.25);
+  overflow: hidden;
+}
+
+.dir-picker-head {
+  display: flex;
+  align-items: center;
+  justify-content: space-between;
+  padding: 12px 16px;
+  border-bottom: 1px solid #e5e7eb;
+}
+
+/* 关闭按钮（✕）：无边框文本按钮。 */
+.dir-picker-head .plain {
+  border: none;
+  background: none;
+  font-size: 16px;
+  cursor: pointer;
+  color: #666;
+}
+
+.dir-tree {
+  flex: 1;
+  overflow: auto;
+  padding: 8px 0;
+  min-height: 160px;
+}
+
+/* 目录节点行：悬停高亮，选中项加深。 */
+.dir-node {
+  display: flex;
+  align-items: center;
+  padding: 6px 12px;
+  cursor: pointer;
+  font-size: 14px;
+}
+
+.dir-node:hover {
+  background: #f3f6fa;
+}
+
+.dir-node.selected {
+  background: #e3edf7;
+  font-weight: 600;
+}
+
+.dir-arrow {
+  width: 20px;
+  color: #888;
+  user-select: none;
+}
+
+.dir-name {
+  margin-left: 4px;
+  white-space: nowrap;
+  overflow: hidden;
+  text-overflow: ellipsis;
+}
+
+.dir-empty {
+  padding: 4px 12px 4px 56px;
+  font-size: 13px;
+}
+
+.dir-picker-foot {
+  display: flex;
+  align-items: center;
+  justify-content: space-between;
+  gap: 12px;
+  padding: 12px 16px;
+  border-top: 1px solid #e5e7eb;
+}
+
+.dir-picker-foot .muted {
+  flex: 1;
+  font-size: 13px;
+  white-space: nowrap;
+  overflow: hidden;
+  text-overflow: ellipsis;
+}

web/batch.html

@@ -0,0 +1,91 @@
+<!doctype html>
+<!-- VRSub 批量处理页：输入文件夹路径批量处理全部视频，不上传副本。 -->
+<html lang="zh-CN">
+  <head>
+    <meta charset="utf-8" />
+    <meta name="viewport" content="width=device-width, initial-scale=1" />
+    <title>VRSub - 批量处理</title>
+    <link rel="stylesheet" href="/assets/styles.css?v=3" />
+  </head>
+  <body>
+    <header class="topbar">
+      <a class="brand" href="/">VRSub</a>
+      <nav>
+        <a href="/">发起任务</a>
+        <a href="/tasks.html">任务管理</a>
+        <a href="/batch.html">批量处理</a>
+        <a href="/admin.html">管理后台</a>
+        <a href="/workflow.html">工作流</a>
+      </nav>
+    </header>
+
+    <main class="container">
+      <h1>批量处理</h1>
+      <p class="muted">
+        直接读取所选文件夹下的全部视频（<b>不上传副本</b>），逐个执行所选流水线；
+        每个视频的中间态数据与最终产物存放在视频旁边的同名文件夹（如 movie.mp4 →
+        movie/）。已处理过的视频自动跳过；暂停后重新开始时，从未完成的视频继续。
+      </p>
+
+      <!-- 创建批量任务：文件夹路径 + 工作流选择 + 是否递归。 -->
+      <section class="panel">
+        <h2>创建批量任务</h2>
+        <div class="form-row">
+          <label for="batchFolder">视频文件夹路径</label>
+          <div class="path-picker">
+            <input id="batchFolder" type="text" readonly placeholder="点击右侧按钮选择文件夹" />
+            <button id="batchBrowse" type="button">选择文件夹…</button>
+          </div>
+        </div>
+        <div class="form-row">
+          <label for="batchWorkflow">处理流水线</label>
+          <select id="batchWorkflow"></select>
+          <label class="inline"><input id="batchRecursive" type="checkbox" checked /> 包含子文件夹</label>
+        </div>
+        <div class="form-row">
+          <button id="batchStart" class="primary">开始批量处理</button>
+          <span id="batchCreateHint" class="muted"></span>
+        </div>
+      </section>
+
+      <!-- 目录树选择器：点击"选择文件夹"按钮弹出，懒加载浏览本地目录。 -->
+      <div id="dirPicker" class="dir-picker hidden">
+        <div class="dir-picker-panel">
+          <div class="dir-picker-head">
+            <b>选择视频文件夹</b>
+            <button id="dirPickerClose" type="button" class="plain">✕</button>
+          </div>
+          <div id="dirTree" class="dir-tree"></div>
+          <div class="dir-picker-foot">
+            <span id="dirPickerSelected" class="muted"></span>
+            <button id="dirPickerUse" type="button" disabled>使用此文件夹</button>
+          </div>
+        </div>
+      </div>
+
+      <!-- 批量任务列表：状态、进度与暂停/继续操作，可展开查看每视频明细。 -->
+      <section class="panel">
+        <div class="table-wrap">
+          <table>
+            <thead>
+              <tr>
+                <th>任务 ID</th>
+                <th>文件夹</th>
+                <th>工作流</th>
+                <th>状态</th>
+                <th>进度</th>
+                <th>当前视频</th>
+                <th>创建时间</th>
+                <th>操作</th>
+              </tr>
+            </thead>
+            <tbody id="batchJobList"></tbody>
+          </table>
+        </div>
+      </section>
+    </main>
+
+    <script src="/assets/app.js?v=3"></script>
+    <script src="/assets/batch.js?v=1"></script>
+  </body>
+</html>

web/index.html

@@ -15,6 +15,7 @@
       <nav>
         <a href="/">发起任务</a>
         <a href="/tasks.html">任务管理</a>
+        <a href="/batch.html">批量处理</a>
         <a href="/admin.html">管理后台</a>
         <a href="/workflow.html">工作流</a>
       </nav>

web/tasks.html

@@ -13,6 +13,7 @@
       <nav>
         <a href="/">发起任务</a>
         <a href="/tasks.html">任务管理</a>
+        <a href="/batch.html">批量处理</a>
         <a href="/admin.html">管理后台</a>
         <a href="/workflow.html">工作流</a>
       </nav>

web/workflow.html

@@ -13,6 +13,7 @@
       <nav>
         <a href="/">应用中心</a>
         <a href="/tasks.html">任务管理</a>
+        <a href="/batch.html">批量处理</a>
         <a href="/admin.html">管理后台</a>
         <a href="/workflow.html">工作流</a>
       </nav>

workflows/ocr-subtitle.json

@@ -14,9 +14,9 @@
           "interval_seconds": 0.5,
           "crop": [
             0,
-            0.82,
+            0.75,
             1,
-            0.18
+            0.25
           ]
         },
         "inputs": {