diff --git a/AGENTS.md b/AGENTS.md index d842f93..c58ac11 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -41,6 +41,10 @@ vrsub/ `data/storage/runs//steps//` 落盘并登记到 artifacts 表。 - **前端**：由 FastAPI 静态挂载 `web/`，节点注册/实例管理页面已移除， 仅保留应用中心、任务管理、管理后台（工作流）与工作流编排。 +- **工作流编排页（web/workflow.html）**：支持新建工作流（空表单预填演示模板）， + 从列表"编辑"加载任一工作流的最新定义（ID 锁定，保存即追加新版本）；"版本" + 查看全部历史版本并可"加载到编辑器"（对比/回滚后另存新版本）；管理后台 + （admin.html）无编辑器，点"编辑"自动跳转 `workflow.html?edit=` 加载。 ## 节点输入/输出协议 @@ -56,7 +60,7 @@ vrsub/ | `vlm-ocr` | `image_uri` | `text`、`text_uri` | 直接调本地 Ollama 多模态模型（glm-ocr）的 `/api/chat` 做视频帧 OCR（流式 + 5s 上限），参数：`model`、`ollama_host`、`prompt`、`timeout_seconds`、`keep_alive`、`num_predict`、`temperature`、`repeat_penalty` | | `frame-extract` | `video_uri` | `frames_manifest`、`frame_count` | 按**帧间隔**抽帧（解析 fps → step=round(间隔秒×fps)，ffmpeg select 按帧号精确取帧，帧时间=帧号/fps 无累计偏差）并 crop 裁切字幕区域，参数：`interval_seconds`（默认 0.5）、`crop`（[x,y,w,h] 0~1）。**帧文件必须按帧号数值排序读取**（`_sorted_frame_files`）：ffmpeg `%04d` 编号超过 9999 帧后扩为 5 位，字典序 `sorted()` 会把 5 位编号排在 4 位之前导致时间与图像错位（真实发生于 run_339ec7ee437f 的 14236 帧任务，回归测试见 `test_frame_files_read_order_matches_frame_number`） | | `subtitle-ocr` | `frames_manifest` | `srt_uri`、`count` | 自适应线程池并发逐帧调 vlm-ocr → 垃圾过滤（无文字帧）→ 相同字幕合并（记录最后可见帧）→ 组装 SRT，消失时间=最后可见帧+采样间隔（间隔从帧清单推导），参数：`min_chars`、`min_alnum_ratio`、`garbage_tokens`、`pool_min_workers`/`pool_max_workers`/`pool_window_seconds`/`pool_fast_threshold`/`pool_slow_threshold` | -| `llm-filter` | `srt_uri` | `srt_uri`、`kept`、`removed` | 两级过滤：①**规则层**（不调 LLM）直接删横线装饰/HTML 水印 token/URL/邮箱/单双 ASCII 字符；②**LLM 五类分类**（garbage/overlay/noise 删，repeat/dialogue 留，未识别回退保留）每条连同前后各 `context_size`（默认 10）条纯文本分批判断，**按文本去重**（忽略空白/大小写，相同文本只调一次 LLM，上下文取首次出现）保证判定一致并省调用，**长文本保护**（≥`min_keep_len` 默认 12 时 noise 不构成删除依据）。参数：`context_size`、`min_keep_len`、`overlay_tokens`（JSON 数组）、`dedupe`（默认开）、`model`、`pool_min_workers`/`pool_max_workers`/`pool_window_seconds`/`pool_fast_threshold`/`pool_slow_threshold`。回归数据：testdata/ocr_srt_run_ac7f480a3ccb.srt（真实任务 1666 条 OCR 输出） | +| `llm-filter` | `srt_uri` | `srt_uri`、`kept`、`removed` | 两级过滤：①**规则层**（不调 LLM）正则确定性删除——横线装饰、URL/邮箱/**裸网址域名**（含中文夹杂的注册地址）、**HTML/水印模式**（html code/标签/javascript 等）、overlay token（html/marketing 等）、单双 ASCII 字符；②**LLM 五类分类**（garbage/overlay/noise 删，repeat/dialogue 留，未识别回退保留）每条连同前后各 `context_size`（默认 10）条纯文本分批判断——**上下文净化**：喂给 LLM 的是**过滤后的字幕**，规则层确定性垃圾从上下文中剔除（原文不进 LLM），避免覆盖层垃圾污染场景判断误删真实对话（回归：run_011d01f19999 曾 190 条含 ≥4 汉字对话被误删）；**长文本保护**：≥`min_keep_len`（默认 12）时 noise 不构成删除依据——LLM 判定不稳定，长度是必要兜底（实测移除保护后新增误删 124 条真实长对话）。**限流自适应**：LLM 调用 429/5xx 指数退避重试（最多 3 次，1s/2s/4s），worker 捕获限流错误时调用线程池 `report_failure()` **内存中临时降低最大线程数并缩容**（连续无错误窗口后逐步回升），失败条目在收紧后的并发下**重试一轮**，二次仍失败才整体失败——20 并发一拥而上触发 429 时自动收敛到配额内而不打挂任务。**按文本去重**（忽略空白/大小写，相同文本只调一次 LLM，上下文取首次出现）保证判定一致并省调用。参数：`context_size`、`min_keep_len`、`overlay_tokens`（JSON 数组）、`dedupe`（默认开）、`model`、`pool_min_workers`/`pool_max_workers`/`pool_window_seconds`/`pool_fast_threshold`/`pool_slow_threshold`。回归数据：testdata/ocr_srt_run_ac7f480a3ccb.srt（真实任务 1666 条 OCR 输出） | | `srt-to-dual-eye-ass` | `cn_srt_uri` | `ass_uri` | 参数：`resolution`，如 `3840x1920` | ### 模型权重解析（本地优先） @@ -238,14 +242,35 @@ http://127.0.0.1:8000/docs API 文档 - **状态机**：`QUEUED / RUNNING / PAUSED / COMPLETED / FAILED`。排队中或运行中的 任务可暂停（`POST /api/runs/{run_id}/pause`），PAUSED 可继续 （`POST /api/runs/{run_id}/resume` → 恢复 QUEUED）。 -- **调度器语义**：`next_queued_run` 同时取 QUEUED 与 PAUSED；`execute_run` 在每个 - 节点边界检查状态，被暂停则停下保持 PAUSED（当前节点执行完后才停）； - 继续时从产物表（`restore_run_outputs`，剥去"节点ID."前缀还原输出名）重建已完成 - 节点的输出，**跳过已完成节点断点续跑**，最后补做 final_outputs 收尾。 +- **调度器语义**：`next_queued_run` **只取 QUEUED**——PAUSED 任务不会被调度器 + 自动拾起（修复回归：此前 PAUSED 被拾起后 `execute_run` 先置 RUNNING 再检查， + 节点循环读到的是刚改的 RUNNING，"暂停检查"永远不成立 → 任务被复活继续跑， + 表现为"点击暂停反而开始任务"）。`execute_run` 以 PAUSED 进入时直接返回保持暂停， + 必须用户显式 resume（PAUSED → QUEUED）后才真正执行；运行中被暂停的任务在每个 + 节点边界检查状态停下保持 PAUSED（当前节点执行完后才停）；继续时从产物表 + （`restore_run_outputs`，剥去"节点ID."前缀还原输出名）重建已完成节点的输出， + **跳过已完成节点断点续跑**，最后补做 final_outputs 收尾。 +- **重启恢复**：进程被杀/重启时遗留的 RUNNING 任务在启动时被 + `recover_interrupted_runs` 恢复为 QUEUED（保留产物），调度器自动断点续跑； + PAUSED 任务保持不变，等待显式 resume。 +- **节点级断点（subtitle-ocr）**：OCR 每帧完成后立即把 `{frame, text}` 追加到 + `steps/ocr/ocr_partial.jsonl`（多线程下加锁串行化）。invoke 启动时读取存档， + 只对未处理帧调用 vlm-ocr，存档文本与新增结果合并后组装 SRT——2 小时视频级 + OCR 任务中断/重启后不重复已处理帧，产物与一次跑完逐字节一致。 +- **节点内暂停响应（subtitle-ocr）**：暂停接口（`POST /api/runs/{run_id}/pause`） + 除置 PAUSED 外还向 run 根目录写入 `paused.flag`；OCR 工作线程**逐帧检查**该 + 信号，存在即立即中止（不 OCR、不入存档，恢复时重跑该帧），invoke 返回 + failed；调度器捕获节点异常时若任务已是 PAUSED 则**保持 PAUSED 不标 FAILED**， + resume 时清除信号并从断点存档继续——点击暂停后 OCR 秒级停下，不再等整个 + 节点跑完。继续/重试接口与调度器执行前都会清理残留信号。 - **前端**：任务管理页为 QUEUED/RUNNING 提供"暂停"、PAUSED 提供"继续"按钮。 - **进度日志（数据处理速度）**： - 调度器：每节点完成打印"任务 X 进度 i/N 节点: Y 耗时 Zs, 运行累计 Ws"； - - subtitle-ocr：`OCR 进度 X/Y 帧 (Z 帧/s)`；llm-filter：`字幕判定进度 X/Y 条 (Z 条/s)` + - subtitle-ocr：`OCR 进度 X/Y 帧 (Z 帧/s, 平均 Ws/帧, 线程 N/M)`；llm-filter： + `字幕判定进度 X/Y 条 (Z 条/s, 平均 Ws/条, 线程 N/M)`——`W` 为最近窗口 + 平均单任务耗时（窗口未满时回退累计平均），`N` 为当前目标线程数、`M` 为 + `pool_max_workers` 上限，用于判断多线程是否因单次处理过慢（窗口平均 ≥ + `pool_fast_threshold`）而未扩容； （线程池 `on_progress` 回调，每任务完成触发）； - whisper：分块转写打印"分块 X/Y 完成 offset=... 耗时 Zs (Nx 实时, 累计 ...s)"； - frame-extract：ffmpeg `-progress` 输出解析 `frame=N`，打印"抽帧进度 X/Y 帧 (Z 帧/s)"。 @@ -340,7 +365,8 @@ http://127.0.0.1:8000/docs API 文档 - 存储、队列、调度器都要通过抽象边界隔离，方便从单机实现替换为分布式实现。 - 用户端永远只看到"输入 -> 进度 -> 结果"，不暴露工作流细节。 - 单体对分布式版的三处降级：无子进程隔离、无空闲 TTL 回收（模型常驻， - 仅懒加载）、慢任务无法强制中断（由节点自身超时兜底）。 + 仅懒加载）、非 OCR 慢任务无法强制中断（由节点自身超时兜底；OCR 节点支持 + 暂停信号逐帧中断）。 ## 单体化说明 diff --git a/nodes/adaptive_pool.py b/nodes/adaptive_pool.py index 14a2bb8..5f163ec 100644 --- a/nodes/adaptive_pool.py +++ b/nodes/adaptive_pool.py @@ -55,7 +55,7 @@ class AdaptiveThreadPool: fast_threshold: float = 0.3, slow_threshold: float = 1.0, clock=time.monotonic, - on_progress: Callable[[int, int, float], None] | None = None, + on_progress: Callable[[int, int, float, float, int], None] | None = None, ) -> None: """初始化；clock 可注入便于测试；on_progress(done,total,rate) 每次完成回调。""" self._worker = worker @@ -73,6 +73,15 @@ class AdaptiveThreadPool: self._results: list = [] self._lock = threading.Lock() self._stop = threading.Event() + # 取消标记：worker 检测到取消（如暂停信号）后设置，后续完成的任务 + # 不再触发进度回调——暂停时队列中剩余大量任务会快速退出，若仍逐项 + # 打印进度会在数秒内打出上万行日志。 + self._cancel_event = threading.Event() + # 有效最大线程数：初始等于 max_workers；消费错误（如 API 限流）时 + # report_failure 临时收紧，连续无错误窗口后逐步回升——并发自适应配额。 + self._effective_max_workers = max_workers + # 当前窗口内消费错误计数：窗口评估时无错误才允许恢复有效上限。 + self._window_failures = 0 # 滚动窗口起点与已记录的单次耗时。 self._window_start = clock() # 观测到的最大并发线程数（供测试与监控）。 @@ -83,6 +92,9 @@ class AdaptiveThreadPool: self._total = 0 self._started_at = 0.0 self._window_times: list[float] = [] + # 最近一次窗口评估的平均单任务耗时（秒）：供进度回调诊断使用， + # 与扩缩容决策共用同一依据；窗口尚未评估时为 None（回退累计平均）。 + self._window_avg_time: float | None = None def _run(self) -> None: """工作线程主循环：取任务 → 执行 → 记录耗时并自适应评估。""" @@ -107,10 +119,16 @@ class AdaptiveThreadPool: self._results.append((seq, result)) # 进度回调：已完成数、总数与平均处理速度（条/秒）。 self._completed += 1 - if self._on_progress is not None: + if self._on_progress is not None and not self._cancel_event.is_set(): elapsed_total = max(self._clock() - self._started_at, 1e-9) + with self._lock: + workers = self._target_workers self._on_progress( - self._completed, self._total, self._completed / elapsed_total + self._completed, + self._total, + self._completed / elapsed_total, + self._current_avg_time(elapsed_total), + workers, ) self._tick(elapsed) self._queue.task_done() @@ -128,15 +146,36 @@ class AdaptiveThreadPool: avg = sum(self._window_times) / len(self._window_times) self._window_start = self._clock() self._window_times.clear() + # 记录本次窗口平均耗时：进度回调据此展示"当前扩缩容依据"。 + self._window_avg_time = avg with self._lock: current = self._target_workers + # 窗口内无消费错误 → 有效上限逐步回升（错误降下来的并发慢慢恢复）。 + if ( + self._window_failures == 0 + and self._effective_max_workers < self.max_workers + ): + self._effective_max_workers += 1 + # 重置窗口错误计数，进入下一窗口。 + self._window_failures = 0 + # 扩容上限用有效最大线程数：错误窗口内即使响应快也不超过收紧后的上限。 self._resize( decide( - current, avg, self.min_workers, self.max_workers, + current, avg, self.min_workers, self._effective_max_workers, self.fast_threshold, self.slow_threshold, ) ) + def _current_avg_time(self, elapsed_total: float) -> float: + """返回供进度回调展示的平均单任务耗时（秒）。 + + 优先使用最近一次窗口评估的平均耗时（与扩缩容决策同一依据）； + 窗口尚未评估过时回退为启动至今的累计平均，避免无数据可看。 + """ + if self._window_avg_time is not None: + return self._window_avg_time + return elapsed_total / max(self._completed, 1) + def _resize(self, target: int) -> None: """调整并发目标：扩容启动新线程；缩容压入等量停止哨兵（幂等）。 @@ -157,6 +196,28 @@ class AdaptiveThreadPool: self._queue.put((None, _POISON)) self._target_workers = target + def cancel(self) -> None: + """请求取消本批任务：后续完成的任务不再触发进度回调。 + + 供调用方在工作线程内检测到外部信号（如暂停）时调用，抑制暂停后 + 队列中剩余任务快速退出导致的进度日志井喷；下一批 map 自动重置。 + """ + self._cancel_event.set() + + def report_failure(self) -> None: + """通知一次消费错误（如 API 限流 429）：临时降低有效最大线程数并缩容。 + + 供工作线程捕获可退避错误（限流/服务端 5xx）后调用：并发立即收紧到 + 新上限，后续请求减少从而避开持续限流；连续无错误窗口后有效上限 + 逐步回升到 max_workers（见 _tick 的恢复逻辑）。 + """ + with self._lock: + self._window_failures += 1 + if self._effective_max_workers > self.min_workers: + self._effective_max_workers -= 1 + # 缩容到新上限（幂等：目标低于当前才放停止哨兵）。 + self._resize(self._effective_max_workers) + def map(self, items) -> list: """按输入顺序返回每个 item 经 worker 处理后的结果列表。""" self._results = [] @@ -164,6 +225,13 @@ class AdaptiveThreadPool: self._total = len(items) self._started_at = self._clock() self._stop.clear() + # 每批任务开始时重置取消状态：上一批的取消不延续到下一批。 + self._cancel_event.clear() + # 上一批任务结束后工作线程已全部退出（_stop 停止）但 _target_workers + # 仍记旧值，_resize 不会重新启动线程——实际无线程时归零后重建。 + with self._lock: + if not self._threads: + self._target_workers = 0 self._resize(self.min_workers) for seq, item in enumerate(items): self._queue.put((seq, item)) diff --git a/nodes/llm_filter.py b/nodes/llm_filter.py index 6bf5e9b..9e544c6 100644 --- a/nodes/llm_filter.py +++ b/nodes/llm_filter.py @@ -19,8 +19,9 @@ 上下文取首次出现位置，结果缓存复用——修复"同一句字幕 5 留 6 删"的 判定不一致，同时把长视频的 LLM 调用量降到唯一文本数。 -4. **长文本保护**：长度 ≥ min_keep_len 的文本，仅 garbage/overlay 两个 - 明确垃圾类别可删，noise 不构成删除依据，防止长对话被误删。 +4. **上下文净化**：喂给 LLM 的上下文是**过滤后的字幕**——规则层确定性的 + 垃圾（横线/HTML/网址/水印等）从上下文中剔除，只留下有意义的对白， + 避免覆盖层垃圾污染 LLM 的场景判断导致误删真实对话。 参考真实任务 run_ac7f480a3ccb（2026-08，OCR 1666 条）：旧实现把 394 条 真实对话当噪声删掉（占删除 35%）、同文本判定不一致；新实现按上述机制 @@ -32,6 +33,8 @@ from __future__ import annotations import json import os import re +import time +import urllib.error import urllib.request from pathlib import Path @@ -72,11 +75,22 @@ DELETE_CATEGORIES = {CATEGORY_GARBAGE, CATEGORY_OVERLAY, CATEGORY_NOISE} _DASH_RE = re.compile(r"^[\s\-—_~=•・。．.、]+$") # 规则层正则：URL / 邮箱。 _URL_OR_MAIL_RE = re.compile(r"^(https?://|www\.)\S+$|^[\w.+-]+@[\w.-]+\.\w+$") +# 规则层正则：裸网址/域名（含中文夹杂的注册地址，如 "水火地址 489155.com"）。 +_DOMAIN_RE = re.compile( + r"[\w-]+\.(?:com|net|org|cn|tv|me|io|xyz|cc|top|info|biz)(?:[/\s.,;:：!?）)]|$)", + re.IGNORECASE, +) +# 规则层正则：HTML/脚本/播放器水印模式（OCR 常把网页界面识别成这类文本）。 +_HTML_MARK_RE = re.compile( + r"html\s*code|<\s*[a-z][^>]*>|javascript|web\s*address|watermark|sign\s*in", + re.IGNORECASE, +) # 默认水印/覆盖层 token（casefold 后比较，可经 overlay_tokens 参数覆盖）。 DEFAULT_OVERLAY_TOKENS = frozenset( - {"html", "background", "___", "cleaning", "buffering", "loading"} + {"html", "background", "___", "cleaning", "buffering", "loading", "marketing"} ) # 长文本保护阈值：≥ 该长度的文本，noise 类别不构成删除依据。 +# LLM 判定不稳定（实测把完整对话句误判 noise），长度是必要兜底而非删除依据。 DEFAULT_MIN_KEEP_LEN = 12 @@ -111,16 +125,19 @@ def serialize_srt(entries: list[dict]) -> str: def _rule_verdict(text: str, overlay_tokens: set[str]) -> bool | None: """确定性规则层：返回 True（删除）/ None（交给 LLM 多维判断）。 - 规则覆盖 OCR 噪声的稳定模式：空文本、纯横线装饰、URL/邮箱、水印 token - （HTML/background 等）、单双 ASCII 字符。含 CJK 的短文本不算垃圾， - 因为"嗯/好"等可能是内容；其余情况返回 None 交由 LLM 分类。 + 规则覆盖 OCR 噪声的稳定模式：空文本、纯横线装饰、URL/邮箱、裸网址 + 域名、HTML/水印模式（html code/标签/javascript 等）、水印 token、单双 + ASCII 字符。含 CJK 的短文本不算垃圾，因为"嗯/好"等可能是内容； + 其余情况返回 None 交由 LLM 分类。 """ t = text.strip() if not t: return True if _DASH_RE.match(t): return True - if _URL_OR_MAIL_RE.match(t): + if _URL_OR_MAIL_RE.match(t) or _DOMAIN_RE.search(t): + return True + if _HTML_MARK_RE.search(t): return True if t.casefold() in overlay_tokens: return True @@ -141,9 +158,10 @@ def _dedup_key(text: str) -> str: def _should_delete(category: str, text: str, min_keep_len: int) -> bool: """按 LLM 类别与长文本保护决定是否删除。 - repeat/dialogue 一律保留；garbage/overlay 一律删除（含长文本—— - 这两个是明确的垃圾信号）；noise 对短文本删除，但 ≥min_keep_len 的 - 长文本不删（noise 太模糊，不足以推翻一句完整台词）。 + repeat/dialogue 一律保留；garbage/overlay 一律删除（明确的垃圾信号）； + noise 对短文本删除，但 ≥min_keep_len 的长文本不删——LLM 判定不稳定， + 完整对话句常被误判 noise，长度保护是必要兜底（实测移除后新增误删 + 124 条真实长对话）。长度只用于"保护"，不用于"删除"。 """ if category not in DELETE_CATEGORIES: return False @@ -151,9 +169,9 @@ def _should_delete(category: str, text: str, min_keep_len: int) -> bool: return False return True - def _judge_category( - entries: list[dict], index: int, context_size: int, params: dict + entries: list[dict], index: int, context_size: int, params: dict, + overlay_tokens: set[str] | None = None, ) -> str: """调用 LLM 把目标字幕归入五类之一，返回类别词（未识别回退 dialogue）。 @@ -164,10 +182,16 @@ def _judge_category( start = max(0, index - context_size) end = min(len(entries), index + context_size + 1) target_pos = index - start - lines = [ - f"{TARGET_MARK}{text}" if pos == target_pos else text - for pos, text in enumerate(entry["text"] for entry in entries[start:end]) - ] + # 上下文净化：喂给 LLM 的是**过滤后的字幕**——相邻条目若被确定性规则层 + # 识别为垃圾（横线/HTML/水印 token/URL/裸域名等）直接从上下文中剔除， + # 避免覆盖层垃圾污染 LLM 对整段场景的判断（误删相邻的真实对话）。 + tokens = overlay_tokens if overlay_tokens is not None else DEFAULT_OVERLAY_TOKENS + lines = [] + for pos, entry in enumerate(entries[start:end]): + if pos != target_pos and _rule_verdict(entry["text"], tokens) is True: + continue + text = entry["text"] + lines.append(f"{TARGET_MARK}{text}" if pos == target_pos else text) # LLM 兼容接口配置：地址/Key/模型/超时均可通过环境变量覆盖（默认 SiliconFlow）。 api_base = os.getenv( @@ -186,6 +210,8 @@ def _judge_category( "noise：与上下文无关、无实际语义的杂项\n" "repeat：内容性重复（如语气词、呻吟、重复的感叹或对话），属于内容本身\n" "dialogue：正常对话\n" + "字幕序列已经过确定性规则过滤（装饰性横线、HTML、网址/水印等已被剔除），" + "请只依据剩下的对话内容判断目标字幕，不要臆测被过滤掉的部分。\n" "只输出一个类别英文单词，不要输出其他内容。" ) body = { @@ -210,8 +236,23 @@ def _judge_category( headers=headers, method="POST", ) - with urllib.request.urlopen(request, timeout=request_timeout) as response: - payload = json.loads(response.read().decode("utf-8")) + # 429（限流）与 5xx（服务端错误）时指数退避重试：请求被限流时让出时间， + # 使窗口平均响应变慢，触发自适应线程池"慢响应减线程"，并发自动回落到 + # 限流配额内；最多尝试 3 次，耗尽仍失败则抛出，由调用方（任务）重新处理。 + max_attempts = 3 + retry_delay = 1.0 + for attempt in range(max_attempts): + try: + with urllib.request.urlopen(request, timeout=request_timeout) as response: + payload = json.loads(response.read().decode("utf-8")) + break + except urllib.error.HTTPError as exc: + if exc.code != 429 and not (500 <= exc.code < 600): + raise + if attempt >= max_attempts - 1: + raise + time.sleep(retry_delay) + retry_delay *= 2 content = str(payload["choices"][0]["message"]["content"]).strip().lower() # 精确匹配五个类别词（兼容"garbagexxx"这类多余输出）。 for category in _ALL_CATEGORIES: @@ -226,8 +267,8 @@ def _judge_category( def invoke(request: InvokeRequest) -> InvokeResponse: """过滤 SRT 中多余/无意义的字幕，产物为 filtered.srt。 - 流程：规则层（确定性删除）→ LLM 层（去重后按唯一文本多维分类）→ - 长文本保护 → 保留条重新编号输出。 + 流程：规则层（确定性删除）→ LLM 层（去重后按唯一文本多维分类， + 上下文为过滤后的字幕）→ 保留条重新编号输出。 """ srt_uri = request.inputs.get("srt_uri") if not srt_uri: @@ -267,13 +308,23 @@ def invoke(request: InvokeRequest) -> InvokeResponse: else: pool_indices = llm_needed - # 单条判断的工作函数：返回类别词。 + # 单条判断的工作函数：返回类别词；overlay_tokens 用于上下文净化。 def judge_one(index: int) -> str: - return _judge_category(entries, index, context_size, request.params) + try: + return _judge_category(entries, index, context_size, request.params, overlay_tokens) + except urllib.error.HTTPError as exc: + # 限流/服务端错误：通知线程池临时降低最大并发，避免持续超配额。 + if exc.code == 429 or 500 <= exc.code < 600: + pool.report_failure() + raise - # 进度日志：打印已判定条数、总数与平均处理速度（条/s）。 - def log_progress(done: int, total: int, rate: float) -> None: - logger.info("字幕判定进度 %d/%d 条 (%.1f 条/s)", done, total, rate) + # 进度日志：打印已判定条数、总数、平均处理速度（条/s）、最近窗口 + # 平均单条耗时与当前线程数（与 OCR 节点同一回调协议）。 + def log_progress(done: int, total: int, rate: float, avg_time: float, workers: int) -> None: + logger.info( + "字幕判定进度 %d/%d 条 (%.1f 条/s, 平均 %.2fs/条, 线程 %d/%d)", + done, total, rate, avg_time, workers, pool.max_workers, + ) # 自适应并发调用 LLM：按实测负载弹性伸缩，避免压垮 LLM 接口。 pool = AdaptiveThreadPool( @@ -287,11 +338,25 @@ def invoke(request: InvokeRequest) -> InvokeResponse: ) categories = pool.map(pool_indices) + # 限流/服务端错误已在 worker 内 report_failure 临时降并发：把失败的 + # 条目在收紧后的并发下重试一轮（_judge_category 内还有 429/5xx 退避）， + # 二次仍失败才整体失败——避免 20 并发一拥而上被限流打挂整个任务。 + failed = [ + index for index, category in zip(pool_indices, categories) + if isinstance(category, Exception) + ] + if failed: + logger.info("判定失败 %d 条，降并发后重试", len(failed)) + retried = pool.map(failed) + for index, category in zip(failed, retried): + if isinstance(category, Exception): + # 二次仍失败（限流持续/非限流错误）：整体失败，由任务重试。 + return InvokeResponse(status="failed", error=str(category)) + cat_by_index[index] = category for index, category in zip(pool_indices, categories): - # 并行下 LLM 异常被线程池隔离为异常结果：任一条失败即整体失败， - # 避免静默输出未过滤结果。 + # 首次失败的条目已在重试分支处理，这里只登记成功结果。 if isinstance(category, Exception): - return InvokeResponse(status="failed", error=str(category)) + continue cat_by_index[index] = category if dedupe: # 去重填充：与首次出现同键的条目复用同一类别，保证判定一致。 @@ -299,11 +364,11 @@ def invoke(request: InvokeRequest) -> InvokeResponse: if i not in cat_by_index: cat_by_index[i] = cat_by_index[first_of_key[_dedup_key(entries[i]["text"])]] - # 阶段 3：合并规则与 LLM 判定，应用长文本保护并输出。 + # 阶段 3：合并规则与 LLM 判定，输出保留条。 kept: list[dict] = [] removed = 0 for i, (entry, rule_verdict) in enumerate(zip(entries, rule_verdicts)): - # 规则层命中即删（True）；未命中则按 LLM 类别与长文本保护判定。 + # 规则层命中即删（True）；未命中则按 LLM 类别判定。 if rule_verdict is True: removed += 1 continue diff --git a/nodes/subtitle_ocr.py b/nodes/subtitle_ocr.py index 4a76793..af7a6f5 100644 --- a/nodes/subtitle_ocr.py +++ b/nodes/subtitle_ocr.py @@ -10,6 +10,7 @@ SRT 基准数据：每条字幕消失时间 = 最后可见帧时间 + 采样间 from __future__ import annotations import json +import threading from pathlib import Path from nodes.adaptive_pool import AdaptiveThreadPool @@ -20,6 +21,44 @@ from wov_sdk.models import InvokeRequest, InvokeResponse logger = get_logger("subtitle-ocr") +# 节点级断点存档的写锁：多线程 OCR 并发完成时串行化追加写，避免行交错。 +_partial_lock = threading.Lock() + +# 断点存档文件名：每行 {"frame": 帧序号(0-based), "text": 该帧OCR文本}。 +# OCR 每帧完成后立即追加一行；进程重启后从存档恢复已处理帧，只对未处理 +# 帧重新调用 vlm-ocr——2 小时视频级任务中断后不浪费已完成的帧。 +_PARTIAL_NAME = "ocr_partial.jsonl" + +# 暂停信号文件名：位于 run 根目录（/runs//paused.flag）， +# 由暂停接口写入、继续/重试/删除时清除；节点逐帧检查，存在即中止。 +_PAUSE_FLAG = "paused.flag" + + +class PauseRequested(Exception): + """节点内暂停信号：OCR 检测到任务被暂停后抛出，由调度器保持 PAUSED。 + + 不把暂停误报为 FAILED：调度器捕获异常时若任务状态已是 PAUSED， + 则保持暂停等待用户 resume，从断点存档继续未处理帧。 + """ + + +def _load_partial(output_dir: Path) -> dict[int, str]: + """读取节点级断点存档，返回 {帧序号: OCR 文本}；无存档时返回空字典。""" + path = output_dir / _PARTIAL_NAME + if not path.is_file(): + return {} + result: dict[int, str] = {} + for line in path.read_text(encoding="utf-8").splitlines(): + if not line.strip(): + continue + try: + item = json.loads(line) + except json.JSONDecodeError: + # 进程被杀时可能残留半行写入：跳过该行，对应帧视为未处理。 + continue + result[int(item["frame"])] = str(item["text"]) + return result + # 默认垃圾词：无文字帧的模型输出可能反复出现这些词。 def _sampling_interval(manifest: list[dict], default: float) -> float: """从帧清单时间轴推导采样间隔（相邻帧时间差的中位数）。 @@ -112,9 +151,30 @@ def invoke(request: InvokeRequest) -> InvokeResponse: if request.params.get(key) is not None } + # 节点级断点：读取已处理帧存档，只对未处理帧调用 vlm-ocr。 + output_dir = Path(request.output_dir) + output_dir.mkdir(parents=True, exist_ok=True) + partial_path = output_dir / _PARTIAL_NAME + partial_texts = _load_partial(output_dir) + pending = [(i, item) for i, item in enumerate(manifest) if i not in partial_texts] + if partial_texts: + logger.info( + "检测到节点级断点：%d/%d 帧已处理，本次只处理剩余 %d 帧", + len(partial_texts), len(manifest), len(pending), + ) + # 单帧 OCR：并行池的工作函数，返回该帧识别文本（失败/空/超长均返回空串）。 + # 每帧无论结果如何都把 {frame, text} 追加到断点存档，重启后不再重跑该帧。 def ocr_frame(payload) -> str: index, item = payload + # 暂停检查：调度器置 PAUSED 并向 run 根写入 paused.flag 后，工作线程 + # 立即中止（不 OCR、不写存档，该帧恢复时重跑），让 map 快速结束。 + if (Path(request.output_dir).parent.parent / _PAUSE_FLAG).exists(): + # 抑制后续进度回调：队列中剩余大量帧会快速失败退出，避免逐项 + # 打印"OCR 进度"导致日志井喷。 + pool.cancel() + raise PauseRequested(f"OCR 被暂停（run {request.run_id}）") + text = "" response = registry.invoke( "vlm-ocr", InvokeRequest( @@ -128,42 +188,62 @@ def invoke(request: InvokeRequest) -> InvokeResponse: if response.status != "completed": # 单帧失败不中断整体，跳过该帧继续汇总。 logger.warning("帧 %d OCR 失败，跳过: %s", index, response.error) - return "" - logger.info("帧 %d/%d OCR 完成: %r", index + 1, len(manifest), response.outputs.get("text")) - text = str(response.outputs.get("text", "")).strip() - if not text: - return "" - # 超长输出视为模型异常（重复循环等），直接报错并跳过该帧。 - if len(text) > max_result_chars: - logger.warning( - "帧 %d OCR 输出超长(%d > %d)，跳过: %r", - index, len(text), max_result_chars, text[:60], - ) - return "" + else: + logger.info("帧 %d/%d OCR 完成: %r", index + 1, len(manifest), response.outputs.get("text")) + text = str(response.outputs.get("text", "")).strip() + if len(text) > max_result_chars: + # 超长输出视为模型异常（重复循环等），直接报错并跳过该帧。 + logger.warning( + "帧 %d OCR 输出超长(%d > %d)，跳过: %r", + index, len(text), max_result_chars, text[:60], + ) + text = "" + # 断点存档：成功/失败/空串都记录"已处理"，恢复时保持一致行为。 + with _partial_lock: + with partial_path.open("a", encoding="utf-8") as fh: + fh.write(json.dumps({"frame": index, "text": text}, ensure_ascii=False) + "\n") return text - # 进度日志：打印已识别帧数、总数与平均处理速度（帧/s）。 - def log_progress(done: int, total: int, rate: float) -> None: - logger.info("OCR 进度 %d/%d 帧 (%.1f 帧/s)", done, total, rate) + if pending: + # 进度日志：打印已识别帧数、总数、平均处理速度（帧/s）、最近窗口平均 + # 单帧耗时与当前线程数——便于判断多线程是否因单帧处理过慢而未启用 + # （窗口平均响应 ≥ fast_threshold 时自适应池不会扩容）。 + def log_progress(done: int, total: int, rate: float, avg_time: float, workers: int) -> None: + logger.info( + "OCR 进度 %d/%d 帧 (%.1f 帧/s, 平均 %.2fs/帧, 线程 %d/%d)", + done, total, rate, avg_time, workers, pool.max_workers, + ) - # 自适应并发调用 vlm-ocr：10s 窗口内平均响应 < 0.3s 则加 1 线程（上限 - # pool_max_workers），> pool_slow_threshold 则减 1 线程（下限 1）， - # 按实测负载弹性伸缩，避免盲目并发压垮本地 Ollama。 - pool = AdaptiveThreadPool( - worker=ocr_frame, - on_progress=log_progress, - min_workers=int(request.params.get("pool_min_workers", 1)), - max_workers=int(request.params.get("pool_max_workers", 16)), - window_seconds=float(request.params.get("pool_window_seconds", 10.0)), - fast_threshold=float(request.params.get("pool_fast_threshold", 0.3)), - slow_threshold=float(request.params.get("pool_slow_threshold", 1.0)), - ) - texts = pool.map(list(enumerate(manifest))) + # 自适应并发调用 vlm-ocr：10s 窗口内平均响应 < 0.3s 则加 1 线程（上限 + # pool_max_workers），> pool_slow_threshold 则减 1 线程（下限 1）， + # 按实测负载弹性伸缩，避免盲目并发压垮本地 Ollama。 + pool = AdaptiveThreadPool( + worker=ocr_frame, + on_progress=log_progress, + min_workers=int(request.params.get("pool_min_workers", 1)), + max_workers=int(request.params.get("pool_max_workers", 16)), + window_seconds=float(request.params.get("pool_window_seconds", 10.0)), + fast_threshold=float(request.params.get("pool_fast_threshold", 0.3)), + slow_threshold=float(request.params.get("pool_slow_threshold", 1.0)), + ) + pending_texts = pool.map(pending) + # 任一工作线程检测到暂停信号即整体中止：已写盘的断点存档保留， + # resume 后从剩余帧续跑；以 failed 返回让调度器保持 PAUSED（不误报失败）。 + if any(isinstance(text, PauseRequested) for text in pending_texts): + return InvokeResponse(status="failed", error=f"OCR 被暂停（run {request.run_id}）") + # 新增结果按帧号归位，与断点存档合并成完整帧序文本列表。 + new_by_index = {i: t for (i, _item), t in zip(pending, pending_texts)} + else: + new_by_index = {} + # 完整帧序文本：优先取存档，其次取本次新增（未处理帧必有其一条目）。 + texts = [ + partial_texts.get(i, new_by_index.get(i, "")) + for i in range(len(manifest)) + ] # 按帧顺序合并连续相同字幕（与重组装旧数据共用 _merge_kept）。 kept = _merge_kept(manifest, texts) - output_dir = Path(request.output_dir) - output_dir.mkdir(parents=True, exist_ok=True) + # output_dir 已在断点初始化时创建（mkdir 幂等），此处直接使用。 output_path = output_dir / "subtitle.srt" output_path.write_text("\n".join(_assemble_srt(kept, interval)), encoding="utf-8") logger.info("字幕汇总完成: %d 条", len(kept)) diff --git a/scripts/probe_llm_rate_limit.py b/scripts/probe_llm_rate_limit.py new file mode 100644 index 0000000..c04dbe6 --- /dev/null +++ b/scripts/probe_llm_rate_limit.py @@ -0,0 +1,86 @@ +"""一次性诊断：实测 SiliconFlow API 在不同并发下的 429 限流情况。 + +用于确定 llm_filter 的安全并发上限（run_011d01f19999 在 20/4 并发下均 429）。 +""" +import json +import os +import threading +import time +import urllib.error +import urllib.request +from pathlib import Path + +from dotenv import load_dotenv + +load_dotenv(Path(".env")) + +API_BASE = os.getenv("LLM_API_BASE", "https://api.siliconflow.cn/v1/chat/completions") +API_KEY = os.getenv("LLM_API_KEY", "") +MODEL = os.getenv("LLM_MODEL", "Qwen/Qwen3.6-35B-A3B") + + +def one_call(idx: int) -> str: + """发送一次最小 LLM 请求，返回结果码：ok / 429 / 其他。""" + body = { + "model": MODEL, + "messages": [ + {"role": "system", "content": "你是字幕质量过滤器。"}, + {"role": "user", "content": f"【目标】测试请求 {idx}"}, + ], + "enable_thinking": False, + "max_tokens": 8, + } + req = urllib.request.Request( + API_BASE, + data=json.dumps(body).encode("utf-8"), + headers={"Content-Type": "application/json", "Authorization": f"Bearer {API_KEY}"}, + method="POST", + ) + try: + with urllib.request.urlopen(req, timeout=30) as resp: + resp.read() + return "ok" + except urllib.error.HTTPError as exc: + # 429 时尝试读取响应体里的限流信息 + try: + detail = exc.read().decode("utf-8")[:120] + except Exception: + detail = "" + retry_after = exc.headers.get("Retry-After") if exc.headers else None + return f"HTTP {exc.code} | Retry-After={retry_after} | {detail}" + except Exception as exc: # noqa: BLE001 + return f"{type(exc).__name__}: {exc}" + + +def probe(concurrency: int, total: int) -> list[str]: + """以给定并发发送 total 个请求，返回结果码列表。""" + results: list[str] = [] + lock = threading.Lock() + index = 0 + + def worker() -> None: + nonlocal index + while True: + with lock: + if index >= total: + return + i = index + index += 1 + with lock: + results.append(one_call(i)) + + threads = [threading.Thread(target=worker) for _ in range(concurrency)] + for t in threads: + t.start() + for t in threads: + t.join() + return results + + +if __name__ == "__main__": + for conc in (1, 2, 4, 8): + results = probe(conc, total=8) + ok = sum(1 for r in results if r == "ok") + other = [r for r in results if r != "ok"][:2] + print(f"并发 {conc:2d}: 8 请求 -> ok {ok}/8 | 其余样例: {other}") + time.sleep(2) # 组间休息，避免误伤 diff --git a/src/wov_app/db.py b/src/wov_app/db.py index 8285fa6..1f96a74 100644 --- a/src/wov_app/db.py +++ b/src/wov_app/db.py @@ -303,18 +303,38 @@ class Database: conn.execute("DELETE FROM artifacts WHERE run_id = ?", (run_id,)) def next_queued_run(self) -> dict[str, Any] | None: - """按创建时间返回最早一条可执行任务（排队或已暂停待续跑）。""" + """按创建时间返回最早一条排队（QUEUED）任务。 + + 只取 QUEUED：PAUSED 任务必须由用户显式 resume（转回 QUEUED）后调度器 + 才重新执行。修复回归——此前把 PAUSED 也当可执行任务拾起，execute_run + 会先置 RUNNING 再检查暂停，导致"点击暂停反而开始任务"。 + """ with self._connect() as conn: row = conn.execute( """ SELECT * FROM workflow_runs - WHERE status IN ('QUEUED', 'PAUSED') + WHERE status = 'QUEUED' ORDER BY created_at ASC LIMIT 1 """ ).fetchone() return self._parse_overrides(row) if row else None + def recover_interrupted_runs(self, updated_at: str) -> int: + """重启恢复：把遗留 RUNNING 任务恢复为 QUEUED，返回恢复数量。 + + 进程被杀/重启时 RUNNING 任务没有机会收尾：保持 RUNNING 会永久孤儿 + （next_queued_run 不拾起）。恢复为 QUEUED 后调度器会从产物表 + （restore_run_outputs）断点续跑，不重复已完成节点；用户主动暂停的 + PAUSED 任务保持不变，等待显式 resume。 + """ + with self._connect() as conn: + cur = conn.execute( + "UPDATE workflow_runs SET status = 'QUEUED', updated_at = ? WHERE status = 'RUNNING'", + (updated_at,), + ) + return cur.rowcount + def pause_run(self, run_id: str, updated_at: str) -> None: """暂停任务：置为 PAUSED；调度器会在节点边界检查并停止推进。""" with self._connect() as conn: diff --git a/src/wov_app/main.py b/src/wov_app/main.py index 1887397..caf1d1b 100644 --- a/src/wov_app/main.py +++ b/src/wov_app/main.py @@ -13,6 +13,7 @@ load_dotenv() import os # noqa: E402 from contextlib import asynccontextmanager # noqa: E402 +from datetime import datetime, timezone # noqa: E402 from pathlib import Path # noqa: E402 from fastapi import FastAPI # noqa: E402 @@ -38,7 +39,9 @@ async def lifespan(app: FastAPI): # 默认创建演示工作流，可关闭便于测试。 if os.getenv("WOV_AUTO_SEED", "1") == "1": seed_default_workflows(db) - + # 重启恢复：上次进程被杀时遗留的 RUNNING 任务恢复为 QUEUED， + # 调度器会从产物表断点续跑（不重复已完成节点）；PAUSED 保持等待显式 resume。 + db.recover_interrupted_runs(datetime.now(timezone.utc).isoformat()) scheduler = WorkflowScheduler(db, STORAGE_DIR) # 调度器默认开启，处理排队中的任务；测试可关闭后手动执行。 if os.getenv("WOV_SCHEDULER_ENABLED", "1") == "1": diff --git a/src/wov_app/routers/apps.py b/src/wov_app/routers/apps.py index 3b2fb93..6f75ef4 100755 --- a/src/wov_app/routers/apps.py +++ b/src/wov_app/routers/apps.py @@ -141,6 +141,10 @@ def retry_run(run_id: str, db: Database = Depends(_get_db)) -> dict: raise HTTPException(status_code=422, detail="only failed runs can be retried") # reset_run 会清空进度、错误和旧产物，确保从头开始。 db.reset_run(run_id, _now_iso()) + # 重试前清除可能残留的暂停信号（任务失败时信号文件可能仍在）。 + from wov_app.config import STORAGE_DIR + + (STORAGE_DIR / "runs" / run_id / "paused.flag").unlink(missing_ok=True) return {"id": run_id, "status": "QUEUED"} @@ -153,6 +157,13 @@ def pause_run(run_id: str, db: Database = Depends(_get_db)) -> dict: if run["status"] not in ("QUEUED", "RUNNING"): raise HTTPException(status_code=422, detail="only queued or running runs can be paused") db.pause_run(run_id, _now_iso()) + # 写入暂停信号文件：运行中的节点（如 OCR）逐帧检查到后立即中止， + # 由调度器保持 PAUSED；resume 时清除。 + from wov_app.config import STORAGE_DIR + + run_dir = STORAGE_DIR / "runs" / run_id + run_dir.mkdir(parents=True, exist_ok=True) + (run_dir / "paused.flag").write_text("", encoding="utf-8") return {"id": run_id, "status": "PAUSED"} @@ -165,6 +176,10 @@ def resume_run(run_id: str, db: Database = Depends(_get_db)) -> dict: if run["status"] != "PAUSED": raise HTTPException(status_code=422, detail="only paused runs can be resumed") db.resume_run(run_id, _now_iso()) + # 清除暂停信号文件，避免节点误判仍处于暂停状态。 + from wov_app.config import STORAGE_DIR + + (STORAGE_DIR / "runs" / run_id / "paused.flag").unlink(missing_ok=True) return {"id": run_id, "status": "QUEUED"} @router.delete("/api/runs/{run_id}") diff --git a/src/wov_app/scheduler.py b/src/wov_app/scheduler.py index c4c41e1..6c7fe2d 100755 --- a/src/wov_app/scheduler.py +++ b/src/wov_app/scheduler.py @@ -96,12 +96,18 @@ class WorkflowScheduler: def _loop(self) -> None: """轮询循环：有排队任务就立即执行，否则休眠一个间隔。""" while not self._stopping: - run = self.db.next_queued_run() - if run is not None: - self.execute_run(run["id"]) - else: + try: + run = self.db.next_queued_run() + if run is not None: + self.execute_run(run["id"]) + else: + time.sleep(self.interval_seconds) + except Exception: # noqa: BLE001 + # 单次轮询异常不杀死调度线程：曾因 next_queued_run/execute_run + # 的未捕获异常导致线程退出，任务永远停留在 QUEUED 不被拾起 + # （run_011d01f19999 实际发生）。记录后跳过本轮，下一轮继续。 + logger.exception("调度器轮询异常，跳过本轮") time.sleep(self.interval_seconds) - def _resolve_ref( self, ref: str, @@ -124,6 +130,12 @@ class WorkflowScheduler: # 任务不存在或不在可执行状态（排队/暂停）时直接返回，避免重复执行。 if run is None or run["status"] not in ("QUEUED", "PAUSED"): return + # 已暂停的任务不自动续跑：直接返回保持 PAUSED，等待用户显式 resume + # （resume 把状态转回 QUEUED 后才会真正执行）。修复回归——此前以 + # PAUSED 进入后立即置 RUNNING，节点循环的暂停检查永远不成立， + # 任务被复活继续执行（"点击暂停反而开始任务"）。 + if run["status"] == "PAUSED": + return # 工作流或版本记录丢失时把任务标记为失败。 workflow = self.db.get_workflow(run["workflow_id"]) @@ -143,6 +155,8 @@ class WorkflowScheduler: # 从已登记产物重建已完成节点的输出，支持暂停后断点续跑。 outputs_by_node = self.db.restore_run_outputs(run_id) run_started = time.monotonic() + # 清除可能残留的暂停信号（重启/异常中断后），避免本次执行误触发节点内暂停。 + (self.storage_dir / "runs" / run_id / "paused.flag").unlink(missing_ok=True) self.db.update_run(run_id, status="RUNNING", progress=0, updated_at=_now_iso()) try: for index, node_id in enumerate(ordered): @@ -249,7 +263,13 @@ class WorkflowScheduler: updated_at=_now_iso(), ) except Exception as exc: # noqa: BLE001 - # 任一步骤异常都结束任务并记录错误，等待用户重试。 + # 节点执行中被暂停（节点内检测到 paused.flag 而中止）：保持 PAUSED + # 等待用户 resume 从断点续跑，而不是把暂停误报为 FAILED。 + current = self.db.get_run(run_id) + if current is not None and current["status"] == "PAUSED": + logger.info("任务 %s 节点内被暂停，保持 PAUSED: %s", run_id, exc) + return + # 其余异常：结束任务并记录错误，等待用户重试。 self.db.update_run( run_id, status="FAILED", diff --git a/tests/test_adaptive_pool.py b/tests/test_adaptive_pool.py index a13af83..6716333 100644 --- a/tests/test_adaptive_pool.py +++ b/tests/test_adaptive_pool.py @@ -51,18 +51,96 @@ def test_pool_map_ordered_results() -> None: def test_pool_on_progress_callback() -> None: - """进度回调：每次完成触发一次，携带已完成数/总数/速度。""" - progress: list[tuple[int, int, float]] = [] + """进度回调：每次完成触发一次，携带已完成数/总数/速度/平均耗时/线程数。""" + progress: list[tuple[int, int, float, float, int]] = [] pool = AdaptiveThreadPool( worker=lambda item: item, - on_progress=lambda done, total, rate: progress.append((done, total, rate)), + on_progress=lambda done, total, rate, avg_time, workers: progress.append( + (done, total, rate, avg_time, workers) + ), ) pool.map([10, 20, 30]) assert [item[0] for item in progress] == [1, 2, 3] # 已完成数递增。 assert all(item[1] == 3 for item in progress) # 总数固定。 assert all(item[2] > 0 for item in progress) # 速度为正值。 + # 窗口未满时平均耗时回退为累计平均（>0）；线程数 ∈ [1, 上限]。 + assert all(item[3] > 0 for item in progress) + assert all(1 <= item[4] <= pool.max_workers for item in progress) + + +def test_pool_progress_reports_window_avg_after_first_window() -> None: + """窗口评估后：回调携带最近窗口平均耗时（扩缩容依据）与扩容后的线程数。 + + 覆盖 `_current_avg_time` 两个分支：窗口评估前回退累计平均，评估后使用 + 最近窗口平均（0.0s，响应远快于 fast_threshold 0.3s → 线程 +1）。 + """ + clock = FakeClock() + seen: list[tuple[int, int, float, float, int]] = [] + pool = AdaptiveThreadPool( + worker=lambda item: item, + min_workers=1, max_workers=16, + window_seconds=10.0, fast_threshold=0.3, + clock=clock, + on_progress=lambda done, total, rate, avg_time, workers: seen.append( + (done, total, rate, avg_time, workers) + ), + ) + clock.advance(11) # 首个任务完成即越过窗口 → 触发评估。 + pool.map(list(range(5))) + # 第一个完成的任务回调在窗口评估前：回退累计平均（>0）。 + assert seen[0][3] > 0 + # 窗口评估后：回调携带最近窗口平均（≈0.0），且线程已扩容到 2。 + assert any(item[3] == 0.0 for item in seen) + assert any(item[4] == 2 for item in seen) + assert pool.max_concurrency == 2 +def test_pool_cancel_suppresses_progress() -> None: + """worker 触发 cancel（如检测到暂停信号）后：剩余任务不再触发进度回调。 + + 暂停场景：队列中剩余的大量帧会逐帧快速失败退出，若每完成一项都打印 + 进度日志，会在数秒内打出上万行日志；cancel 后抑制后续进度回调。 + """ + progress: list[tuple[int, int, float, float, int]] = [] + + def worker(item): + if item == 1: + pool.cancel() # 模拟某帧检测到暂停信号。 + return item + + pool = AdaptiveThreadPool( + worker=worker, + on_progress=lambda done, total, rate, avg_time, workers: progress.append( + (done, total, rate, avg_time, workers) + ), + ) + pool.map([0, 1, 2, 3]) + # 只有 cancel 之前的任务（item=0）触发了进度回调。 + assert len(progress) == 1 + assert progress[0][1] == 4 # 总数仍是 4。 + + +def test_pool_cancel_resets_between_maps() -> None: + """取消状态按批（map）重置：下一批任务进度回调恢复正常。""" + progress: list[tuple[int, int, float, float, int]] = [] + + def worker(item): + if item == "stop": + pool.cancel() + return item + + pool = AdaptiveThreadPool( + worker=worker, + on_progress=lambda done, total, rate, avg_time, workers: progress.append( + (done, total, rate, avg_time, workers) + ), + ) + pool.map(["a", "stop", "b"]) + assert len(progress) == 1 # 只有 cancel 前的 a 触发回调。 + pool.map(["c", "d"]) + # 新一批恢复回调（done 从 1 重新计数）：共 3 次回调（1 + 2）。 + assert [item[0] for item in progress] == [1, 1, 2] def test_pool_map_empty() -> None: + """空输入：不启动任务，直接返回空列表。""" pool = AdaptiveThreadPool(worker=lambda item: item) assert pool.map([]) == [] @@ -146,3 +224,52 @@ def test_pool_survives_mixed_grow_shrink() -> None: ) out = pool.map(list(range(60))) assert out == list(range(60)) + + +def test_pool_report_failure_lowers_effective_max() -> None: + """消费错误（如 API 限流）临时降低有效最大线程数，下限为 min_workers。 + + 自适应：并发打到配额线触发 429 时，report_failure 收紧有效上限， + 后续请求减少从而避开持续限流。 + """ + pool = AdaptiveThreadPool(worker=lambda item: item, min_workers=1, max_workers=16) + assert pool._effective_max_workers == 16 + pool.report_failure() + assert pool._effective_max_workers == 15 + for _ in range(30): + pool.report_failure() + assert pool._effective_max_workers == 1 # 下限 min_workers。 + + +def test_pool_effective_max_recovers_after_clean_window() -> None: + """连续无错误窗口后有效上限逐步回升到 max_workers。""" + clock = FakeClock() + pool = AdaptiveThreadPool( + worker=lambda item: item, min_workers=1, max_workers=16, + window_seconds=10.0, fast_threshold=0.3, clock=clock, + ) + pool.report_failure() # 有效上限 16 -> 15。 + clock.advance(11) + pool._tick(0.01) # 错误所在窗口：上限不恢复。 + assert pool._effective_max_workers == 15 + clock.advance(11) + pool._tick(0.01) # 下一个干净窗口：恢复 +1。 + assert pool._effective_max_workers == 16 + pool._tick(0.01) # 窗口未满早退，上限不变。 + assert pool._effective_max_workers == 16 + + +def test_pool_decide_uses_effective_max() -> None: + """扩容上限按有效最大线程数：错误窗口内即使响应快也不超过收紧后的上限。""" + clock = FakeClock() + pool = AdaptiveThreadPool( + worker=lambda item: item, min_workers=1, max_workers=16, + window_seconds=10.0, fast_threshold=0.3, clock=clock, + ) + pool.report_failure() # 有效上限 16 -> 15。 + pool._resize(15) + pool._window_failures = 1 # 本窗口内仍有错误 → 不恢复上限。 + clock.advance(11) + pool._tick(0.01) # 响应快，但 15 已是有效上限 → 不扩。 + assert pool._target_workers == 15 + assert pool._effective_max_workers == 15 diff --git a/tests/test_apps_api.py b/tests/test_apps_api.py index 4567f3e..7775216 100644 --- a/tests/test_apps_api.py +++ b/tests/test_apps_api.py @@ -196,15 +196,22 @@ def test_pause_resume_run_api() -> None: run_id = uploaded.json()["id"] assert uploaded.json()["status"] == "QUEUED" + from wov_app.config import STORAGE_DIR + flag = STORAGE_DIR / "runs" / run_id / "paused.flag" + paused = client.post(f"/api/runs/{run_id}/pause") assert paused.status_code == 200 assert paused.json() == {"id": run_id, "status": "PAUSED"} assert client.get(f"/api/runs/{run_id}").json()["status"] == "PAUSED" + # 暂停时写入暂停信号文件，供运行中的节点（如 OCR）逐帧检查并中止。 + assert flag.exists() resumed = client.post(f"/api/runs/{run_id}/resume") assert resumed.status_code == 200 assert resumed.json() == {"id": run_id, "status": "QUEUED"} assert client.get(f"/api/runs/{run_id}").json()["status"] == "QUEUED" + # 继续时清除暂停信号，避免误触发节点内暂停。 + assert not flag.exists() # 非 PAUSED 任务不可继续。 assert client.post(f"/api/runs/{run_id}/resume").status_code == 422 diff --git a/tests/test_db.py b/tests/test_db.py index 18c3e47..f4ecb99 100644 --- a/tests/test_db.py +++ b/tests/test_db.py @@ -238,7 +238,12 @@ def test_db_migration_adds_param_overrides(tmp_path) -> None: def test_pause_resume_run(tmp_path) -> None: - """验证 pause_run/resume_run 的状态流转与 PAUSED 任务可被调度器取到。""" + """验证 pause_run/resume_run 的状态流转与 PAUSED 任务不被调度器自动拾起。 + + 修复回归：PAUSED 任务若被 next_queued_run 取到，execute_run 会把它复活为 + RUNNING 继续执行——"点击暂停反而开始任务"。暂停必须由用户显式 resume + （PAUSED → QUEUED）后调度器才重新执行。 + """ 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" @@ -255,13 +260,45 @@ def test_pause_resume_run(tmp_path) -> None: ) db.pause_run("run_p", now) assert db.get_run("run_p")["status"] == "PAUSED" - # PAUSED 任务会被 next_queued_run 取到（等待续跑）。 - assert db.next_queued_run()["id"] == "run_p" + # 已暂停的任务不会被调度器拾起（等待用户显式 resume）。 + assert db.next_queued_run() is None db.resume_run("run_p", now) assert db.get_run("run_p")["status"] == "QUEUED" assert db.next_queued_run()["id"] == "run_p" +def test_recover_interrupted_runs(tmp_path) -> None: + """重启恢复：遗留 RUNNING 任务恢复为 QUEUED（保留产物供断点续跑）。 + + 进程被杀/重启时 RUNNING 任务不会自动收尾，若保持 RUNNING 将永久孤儿 + （next_queued_run 不拾起、暂停后又被 execute_run 复活）。恢复为 QUEUED + 后调度器会从产物表断点续跑；用户主动暂停的 PAUSED 任务保持不变。 + """ + 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" + for run_id, status in (("run_orphan", "RUNNING"), ("run_paused", "PAUSED"), + ("run_done", "COMPLETED")): + db.create_run( + { + "id": run_id, + "workflow_id": "demo", + "workflow_version": 1, + "status": status, + "progress": 0.5, + "created_at": now, + "updated_at": now, + } + ) + recovered = db.recover_interrupted_runs("2026-01-02T00:00:00+00:00") + assert recovered == 1 # 只有 RUNNING 被恢复。 + assert db.get_run("run_orphan")["status"] == "QUEUED" + assert db.get_run("run_orphan")["updated_at"] == "2026-01-02T00:00:00+00:00" + assert db.get_run("run_paused")["status"] == "PAUSED" + assert db.get_run("run_done")["status"] == "COMPLETED" + # 恢复后调度器可拾起并断点续跑。 + assert db.next_queued_run()["id"] == "run_orphan" + def test_restore_run_outputs(tmp_path) -> None: """验证从产物重建节点输出（断点续跑的依据）。""" db = Database(tmp_path / "wov.db") diff --git a/tests/test_llm_filter.py b/tests/test_llm_filter.py index b0e8ed8..3f319b5 100644 --- a/tests/test_llm_filter.py +++ b/tests/test_llm_filter.py @@ -17,6 +17,7 @@ from pathlib import Path import pytest from nodes.llm_filter import ( + DEFAULT_OVERLAY_TOKENS, _dedup_key, _judge_category, _rule_verdict, @@ -81,7 +82,31 @@ class FakeLLM: payload = json.dumps({"choices": [{"message": {"content": content}}]}).encode() return FakeResponse(payload) +class FlakyLLM: + """模拟限流/服务端错误：前 failures 次抛 HTTPError(429/503)，之后正常返回。 + 用于验证 _judge_category 的退避重试：429/5xx 时让出时间重试，配合 + 自适应线程池"慢响应减线程"弹性，让并发自动回落到限流配额内。 + """ + + def __init__(self, failures: int, code: int = 429, answer: str = "dialogue") -> None: + self.failures = failures + self.code = code + self.answer = answer + self.calls = 0 + self.bodies: list[dict] = [] + + def __call__(self, request, timeout=None): + self.calls += 1 + self.bodies.append(json.loads(request.data.decode("utf-8"))) + if self.calls <= self.failures: + raise urllib.error.HTTPError( + request.full_url, self.code, "flaky", {}, None + ) + payload = json.dumps( + {"choices": [{"message": {"content": self.answer}}]} + ).encode() + return FakeResponse(payload) def _patch_llm(monkeypatch, contents: list[str] | None = None, decision_fn=None) -> FakeLLM: """替换 nodes.llm_filter 的 urlopen 为 FakeLLM 并返回实例。""" fake = FakeLLM(contents=contents, decision_fn=decision_fn) @@ -191,6 +216,34 @@ def test_judge_category_delete_classes(monkeypatch) -> None: for cat in ("garbage", "overlay", "noise"): _patch_llm(monkeypatch, [cat]) assert _judge_category(entries, 2, context_size=10, params={}) == cat +def test_judge_category_sanitizes_noise_context(monkeypatch) -> None: + """上下文净化：规则层可确定性识别的垃圾（横线/HTML 等）在喂给 LLM 前 + 替换为 [噪音] 占位，避免污染对目标条目的场景判断。 + + 修复回归：OCR 输出中相邻字幕混有大量覆盖层垃圾（---------------、HTML、 + Marketing 等），原样进入 LLM 上下文会让模型误判整段为"水印覆盖层"， + 把相邻的真实对话误删（run_011d01f19999 中 190 条含 ≥4 汉字的对话被删）。 + """ + entries = [ + {"start": "00:00:01,000", "end": "00:00:02,000", "text": "正常对话一"}, + {"start": "00:00:03,000", "end": "00:00:04,000", "text": "---------------"}, + {"start": "00:00:05,000", "end": "00:00:06,000", "text": "HTML"}, + {"start": "00:00:07,000", "end": "00:00:08,000", "text": "我是目标对话"}, + {"start": "00:00:09,000", "end": "00:00:10,000", "text": "---"}, + {"start": "00:00:11,000", "end": "00:00:12,000", "text": "正常对话二"}, + ] + fake = _patch_llm(monkeypatch, ["dialogue"]) + assert _judge_category(entries, 3, context_size=10, params={}) == "dialogue" + content = fake.bodies[0]["messages"][1]["content"] + lines = content.splitlines() + # 上下文只含过滤后的字幕：确定性垃圾条目被剔除，原文不进入 LLM。 + assert lines == ["正常对话一", "【目标】我是目标对话", "正常对话二"] + assert "---------------" not in content + assert "HTML" not in content + assert "---" not in content + # 系统提示词明确说明上下文已过滤装饰/水印符号。 + assert "过滤" in fake.bodies[0]["messages"][0]["content"] + def test_judge_category_repeat_and_unknown_kept(monkeypatch) -> None: @@ -215,6 +268,47 @@ def test_judge_category_model_and_auth(monkeypatch) -> None: assert fake.headers[0]["Authorization"] == "Bearer sk-test" +def test_judge_category_retries_on_429(monkeypatch) -> None: + """429 限流时指数退避重试，最终成功判定（配合弹性把并发压回配额内）。""" + entries = parse_srt(_SRT) + fake = FlakyLLM(failures=2, answer="dialogue") + monkeypatch.setattr("nodes.llm_filter.time.sleep", lambda s: None) + monkeypatch.setattr("nodes.llm_filter.urllib.request.urlopen", fake) + assert _judge_category(entries, 2, context_size=10, params={}) == "dialogue" + assert fake.calls == 3 # 失败 2 次 + 成功 1 次。 + + +def test_judge_category_retries_on_5xx(monkeypatch) -> None: + """服务端 5xx 同样退避重试。""" + entries = parse_srt(_SRT) + fake = FlakyLLM(failures=1, code=503, answer="dialogue") + monkeypatch.setattr("nodes.llm_filter.time.sleep", lambda s: None) + monkeypatch.setattr("nodes.llm_filter.urllib.request.urlopen", fake) + assert _judge_category(entries, 2, context_size=10, params={}) == "dialogue" + assert fake.calls == 2 + + +def test_judge_category_gives_up_after_retries(monkeypatch) -> None: + """重试耗尽仍失败时抛错：该条判定失败，任务失败后可重新处理数据。""" + entries = parse_srt(_SRT) + fake = FlakyLLM(failures=99) + monkeypatch.setattr("nodes.llm_filter.time.sleep", lambda s: None) + monkeypatch.setattr("nodes.llm_filter.urllib.request.urlopen", fake) + with pytest.raises(urllib.error.HTTPError): + _judge_category(entries, 2, context_size=10, params={}) + assert fake.calls == 3 # 最多尝试 3 次。 + + +def test_judge_category_other_errors_no_retry(monkeypatch) -> None: + """非 429/5xx 错误（如 400）不重试，直接抛出。""" + entries = parse_srt(_SRT) + fake = FlakyLLM(failures=1, code=400, answer="dialogue") + monkeypatch.setattr("nodes.llm_filter.time.sleep", lambda s: None) + monkeypatch.setattr("nodes.llm_filter.urllib.request.urlopen", fake) + with pytest.raises(urllib.error.HTTPError): + _judge_category(entries, 2, context_size=10, params={}) + assert fake.calls == 1 # 只调用一次。 + # --------------------------------------------------------------------------- # 去重键与删除判定 # --------------------------------------------------------------------------- @@ -229,8 +323,22 @@ def test_dedup_key_normalizes_whitespace_and_case() -> None: assert _dedup_key("你好 世界") != _dedup_key("你好世界2") +def test_should_delete_by_category_only() -> None: + """删除判定：garbage/overlay/noise 删，repeat/dialogue 留（短文本无保护）。""" + assert _should_delete("garbage", "x", min_keep_len=12) is True + assert _should_delete("overlay", "x", min_keep_len=12) is True + assert _should_delete("noise", "x", min_keep_len=12) is True + assert _should_delete("repeat", "x", min_keep_len=12) is False + assert _should_delete("dialogue", "x", min_keep_len=12) is False + + def test_should_delete_long_text_noise_protected() -> None: - """长文本保护：≥min_keep_len 时 noise 不删，garbage/overlay 仍删；短文本三类都删。""" + """长文本保护：≥min_keep_len 时 noise 不删（LLM 判定不稳的兜底）， + garbage/overlay 仍删；短文本 noise 可删。 + + 回归：移除保护后 run_011d01f19999 新增误删 124 条真实长对话 + （'很棒的表情呢 看 拍下来了吗' 等被 LLM 误判 noise），恢复保护。 + """ long_text = "不这么做的话 可没法胜任患者的对象" assert _should_delete("noise", long_text, min_keep_len=12) is False assert _should_delete("garbage", long_text, min_keep_len=12) is True @@ -240,6 +348,25 @@ def test_should_delete_long_text_noise_protected() -> None: assert _should_delete("dialogue", long_text, min_keep_len=12) is False +def test_rule_verdict_removes_domain_and_html_watermark() -> None: + """正则确定性过滤：网址域名/HTML 水印等"一定需要移除"的模式直接删除。 + + 覆盖真实案例：'98室[巴花堂] 水火地址 489155.com'（广告）、 + 'HTML code for a simple blue background'（OCR 识别出的网页水印）。 + """ + tokens = set(DEFAULT_OVERLAY_TOKENS) + for text in ( + "98室[巴花堂] 水火地址 489155.com", + "HTML code for a simple blue background", + "http://example.com/path", + "联系我们 admin@example.com", + ): + assert _rule_verdict(text, tokens) is True, text + # 正常对话不含垃圾模式 → 交 LLM 判断。 + assert _rule_verdict("青沼君 好可爱", tokens) is None + assert _rule_verdict("再见了 再见", tokens) is None + + # --------------------------------------------------------------------------- # invoke 全链路 # --------------------------------------------------------------------------- @@ -454,6 +581,39 @@ def test_invoke_llm_error(monkeypatch, tmp_path) -> None: ) assert response.status == "failed" assert "llm down" in response.error +def test_invoke_retries_rate_limited_entries(monkeypatch, tmp_path) -> None: + """判定遇 429（限流）时临时降并发并对失败条目重试，最终正常完成。 + + 自适应：多线程并发打到 SiliconFlow 配额线触发 429 时，worker 通知线程池 + report_failure 临时收紧最大并发，失败条目在收紧后重试一轮，避免整体失败 + （run_011d01f19999 在 20 并发下因 429 重试耗尽而 FAILED 的修复）。 + """ + srt = tmp_path / "in.srt" + srt.write_text( + "1\n00:00:01,000 --> 00:00:04,000\n第一句对话\n\n" + "2\n00:00:05,000 --> 00:00:08,000\n---------------\n\n" + "3\n00:00:09,000 --> 00:00:12,000\n第二句对话\n\n" + "4\n00:00:13,000 --> 00:00:16,000\nHTML\n", + encoding="utf-8", + ) + # 前 3 次调用都 429（重试耗尽抛错），之后成功：首次 map 部分条目失败， + # 二次重试在降并发后成功。 + fake = FlakyLLM(failures=3, answer="dialogue") + monkeypatch.setattr("nodes.llm_filter.time.sleep", lambda s: None) + monkeypatch.setattr("nodes.llm_filter.urllib.request.urlopen", fake) + response = invoke( + InvokeRequest( + run_id="r", node_instance_id="", + inputs={"srt_uri": str(srt)}, + params={}, + output_dir=str(tmp_path / "out"), + ) + ) + assert response.status == "completed", response.error + kept = parse_srt((tmp_path / "out" / "filtered.srt").read_text(encoding="utf-8")) + assert [e["text"] for e in kept] == ["第一句对话", "第二句对话"] + assert fake.calls >= 5 # 首次 map + 二次重试均有调用。 + def test_invoke_empty_srt(monkeypatch, tmp_path) -> None: diff --git a/tests/test_ocr_flow.py b/tests/test_ocr_flow.py index c27aa86..bee450d 100644 --- a/tests/test_ocr_flow.py +++ b/tests/test_ocr_flow.py @@ -497,3 +497,45 @@ def test_ocr_passes_short_text_through(monkeypatch, tmp_path) -> None: assert response.status == "completed", response.error srt = Path(response.outputs["srt_uri"]).read_text(encoding="utf-8") assert "SUB 001" in srt + + +def test_ocr_interrupts_on_pause_flag(monkeypatch, tmp_path) -> None: + """暂停信号：任务被暂停（paused.flag 存在）时 OCR 立即中断。 + + output_dir = /runs//steps/ocr，run 根目录为其父父目录； + 调度器/API 在暂停时向 run 根写入 paused.flag，节点逐帧检查到即中止： + 不调用 vlm-ocr、不写该帧断点存档，invoke 返回 failed（由调度器识别为 + "被暂停"并保持 PAUSED，等待 resume 后从断点续跑）。 + """ + image = tmp_path / "s0.png" + image.write_bytes(TEXT_IMG.read_bytes()) + frames = [(i * 2.0, image) for i in range(4)] + + # 暂停信号位于 run 根目录（steps/ocr 的父父目录）。 + run_root = tmp_path / "run_root" + out_dir = run_root / "steps" / "ocr" + out_dir.mkdir(parents=True) + (run_root / "paused.flag").write_text("", encoding="utf-8") + + calls: list[str] = [] + + def fake_vlm(node_id, request): + calls.append(request.inputs["image_uri"]) + return InvokeResponse(status="completed", outputs={"text": "SUB 001"}) + + monkeypatch.setattr("wov_app.registry.invoke", fake_vlm) + manifest = _frames_manifest(tmp_path, frames) + response = ocr_invoke( + InvokeRequest( + run_id="run_paused_flag", + node_instance_id="", + inputs={"frames_manifest": str(manifest)}, + params={}, + output_dir=str(out_dir), + ) + ) + # 节点以"被暂停"失败：调度器会据此保持 PAUSED 而不标 FAILED。 + assert response.status == "failed" + assert "暂停" in (response.error or "") + assert calls == [] # 一帧都没有真正 OCR。 + assert not (out_dir / "ocr_partial.jsonl").exists() # 未处理帧不入存档。 diff --git a/tests/test_scheduler.py b/tests/test_scheduler.py index e52fac8..9092c23 100644 --- a/tests/test_scheduler.py +++ b/tests/test_scheduler.py @@ -612,6 +612,132 @@ def test_execute_paused_run_not_run(tmp_path, monkeypatch) -> None: assert called == [] +def test_execute_paused_run_stays_paused(tmp_path, monkeypatch) -> None: + """PAUSED 任务不被 execute_run 复活：不置 RUNNING、不执行任何节点。 + + 修复回归：PAUSED 任务被调度器拾起后曾先置 RUNNING 再检查，节点循环 + 读到的是刚改的 RUNNING 状态，"暂停检查"永远不成立 → 任务被复活继续跑 + （用户观察到的"点击暂停反而开始任务"）。修复后 PAUSED 直接返回保持暂停。 + """ + db = _db(tmp_path) + db.upsert_workflow({"id": "flow", "name": "Flow", "published": 1, "latest_version": 1}) + db.create_workflow_version("flow", 1, _two_node_definition().to_dict()) + now = "2026-01-01T00:00:00+00:00" + db.create_run( + { + "id": "run_paused", + "workflow_id": "flow", + "workflow_version": 1, + "status": "PAUSED", + "progress": 0.5, + "current_node_id": "a", + "created_at": now, + "updated_at": now, + } + ) + called = [] + + def fake_invoke(node_type, request): + called.append(node_type) + return InvokeResponse(status="completed", outputs={}) + + monkeypatch.setattr(registry, "invoke", fake_invoke) + scheduler = WorkflowScheduler(db, tmp_path / "storage") + scheduler.execute_run("run_paused") + # 状态保持 PAUSED（未被置为 RUNNING），节点一个都不执行。 + assert db.get_run("run_paused")["status"] == "PAUSED" + assert called == [] + +def test_execute_paused_during_node_keeps_paused(tmp_path, monkeypatch) -> None: + """节点内被暂停（节点检测到暂停信号后中止）：保持 PAUSED 不标 FAILED。 + + 节点内暂停响应：subtitle-ocr 检查到 paused.flag 后中止并返回失败； + 调度器捕获节点异常时应检查任务状态——若已被置为 PAUSED（用户点了暂停）， + 则保持 PAUSED 等待 resume 从断点续跑，而不是覆盖为 FAILED。 + """ + db = _db(tmp_path) + db.upsert_workflow({"id": "flow", "name": "Flow", "published": 1, "latest_version": 1}) + db.create_workflow_version("flow", 1, _two_node_definition().to_dict()) + now = "2026-01-01T00:00:00+00:00" + db.create_run( + { + "id": "run_paused", + "workflow_id": "flow", + "workflow_version": 1, + "status": "QUEUED", + "progress": 0, + "input_uri": str(tmp_path / "in.txt"), + "created_at": now, + "updated_at": now, + } + ) + + def fake_invoke(node_type, request): + # 模拟节点内暂停：任务已被置 PAUSED，节点随后中止并抛异常。 + db.pause_run("run_paused", now) + raise RuntimeError("OCR interrupted by pause") + + monkeypatch.setattr(registry, "invoke", fake_invoke) + WorkflowScheduler(db, tmp_path / "storage").execute_run("run_paused") + # 保持 PAUSED，不标 FAILED、不写 error（等待用户 resume 断点续跑）。 + run = db.get_run("run_paused") + assert run["status"] == "PAUSED" + assert run["error"] is None + + +def test_scheduler_loop_survives_poll_exception(tmp_path, monkeypatch) -> None: + """调度轮询遇异常不退出线程：下一轮继续执行排队任务。 + + 修复回归：_loop 中 next_queued_run/execute_run 的未捕获异常曾杀死调度 + 线程（worker 进程只剩 uvicorn 主线程），任务永远停留在 QUEUED—— + run_011d01f19999 实际发生：回退 QUEUED 后调度器不再拾起，新配置 + （pool_max_workers=20）因此从未执行。 + """ + db = _db(tmp_path) + input_file = tmp_path / "input.txt" + input_file.write_text("resilient", encoding="utf-8") + _register_echo() + db.upsert_workflow({"id": "flow", "name": "Flow", "published": 1, "latest_version": 1}) + db.create_workflow_version("flow", 1, _echo_definition().to_dict()) + now = "2026-01-01T00:00:00+00:00" + db.create_run( + { + "id": "run_resilient", + "workflow_id": "flow", + "workflow_version": 1, + "status": "QUEUED", + "progress": 0, + "input_uri": str(input_file), + "created_at": now, + "updated_at": now, + } + ) + + # 第一次轮询抛异常（模拟数据库抖动等），后续正常。 + calls = {"n": 0} + real_next = db.next_queued_run + + def flaky_next(): + calls["n"] += 1 + if calls["n"] == 1: + raise RuntimeError("transient db error") + return real_next() + + monkeypatch.setattr(db, "next_queued_run", flaky_next) + scheduler = WorkflowScheduler(db, tmp_path / "storage", interval_seconds=0.05) + scheduler.start() + try: + deadline = time.monotonic() + 10 + while time.monotonic() < deadline: + if db.get_run("run_resilient")["status"] in {"COMPLETED", "FAILED"}: + break + time.sleep(0.1) + finally: + scheduler.stop() + # 第一次异常后调度线程仍存活，第二轮回合把任务执行完成。 + assert db.get_run("run_resilient")["status"] == "COMPLETED" + assert calls["n"] >= 2 + def test_execute_pause_after_last_node_keeps_paused(tmp_path, monkeypatch) -> None: """验证全部节点完成但运行中被暂停时保持 PAUSED；续跑补做收尾后完成。""" db = _db(tmp_path) diff --git a/tests/test_subtitle_ocr_order_threading.py b/tests/test_subtitle_ocr_order_threading.py index e15ac69..e8e08c1 100644 --- a/tests/test_subtitle_ocr_order_threading.py +++ b/tests/test_subtitle_ocr_order_threading.py @@ -213,3 +213,83 @@ class TestSubtitleOcrOrderUnderThreading: # ⑤ 已知真实内容存在于结果中（确认结果的代表性条目）。 for line in ("北冈小姐", "这是特别病房患者的病历表", "应该已经察觉到 至今为止的一切了吧"): assert line in confirmed, line + + +def test_ocr_resumes_from_partial_checkpoint(monkeypatch, tmp_path) -> None: + """节点级断点：预写部分帧存档后运行，只处理未处理帧，产物与全量一致。 + + 模拟中断时已落盘的 ocr_partial.jsonl（前 100 帧已处理）：invoke 应只对 + 未处理帧调用 vlm-ocr，并把存档文本与新增文本合并，最终 SRT 与全量一次 + 跑完逐字节一致——重启不浪费已处理的帧。 + """ + manifest, texts_by_frame = _load_full_data() + confirmed = CONFIRMED_SRT.read_text(encoding="utf-8") + out_dir = tmp_path / "resume" + partial_path = out_dir / "ocr_partial.jsonl" + partial_path.parent.mkdir(parents=True) + lines = [] + for i in range(100): + # 存档按 0-based 帧序号记录；manifest[i] 的帧号 = i+1。 + lines.append( + json.dumps({"frame": i, "text": texts_by_frame[i + 1]}, ensure_ascii=False) + ) + if i == 50: + lines.append("") # 空行：_load_partial 必须跳过，不视为一条记录。 + if i == 51: + lines.append("broken-json-line") # 损坏行（进程被杀残留）：跳过。 + partial_path.write_text("\n".join(lines) + "\n", encoding="utf-8") + + from nodes.subtitle_ocr import invoke as ocr_invoke + + fake = FakeVlmOcrApi(texts_by_frame, seed=99) + monkeypatch.setattr("wov_app.registry.invoke", fake) + response = ocr_invoke( + InvokeRequest( + run_id="resume_test", + node_instance_id="", + inputs={"frames_manifest": str(FULL_MANIFEST)}, + params={"pool_min_workers": 4, "pool_max_workers": 4}, + output_dir=str(out_dir), + ) + ) + assert response.status == "completed", response.error + srt = Path(response.outputs["srt_uri"]).read_text(encoding="utf-8") + assert srt == confirmed # 存档 + 新增合并结果与全量一致。 + # 只处理未处理帧：前 100 帧不再调用 vlm-ocr。 + + # 场景 2：先用真实逻辑完整跑一遍（生成存档），再以同一目录续跑—— + # 第二次 pending 为空，完全不调用 vlm-ocr，产物与第一次逐字节一致 + # （覆盖 _load_partial 全恢复路径，存档文本与处理逻辑天然一致）。 + out_dir_all = tmp_path / "resume_all" + fake_first = FakeVlmOcrApi(texts_by_frame, seed=7) + monkeypatch.setattr("wov_app.registry.invoke", fake_first) + resp_first = ocr_invoke( + InvokeRequest( + run_id="resume_all_test", + node_instance_id="", + inputs={"frames_manifest": str(FULL_MANIFEST)}, + params={"pool_min_workers": 4, "pool_max_workers": 4}, + output_dir=str(out_dir_all), + ) + ) + assert resp_first.status == "completed", resp_first.error + srt_first = Path(resp_first.outputs["srt_uri"]).read_text(encoding="utf-8") + assert srt_first == confirmed + assert len(fake_first.completed_frames) == TOTAL_FRAMES + + fake_second = FakeVlmOcrApi(texts_by_frame, seed=8) + monkeypatch.setattr("wov_app.registry.invoke", fake_second) + resp_second = ocr_invoke( + InvokeRequest( + run_id="resume_all_test", + node_instance_id="", + inputs={"frames_manifest": str(FULL_MANIFEST)}, + params={"pool_min_workers": 4, "pool_max_workers": 4}, + output_dir=str(out_dir_all), + ) + ) + assert resp_second.status == "completed", resp_second.error + srt_second = Path(resp_second.outputs["srt_uri"]).read_text(encoding="utf-8") + assert srt_second == srt_first # 存档恢复与一次跑完逐字节一致。 + assert len(fake_second.completed_frames) == 0 # 一帧都不重新调用。 + assert len(fake.completed_frames) == TOTAL_FRAMES - 100 diff --git a/web/assets/app.js b/web/assets/app.js index fbdb2ca..54c9757 100644 --- a/web/assets/app.js +++ b/web/assets/app.js @@ -124,6 +124,8 @@ async function loadWorkflows() { ${escapeHtml(workflow.latest_version)} ${workflow.published ? badge("ok") : badge("draft")} + + ${workflow.published ? "" : ``} @@ -480,7 +482,109 @@ async function deleteRun(runId) { } } -// 创建或更新工作流：解析 DAG JSON 后提交，随后刷新列表。 +// 当前编辑模式：null=新建；否则为正在编辑的工作流 ID（保存追加新版本）。 +let editingWorkflowId = null; + +// 切换编辑器模式：id 为空进入新建，否则锁定 ID 并提示保存将追加新版本。 +function setEditMode(id) { + editingWorkflowId = id; + const idInput = document.getElementById("workflowId"); + const mode = document.getElementById("editMode"); + const save = document.getElementById("saveWorkflow"); + if (!idInput || !mode || !save) return; + if (id) { + idInput.value = id; + idInput.disabled = true; // 编辑模式下锁定 ID，防止误建新工作流。 + mode.textContent = `编辑中：${id} —— 保存将追加为新版本。`; + save.textContent = "保存为新版本"; + } else { + idInput.disabled = false; + mode.textContent = "新建工作流：填写 ID 与 DAG 后保存。"; + save.textContent = "创建工作流"; + } +} + +// 新建模式：清空表单并预填演示 DAG 模板。 +function resetWorkflowForm() { + document.getElementById("workflowId").value = ""; + document.getElementById("workflowName").value = ""; + document.getElementById("workflowDescription").value = ""; + document.getElementById("workflowDefinition").value = JSON.stringify(DEMO_WORKFLOW, null, 2); + setEditMode(null); +} + +// 加载指定工作流的最新定义到编辑器（列表"编辑"按钮）。 +// 当前页面没有编辑器表单（如管理后台列表）时跳转到编排页并自动加载。 +async function loadWorkflowForEdit(workflowId) { + if (!document.getElementById("workflowName")) { + window.location.href = `/workflow.html?edit=${encodeURIComponent(workflowId)}`; + return; + } + try { + const workflow = await api(`/api/admin/workflows/${workflowId}`); + document.getElementById("workflowName").value = workflow.name || ""; + document.getElementById("workflowDescription").value = workflow.description || ""; + const latest = workflow.latest_version_data; + document.getElementById("workflowDefinition").value = latest + ? JSON.stringify(latest.definition, null, 2) + : JSON.stringify({ name: "", version: 1, nodes: [], edges: [], entry_inputs: {}, final_outputs: {} }, null, 2); + setEditMode(workflowId); + } catch (error) { + alert(`加载失败：${error.message}`); + } +} + +// 展示指定工作流的版本历史面板。 +async function loadWorkflowVersions(workflowId) { + try { + const versions = await api(`/api/admin/workflows/${workflowId}/versions`); + const tbody = document.getElementById("versionList"); + document.getElementById("versionPanelTitle").textContent = `版本历史：${workflowId}`; + tbody.innerHTML = versions.length + ? versions + .map( + (version) => ` + + v${escapeHtml(version.version)} + ${escapeHtml(formatTime(version.created_at))} + + `, + ) + .join("") + : '暂无版本'; + document.getElementById("versionPanel").hidden = false; + } catch (error) { + alert(`版本加载失败：${error.message}`); + } +} + +// 关闭版本历史面板。 +function closeVersionPanel() { + const panel = document.getElementById("versionPanel"); + if (panel) panel.hidden = true; +} + +// 把指定工作流的某历史版本定义加载回编辑器（可对比、回滚后保存为新版本）。 +async function loadVersionIntoEditor(workflowId, version) { + try { + const versions = await api(`/api/admin/workflows/${workflowId}/versions`); + const target = versions.find((item) => item.version === Number(version)); + if (!target) { + alert(`版本 v${version} 不存在`); + return; + } + const workflow = await api(`/api/admin/workflows/${workflowId}`); + document.getElementById("workflowName").value = workflow.name || ""; + document.getElementById("workflowDescription").value = workflow.description || ""; + document.getElementById("workflowDefinition").value = JSON.stringify(target.definition, null, 2); + setEditMode(workflowId); + closeVersionPanel(); + } catch (error) { + alert(`版本加载失败：${error.message}`); + } +} + +// 创建或保存工作流：新建时创建首个版本；编辑时（ID 已锁定）追加新版本。 async function createWorkflow() { const workflowId = document.getElementById("workflowId").value.trim(); const name = document.getElementById("workflowName").value.trim(); @@ -493,7 +597,7 @@ async function createWorkflow() { return; } try { - await api("/api/admin/workflows", { + const result = await api("/api/admin/workflows", { method: "POST", body: JSON.stringify({ id: workflowId || undefined, @@ -502,8 +606,14 @@ async function createWorkflow() { definition, }), }); - alert("工作流已保存"); + alert(`工作流已保存：${result.id} v${result.latest_version}`); await loadWorkflows(); + // 编辑模式保持编辑状态并刷新最新版本；新建模式回到空表单。 + if (editingWorkflowId) { + await loadWorkflowForEdit(editingWorkflowId); + } else { + resetWorkflowForm(); + } } catch (error) { alert(`保存失败：${error.message}`); } @@ -564,6 +674,22 @@ async function uploadVideo() { // 全局点击委托：按按钮上的 data-* 属性分发到对应操作。 document.addEventListener("click", (event) => { + const editButton = event.target.closest("[data-edit-workflow]"); + if (editButton) { + loadWorkflowForEdit(editButton.dataset.editWorkflow); + return; + } + const versionsButton = event.target.closest("[data-versions-workflow]"); + if (versionsButton) { + loadWorkflowVersions(versionsButton.dataset.versionsWorkflow); + return; + } + const loadVersionButton = event.target.closest("[data-load-version]"); + if (loadVersionButton) { + const [workflowId, version] = loadVersionButton.dataset.loadVersion.split(":"); + loadVersionIntoEditor(workflowId, version); + return; + } const deleteWorkflowButton = event.target.closest("[data-delete-workflow]"); if (deleteWorkflowButton) { deleteWorkflow(deleteWorkflowButton.dataset.deleteWorkflow); @@ -597,13 +723,27 @@ document.addEventListener("click", (event) => { // 页面初始化：预填 DAG、绑定按钮事件并加载对应页面数据。 document.addEventListener("DOMContentLoaded", async () => { - const workflowDefinition = document.getElementById("workflowDefinition"); - if (workflowDefinition) { - workflowDefinition.value = JSON.stringify(DEMO_WORKFLOW, null, 2); + // 工作流编辑器：默认进入新建模式并预填演示模板； + // 支持 URL 参数 ?edit= 直接加载指定工作流进行编辑。 + if (document.getElementById("workflowDefinition")) { + resetWorkflowForm(); + const params = new URLSearchParams(window.location.search); + const editId = params.get("edit"); + if (editId) { + await loadWorkflowForEdit(editId); + } } - const createWorkflowButton = document.getElementById("createWorkflow"); - if (createWorkflowButton) { - createWorkflowButton.addEventListener("click", createWorkflow); + const saveWorkflowButton = document.getElementById("saveWorkflow"); + if (saveWorkflowButton) { + saveWorkflowButton.addEventListener("click", createWorkflow); + } + const resetWorkflowButton = document.getElementById("resetWorkflow"); + if (resetWorkflowButton) { + resetWorkflowButton.addEventListener("click", resetWorkflowForm); + } + const closeVersionsButton = document.getElementById("closeVersions"); + if (closeVersionsButton) { + closeVersionsButton.addEventListener("click", closeVersionPanel); } const publishWorkflowButton = document.getElementById("publishWorkflow"); if (publishWorkflowButton) { diff --git a/web/workflow.html b/web/workflow.html index 02cdf0b..fcbf018 100755 --- a/web/workflow.html +++ b/web/workflow.html @@ -21,26 +21,49 @@

