diff --git a/src/wov_app/batch.py b/src/wov_app/batch.py new file mode 100644 index 0000000..5d000bf --- /dev/null +++ b/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 根目录（/runs//）。 +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，中间态落在 + /runs//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()) diff --git a/src/wov_app/config.py b/src/wov_app/config.py index 80cda51..b2fea8a 100644 --- a/src/wov_app/config.py +++ b/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" # 清理扫描周期（秒），默认每小时一次。 diff --git a/src/wov_app/db.py b/src/wov_app/db.py index 1f96a74..c44daee 100644 --- a/src/wov_app/db.py +++ b/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) diff --git a/src/wov_app/main.py b/src/wov_app/main.py index caf1d1b..7c3cb87 100644 --- a/src/wov_app/main.py +++ b/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") diff --git a/src/wov_app/maintenance.py b/src/wov_app/maintenance.py index d4cc00c..0c444dc 100644 --- a/src/wov_app/maintenance.py +++ b/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")): diff --git a/src/wov_app/routers/batch.py b/src/wov_app/routers/batch.py new file mode 100644 index 0000000..34f72fd --- /dev/null +++ b/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} diff --git a/src/wov_app/schemas.py b/src/wov_app/schemas.py index b15bc86..84298b6 100644 --- a/src/wov_app/schemas.py +++ b/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 diff --git a/tests/test_batch.py b/tests/test_batch.py new file mode 100644 index 0000000..9c56aa7 --- /dev/null +++ b/tests/test_batch.py @@ -0,0 +1,1003 @@ +"""文件夹批量处理引擎与 API 测试。 + +覆盖：视频扫描/同名文件夹推导、完成标记读写、批量任务创建校验、 +引擎逐视频执行（建 run、复用现有调度器、复制最终产物、跳过已处理）、 +暂停/继续断点续跑、失败视频不中断、孤儿清理不删批量运行、 +批量 API 全端点（创建/列表/详情/暂停/继续/删除/下载）与 404/422 分支。 +""" + +import json +import time +from datetime import datetime, timezone +from pathlib import Path + +import pytest +from fastapi.testclient import TestClient + +from wov_app import batch as batch_engine +from wov_app.batch import ( + MARKER_NAME, + PAUSE_FLAG, + BatchWorker, + create_job, + load_marker, + scan_videos, + work_dir_for, +) +from wov_app.db import Database +from wov_app.main import app + + +def _now_iso() -> str: + """返回当前 UTC 时间的 ISO 格式字符串。""" + return datetime.now(timezone.utc).isoformat() + + +def _db(tmp_path) -> Database: + """在临时目录创建独立数据库。""" + return Database(tmp_path / "wov.db") + + +def _seed_echo_workflow(db: Database, workflow_id: str = "echo-app", published: bool = True) -> None: + """创建引用 echo 节点的单节点工作流（真实数据流：输入文件复制为产物）。 + + 幂等：先删除同 ID 的旧工作流（含版本与任务），再重新创建。 + """ + definition = { + "name": "echo-flow", + "version": 1, + "nodes": [ + {"id": "step", "node_type": "echo", "inputs": {"file_uri": "input.video_uri"}} + ], + "edges": [], + "entry_inputs": {"video_uri": "file"}, + "final_outputs": {"result": "step.file_uri"}, + } + # 批量测试会真实执行 echo 节点，注册表由 conftest 每测试隔离，需显式注册。 + _register_echo() + if db.get_workflow(workflow_id) is not None: + db.delete_workflow(workflow_id) + db.upsert_workflow( + { + "id": workflow_id, + "name": "Echo", + "description": "", + "published": 1 if published else 0, + "latest_version": 1, + } + ) + db.create_workflow_version(workflow_id, 1, definition) + + +def _register_echo() -> None: + """把内置 echo 节点注册到进程内注册表（conftest 每个测试隔离注册表）。""" + from wov_app import registry + from wov_sdk.models import NodeManifest + + root = Path(__file__).resolve().parent.parent + from nodes.echo import invoke + + registry.register(NodeManifest.load(str(root / "manifests" / "echo.json")), invoke) + +def _video_folder(tmp_path, names=("a.mp4", "b.mp4")) -> Path: + """创建含视频文件的文件夹：内容为真实文本（echo 节点按文本读入）。""" + folder = tmp_path / "videos" + folder.mkdir() + for index, name in enumerate(names, start=1): + (folder / name).write_text(f"视频 {name} 的测试内容 {index}\n", encoding="utf-8") + return folder + + +def _make_job(db: Database, folder: Path, workflow_id: str = "echo-app", recursive: bool = True) -> dict: + """通过 create_job 创建批量任务并返回任务记录。""" + job_id = create_job(db, str(folder), workflow_id, recursive) + return db.get_batch_job(job_id) + + +# --------------------------------------------------------------------------- +# 扫描 / 目录推导 / 完成标记 +# --------------------------------------------------------------------------- + + +def test_scan_videos_recursive_and_work_dir(tmp_path) -> None: + """递归/非递归扫描只返回视频文件，同名文件夹去掉扩展名。""" + folder = tmp_path / "media" + (folder / "sub").mkdir(parents=True) + (folder / "a.mp4").write_text("a", encoding="utf-8") + (folder / "b.MKV").write_text("b", encoding="utf-8") + (folder / "readme.txt").write_text("c", encoding="utf-8") + (folder / "sub" / "c.avi").write_text("d", encoding="utf-8") + (folder / "sub" / "notes.md").write_text("e", encoding="utf-8") + + recursive = scan_videos(folder, recursive=True) + assert [p.name for p in recursive] == ["a.mp4", "b.MKV", "c.avi"] + flat = scan_videos(folder, recursive=False) + assert [p.name for p in flat] == ["a.mp4", "b.MKV"] + # 同名文件夹：去掉扩展名，位于视频所在目录。 + assert work_dir_for(folder / "sub" / "c.avi") == folder / "sub" / "c" + + +def test_load_marker_variants(tmp_path) -> None: + """完成标记缺失/损坏/非字典时返回 None。""" + work = tmp_path / "movie" + work.mkdir() + assert load_marker(work) is None + (work / MARKER_NAME).write_text("{broken json", encoding="utf-8") + assert load_marker(work) is None + (work / MARKER_NAME).write_text("[1,2]", encoding="utf-8") + assert load_marker(work) is None + (work / MARKER_NAME).write_text('{"workflow_id": "w", "finals": {"r": "m.txt"}}', encoding="utf-8") + marker = load_marker(work) + assert marker["workflow_id"] == "w" + + +# --------------------------------------------------------------------------- +# create_job 校验 +# --------------------------------------------------------------------------- + + +def test_create_job_validation_errors(tmp_path) -> None: + """文件夹不存在、未发布工作流、无版本、无视频都拒绝创建。""" + db = _db(tmp_path) + folder = _video_folder(tmp_path) + with pytest.raises(ValueError, match="folder not found"): + create_job(db, str(tmp_path / "missing"), "echo-app") + with pytest.raises(ValueError, match="published workflow not found"): + create_job(db, str(folder), "ghost-flow") + _seed_echo_workflow(db, published=False) + with pytest.raises(ValueError, match="published workflow not found"): + create_job(db, str(folder), "echo-app") + # 有版本但文件夹里没有视频。 + _seed_echo_workflow(db, published=True) + empty = tmp_path / "empty" + empty.mkdir() + with pytest.raises(ValueError, match="no videos found in folder"): + create_job(db, str(empty), "echo-app") + # 已发布但没有版本的工作流。 + db.delete_workflow("echo-app") + db.upsert_workflow({"id": "echo-app", "name": "E", "description": "", "published": 1, "latest_version": 0}) + with pytest.raises(ValueError, match="workflow has no version"): + create_job(db, str(folder), "echo-app") + + +def test_create_job_success_creates_rows(tmp_path) -> None: + """创建成功：任务入队、视频明细齐全、递归标志与文件夹路径落库。""" + db = _db(tmp_path) + _seed_echo_workflow(db) + (tmp_path / "videos" / "sub").mkdir(parents=True) + (tmp_path / "videos" / "a.mp4").write_text("a", encoding="utf-8") + (tmp_path / "videos" / "sub" / "b.mkv").write_text("b", encoding="utf-8") + job = _make_job(db, tmp_path / "videos", recursive=True) + assert job["status"] == "QUEUED" + videos = db.list_batch_videos(job["id"]) + assert {v["video_path"] for v in videos} == { + str(tmp_path / "videos" / "a.mp4"), + str(tmp_path / "videos" / "sub" / "b.mkv"), + } + assert all(v["status"] == "PENDING" for v in videos) + assert db.next_queued_batch_job()["id"] == job["id"] + + +# --------------------------------------------------------------------------- +# 引擎：完整处理 / 跳过 / 失败 / 暂停续跑 +# --------------------------------------------------------------------------- + + +def test_batch_worker_processes_all_videos(tmp_path) -> None: + """批量引擎逐个处理视频：建 source=batch 的 run、中间态进同名文件夹、 + 最终产物复制到同名文件夹根目录并写完成标记，任务最终 COMPLETED。""" + db = _db(tmp_path) + _seed_echo_workflow(db) + folder = _video_folder(tmp_path) + job = _make_job(db, folder) + worker = BatchWorker(db, interval_seconds=0.05) + worker._process_job(job) + + job = db.get_batch_job(job["id"]) + assert job["status"] == "COMPLETED" + assert job["done"] == 2 and job["failed"] == 0 + for name in ("a", "b"): + work = folder / name + marker = load_marker(work) + assert marker is not None and marker["workflow_id"] == "echo-app" + final_name = marker["finals"]["result"] + final = work / final_name + # 最终产物复制到同名文件夹根目录，内容与输入一致（真实数据流）。 + assert final.is_file() + assert final.read_text(encoding="utf-8") == (folder / f"{name}.mp4").read_text(encoding="utf-8") + # 中间态落在 /runs//steps/ 下。 + # 中间态落在 /runs//steps/ 下（收尾时产物已重命名）。 + assert list(work.glob("runs/*/steps/step/*.txt")) + # run 记录为 batch 来源，且主调度器不会抢占。 + runs = [db.get_run(v["run_id"]) for v in db.list_batch_videos(job["id"])] + assert all(run["source"] == "batch" and run["status"] == "COMPLETED" for run in runs) + assert db.next_queued_run() is None + + +def test_batch_worker_skips_already_done_videos(tmp_path) -> None: + """已有同工作流完成标记且产物齐全的视频直接跳过（不再处理）。""" + db = _db(tmp_path) + _seed_echo_workflow(db) + folder = _video_folder(tmp_path, names=("a.mp4", "b.mp4")) + work_a = folder / "a" + work_a.mkdir() + # 伪造 a.mp4 的完成标记与最终产物（模拟上次任务已处理完）。 + marker = {"workflow_id": "echo-app", "workflow_version": 1, "run_id": "run_old", "finals": {"result": "a.result.txt"}} + (work_a / MARKER_NAME).write_text(json.dumps(marker), encoding="utf-8") + (work_a / "a.result.txt").write_text("已处理", encoding="utf-8") + + job = _make_job(db, folder) + BatchWorker(db, interval_seconds=0.05)._process_job(job) + videos = {Path(v["video_path"]).name: v for v in db.list_batch_videos(job["id"])} + assert videos["a.mp4"]["status"] == "SKIPPED" + assert videos["a.mp4"]["run_id"] is None + assert videos["b.mp4"]["status"] == "COMPLETED" + assert db.get_batch_job(job["id"])["done"] == 2 + + +def test_batch_worker_marker_workflow_mismatch_reprocesses(tmp_path) -> None: + """不同工作流的完成标记不互相误判：换工作流后视频重新处理。""" + db = _db(tmp_path) + _seed_echo_workflow(db, workflow_id="echo-app") + _seed_echo_workflow(db, workflow_id="echo-other") + folder = _video_folder(tmp_path, names=("a.mp4",)) + work_a = folder / "a" + work_a.mkdir() + marker = {"workflow_id": "echo-other", "finals": {"result": "x.txt"}} + (work_a / MARKER_NAME).write_text(json.dumps(marker), encoding="utf-8") + (work_a / "x.txt").write_text("x", encoding="utf-8") + + job = _make_job(db, folder, workflow_id="echo-app") + BatchWorker(db, interval_seconds=0.05)._process_job(job) + video = db.list_batch_videos(job["id"])[0] + # 标记工作流不匹配 → 重新处理并覆盖为新工作流的标记。 + assert video["status"] == "COMPLETED" + new_marker = load_marker(work_a) + assert new_marker["workflow_id"] == "echo-app" + + +def test_batch_worker_failed_video_continues(tmp_path) -> None: + """缺失视频文件与失败 run 都记为 FAILED，任务继续处理后续视频。""" + db = _db(tmp_path) + _seed_echo_workflow(db) + folder = _video_folder(tmp_path, names=("gone.mp4", "b.mp4")) + job = _make_job(db, folder) + # 任务创建后、处理前删除第一个视频（模拟外部移除），第二个正常处理。 + (folder / "gone.mp4").unlink() + BatchWorker(db, interval_seconds=0.05)._process_job(job) + + videos = {Path(v["video_path"]).name: v for v in db.list_batch_videos(job["id"])} + assert videos["gone.mp4"]["status"] == "FAILED" + assert videos["gone.mp4"]["error"] == "video file not found" + assert videos["b.mp4"]["status"] == "COMPLETED" + job = db.get_batch_job(job["id"]) + assert job["status"] == "COMPLETED" + assert job["done"] == 1 and job["failed"] == 1 + + +class _PausingScheduler: + """把 execute_run 模拟为"被暂停"的假调度器：置 run 为 PAUSED。""" + + def __init__(self, db: Database, work_dir: Path) -> None: + self.db = db + self.work_dir = work_dir + + def execute_run(self, run_id: str) -> None: + self.db.update_run(run_id, status="PAUSED", updated_at=_now_iso()) + + +def test_batch_worker_pause_then_resume_continues(tmp_path, monkeypatch) -> None: + """暂停后重新开始：PAUSED 视频从断点续跑，未开始的视频接着处理。""" + db = _db(tmp_path) + _seed_echo_workflow(db) + folder = _video_folder(tmp_path, names=("a.mp4", "b.mp4")) + job = _make_job(db, folder) + worker = BatchWorker(db, interval_seconds=0.05) + + # 第一次执行：第一个视频处理中被暂停（假调度器把 run 置为 PAUSED）。 + monkeypatch.setattr("wov_app.batch.WorkflowScheduler", _PausingScheduler) + worker._process_job(job) + videos = {Path(v["video_path"]).name: v for v in db.list_batch_videos(job["id"])} + assert videos["a.mp4"]["status"] == "PAUSED" + assert videos["b.mp4"]["status"] == "PENDING" + assert db.get_batch_job(job["id"])["status"] == "PAUSED" + + # 恢复真实调度器并继续：a 从断点完成，b 接着处理，任务 COMPLETED。 + monkeypatch.undo() + worker.resume_job(job["id"]) + worker._process_job(db.get_batch_job(job["id"])) + videos = {Path(v["video_path"]).name: v for v in db.list_batch_videos(job["id"])} + assert videos["a.mp4"]["status"] == "COMPLETED" + assert videos["b.mp4"]["status"] == "COMPLETED" + assert db.get_batch_job(job["id"])["status"] == "COMPLETED" + + # 再次处理（已完成视频在循环里直接 continue）：结果不变，幂等。 + worker._process_job(db.get_batch_job(job["id"])) + videos = {Path(v["video_path"]).name: v for v in db.list_batch_videos(job["id"])} + assert videos["a.mp4"]["status"] == "COMPLETED" + assert videos["b.mp4"]["status"] == "COMPLETED" + assert db.get_batch_job(job["id"])["done"] == 2 + + +def test_batch_worker_paused_job_does_not_start_new_video(tmp_path, monkeypatch) -> None: + """任务在视频之间被暂停：后续视频不开始，不创建 run。""" + db = _db(tmp_path) + _seed_echo_workflow(db) + folder = _video_folder(tmp_path, names=("a.mp4", "b.mp4")) + job = _make_job(db, folder) + job_id = job["id"] + from wov_app.scheduler import WorkflowScheduler + + class _PauseAfterFirstScheduler: + """真实执行第一个视频后把批量任务置为 PAUSED（模拟用户处理中暂停）。""" + + def __init__(self, db: Database, work_dir: Path) -> None: + self.db = db + self.work_dir = work_dir + + def execute_run(self, run_id: str) -> None: + WorkflowScheduler(self.db, self.work_dir).execute_run(run_id) + self.db.update_batch_job(job_id, status="PAUSED", updated_at=_now_iso()) + + monkeypatch.setattr("wov_app.batch.WorkflowScheduler", _PauseAfterFirstScheduler) + BatchWorker(db, interval_seconds=0.05)._process_job(db.get_batch_job(job_id)) + videos = {Path(v["video_path"]).name: v for v in db.list_batch_videos(job_id)} + # 第一个视频真实完成；第二个视频在开始前因任务已暂停而不处理。 + assert videos["a.mp4"]["status"] == "COMPLETED" + assert videos["b.mp4"]["status"] == "PENDING" + assert videos["b.mp4"]["run_id"] is None + assert db.get_batch_job(job_id)["status"] == "PAUSED" + +class _FailingScheduler: + """把 execute_run 模拟为"节点失败"的假调度器。""" + + def __init__(self, db: Database, work_dir: Path) -> None: + self.db = db + + def execute_run(self, run_id: str) -> None: + self.db.update_run(run_id, status="FAILED", error="node boom", updated_at=_now_iso()) + + +def test_batch_worker_run_failed_marks_video_failed(tmp_path, monkeypatch) -> None: + """节点执行失败：run FAILED → 视频 FAILED 带错误信息，任务继续。""" + db = _db(tmp_path) + _seed_echo_workflow(db) + folder = _video_folder(tmp_path, names=("a.mp4", "b.mp4")) + job = _make_job(db, folder) + monkeypatch.setattr("wov_app.batch.WorkflowScheduler", _FailingScheduler) + BatchWorker(db, interval_seconds=0.05)._process_job(job) + videos = {Path(v["video_path"]).name: v for v in db.list_batch_videos(job["id"])} + assert videos["a.mp4"]["status"] == "FAILED" + assert videos["a.mp4"]["error"] == "node boom" + assert videos["b.mp4"]["status"] == "FAILED" + assert db.get_batch_job(job["id"])["status"] == "COMPLETED" + assert db.get_batch_job(job["id"])["failed"] == 2 + + +def test_batch_worker_resumes_failed_and_running_runs(tmp_path) -> None: + """已失败的 run 重跑、上次进程残留的 RUNNING run 恢复后继续。""" + db = _db(tmp_path) + _seed_echo_workflow(db) + folder = _video_folder(tmp_path, names=("a.mp4", "b.mp4")) + job = _make_job(db, folder) + now = _now_iso() + # 手动构造 run：a 失败（重跑）、b 残留 RUNNING（进程被杀后恢复）。 + for name, status in (("a", "FAILED"), ("b", "RUNNING")): + run_id = f"run_{name}" + db.create_run( + { + "id": run_id, + "workflow_id": "echo-app", + "workflow_version": 1, + "status": status, + "progress": 0, + "input_uri": str(folder / f"{name}.mp4"), + "param_overrides": None, + "source": "batch", + "created_at": now, + "updated_at": now, + } + ) + video = next(v for v in db.list_batch_videos(job["id"]) if v["video_path"].endswith(f"{name}.mp4")) + db.update_batch_video(video["id"], run_id=run_id, updated_at=now) + + BatchWorker(db, interval_seconds=0.05)._process_job(db.get_batch_job(job["id"])) + videos = {Path(v["video_path"]).name: v for v in db.list_batch_videos(job["id"])} + assert videos["a.mp4"]["status"] == "COMPLETED" + assert videos["b.mp4"]["status"] == "COMPLETED" + + +def test_batch_worker_retry_failed_keeps_completed_node_artifacts(tmp_path) -> None: + """批量重试 FAILED 视频**保留已完成节点产物**：只重跑失败节点。 + + 回归：此前 FAILED 走 reset_run 清空全部产物记录，重跑时 extract/ocr 等 + 长耗时节点从头重做（run_e2b74e89e232 的 22222 帧 OCR 被白白丢弃）。 + 现改为保留产物恢复 QUEUED，execute_run 从产物表跳过已完成节点。 + """ + db = _db(tmp_path) + _register_echo() + # 双节点串联工作流：step1 成功、step2 失败的场景下验证只重跑 step2。 + definition = { + "name": "two-step", + "version": 1, + "nodes": [ + {"id": "s1", "node_type": "echo", "inputs": {"file_uri": "input.video_uri"}}, + {"id": "s2", "node_type": "echo", "inputs": {"file_uri": "s1.file_uri"}}, + ], + "edges": [{"from": "s1", "to": "s2"}], + "entry_inputs": {"video_uri": "file"}, + "final_outputs": {"result": "s2.file_uri"}, + } + db.upsert_workflow({"id": "echo-app", "name": "Echo", "description": "", "published": 1, "latest_version": 1}) + db.create_workflow_version("echo-app", 1, definition) + folder = _video_folder(tmp_path, names=("a.mp4",)) + job = _make_job(db, folder) + now = _now_iso() + run_id = "run_retry" + db.create_run( + { + "id": run_id, + "workflow_id": "echo-app", + "workflow_version": 1, + "status": "FAILED", + "progress": 0.5, + "error": "节点2炸了", + "input_uri": str(folder / "a.mp4"), + "param_overrides": None, + "source": "batch", + "created_at": now, + "updated_at": now, + } + ) + # 模拟 step1 已完成并登记产物（step2 失败时 step1 的成果）。 + s1_out = folder / "a" / "runs" / run_id / "steps" / "s1" / "echo.txt" + s1_out.parent.mkdir(parents=True) + s1_out.write_text("step1 产物", encoding="utf-8") + db.create_artifact( + { + "run_id": run_id, + "node_id": "s1", + "name": "s1.file_uri", + "uri": str(s1_out), + "mime_type": "text/plain", + "size": 10, + } + ) + video = db.list_batch_videos(job["id"])[0] + db.update_batch_video(video["id"], run_id=run_id, updated_at=now) + + BatchWorker(db, interval_seconds=0.05)._process_job(db.get_batch_job(job["id"])) + video = db.list_batch_videos(job["id"])[0] + assert video["status"] == "COMPLETED" + run = db.get_run(run_id) + assert run["status"] == "COMPLETED" and run["error"] is None + # step1 产物记录保留且未被重写（节点没有重新执行）。 + artifacts = {a["name"]: a for a in db.list_artifacts(run_id)} + assert artifacts["s1.file_uri"]["uri"] == str(s1_out) + # step2 本次补做完成；收尾时最终产物被重命名为 <片名>.result.<时间戳>.txt。 + assert "s2.file_uri" in artifacts + assert list((folder / "a" / "runs" / run_id / "steps" / "s2").glob("*")) + finals = {a["name"]: a for a in db.list_artifacts(run_id)} + assert "result" in finals + assert Path(finals["result"]["uri"]).is_file() + +def test_batch_worker_already_completed_run_copies_finals(tmp_path) -> None: + """run 已完成但视频未标记（收尾前中断）：直接复制产物并标记完成。 + + 覆盖 _copy_finals 的三个分支：产物齐全（复制）、产物记录存在但文件丢失 + （跳过）、无产物记录（跳过）——完成后均写出完成标记。 + """ + db = _db(tmp_path) + _seed_echo_workflow(db) + folder = _video_folder(tmp_path, names=("a.mp4", "b.mp4", "c.mp4")) + job = _make_job(db, folder) + now = _now_iso() + videos = {Path(v["video_path"]).stem: v for v in db.list_batch_videos(job["id"])} + + def complete_run(run_id: str, name: str, with_file: bool, with_artifact: bool) -> None: + """创建 COMPLETED run；可选产物文件与产物记录。""" + db.create_run( + { + "id": run_id, + "workflow_id": "echo-app", + "workflow_version": 1, + "status": "COMPLETED", + "progress": 1, + "input_uri": str(folder / f"{name}.mp4"), + "param_overrides": None, + "source": "batch", + "created_at": now, + "updated_at": now, + } + ) + if with_artifact: + db.create_artifact( + { + "run_id": run_id, + "node_id": "step", + "name": "result", + "uri": str(folder / name / "runs" / run_id / "steps" / "step" / f"{name}.result.txt"), + "mime_type": "text/plain", + "size": 6, + } + ) + if with_file: + path = folder / name / "runs" / run_id / "steps" / "step" / f"{name}.result.txt" + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text("产物", encoding="utf-8") + db.update_batch_video(videos[name]["id"], run_id=run_id, updated_at=now) + + # a：产物齐全；b：产物记录存在但文件丢失；c：没有任何产物记录。 + complete_run("run_done_a", "a", with_file=True, with_artifact=True) + complete_run("run_done_b", "b", with_file=False, with_artifact=True) + complete_run("run_done_c", "c", with_file=False, with_artifact=False) + + BatchWorker(db, interval_seconds=0.05)._process_job(db.get_batch_job(job["id"])) + videos = {Path(v["video_path"]).stem: v for v in db.list_batch_videos(job["id"])} + assert all(videos[name]["status"] == "COMPLETED" for name in ("a", "b", "c")) + # a 的产物被复制到同名文件夹根目录；b/c 无产物可复制，标记为空。 + marker_a = load_marker(folder / "a") + assert marker_a["finals"]["result"] == "a.result.txt" + assert (folder / "a" / "a.result.txt").is_file() + assert load_marker(folder / "b")["finals"] == {} + assert load_marker(folder / "c")["finals"] == {} + + +# --------------------------------------------------------------------------- +# 引擎：任务级异常与校验失败 +# --------------------------------------------------------------------------- + + +def test_batch_worker_job_validation_failures(tmp_path) -> None: + """文件夹缺失/工作流未发布/无版本时任务置为 FAILED 并记录错误。""" + db = _db(tmp_path) + now = _now_iso() + + def add_job(job_id, folder, workflow_id="echo-app"): + db.create_batch_job( + { + "id": job_id, "folder_path": folder, "workflow_id": workflow_id, + "recursive": 1, "status": "QUEUED", "progress": 0, "total": 0, + "done": 0, "failed": 0, "error": None, + "created_at": now, "updated_at": now, + } + ) + + worker = BatchWorker(db, interval_seconds=0.05) + add_job("job_nofolder", str(tmp_path / "missing")) + worker._process_job(db.get_batch_job("job_nofolder")) + assert db.get_batch_job("job_nofolder")["status"] == "FAILED" + assert "folder not found" in db.get_batch_job("job_nofolder")["error"] + + folder = _video_folder(tmp_path) + _seed_echo_workflow(db, published=False) + add_job("job_unpublished", str(folder)) + worker._process_job(db.get_batch_job("job_unpublished")) + assert db.get_batch_job("job_unpublished")["status"] == "FAILED" + assert "not found or unpublished" in db.get_batch_job("job_unpublished")["error"] + + # 已发布但没有任何版本。 + _seed_echo_workflow(db, published=True) + db.delete_workflow("echo-app") + db.upsert_workflow({"id": "echo-app", "name": "E", "description": "", "published": 1, "latest_version": 0}) + add_job("job_noversion", str(folder)) + worker._process_job(db.get_batch_job("job_noversion")) + assert db.get_batch_job("job_noversion")["status"] == "FAILED" + assert "has no version" in db.get_batch_job("job_noversion")["error"] + + +def test_batch_worker_catches_unexpected_job_error(tmp_path) -> None: + """任务级兜底：DAG 解析失败（缺 name）时任务 FAILED 而不是卡死在 RUNNING。""" + db = _db(tmp_path) + _seed_echo_workflow(db) + folder = _video_folder(tmp_path, names=("a.mp4",)) + # 覆盖为缺失 name 的非法定义：from_dict 抛 KeyError。 + bad_definition = {"version": 1, "nodes": [], "edges": []} + db.create_workflow_version("echo-app", 2, bad_definition) + db.upsert_workflow({"id": "echo-app", "name": "E", "description": "", "published": 1, "latest_version": 2}) + job = _make_job(db, folder) + BatchWorker(db, interval_seconds=0.05)._process_job(job) + assert db.get_batch_job(job["id"])["status"] == "FAILED" + assert "name" in db.get_batch_job(job["id"])["error"] + + +def test_batch_worker_cycle_fails_video_not_job(tmp_path) -> None: + """DAG 环在执行期抛错：单个视频 FAILED 记录错误，批量任务继续并完成。""" + db = _db(tmp_path) + _seed_echo_workflow(db) + folder = _video_folder(tmp_path, names=("a.mp4",)) + # 覆盖为带环的定义：validate 通过、调度器拓扑排序时抛"contains a cycle"。 + cycle_definition = { + "name": "bad", + "version": 2, + "nodes": [ + {"id": "x", "node_type": "echo", "inputs": {"file_uri": "y.file_uri"}}, + {"id": "y", "node_type": "echo", "inputs": {"file_uri": "x.file_uri"}}, + ], + "edges": [{"from": "x", "to": "y"}, {"from": "y", "to": "x"}], + "entry_inputs": {"video_uri": "file"}, + "final_outputs": {"result": "x.file_uri"}, + } + db.create_workflow_version("echo-app", 2, cycle_definition) + db.upsert_workflow({"id": "echo-app", "name": "E", "description": "", "published": 1, "latest_version": 2}) + job = _make_job(db, folder) + BatchWorker(db, interval_seconds=0.05)._process_job(job) + video = db.list_batch_videos(job["id"])[0] + assert video["status"] == "FAILED" + assert "cycle" in video["error"] + assert db.get_batch_job(job["id"])["status"] == "COMPLETED" + assert db.get_batch_job(job["id"])["failed"] == 1 + + +def test_batch_worker_loop_processes_and_survives_exceptions(tmp_path, monkeypatch) -> None: + """轮询线程：处理排队任务；轮询异常不杀死线程；重复启动/幽灵任务无害。""" + db = _db(tmp_path) + _seed_echo_workflow(db) + folder = _video_folder(tmp_path, names=("a.mp4",)) + job = _make_job(db, folder) + # 不存在的任务 ID：_run_job 直接返回，不报错。 + BatchWorker(db, interval_seconds=0.05)._process_job({"id": "ghost_job"}) + # 第一次轮询抛异常（模拟数据库抖动），后续正常。 + # 第一次轮询抛异常（模拟数据库抖动），后续正常。 + calls = {"n": 0} + real_next = db.next_queued_batch_job + + def flaky_next(): + calls["n"] += 1 + if calls["n"] == 1: + raise RuntimeError("transient error") + return real_next() + + monkeypatch.setattr(db, "next_queued_batch_job", flaky_next) + worker = BatchWorker(db, interval_seconds=0.05) + worker.start() + # 重复启动无副作用：线程已存在时直接返回。 + worker.start() + try: + deadline = time.monotonic() + 10 + while time.monotonic() < deadline: + if db.get_batch_job(job["id"])["status"] in {"COMPLETED", "FAILED"}: + break + time.sleep(0.1) + finally: + worker.stop() + assert db.get_batch_job(job["id"])["status"] == "COMPLETED" + + +def test_batch_worker_pause_job_writes_flag_and_pauses_run(tmp_path) -> None: + """pause_job：任务置 PAUSED、排队/运行中的 run 暂停并写 paused.flag。""" + db = _db(tmp_path) + _seed_echo_workflow(db) + folder = _video_folder(tmp_path, names=("a.mp4", "b.mp4", "c.mp4")) + job = _make_job(db, folder) + now = _now_iso() + videos = {Path(v["video_path"]).stem: v for v in db.list_batch_videos(job["id"])} + + # a：QUEUED 运行（应被暂停并写 flag）；b：run 记录不存在；c：已完成的 run。 + db.create_run( + { + "id": "run_pause_me", + "workflow_id": "echo-app", + "workflow_version": 1, + "status": "QUEUED", + "progress": 0, + "input_uri": str(folder / "a.mp4"), + "param_overrides": None, + "source": "batch", + "created_at": now, + "updated_at": now, + } + ) + db.create_run( + { + "id": "run_done_c", + "workflow_id": "echo-app", + "workflow_version": 1, + "status": "COMPLETED", + "progress": 1, + "input_uri": str(folder / "c.mp4"), + "param_overrides": None, + "source": "batch", + "created_at": now, + "updated_at": now, + } + ) + db.update_batch_video(videos["a"]["id"], run_id="run_pause_me", updated_at=now) + db.update_batch_video(videos["b"]["id"], run_id="run_ghost", updated_at=now) + db.update_batch_video(videos["c"]["id"], run_id="run_done_c", updated_at=now) + + worker = BatchWorker(db, interval_seconds=0.05) + worker.pause_job(job["id"]) + assert db.get_batch_job(job["id"])["status"] == "PAUSED" + assert db.get_run("run_pause_me")["status"] == "PAUSED" + assert (folder / "a" / "runs" / "run_pause_me" / PAUSE_FLAG).is_file() + # b 的 run 不存在、c 的 run 已完成：都被跳过，不写 flag。 + assert not (folder / "b" / "runs" / "run_ghost" / PAUSE_FLAG).exists() + assert not (folder / "c" / "runs" / "run_done_c" / PAUSE_FLAG).exists() + + +# --------------------------------------------------------------------------- +# 批量 API +# --------------------------------------------------------------------------- + + +def _client_with_echo_workflow(tmp_path): + """返回 TestClient 与包含真实视频文件的文件夹（echo 工作流已发布）。""" + folder = _video_folder(tmp_path, names=("a.mp4", "b.mp4")) + client = TestClient(app) + client.__enter__() + client.post( + "/api/admin/workflows", + json={ + "id": "echo-app", + "name": "Echo App", + "description": "batch test", + "definition": { + "name": "echo-flow", + "version": 1, + "nodes": [ + {"id": "step", "node_type": "echo", "inputs": {"file_uri": "input.video_uri"}} + ], + "edges": [], + "entry_inputs": {"video_uri": "file"}, + "final_outputs": {"result": "step.file_uri"}, + }, + }, + ) + client.post("/api/admin/workflows/echo-app/publish") + return client, folder + + +def test_batch_api_create_list_detail(tmp_path) -> None: + """批量 API：创建任务、列表、详情（含视频明细与完成标记产物）。""" + client, folder = _client_with_echo_workflow(tmp_path) + try: + response = client.post( + "/api/batch/jobs", + json={"folder": str(folder), "workflow_id": "echo-app", "recursive": True}, + ) + assert response.status_code == 200 + job = response.json() + assert job["status"] == "QUEUED" + assert len(job["videos"]) == 2 + assert job["videos"][0]["finals"] == {} + + listed = client.get("/api/batch/jobs").json() + assert any(item["id"] == job["id"] for item in listed) + + detail = client.get(f"/api/batch/jobs/{job['id']}").json() + assert detail["workflow_id"] == "echo-app" + assert len(detail["videos"]) == 2 + + missing = client.get("/api/batch/jobs/ghost") + assert missing.status_code == 404 + finally: + client.__exit__(None, None, None) + + +def test_batch_api_creation_errors(tmp_path) -> None: + """批量 API 拒绝：文件夹不存在、无视频、未发布工作流。""" + client, folder = _client_with_echo_workflow(tmp_path) + try: + bad_folder = client.post( + "/api/batch/jobs", + json={"folder": str(tmp_path / "missing"), "workflow_id": "echo-app"}, + ) + assert bad_folder.status_code == 422 + empty = tmp_path / "empty" + empty.mkdir() + no_videos = client.post( + "/api/batch/jobs", + json={"folder": str(empty), "workflow_id": "echo-app"}, + ) + assert no_videos.status_code == 422 + not_published = client.post( + "/api/batch/jobs", + json={"folder": str(folder), "workflow_id": "ghost"}, + ) + assert not_published.status_code == 422 + finally: + client.__exit__(None, None, None) + + +def test_batch_api_pause_resume_delete(tmp_path) -> None: + """批量 API：暂停/继续切换任务状态，删除清理数据库记录（含 run）。""" + client, folder = _client_with_echo_workflow(tmp_path) + try: + job = client.post( + "/api/batch/jobs", + json={"folder": str(folder), "workflow_id": "echo-app"}, + ).json() + + paused = client.post(f"/api/batch/jobs/{job['id']}/pause") + assert paused.status_code == 200 + assert client.get(f"/api/batch/jobs/{job['id']}").json()["status"] == "PAUSED" + + resumed = client.post(f"/api/batch/jobs/{job['id']}/resume") + assert resumed.status_code == 200 + assert client.get(f"/api/batch/jobs/{job['id']}").json()["status"] == "QUEUED" + + # 给第一个视频挂一个 run，验证删除任务时级联删除 run 记录。 + db = app.state.db + db.create_run( + { + "id": "run_del", + "workflow_id": "echo-app", + "workflow_version": 1, + "status": "QUEUED", + "progress": 0, + "input_uri": str(folder / "a.mp4"), + "param_overrides": None, + "source": "batch", + "created_at": _now_iso(), + "updated_at": _now_iso(), + } + ) + video = job["videos"][0] + db.update_batch_video(video["id"], run_id="run_del", updated_at=_now_iso()) + assert db.get_run("run_del") is not None + + deleted = client.delete(f"/api/batch/jobs/{job['id']}") + assert deleted.status_code == 200 + assert client.get(f"/api/batch/jobs/{job['id']}").status_code == 404 + assert db.get_run("run_del") is None + + assert client.post("/api/batch/jobs/ghost/pause").status_code == 404 + assert client.post("/api/batch/jobs/ghost/resume").status_code == 404 + assert client.delete("/api/batch/jobs/ghost").status_code == 404 + finally: + client.__exit__(None, None, None) + + +def test_batch_api_worker_unavailable(tmp_path, monkeypatch) -> None: + """批量引擎不可用时，暂停/继续接口返回 503。""" + client, folder = _client_with_echo_workflow(tmp_path) + try: + job = client.post( + "/api/batch/jobs", + json={"folder": str(folder), "workflow_id": "echo-app"}, + ).json() + monkeypatch.setattr("wov_app.routers.batch._get_worker", lambda: None) + assert client.post(f"/api/batch/jobs/{job['id']}/pause").status_code == 503 + assert client.post(f"/api/batch/jobs/{job['id']}/resume").status_code == 503 + finally: + client.__exit__(None, None, None) + + +def test_batch_api_download_artifact(tmp_path) -> None: + """批量 API：从完成标记下载最终产物，缺失别名/文件返回 404。""" + client, folder = _client_with_echo_workflow(tmp_path) + try: + job = client.post( + "/api/batch/jobs", + json={"folder": str(folder), "workflow_id": "echo-app"}, + ).json() + video = job["videos"][0] + db = app.state.db + # 直接构造完成状态：run + 产物 + 同名文件夹完成标记。 + # 产物文件名与下载端点约定一致：完成标记里的文件名位于同名文件夹根。 + run_id = "run_dl" + work = folder / "a" + steps = work / "runs" / run_id / "steps" / "step" + steps.mkdir(parents=True) + (steps / "a.result.20260819000000.txt").write_text("下载内容", encoding="utf-8") + (work / "a.result.20260819000000.txt").write_text("下载内容", encoding="utf-8") + db.create_run( + { + "id": run_id, + "workflow_id": "echo-app", + "workflow_version": 1, + "status": "COMPLETED", + "progress": 1, + "input_uri": str(folder / "a.mp4"), + "param_overrides": None, + "source": "batch", + "created_at": _now_iso(), + "updated_at": _now_iso(), + } + ) + db.create_artifact( + { + "run_id": run_id, + "node_id": "step", + "name": "result", + "uri": str(steps / "a.result.20260819000000.txt"), + "mime_type": "text/plain", + "size": 8, + } + ) + marker = {"workflow_id": "echo-app", "workflow_version": 1, "run_id": run_id, "finals": {"result": "a.result.20260819000000.txt"}} + (work / MARKER_NAME).write_text(json.dumps(marker), encoding="utf-8") + + ok = client.get(f"/api/batch/jobs/{job['id']}/videos/{video['id']}/download?alias=result") + assert ok.status_code == 200 + assert ok.content == "下载内容".encode("utf-8") + + bad_alias = client.get(f"/api/batch/jobs/{job['id']}/videos/{video['id']}/download?alias=nope") + assert bad_alias.status_code == 404 + bad_video = client.get(f"/api/batch/jobs/{job['id']}/videos/bv_ghost/download?alias=result") + assert bad_video.status_code == 404 + # 产物文件缺失 → 404。 + (work / "a.result.20260819000000.txt").unlink() + gone = client.get(f"/api/batch/jobs/{job['id']}/videos/{video['id']}/download?alias=result") + assert gone.status_code == 404 + finally: + client.__exit__(None, None, None) + + +def test_batch_api_roots(tmp_path, monkeypatch) -> None: + """目录树选择器：返回可浏览根目录（含根/家目录与 Windows 盘符分支）。""" + client, _folder = _client_with_echo_workflow(tmp_path) + try: + roots = client.get("/api/batch/roots").json() + assert isinstance(roots, list) and len(roots) >= 1 + # POSIX 必有 /；Windows 必有盘符；两者都有家目录。 + assert any(item["path"] in ("/", str(Path.home())) for item in roots) + assert all(item["name"] for item in roots) + + # Windows 分支：模拟 os.name=nt，存在盘符时返回该驱动器。 + import os + + real_exists = Path.exists + + def fake_exists(path): + # 盘符形式（如 C:\）视为存在，其余走真实判断。 + return str(path).endswith(":\\") or real_exists(path) + + monkeypatch.setattr(os, "name", "nt") + monkeypatch.setattr(Path, "exists", fake_exists) + nt_roots = client.get("/api/batch/roots").json() + assert any(str(item["path"]).endswith(":\\") for item in nt_roots) + finally: + client.__exit__(None, None, None) + + +def test_batch_api_dirs(tmp_path, monkeypatch) -> None: + """目录树选择器：列出子目录、隐藏目录过滤、不存在/不可读返回空。""" + client, _folder = _client_with_echo_workflow(tmp_path) + try: + # 真实目录结构：普通子目录、隐藏目录、文件。 + target = tmp_path / "media" + (target / "movies").mkdir(parents=True) + (target / "series").mkdir(parents=True) + (target / ".hidden").mkdir(parents=True) + (target / "note.txt").write_text("x", encoding="utf-8") + + data = client.get(f"/api/batch/dirs?path={target}").json() + assert [item["name"] for item in data["dirs"]] == ["movies", "series"] + assert data["path"] == str(target) + + # 路径指向文件 → 空列表。 + file_data = client.get(f"/api/batch/dirs?path={target / 'note.txt'}").json() + assert file_data["dirs"] == [] + # 目录不存在 → 空列表。 + missing = client.get(f"/api/batch/dirs?path={tmp_path / 'ghost'}").json() + assert missing["dirs"] == [] + + from pathlib import Path as RealPath + + # 单个子项不可读（is_dir 抛 OSError）→ 跳过该项，其余目录正常返回。 + class _PoisonPath(RealPath): + """is_dir 恒抛权限错误的子类，模拟不可读的子目录。""" + + def is_dir(self): + raise OSError("denied") + + real_iterdir = RealPath.iterdir + + def mixed_iterdir(path): + return list(real_iterdir(path)) + [_PoisonPath(str(tmp_path / "poison"))] + + monkeypatch.setattr(RealPath, "iterdir", mixed_iterdir) + mixed = client.get(f"/api/batch/dirs?path={target}").json() + assert [item["name"] for item in mixed["dirs"]] == ["movies", "series"] + + # 整个目录不可读（iterdir 抛 OSError）→ 空列表而不是 500。 + def deny(path): + raise OSError("denied") + + monkeypatch.setattr(RealPath, "iterdir", deny) + denied = client.get(f"/api/batch/dirs?path={target}").json() + assert denied["dirs"] == [] + finally: + client.__exit__(None, None, None) diff --git a/tests/test_db.py b/tests/test_db.py index f4ecb99..75f8276 100644 --- a/tests/test_db.py +++ b/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 diff --git a/tests/test_maintenance.py b/tests/test_maintenance.py index 437d9f6..39188ef 100644 --- a/tests/test_maintenance.py +++ b/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)