From 52512c26e55abd69bbdfadd69026e430d6bfafd6 Mon Sep 17 00:00:00 2001 From: catShark <1716967236@qq.com> Date: Sun, 6 Sep 2026 16:57:05 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E6=89=B9=E9=87=8F=E4=BB=BB=E5=8A=A1?= =?UTF-8?q?=E8=BF=9B=E5=BA=A6=E5=AE=9E=E6=97=B6=E6=B1=87=E6=80=BB=EF=BC=8C?= =?UTF-8?q?=E6=9A=82=E5=81=9C/=E4=B8=AD=E6=96=AD=E6=97=B6=E5=89=8D?= =?UTF-8?q?=E7=AB=AF=E4=B8=8D=E5=86=8D=E6=98=BE=E7=A4=BA=200/=E6=80=BB?= =?UTF-8?q?=E6=95=B0=200%?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit batch_jobs.done 此前只在任务整体完成时一次性汇总，处理中途（尤其被暂停） 恒为 0，导致前端显示 "0/431 完成 0%" 而实际已处理多个视频 （回归 batch_fee668175444：已处理 15 个仍显示 0/431）。 - db: 新增 sync_batch_job_progress（按明细实时重算 done=COMPLETED+SKIPPED、 failed=FAILED 并落库）与 refresh_batch_job（对齐后返回最新任务） - batch 引擎: 暂停返回、SKIPPED/COMPLETED continue、视频缺失、单视频异常后、 任务收尾等边界统一调用实时对齐，替代收尾一次性 sum - routers/batch: 列表/详情读取前实时对齐，即使引擎不在运行也返回真实进度 - tests: 新增 6 条 TDD 回归测试覆盖 db 助手、引擎暂停、SKIPPED、API 读取侧 同时按用户要求：移除 100% 行覆盖率强制门槛（pytest 不再 --cov-fail-under）， 约定小改动只跑相关测试、保证功能可用即可（AGENTS.md 与 pyproject.toml）。 --- AGENTS.md | 16 ++-- pyproject.toml | 8 +- src/wov_app/batch.py | 21 +++-- src/wov_app/db.py | 46 +++++++++++ src/wov_app/routers/batch.py | 19 ++++- tests/test_batch.py | 151 +++++++++++++++++++++++++++++++++++ 6 files changed, 243 insertions(+), 18 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index b73c008..e063287 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -28,7 +28,7 @@ vrsub/ ├── web/ # 静态前端（index/tasks/admin/workflow + assets） ├── model/ # 本地 whisper 权重（gitignored） ├── data/ # SQLite + 上传/产物存储（gitignored） -└── tests/ # 全部单元/API/冒烟测试（100% 覆盖率） +└── tests/ # 单元/API/冒烟测试（覆盖核心路径，不强制 100%） ``` ### 核心机制 @@ -333,10 +333,14 @@ http://127.0.0.1:8000/docs API 文档 （默认 1.0）。 -## 测试与覆盖率 +## 测试 -- 必须达到 100% 行覆盖率（pytest 已配置 `--cov-fail-under=100`，范围 - `src/` 与 `nodes/`）。 +- 测试以保证功能可用为目标，**不强制 100% 行覆盖率**（pytest 已移除 + `--cov-fail-under=100` 门槛）；需要查看覆盖率时可手动追加 + `uv run pytest --cov=src --cov=nodes`。 +- **小改动只跑相关测试**，避免每次都完整跑全量测试浪费时间；改动涉及 + 哪个模块就跑对应测试文件（如 `uv run pytest tests/test_batch.py`）， + 确认相关用例通过、功能可用即可。完整跑全量测试只在改动影响面大时进行。 - 测试必须调用真实代码路径，不得在测试类中重写业务逻辑来模拟被测功能。 - **测试必须使用真实数据**：真实音频（合法 WAV/PCM）、真实 JSON/数据库/文件； 禁止用占位字节（如 `b"x"`）或伪造结构冒充被测数据——假数据测试只能凑覆盖率， @@ -361,8 +365,8 @@ http://127.0.0.1:8000/docs API 文档 - **开发流程强制 TDD（红-绿-重构）**：任何新功能/修复必须先写失败测试（红）， 再实现最小代码让其通过（绿），最后重构保持整洁；不允许先写实现后补测试。 - 测试运行：`uv run pytest`；全部测试位于 `tests/`。 -- 100% 行覆盖率只保证代码路径被覆盖，不覆盖端口占用、防火墙、权限等 - 外部环境状态；端口问题用启动检查、端口检查与 uvicorn 冒烟测试补充。 +- 测试（无论是否测覆盖率）只保证代码路径被执行，不覆盖端口占用、防火墙、 + 权限等外部环境状态；端口问题用启动检查、端口检查与 uvicorn 冒烟测试补充。 - 本地出现 `WinError 10013` / `WinError 10048` 时，先用 `netstat -ano | findstr :` 确认是否有残留监听进程。 diff --git a/pyproject.toml b/pyproject.toml index 0599cc0..6597002 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -33,12 +33,14 @@ build-backend = "setuptools.build_meta" where = ["src", "."] include = ["wov_sdk*", "wov_app*", "nodes*"] -# pytest 配置：扫描 tests 目录并强制 100% 行覆盖率；同时把仓库根加入 -# sys.path，保证 nodes 包在未重新安装时也能被导入。 +# pytest 配置：扫描 tests 目录并把仓库根加入 sys.path，保证 nodes 包在 +# 未重新安装时也能被导入。 +# 注意：默认不再强制 100% 覆盖率（开发约定：小改动只跑相关测试保证功能 +# 可用即可）；需要查看覆盖率时手动追加 --cov=src --cov=nodes。 [tool.pytest.ini_options] testpaths = ["tests"] pythonpath = ["."] -addopts = "--cov=src --cov=nodes --cov-fail-under=100 -p no:cacheprovider" +addopts = "-p no:cacheprovider" # 集成测试标记（真实模型/真实 LLM/真实数据，默认随全套执行，缺数据自动跳过）。 markers = [ "integration: 需要真实模型/真实音频/真实 LLM API 或用户提供的真实数据，数据或环境缺失时跳过", diff --git a/src/wov_app/batch.py b/src/wov_app/batch.py index 746f1db..10d9f38 100644 --- a/src/wov_app/batch.py +++ b/src/wov_app/batch.py @@ -350,16 +350,23 @@ class BatchWorker: # 暂停检查：批量任务被暂停后停止处理后续视频，等待用户继续。 current = self.db.get_batch_job(job_id) if current is None or current["status"] == "PAUSED": + # 停下前把已完成的视频实时入账：任务可能已处理多个视频才被暂停， + # 若不在暂停边界同步，前端会一直看到 0/总数 0%（回归 batch_fee668175444）。 + self.db.sync_batch_job_progress(job_id) logger.info("批量任务 %s 已暂停，停止在视频 %s", job_id, item["video_path"]) return # 已完成/已跳过的视频不再处理（跳过决策在创建任务时已定）。 if item["status"] in ("COMPLETED", "SKIPPED"): + # 已完成的视频同样是任务进度的一部分：continue 前实时同步汇总， + # 避免长任务（大量 SKIPPED）中途汇总停留在 0。 + self.db.sync_batch_job_progress(job_id) 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()) + self.db.sync_batch_job_progress(job_id) continue work_dir = Path(item["work_dir"]) @@ -374,6 +381,8 @@ class BatchWorker: # 单视频兜底：不中断整个批量任务，记录错误后继续下一个视频。 logger.exception("批量任务 %s 视频 %s 处理异常", job_id, video) self.db.update_batch_video(item["id"], status="FAILED", error=str(exc), updated_at=_now_iso()) + # 本视频处理完（成功/失败/暂停）后实时同步一次汇总，让进度尽快入账。 + self.db.sync_batch_job_progress(job_id) # 重新读取视频明细：_process_video 可能刚创建 run 或已收尾清理 # （快照里 run_id 可能是旧值），必须取最新记录判断暂停状态。 @@ -382,15 +391,17 @@ class BatchWorker: 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.sync_batch_job_progress(job_id) 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") + # 全部视频处理完成：先用明细实时对齐汇总（含已跳过），再置 COMPLETED。 + self.db.sync_batch_job_progress(job_id) + job = self.db.get_batch_job(job_id) + done = int(job["done"]) if job else 0 + failed = int(job["failed"]) if job else 0 self.db.update_batch_job( - job_id, status="COMPLETED", progress=1.0, done=done, failed=failed, + job_id, status="COMPLETED", progress=1.0, current_video=None, error=None, updated_at=_now_iso(), ) logger.info( diff --git a/src/wov_app/db.py b/src/wov_app/db.py index c44daee..c68f899 100644 --- a/src/wov_app/db.py +++ b/src/wov_app/db.py @@ -9,10 +9,16 @@ from __future__ import annotations import json import sqlite3 from contextlib import contextmanager +from datetime import datetime, timezone from pathlib import Path from typing import Any, Iterator +def _now_iso() -> str: + """返回当前 UTC 时间的 ISO 格式字符串（与批量引擎的时间戳一致）。""" + return datetime.now(timezone.utc).isoformat() + + class Database: """SQLite 数据库封装：负责建表以及工作流/任务/产物的 CRUD。""" @@ -551,6 +557,46 @@ class Database: with self._connect() as conn: conn.execute(f"UPDATE batch_jobs SET {assignments} WHERE id = ?", values) + + def sync_batch_job_progress(self, job_id: str) -> None: + """按视频明细实时重算任务的 done/failed 汇总并落库。 + + 明细里 COMPLETED 与 SKIPPED 都计入 done（它们都不再需要处理，是任务 + 已完成的工作量）；FAILED 计入 failed；PAUSED/PENDING 等状态不计。 + 引擎在暂停、视频间检查、收尾等边界调用，router 在读取前也调用，保证 + 前端看到的进度始终与明细一致——即使任务被暂停或进程被终止，汇总字段 + 也不会停留在创建时的 0（回归 batch_fee668175444：处理了 15 个仍显示 + 0/431）。任务不存在时静默返回。 + """ + with self._connect() as conn: + counts = conn.execute( + """ + SELECT + SUM(CASE WHEN status IN ('COMPLETED', 'SKIPPED') THEN 1 ELSE 0 END) AS done, + SUM(CASE WHEN status = 'FAILED' THEN 1 ELSE 0 END) AS failed + FROM batch_videos WHERE job_id = ? + """, + (job_id,), + ).fetchone() + if counts is None or counts["done"] is None: + # 任务不存在或没有任何明细：无需更新。 + return + conn.execute( + "UPDATE batch_jobs SET done = ?, failed = ?, updated_at = ? WHERE id = ?", + (int(counts["done"]), int(counts["failed"]), _now_iso(), job_id), + ) + + def refresh_batch_job(self, job_id: str) -> dict[str, Any] | None: + """实时对齐任务汇总后返回最新记录（读取侧统一入口）。 + + 先 sync_batch_job_progress 让 done/failed 与明细一致，再返回最新 job； + 任务不存在返回 None。批量引擎与 router 共用此入口，保证各处看到的 + 进度数字一致。 + """ + self.sync_batch_job_progress(job_id) + return self.get_batch_job(job_id) + + def delete_batch_job(self, job_id: str) -> None: """删除批量任务记录及其全部视频明细（不含 workflow_runs）。""" with self._connect() as conn: diff --git a/src/wov_app/routers/batch.py b/src/wov_app/routers/batch.py index a076d47..36c6145 100644 --- a/src/wov_app/routers/batch.py +++ b/src/wov_app/routers/batch.py @@ -79,20 +79,31 @@ def create_batch_job( @router.get("/api/batch/jobs") def list_batch_jobs(db: Database = Depends(_get_db)) -> list[dict]: - """返回最近的批量任务列表（不含视频明细，明细按需单独查询）。""" + """返回最近的批量任务列表（不含视频明细，明细按需单独查询）。 + + 返回前对每个任务实时对齐 done/failed：任务被暂停或引擎不在运行时，汇总 + 字段也能与明细一致，前端列表的进度数字不会停留在 0（回归 batch_fee668175444）。 + """ + jobs = db.list_batch_jobs() + for job in jobs: + db.sync_batch_job_progress(str(job["id"])) + # 对齐会刷新 updated_at 与顺序无关，直接返回更新后的最新列表。 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) + """返回批量任务详情，附带每个视频的处理状态与最终产物清单。 + + 详情读取前也实时对齐汇总（与列表接口一致），保证暂停/中断中的任务展示 + 真实进度。任务不存在返回 404。 + """ + job = db.refresh_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, diff --git a/tests/test_batch.py b/tests/test_batch.py index d0ea150..c2ce462 100644 --- a/tests/test_batch.py +++ b/tests/test_batch.py @@ -882,6 +882,120 @@ def test_batch_worker_pause_job_writes_flag_and_pauses_run(tmp_path) -> None: assert db.get_batch_job(job["id"])["status"] == "QUEUED" + +def test_batch_progress_sync_recounts_done_failed(tmp_path) -> None: + """sync_batch_job_progress 按明细实时重算 done(=完成+跳过) 与 failed。 + + 回归：此前 batch_jobs.done 只在任务收尾一次性汇总，处理中途（尤其暂停） + 恒为 0——前端显示 0/431 0%，与实际已处理数量严重不符（真实任务 + batch_fee668175444 已处理 15 个仍显示 0/431）。本测试要求无论任务处于 + 哪种状态，汇总字段都与明细实时一致。 + """ + db = _db(tmp_path) + _seed_echo_workflow(db) + folder = _video_folder(tmp_path, names=("a.mp4", "b.mp4", "c.mp4", "d.mp4")) + job = _make_job(db, folder) + job_id = job["id"] + videos = {Path(v["video_path"]).stem: v for v in db.list_batch_videos(job_id)} + + # 手工构造典型中途态：a 已完成、b 处理中被暂停、c 失败、d 仍排队。 + db.update_batch_video(videos["a"]["id"], status="COMPLETED", updated_at=_now_iso()) + db.update_batch_video(videos["b"]["id"], status="PAUSED", updated_at=_now_iso()) + db.update_batch_video(videos["c"]["id"], status="FAILED", error="boom", updated_at=_now_iso()) + # 任务保持 PAUSED（引擎在视频间停下时的真实状态）。 + db.update_batch_job(job_id, status="PAUSED", updated_at=_now_iso()) + + db.sync_batch_job_progress(job_id) + synced = db.get_batch_job(job_id) + # done 计入已完成 + 已跳过（与收尾口径一致）；失败单独计；进行中不计。 + assert synced["done"] == 1 + assert synced["failed"] == 1 + + # 幂等：重复对齐结果不变（引擎在多个边界可能重复调用）。 + db.sync_batch_job_progress(job_id) + assert db.get_batch_job(job_id)["done"] == 1 + + +def test_batch_progress_sync_counts_skipped(tmp_path) -> None: + """SKIPPED 视频同样计入 done（它们不需要处理，属于已完成的工作量）。""" + db = _db(tmp_path) + _seed_echo_workflow(db) + folder = _video_folder(tmp_path, names=("a.mp4",)) + job = _make_job(db, folder) + # 把 a 改为 SKIPPED（等同创建时旁挂字幕被跳过的语义）。 + db.update_batch_video(db.list_batch_videos(job["id"])[0]["id"], status="SKIPPED", updated_at=_now_iso()) + db.sync_batch_job_progress(job["id"]) + assert db.get_batch_job(job["id"])["done"] == 1 + + +def test_batch_progress_sync_returns_refreshed_job(tmp_path) -> None: + """refresh_batch_job 返回对齐后的最新任务记录（供 router 读取侧使用）。""" + 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"] + db.update_batch_video(db.list_batch_videos(job_id)[0]["id"], status="COMPLETED", updated_at=_now_iso()) + refreshed = db.refresh_batch_job(job_id) + assert refreshed is not None + assert refreshed["done"] == 1 + # 幽灵任务：返回 None，不报错。 + assert db.refresh_batch_job("batch_ghost") is None + + +def test_batch_worker_paused_job_reports_real_done(tmp_path, monkeypatch) -> None: + """引擎在视频之间暂停后，job 汇总实时反映已完成数量（不等任务收尾）。 + + 回归：真实任务 batch_fee668175444 手动暂停后前端仍显示 0/431 0%， + 因为 done 只在任务整体 COMPLETED 时汇总一次。修复后引擎每次停下都要 + 用明细实时对齐 done/failed，暂停中前端即可看到真实进度。 + """ + 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 db.get_batch_job(job_id)["status"] == "PAUSED" + # 关键断言：暂停瞬间 done 已是 1（a 完成），而不是停留在创建时的 0。 + assert db.get_batch_job(job_id)["done"] == 1 + + +def test_batch_worker_skipped_continue_syncs_done(tmp_path) -> None: + """循环里遇到 SKIPPED/COMPLETED 的 continue 分支也会把汇总实时对齐。""" + db = _db(tmp_path) + _seed_echo_workflow(db) + folder = _video_folder(tmp_path, names=("a.mp4", "b.mp4")) + (folder / "a.CN.srt").write_text("x", encoding="utf-8") # a 创建即 SKIPPED + job = _make_job(db, folder) + job_id = job["id"] + # b 也手工置为 COMPLETED（引擎会因 continue 分支跳过它）。 + db.update_batch_video( + [v for v in db.list_batch_videos(job_id) if v["video_path"].endswith("b.mp4")][0]["id"], + status="COMPLETED", updated_at=_now_iso(), + ) + # 任务先置 PAUSED：首轮循环在 a（SKIPPED）即停下，若 continue 前未对齐则 done=0。 + db.update_batch_job(job_id, status="PAUSED", updated_at=_now_iso()) + BatchWorker(db, interval_seconds=0.05)._run_job(job_id) + assert db.get_batch_job(job_id)["done"] == 2 + def test_batch_worker_ghost_job_id_returns(tmp_path) -> None: """_run_job 在任务不存在时直接返回（幽灵任务处理无副作用）。""" db = _db(tmp_path) @@ -977,6 +1091,43 @@ def test_batch_api_creation_errors(tmp_path) -> None: client.__exit__(None, None, None) +def test_batch_api_list_detail_syncs_real_progress(tmp_path) -> None: + """列表/详情读取前自动对齐 done/failed：暂停中的任务也显示真实进度。 + + 回归：服务运行中任务被暂停（batch_fee668175444），前端轮询列表看到 + 0/431 0%——因为 done 只在收尾时汇总、暂停后无人再写。读取侧兜底对齐 + 保证前端拿到明细真实状态，即使引擎不在运行（进程被杀/暂停中）。 + """ + client, folder = _client_with_echo_workflow(tmp_path) + try: + job = client.post( + "/api/batch/jobs", + json={"folder": str(folder), "workflow_id": "echo-app"}, + ).json() + job_id = job["id"] + db = app.state.db + + # 模拟暂停中已处理 1 个（COMPLETED）+ 1 个失败：明细变了，但 job.done 仍是 0。 + videos = db.list_batch_videos(job_id) + db.update_batch_video(videos[0]["id"], status="COMPLETED", updated_at=_now_iso()) + db.update_batch_video(videos[1]["id"], status="FAILED", error="x", updated_at=_now_iso()) + db.update_batch_job(job_id, status="PAUSED", updated_at=_now_iso()) + assert db.get_batch_job(job_id)["done"] == 0 # 修复前：旧值 + + # 读取详情：返回的 job 已完成实时对齐。 + detail = client.get(f"/api/batch/jobs/{job_id}").json() + assert detail["done"] == 1 and detail["failed"] == 1 + # 数据库里的汇总也一并修正（读取副作用：后续列表/引擎都看到正确值）。 + assert db.get_batch_job(job_id)["done"] == 1 + + # 列表接口同样实时对齐。 + listed = client.get("/api/batch/jobs").json() + mine = next(item for item in listed if item["id"] == job_id) + assert mine["done"] == 1 and mine["failed"] == 1 + finally: + client.__exit__(None, None, None) + + def test_batch_api_pause_resume_delete(tmp_path) -> None: """批量 API：暂停/继续切换任务状态，删除清理 DB 记录与私有工作空间。""" client, folder = _client_with_echo_workflow(tmp_path)