工作流编排

- +
-

工作流定义

- - +

工作流编辑器

+

新建工作流：填写 ID 与 DAG 后保存。

+ + - + - - + +
- + +
- + + + +
-

已发布工作流

+

工作流列表

diff --git a/workflows/ocr-subtitle.json b/workflows/ocr-subtitle.json index ba1a59b..2d6e68c 100644 --- a/workflows/ocr-subtitle.json +++ b/workflows/ocr-subtitle.json @@ -2,10 +2,10 @@ "id": "ocr-subtitle", "name": "字幕OCR提取", "description": "抽帧并 OCR 提取视频烧录字幕，经 LLM 过滤无意义内容后生成带时间轴的 SRT 基准数据。", - "version": 4, + "version": 7, "definition": { "name": "字幕OCR提取", - "version": 4, + "version": 7, "nodes": [ { "id": "extract", @@ -29,7 +29,7 @@ "params": { "prompt": "提取图像中的文字，不要描述图片中的内容", "pool_min_workers": 1, - "pool_max_workers": 1 + "pool_max_workers": 20 }, "inputs": { "frames_manifest": "extract.frames_manifest" @@ -40,7 +40,8 @@ "node_type": "llm-filter", "params": { "pool_min_workers": 1, - "pool_max_workers": 1 + "pool_max_workers": 20, + "pool_fast_threshold": 1 }, "inputs": { "srt_uri": "ocr.srt_uri" @@ -64,4 +65,4 @@ "srt": "filter.srt_uri" } } -} +} \ No newline at end of file