feat: 文件夹批量处理引擎（后端）

- BatchWorker 单线程轮询 batch_jobs 表，处理 source=batch 的运行，
  与主调度器互不抢占（next_queued_run 排除 batch 来源）
- 直接读取用户所选文件夹下的视频逐个执行流水线，不上传到工作目录；
  中间态与产物落在视频旁同名文件夹，batch.done.json 完成标记去重
- 支持暂停/继续、失败容错（单视频失败不阻塞后续）、删除任务只清库
- 孤儿清理跳过 source=batch 运行，防止误删用户视频文件夹
- workflow_runs 新增 source 列（upload/batch），旧库自动迁移

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_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_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)