Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
4ebbfc5198 | ||
|
|
5ffa2ac39e | ||
|
|
2b3a650612 |
@@ -54,9 +54,9 @@ vrsub/
|
|||||||
| `faster-whisper` | `audio_uri`(16kHz 单声道) | `srt_uri` | 参数:`language`、`task`、`model_path`、`device`、`compute_type`、`beam_size`、`vad_filter`(默认开)、`condition_on_previous_text`、`chunk_seconds` |
|
| `faster-whisper` | `audio_uri`(16kHz 单声道) | `srt_uri` | 参数:`language`、`task`、`model_path`、`device`、`compute_type`、`beam_size`、`vad_filter`(默认开)、`condition_on_previous_text`、`chunk_seconds` |
|
||||||
| `llm-translate` | `srt_uri` | `cn_srt_uri` | 参数:`target_language`、`model` |
|
| `llm-translate` | `srt_uri` | `cn_srt_uri` | 参数:`target_language`、`model` |
|
||||||
| `vlm-ocr` | `image_uri` | `text`、`text_uri` | 直接调本地 Ollama 多模态模型(glm-ocr)的 `/api/chat` 做视频帧 OCR(流式 + 5s 上限),参数:`model`、`ollama_host`、`prompt`、`timeout_seconds`、`keep_alive`、`num_predict`、`temperature`、`repeat_penalty` |
|
| `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) |
|
| `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` |
|
| `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 过滤无意义字幕(自适应线程池并发判断):每条连同前后各 `context_size`(默认 10)条纯文本(不含时间戳)分批给 LLM,仅判断目标字幕是否多余/无意义,判定删除则该条连同时间戳移除并重新编号,参数:`context_size`、`model`、`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 输出) |
|
||||||
| `srt-to-dual-eye-ass` | `cn_srt_uri` | `ass_uri` | 参数:`resolution`,如 `3840x1920` |
|
| `srt-to-dual-eye-ass` | `cn_srt_uri` | `ass_uri` | 参数:`resolution`,如 `3840x1920` |
|
||||||
|
|
||||||
### 模型权重解析(本地优先)
|
### 模型权重解析(本地优先)
|
||||||
@@ -269,7 +269,12 @@ http://127.0.0.1:8000/docs API 文档
|
|||||||
缺失时测试跳过而非现场生成。大体积视频素材放 `data/testdata/`(gitignored)。OCR 相关资产:
|
缺失时测试跳过而非现场生成。大体积视频素材放 `data/testdata/`(gitignored)。OCR 相关资产:
|
||||||
`ocr_text.png`(有文字)、`ocr_notext.png`(无文字帧)、`subtitle_10s.mp4`
|
`ocr_text.png`(有文字)、`ocr_notext.png`(无文字帧)、`subtitle_10s.mp4`
|
||||||
(烧录 SUB 001@1-4s / SUB 002@6-9s 的 10s 测试视频)、`test_real_hav_sub.png`
|
(烧录 SUB 001@1-4s / SUB 002@6-9s 的 10s 测试视频)、`test_real_hav_sub.png`
|
||||||
(真实视频字幕截图,VLM 集成测试期望识别出"还有没有什么困扰 或者奇怪的地方吗")。
|
(真实视频字幕截图,VLM 集成测试期望识别出"还有没有什么困扰 或者奇怪的地方吗")、
|
||||||
|
`ocr_srt_run_ac7f480a3ccb.srt`(真实任务 1666 条 OCR 输出,llm-filter 回归)、
|
||||||
|
`frames_manifest_full.json` + `ocr_frames_full.json`(真实任务 run_ac7f480a3ccb
|
||||||
|
**全部 14236 帧**的帧清单与逐帧 OCR 文本,多线程顺序测试常驻夹具;配合
|
||||||
|
`ocr_srt_run_ac7f480a3ccb.srt` 作为单线程确认基线,见
|
||||||
|
`tests/test_subtitle_ocr_order_threading.py`)。
|
||||||
- **开发流程强制 TDD(红-绿-重构)**:任何新功能/修复必须先写失败测试(红),
|
- **开发流程强制 TDD(红-绿-重构)**:任何新功能/修复必须先写失败测试(红),
|
||||||
再实现最小代码让其通过(绿),最后重构保持整洁;不允许先写实现后补测试。
|
再实现最小代码让其通过(绿),最后重构保持整洁;不允许先写实现后补测试。
|
||||||
- 测试运行:`uv run pytest`;全部测试位于 `tests/`。
|
- 测试运行:`uv run pytest`;全部测试位于 `tests/`。
|
||||||
|
|||||||
@@ -111,6 +111,24 @@ def _parse_progress_line(line: str) -> int | None:
|
|||||||
except ValueError:
|
except ValueError:
|
||||||
return None
|
return None
|
||||||
|
|
||||||
|
|
||||||
|
def _sorted_frame_files(frames_dir: Path) -> list[Path]:
|
||||||
|
"""按文件名中的帧号数值排序返回帧文件列表(自然排序,非字典序)。
|
||||||
|
|
||||||
|
关键点:ffmpeg 的 %04d 编号在超过 9999 帧后会自动扩为 5 位
|
||||||
|
(frame_10000.png 等),此时 sorted() 默认的字典序会把 5 位编号排在
|
||||||
|
4 位编号之前(如 "frame_10009" < "frame_1009"),导致帧号回退、
|
||||||
|
manifest 时间与图像错位(曾真实发生于 run_339ec7ee437f 的 14236 帧
|
||||||
|
任务,全片后半段时间轴全部错乱)。必须解析出帧号按数值排序,
|
||||||
|
才能保证"第 k 个文件 = 第 k 个选中帧 = 时间 index*step/fps"成立。
|
||||||
|
"""
|
||||||
|
def frame_number(path: Path) -> int:
|
||||||
|
# 文件名形如 frame_0001.png,取下划线后的数字部分。
|
||||||
|
return int(path.stem.split("_", 1)[1])
|
||||||
|
|
||||||
|
return sorted(frames_dir.glob("frame_*.png"), key=frame_number)
|
||||||
|
|
||||||
|
|
||||||
def invoke(request: InvokeRequest) -> InvokeResponse:
|
def invoke(request: InvokeRequest) -> InvokeResponse:
|
||||||
"""按帧间隔抽取并裁切视频帧,输出 frames.json 清单。"""
|
"""按帧间隔抽取并裁切视频帧,输出 frames.json 清单。"""
|
||||||
video_uri = request.inputs.get("video_uri")
|
video_uri = request.inputs.get("video_uri")
|
||||||
@@ -203,7 +221,7 @@ def invoke(request: InvokeRequest) -> InvokeResponse:
|
|||||||
return InvokeResponse(status="failed", error=stderr[-500:] or "ffmpeg failed")
|
return InvokeResponse(status="failed", error=stderr[-500:] or "ffmpeg failed")
|
||||||
|
|
||||||
# 第 k 个输出文件对应原始帧号 k×step,时间 = 帧号 / fps(帧精确,无累计偏差)。
|
# 第 k 个输出文件对应原始帧号 k×step,时间 = 帧号 / fps(帧精确,无累计偏差)。
|
||||||
files = sorted(frames_dir.glob("frame_*.png"))
|
files = _sorted_frame_files(frames_dir)
|
||||||
manifest = [
|
manifest = [
|
||||||
{"time": round((index * step) / fps, 3), "image_uri": str(path)}
|
{"time": round((index * step) / fps, 3), "image_uri": str(path)}
|
||||||
for index, path in enumerate(files)
|
for index, path in enumerate(files)
|
||||||
|
|||||||
@@ -1,9 +1,30 @@
|
|||||||
"""LLM 字幕过滤节点。
|
"""LLM 字幕过滤节点。
|
||||||
|
|
||||||
对 OCR 识别出的 SRT 字幕做二次过滤:为避免字幕上下文过长,把每条字幕连同其
|
对 OCR 识别出的 SRT 字幕做二次过滤,两级判断:
|
||||||
前后各 context_size 条字幕(纯文本,**不含时间戳**)分批提供给 LLM,模型仅判断
|
|
||||||
目标字幕是否属于多余、无意义的字符(如重复、残缺、无实际语义的杂项);
|
1. **确定性规则层**(不调用 LLM):横线装饰、HTML/水印 token、URL/邮箱、
|
||||||
判定无意义则删除该条(连同其时间戳),其余字幕保持原样并重新编号输出。
|
单双 ASCII 字符等 OCR 噪声直接删除——这些模式是稳定可判的,走规则
|
||||||
|
既省 token 又保证结果确定(真实数据中占删除量的 65%+)。
|
||||||
|
|
||||||
|
2. **LLM 五类分类层**:把目标字幕连同前后各 context_size 条纯文本分批
|
||||||
|
提供给 LLM,模型输出五个类别之一:
|
||||||
|
- garbage:垃圾字符(乱码、残缺、装饰性符号)→ 删除
|
||||||
|
- overlay:水印/网页/播放器等覆盖层文本 → 删除
|
||||||
|
- noise:与上下文无关、无实际语义的杂项 → 删除
|
||||||
|
- repeat:内容性重复(语气词、呻吟、重复感叹,属于内容本身)→ 保留
|
||||||
|
- dialogue:正常对话 → 保留
|
||||||
|
未识别输出一律回退 dialogue(宁滥勿缺,避免误删真实对话)。
|
||||||
|
|
||||||
|
3. **文本去重**:相同文本(忽略全部空白与大小写差异)只调一次 LLM,
|
||||||
|
上下文取首次出现位置,结果缓存复用——修复"同一句字幕 5 留 6 删"的
|
||||||
|
判定不一致,同时把长视频的 LLM 调用量降到唯一文本数。
|
||||||
|
|
||||||
|
4. **长文本保护**:长度 ≥ min_keep_len 的文本,仅 garbage/overlay 两个
|
||||||
|
明确垃圾类别可删,noise 不构成删除依据,防止长对话被误删。
|
||||||
|
|
||||||
|
参考真实任务 run_ac7f480a3ccb(2026-08,OCR 1666 条):旧实现把 394 条
|
||||||
|
真实对话当噪声删掉(占删除 35%)、同文本判定不一致;新实现按上述机制
|
||||||
|
回归测试已固化在 tests/test_llm_filter.py。
|
||||||
"""
|
"""
|
||||||
|
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
@@ -13,6 +34,7 @@ import os
|
|||||||
import re
|
import re
|
||||||
import urllib.request
|
import urllib.request
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
|
|
||||||
from nodes.adaptive_pool import AdaptiveThreadPool
|
from nodes.adaptive_pool import AdaptiveThreadPool
|
||||||
from wov_app.logging import get_logger
|
from wov_app.logging import get_logger
|
||||||
from wov_sdk.models import InvokeRequest, InvokeResponse
|
from wov_sdk.models import InvokeRequest, InvokeResponse
|
||||||
@@ -31,6 +53,37 @@ TARGET_MARK = "【目标】"
|
|||||||
# 默认上下文窗口:目标字幕前后各取 10 条。
|
# 默认上下文窗口:目标字幕前后各取 10 条。
|
||||||
DEFAULT_CONTEXT_SIZE = 10
|
DEFAULT_CONTEXT_SIZE = 10
|
||||||
|
|
||||||
|
# LLM 输出类别:garbage/overlay/noise 删除;repeat/dialogue 保留。
|
||||||
|
CATEGORY_GARBAGE = "garbage"
|
||||||
|
CATEGORY_OVERLAY = "overlay"
|
||||||
|
CATEGORY_NOISE = "noise"
|
||||||
|
CATEGORY_REPEAT = "repeat"
|
||||||
|
CATEGORY_DIALOGUE = "dialogue"
|
||||||
|
_ALL_CATEGORIES = (
|
||||||
|
CATEGORY_GARBAGE,
|
||||||
|
CATEGORY_OVERLAY,
|
||||||
|
CATEGORY_NOISE,
|
||||||
|
CATEGORY_REPEAT,
|
||||||
|
CATEGORY_DIALOGUE,
|
||||||
|
)
|
||||||
|
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+$")
|
||||||
|
# 默认水印/覆盖层 token(casefold 后比较,可经 overlay_tokens 参数覆盖)。
|
||||||
|
DEFAULT_OVERLAY_TOKENS = frozenset(
|
||||||
|
{"html", "background", "___", "cleaning", "buffering", "loading"}
|
||||||
|
)
|
||||||
|
# 长文本保护阈值:≥ 该长度的文本,noise 类别不构成删除依据。
|
||||||
|
DEFAULT_MIN_KEEP_LEN = 12
|
||||||
|
|
||||||
|
|
||||||
|
def _has_cjk(text: str) -> bool:
|
||||||
|
"""是否含 CJK 汉字:单字"嗯/好"等可能是内容,规则层不直接删。"""
|
||||||
|
return any("\u4e00" <= ch <= "\u9fff" for ch in text)
|
||||||
|
|
||||||
|
|
||||||
def parse_srt(text: str) -> list[dict]:
|
def parse_srt(text: str) -> list[dict]:
|
||||||
"""解析 SRT 文本为条目列表:[{"start", "end", "text"}]。"""
|
"""解析 SRT 文本为条目列表:[{"start", "end", "text"}]。"""
|
||||||
@@ -55,13 +108,58 @@ def serialize_srt(entries: list[dict]) -> str:
|
|||||||
return "\n\n".join(blocks) + "\n"
|
return "\n\n".join(blocks) + "\n"
|
||||||
|
|
||||||
|
|
||||||
def _judge_target(
|
def _rule_verdict(text: str, overlay_tokens: set[str]) -> bool | None:
|
||||||
|
"""确定性规则层:返回 True(删除)/ None(交给 LLM 多维判断)。
|
||||||
|
|
||||||
|
规则覆盖 OCR 噪声的稳定模式:空文本、纯横线装饰、URL/邮箱、水印 token
|
||||||
|
(HTML/background 等)、单双 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):
|
||||||
|
return True
|
||||||
|
if t.casefold() in overlay_tokens:
|
||||||
|
return True
|
||||||
|
if len(t) <= 2 and not _has_cjk(t):
|
||||||
|
return True
|
||||||
|
return None
|
||||||
|
|
||||||
|
|
||||||
|
def _dedup_key(text: str) -> str:
|
||||||
|
"""文本去重键:去掉全部空白并统一大小写。
|
||||||
|
|
||||||
|
OCR 同一句字幕常带/不带空格("可没法胜任" vs "可 没法胜任"),
|
||||||
|
视为同一文本以保证判定一致;也用于把 LLM 调用量降到唯一文本数。
|
||||||
|
"""
|
||||||
|
return "".join(text.split()).casefold()
|
||||||
|
|
||||||
|
|
||||||
|
def _should_delete(category: str, text: str, min_keep_len: int) -> bool:
|
||||||
|
"""按 LLM 类别与长文本保护决定是否删除。
|
||||||
|
|
||||||
|
repeat/dialogue 一律保留;garbage/overlay 一律删除(含长文本——
|
||||||
|
这两个是明确的垃圾信号);noise 对短文本删除,但 ≥min_keep_len 的
|
||||||
|
长文本不删(noise 太模糊,不足以推翻一句完整台词)。
|
||||||
|
"""
|
||||||
|
if category not in DELETE_CATEGORIES:
|
||||||
|
return False
|
||||||
|
if category == CATEGORY_NOISE and len(text) >= min_keep_len:
|
||||||
|
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
|
||||||
) -> bool:
|
) -> str:
|
||||||
"""调用 LLM 判断目标字幕是否多余/无意义;返回 True 表示应删除。
|
"""调用 LLM 把目标字幕归入五类之一,返回类别词(未识别回退 dialogue)。
|
||||||
|
|
||||||
请求体只含目标字幕及其前后各 context_size 条字幕的纯文本(无时间戳),
|
请求体只含目标字幕及其前后各 context_size 条字幕的纯文本(无时间戳),
|
||||||
目标字幕用 TARGET_MARK 标记;模型只需回答"保留"或"删除"。
|
目标字幕用 TARGET_MARK 标记;模型只输出一个类别英文单词。
|
||||||
|
输出无法识别(空/乱码/旧式"保留")时回退 dialogue,宁滥勿缺。
|
||||||
"""
|
"""
|
||||||
start = max(0, index - context_size)
|
start = max(0, index - context_size)
|
||||||
end = min(len(entries), index + context_size + 1)
|
end = min(len(entries), index + context_size + 1)
|
||||||
@@ -81,9 +179,14 @@ def _judge_target(
|
|||||||
model = str(params.get("model") or os.getenv("LLM_MODEL", "Qwen/Qwen3.6-35B-A3B"))
|
model = str(params.get("model") or os.getenv("LLM_MODEL", "Qwen/Qwen3.6-35B-A3B"))
|
||||||
system_prompt = (
|
system_prompt = (
|
||||||
"你是字幕质量过滤器。用户会提供一段字幕序列(纯文本,不含时间戳),"
|
"你是字幕质量过滤器。用户会提供一段字幕序列(纯文本,不含时间戳),"
|
||||||
f"其中用{TARGET_MARK}标记的字幕是需要判断的目标。请判断该字幕是否属于"
|
f"其中用{TARGET_MARK}标记的字幕是需要判断的目标。"
|
||||||
"多余、无意义的字符(如重复、残缺、无实际语义的杂项)。"
|
"请把目标字幕归入以下五个类别之一:\n"
|
||||||
"只回答两个字:保留 或 删除,不要输出其他内容。"
|
"garbage:垃圾字符(乱码、残缺、装饰性符号、横线)\n"
|
||||||
|
"overlay:水印、网页/播放器/字幕组等覆盖层文本,不是视频对白\n"
|
||||||
|
"noise:与上下文无关、无实际语义的杂项\n"
|
||||||
|
"repeat:内容性重复(如语气词、呻吟、重复的感叹或对话),属于内容本身\n"
|
||||||
|
"dialogue:正常对话\n"
|
||||||
|
"只输出一个类别英文单词,不要输出其他内容。"
|
||||||
)
|
)
|
||||||
body = {
|
body = {
|
||||||
"model": model,
|
"model": model,
|
||||||
@@ -94,7 +197,7 @@ def _judge_target(
|
|||||||
# 关闭推理模式:Qwen3 等模型默认会把思考过程写入 reasoning_content,
|
# 关闭推理模式:Qwen3 等模型默认会把思考过程写入 reasoning_content,
|
||||||
# 导致 content 为空或包含多余内容。
|
# 导致 content 为空或包含多余内容。
|
||||||
"enable_thinking": False,
|
"enable_thinking": False,
|
||||||
# 只输出"保留/删除",输出上限给得很小即可。
|
# 类别词很短,输出上限给得很小即可。
|
||||||
"max_tokens": 16,
|
"max_tokens": 16,
|
||||||
}
|
}
|
||||||
headers = {"Content-Type": "application/json"}
|
headers = {"Content-Type": "application/json"}
|
||||||
@@ -109,13 +212,23 @@ def _judge_target(
|
|||||||
)
|
)
|
||||||
with urllib.request.urlopen(request, timeout=request_timeout) as response:
|
with urllib.request.urlopen(request, timeout=request_timeout) as response:
|
||||||
payload = json.loads(response.read().decode("utf-8"))
|
payload = json.loads(response.read().decode("utf-8"))
|
||||||
content = str(payload["choices"][0]["message"]["content"])
|
content = str(payload["choices"][0]["message"]["content"]).strip().lower()
|
||||||
# 模型回答含"删除"即视为该条无意义;其余情况(保留/异常)一律保留,宁多勿删。
|
# 精确匹配五个类别词(兼容"garbagexxx"这类多余输出)。
|
||||||
return "删除" in content
|
for category in _ALL_CATEGORIES:
|
||||||
|
if content == category or content.startswith(category):
|
||||||
|
return category
|
||||||
|
# 旧式"删除/保留"回答兼容:含"删除"视为垃圾,其余一律保留。
|
||||||
|
if "删除" in content:
|
||||||
|
return CATEGORY_GARBAGE
|
||||||
|
return CATEGORY_DIALOGUE
|
||||||
|
|
||||||
|
|
||||||
def invoke(request: InvokeRequest) -> InvokeResponse:
|
def invoke(request: InvokeRequest) -> InvokeResponse:
|
||||||
"""过滤 SRT 中多余/无意义的字幕,产物为 filtered.srt。"""
|
"""过滤 SRT 中多余/无意义的字幕,产物为 filtered.srt。
|
||||||
|
|
||||||
|
流程:规则层(确定性删除)→ LLM 层(去重后按唯一文本多维分类)→
|
||||||
|
长文本保护 → 保留条重新编号输出。
|
||||||
|
"""
|
||||||
srt_uri = request.inputs.get("srt_uri")
|
srt_uri = request.inputs.get("srt_uri")
|
||||||
if not srt_uri:
|
if not srt_uri:
|
||||||
return InvokeResponse(status="failed", error="srt_uri is required")
|
return InvokeResponse(status="failed", error="srt_uri is required")
|
||||||
@@ -125,17 +238,44 @@ def invoke(request: InvokeRequest) -> InvokeResponse:
|
|||||||
|
|
||||||
entries = parse_srt(srt_path.read_text(encoding="utf-8"))
|
entries = parse_srt(srt_path.read_text(encoding="utf-8"))
|
||||||
context_size = int(request.params.get("context_size", DEFAULT_CONTEXT_SIZE))
|
context_size = int(request.params.get("context_size", DEFAULT_CONTEXT_SIZE))
|
||||||
# 单条判断的工作函数:返回 True 表示该条应删除。
|
min_keep_len = int(request.params.get("min_keep_len", DEFAULT_MIN_KEEP_LEN))
|
||||||
def judge_one(index) -> bool:
|
# 水印 token 可经参数覆盖(JSON 数组字符串或列表),默认内置常见覆盖层词。
|
||||||
return _judge_target(entries, index, context_size, request.params)
|
raw_tokens = request.params.get("overlay_tokens")
|
||||||
|
if isinstance(raw_tokens, str) and raw_tokens.strip():
|
||||||
|
raw_tokens = json.loads(raw_tokens)
|
||||||
|
overlay_tokens = {
|
||||||
|
str(t).casefold() for t in (raw_tokens or DEFAULT_OVERLAY_TOKENS)
|
||||||
|
}
|
||||||
|
# 去重开关:默认开;关掉时每个条目独立调用 LLM(不省调用,判定各自独立)。
|
||||||
|
dedupe = str(request.params.get("dedupe", "1")) not in ("0", "false", "False")
|
||||||
|
|
||||||
|
# 阶段 1:确定性规则层(不调 LLM)。
|
||||||
|
rule_verdicts = [_rule_verdict(entry["text"], overlay_tokens) for entry in entries]
|
||||||
|
llm_needed = [i for i, verdict in enumerate(rule_verdicts) if verdict is None]
|
||||||
|
|
||||||
|
# 阶段 2:LLM 分类层(去重:相同文本只判一次,上下文取首次出现)。
|
||||||
|
cat_by_index: dict[int, str] = {}
|
||||||
|
if llm_needed:
|
||||||
|
if dedupe:
|
||||||
|
first_of_key: dict[str, int] = {}
|
||||||
|
pool_indices: list[int] = []
|
||||||
|
for i in llm_needed:
|
||||||
|
key = _dedup_key(entries[i]["text"])
|
||||||
|
if key not in first_of_key:
|
||||||
|
first_of_key[key] = i
|
||||||
|
pool_indices.append(i)
|
||||||
|
else:
|
||||||
|
pool_indices = llm_needed
|
||||||
|
|
||||||
|
# 单条判断的工作函数:返回类别词。
|
||||||
|
def judge_one(index: int) -> str:
|
||||||
|
return _judge_category(entries, index, context_size, request.params)
|
||||||
|
|
||||||
# 进度日志:打印已判定条数、总数与平均处理速度(条/s)。
|
# 进度日志:打印已判定条数、总数与平均处理速度(条/s)。
|
||||||
def log_progress(done: int, total: int, rate: float) -> None:
|
def log_progress(done: int, total: int, rate: float) -> None:
|
||||||
logger.info("字幕判定进度 %d/%d 条 (%.1f 条/s)", done, total, rate)
|
logger.info("字幕判定进度 %d/%d 条 (%.1f 条/s)", done, total, rate)
|
||||||
|
|
||||||
# 自适应并发调用 LLM:10s 窗口内平均响应 < 0.3s 则加 1 线程(上限
|
# 自适应并发调用 LLM:按实测负载弹性伸缩,避免压垮 LLM 接口。
|
||||||
# pool_max_workers),> pool_slow_threshold 则减 1 线程(下限 1),
|
|
||||||
# 按实测负载弹性伸缩,避免盲目并发压垮 LLM 接口。
|
|
||||||
pool = AdaptiveThreadPool(
|
pool = AdaptiveThreadPool(
|
||||||
worker=judge_one,
|
worker=judge_one,
|
||||||
on_progress=log_progress,
|
on_progress=log_progress,
|
||||||
@@ -145,18 +285,31 @@ def invoke(request: InvokeRequest) -> InvokeResponse:
|
|||||||
fast_threshold=float(request.params.get("pool_fast_threshold", 0.3)),
|
fast_threshold=float(request.params.get("pool_fast_threshold", 0.3)),
|
||||||
slow_threshold=float(request.params.get("pool_slow_threshold", 1.0)),
|
slow_threshold=float(request.params.get("pool_slow_threshold", 1.0)),
|
||||||
)
|
)
|
||||||
verdicts = pool.map(range(len(entries)))
|
categories = pool.map(pool_indices)
|
||||||
|
|
||||||
kept: list[dict] = []
|
for index, category in zip(pool_indices, categories):
|
||||||
removed = 0
|
|
||||||
for index, (entry, verdict) in enumerate(zip(entries, verdicts)):
|
|
||||||
# 并行下 LLM 异常被线程池隔离为异常结果:任一条失败即整体失败,
|
# 并行下 LLM 异常被线程池隔离为异常结果:任一条失败即整体失败,
|
||||||
# 避免静默输出未过滤结果。
|
# 避免静默输出未过滤结果。
|
||||||
if isinstance(verdict, Exception):
|
if isinstance(category, Exception):
|
||||||
return InvokeResponse(status="failed", error=str(verdict))
|
return InvokeResponse(status="failed", error=str(category))
|
||||||
if verdict:
|
cat_by_index[index] = category
|
||||||
|
if dedupe:
|
||||||
|
# 去重填充:与首次出现同键的条目复用同一类别,保证判定一致。
|
||||||
|
for i in llm_needed:
|
||||||
|
if i not in cat_by_index:
|
||||||
|
cat_by_index[i] = cat_by_index[first_of_key[_dedup_key(entries[i]["text"])]]
|
||||||
|
|
||||||
|
# 阶段 3:合并规则与 LLM 判定,应用长文本保护并输出。
|
||||||
|
kept: list[dict] = []
|
||||||
|
removed = 0
|
||||||
|
for i, (entry, rule_verdict) in enumerate(zip(entries, rule_verdicts)):
|
||||||
|
# 规则层命中即删(True);未命中则按 LLM 类别与长文本保护判定。
|
||||||
|
if rule_verdict is True:
|
||||||
removed += 1
|
removed += 1
|
||||||
logger.info("删除无意义字幕 %d: %r", index + 1, entry["text"][:40])
|
continue
|
||||||
|
if _should_delete(cat_by_index[i], entry["text"], min_keep_len):
|
||||||
|
removed += 1
|
||||||
|
logger.info("删除无意义字幕 %d: %r", i + 1, entry["text"][:40])
|
||||||
else:
|
else:
|
||||||
kept.append(entry)
|
kept.append(entry)
|
||||||
|
|
||||||
|
|||||||
@@ -63,6 +63,28 @@ def _assemble_srt(
|
|||||||
return lines
|
return lines
|
||||||
|
|
||||||
|
|
||||||
|
def _merge_kept(
|
||||||
|
manifest: list[dict], texts: list[str]
|
||||||
|
) -> list[tuple[float, float, str]]:
|
||||||
|
"""按 manifest 时间轴把逐帧 OCR 文本合并为字幕条目。
|
||||||
|
|
||||||
|
kept 元素为 (起始帧时间, 最后可见帧时间, 文本):空文本(无文字帧)跳过;
|
||||||
|
连续帧相同字幕合并为一条(字幕停留多帧属正常现象),仅更新最后可见帧
|
||||||
|
时间,起始时间保持首次出现。要求 texts 与 manifest 按帧顺序一一对应
|
||||||
|
(调用方保证),重组装旧数据时也复用此逻辑保证行为一致。
|
||||||
|
"""
|
||||||
|
kept: list[tuple[float, float, str]] = []
|
||||||
|
for index, text in enumerate(texts):
|
||||||
|
if not text:
|
||||||
|
continue
|
||||||
|
time = float(manifest[index]["time"])
|
||||||
|
if kept and kept[-1][2] == text:
|
||||||
|
kept[-1] = (kept[-1][0], time, text)
|
||||||
|
continue
|
||||||
|
kept.append((time, time, text))
|
||||||
|
return kept
|
||||||
|
|
||||||
|
|
||||||
def invoke(request: InvokeRequest) -> InvokeResponse:
|
def invoke(request: InvokeRequest) -> InvokeResponse:
|
||||||
"""逐帧 OCR 并汇总字幕,产物为 subtitle.srt。"""
|
"""逐帧 OCR 并汇总字幕,产物为 subtitle.srt。"""
|
||||||
manifest_uri = request.inputs.get("frames_manifest")
|
manifest_uri = request.inputs.get("frames_manifest")
|
||||||
@@ -138,19 +160,8 @@ def invoke(request: InvokeRequest) -> InvokeResponse:
|
|||||||
)
|
)
|
||||||
texts = pool.map(list(enumerate(manifest)))
|
texts = pool.map(list(enumerate(manifest)))
|
||||||
|
|
||||||
# kept 元素为 (起始帧时间, 最后可见帧时间, 文本);按帧顺序合并连续相同字幕。
|
# 按帧顺序合并连续相同字幕(与重组装旧数据共用 _merge_kept)。
|
||||||
kept: list[tuple[float, float, str]] = []
|
kept = _merge_kept(manifest, texts)
|
||||||
for index, text in enumerate(texts):
|
|
||||||
if not text:
|
|
||||||
continue
|
|
||||||
time = float(manifest[index]["time"])
|
|
||||||
# 连续帧相同字幕合并为一条(字幕停留多帧属正常现象):
|
|
||||||
# 仅更新最后可见帧时间,起始时间保持首次出现。
|
|
||||||
if kept and kept[-1][2] == text:
|
|
||||||
kept[-1] = (kept[-1][0], time, text)
|
|
||||||
continue
|
|
||||||
kept.append((time, time, text))
|
|
||||||
|
|
||||||
output_dir = Path(request.output_dir)
|
output_dir = Path(request.output_dir)
|
||||||
output_dir.mkdir(parents=True, exist_ok=True)
|
output_dir.mkdir(parents=True, exist_ok=True)
|
||||||
output_path = output_dir / "subtitle.srt"
|
output_path = output_dir / "subtitle.srt"
|
||||||
|
|||||||
@@ -0,0 +1,45 @@
|
|||||||
|
# -*- coding: utf-8 -*-
|
||||||
|
"""用修复后的 llm_filter 对 run_ac7f480a3ccb 的真实 OCR 输出重新过滤。
|
||||||
|
|
||||||
|
- 加载 .env(LLM Key/模型配置)
|
||||||
|
- 复用节点 invoke 真实代码路径(与调度器执行完全一致)
|
||||||
|
- 池参数按该 run 的 workflow v4 配置(单线程 pool 1/1)
|
||||||
|
- 产物写入 run 的 filter 步骤目录(filtered.srt)
|
||||||
|
"""
|
||||||
|
import sys
|
||||||
|
import time
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
from dotenv import load_dotenv
|
||||||
|
|
||||||
|
load_dotenv(".env")
|
||||||
|
|
||||||
|
from wov_sdk.models import InvokeRequest
|
||||||
|
from nodes.llm_filter import invoke
|
||||||
|
|
||||||
|
RUN = Path("data/storage/runs/run_ac7f480a3ccb")
|
||||||
|
SRT_URI = RUN / "steps/ocr/subtitle.srt"
|
||||||
|
OUT_DIR = RUN / "steps/filter"
|
||||||
|
|
||||||
|
print(f"输入: {SRT_URI}", flush=True)
|
||||||
|
print(f"输出目录: {OUT_DIR}", flush=True)
|
||||||
|
|
||||||
|
started = time.monotonic()
|
||||||
|
response = invoke(
|
||||||
|
InvokeRequest(
|
||||||
|
run_id="run_ac7f480a3ccb",
|
||||||
|
node_instance_id="",
|
||||||
|
inputs={"srt_uri": str(SRT_URI)},
|
||||||
|
# 与 workflow v4(单线程)保持一致。
|
||||||
|
params={"pool_min_workers": 1, "pool_max_workers": 1},
|
||||||
|
output_dir=str(OUT_DIR),
|
||||||
|
)
|
||||||
|
)
|
||||||
|
elapsed = time.monotonic() - started
|
||||||
|
print(f"耗时 {elapsed:.0f}s", flush=True)
|
||||||
|
if response.status != "completed":
|
||||||
|
print(f"FAILED: {response.error}", flush=True)
|
||||||
|
sys.exit(1)
|
||||||
|
|
||||||
|
print(f"kept={response.outputs['kept']} removed={response.outputs['removed']}", flush=True)
|
||||||
|
print(f"产物: {response.outputs['srt_uri']}", flush=True)
|
||||||
|
After Width: | Height: | Size: 28 KiB |
|
After Width: | Height: | Size: 166 KiB |
|
After Width: | Height: | Size: 165 KiB |
|
After Width: | Height: | Size: 104 KiB |
|
After Width: | Height: | Size: 99 KiB |
|
After Width: | Height: | Size: 105 KiB |
|
After Width: | Height: | Size: 126 KiB |
|
After Width: | Height: | Size: 88 KiB |
|
After Width: | Height: | Size: 92 KiB |
|
After Width: | Height: | Size: 86 KiB |
|
After Width: | Height: | Size: 100 KiB |
|
After Width: | Height: | Size: 136 KiB |
|
After Width: | Height: | Size: 101 KiB |
@@ -0,0 +1,97 @@
|
|||||||
|
"""真实任务 OCR 数据重组装集成测试。
|
||||||
|
|
||||||
|
run_339ec7ee437f 是 2026-08 真实跑过的 ocr-subtitle 任务(14236 帧 / 2 小时
|
||||||
|
视频)。该任务产出时期存在 frame-extract 帧文件排序 bug:ffmpeg 的 %04d 编号
|
||||||
|
超过 9999 帧后扩为 5 位,sorted() 字典序把 5 位编号排在 4 位之前,导致
|
||||||
|
frames.json 中 time 与图像错位(13236/14236 条位置↔帧号错位),最终 SRT
|
||||||
|
后半段时间轴全部错乱。
|
||||||
|
|
||||||
|
本测试读取该任务**已落盘的逐帧 OCR 文本**(ocr_frames/<位置>/ocr.txt,由
|
||||||
|
vlm-ocr 节点写入的清洗后文本),用 image_uri 解析真实帧号重建正确时间轴,
|
||||||
|
复用生产合并/组装逻辑生成 SRT,并断言"每条字幕的起始时刻 = 其文本来源帧的
|
||||||
|
真实时间"这一核心正确性。
|
||||||
|
|
||||||
|
数据位于 gitignored 的 data/storage,缺失时跳过(与
|
||||||
|
test_integration_subtitle_ocr 的真实资产约定一致)。
|
||||||
|
"""
|
||||||
|
|
||||||
|
import json
|
||||||
|
import re
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
from nodes.subtitle_ocr import _assemble_srt
|
||||||
|
from nodes.subtitle_ocr import _merge_kept
|
||||||
|
from nodes.subtitle_ocr import _sampling_interval
|
||||||
|
|
||||||
|
# 单体根目录:tests/ 的上一级。
|
||||||
|
WORKSPACE = Path(__file__).resolve().parent.parent
|
||||||
|
RUN_DIR = WORKSPACE / "data/storage/runs/run_339ec7ee437f"
|
||||||
|
MANIFEST = RUN_DIR / "steps/extract/frames.json"
|
||||||
|
OCR_DIR = RUN_DIR / "steps/ocr/ocr_frames"
|
||||||
|
|
||||||
|
# 源视频时长(ffprobe: 02:00:24.80)。
|
||||||
|
VIDEO_DURATION = 7224.8
|
||||||
|
# 该视频烧录过的字幕行,被 test_real_hav_sub.png 集成测试确认过识别结果。
|
||||||
|
KNOWN_LINE = "还有没有什么困扰"
|
||||||
|
|
||||||
|
|
||||||
|
def _frame_number(uri: str) -> int:
|
||||||
|
"""从 image_uri 文件名解析真实帧号(frame_0001.png -> 1)。"""
|
||||||
|
return int(re.match(r".*frame_(\d+)\.png", Path(uri).name).group(1))
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.integration
|
||||||
|
def test_reassemble_real_run_ocr_data(tmp_path) -> None:
|
||||||
|
"""真实任务逐帧 OCR 数据按真实帧时间轴重组装为正确 SRT。"""
|
||||||
|
if not (MANIFEST.is_file() and OCR_DIR.is_dir()):
|
||||||
|
pytest.skip("缺少真实任务 run_339ec7ee437f 数据,跳过集成测试")
|
||||||
|
|
||||||
|
manifest = json.loads(MANIFEST.read_text(encoding="utf-8"))
|
||||||
|
total = len(manifest)
|
||||||
|
# 等间隔采样:step/fps 从旧 manifest 相邻 time 差恢复(本任务 0.507s/帧)。
|
||||||
|
dt = manifest[1]["time"] - manifest[0]["time"]
|
||||||
|
assert dt > 0
|
||||||
|
|
||||||
|
# ① 读取逐帧 OCR 文本:位置 i -> ocr_frames/{i:04d}/ocr.txt(真实数据)。
|
||||||
|
texts_by_pos: list[str] = []
|
||||||
|
for i in range(total):
|
||||||
|
p = OCR_DIR / f"{i:04d}" / "ocr.txt"
|
||||||
|
if not p.is_file():
|
||||||
|
pytest.fail(f"缺少逐帧 OCR 数据: {p}")
|
||||||
|
texts_by_pos.append(p.read_text(encoding="utf-8").strip())
|
||||||
|
|
||||||
|
# ② 旧 manifest 位置 -> 真实帧号:image_uri 才是实际被 OCR 的图像,
|
||||||
|
# 旧 time 字段按位置推导已错位,不可用。
|
||||||
|
frame_by_pos = [_frame_number(item["image_uri"]) for item in manifest]
|
||||||
|
assert sorted(frame_by_pos) == list(range(1, total + 1)) # 一一对应,无缺无重
|
||||||
|
|
||||||
|
# ③ 按真实帧号重建正确时间轴:第 k 帧(1-based)时间 = (k-1)*dt。
|
||||||
|
ocr_by_frame = {n: texts_by_pos[i] for i, n in enumerate(frame_by_pos)}
|
||||||
|
manifest_corrected = [{"time": round((k - 1) * dt, 3)} for k in range(1, total + 1)]
|
||||||
|
texts_corrected = [ocr_by_frame[k] for k in range(1, total + 1)]
|
||||||
|
|
||||||
|
# ④ 复用生产合并 + 组装逻辑(与 subtitle_ocr.invoke 完全同一路径)。
|
||||||
|
kept = _merge_kept(manifest_corrected, texts_corrected)
|
||||||
|
interval = _sampling_interval(manifest_corrected, dt)
|
||||||
|
srt = "\n".join(_assemble_srt(kept, interval)) + "\n"
|
||||||
|
out = tmp_path / "subtitle.srt"
|
||||||
|
out.write_text(srt, encoding="utf-8")
|
||||||
|
|
||||||
|
# ⑤ 核心正确性:每条字幕起始时刻对应的帧,其 OCR 文本必须就是本条字幕
|
||||||
|
# 文本(修复前旧 SRT 此处大面积不一致:文本来自其他时刻的帧)。
|
||||||
|
for start, _end, text in kept:
|
||||||
|
frame_no = int(round(start / dt)) + 1
|
||||||
|
assert ocr_by_frame[frame_no] == text, f"帧 {frame_no} 时间错位: {text!r}"
|
||||||
|
|
||||||
|
# ⑥ 时间轴严格递增且不超出片长(含尾部一个采样间隔的消失余量)。
|
||||||
|
assert all(
|
||||||
|
s1 < s2 for (s1, _1, _), (s2, _2, _) in zip(kept, kept[1:])
|
||||||
|
)
|
||||||
|
assert kept[0][0] >= 0.0
|
||||||
|
assert kept[-1][1] + interval <= VIDEO_DURATION + interval + 1e-6
|
||||||
|
|
||||||
|
# ⑦ 已知烧录字幕行必须出现,且字幕条数合理(空帧被跳过、连续相同字幕合并)。
|
||||||
|
assert KNOWN_LINE in srt
|
||||||
|
assert len(kept) > 100
|
||||||
@@ -1,14 +1,30 @@
|
|||||||
"""LLM 字幕过滤节点测试。
|
"""LLM 字幕过滤节点测试。
|
||||||
|
|
||||||
覆盖 SRT 解析/序列化、±N 上下文窗口组装(纯文本无时间戳、目标标记)、
|
覆盖:
|
||||||
LLM 调用(按 I/O 边界 mock urlopen)与删除判定、invoke 全链路与异常路径。
|
- SRT 解析/序列化(多行、末条无空行、重新编号);
|
||||||
|
- 确定性规则层(横线装饰、HTML/水印 token、URL、单双 ASCII 字符直接删,不调 LLM);
|
||||||
|
- LLM 五类分类判断(garbage/overlay/noise 删,repeat/dialogue 留,未识别回退保留);
|
||||||
|
- 文本去重(相同文本只调一次 LLM,上下文取首次出现,判定结果一致);
|
||||||
|
- 长文本保护(≥min_keep_len 时 noise 类不删,需明确垃圾类别);
|
||||||
|
- 真实任务 run_ac7f480a3ccb 留存数据回归(OCR 1666 条:垃圾删除、对话保留、时间轴单调、去重后 LLM 调用数);
|
||||||
|
- invoke 全链路与异常路径。
|
||||||
"""
|
"""
|
||||||
|
|
||||||
import json
|
import json
|
||||||
import urllib.error
|
import urllib.error
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
|
|
||||||
from nodes.llm_filter import invoke, parse_srt, serialize_srt
|
import pytest
|
||||||
|
|
||||||
|
from nodes.llm_filter import (
|
||||||
|
_dedup_key,
|
||||||
|
_judge_category,
|
||||||
|
_rule_verdict,
|
||||||
|
_should_delete,
|
||||||
|
invoke,
|
||||||
|
parse_srt,
|
||||||
|
serialize_srt,
|
||||||
|
)
|
||||||
from wov_sdk.models import InvokeRequest
|
from wov_sdk.models import InvokeRequest
|
||||||
|
|
||||||
# 4 条字幕的 SRT:第 3 条为"答:"开头的无意义杂项,模拟 OCR 噪声。
|
# 4 条字幕的 SRT:第 3 条为"答:"开头的无意义杂项,模拟 OCR 噪声。
|
||||||
@@ -19,6 +35,10 @@ _SRT = (
|
|||||||
"4\n00:00:13,000 --> 00:00:16,000\n第二句正常字幕\n"
|
"4\n00:00:13,000 --> 00:00:16,000\n第二句正常字幕\n"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
# 真实任务 run_ac7f480a3ccb 的 OCR 输出(1666 条,2026-08 单线程 v4 跑完整 2 小时视频)。
|
||||||
|
WORKSPACE = Path(__file__).resolve().parent.parent
|
||||||
|
REAL_SRT = WORKSPACE / "testdata" / "ocr_srt_run_ac7f480a3ccb.srt"
|
||||||
|
|
||||||
|
|
||||||
class FakeResponse:
|
class FakeResponse:
|
||||||
"""模拟 urllib 响应:read() 返回 LLM 兼容接口的 JSON 载荷。"""
|
"""模拟 urllib 响应:read() 返回 LLM 兼容接口的 JSON 载荷。"""
|
||||||
@@ -37,9 +57,9 @@ class FakeResponse:
|
|||||||
|
|
||||||
|
|
||||||
class FakeLLM:
|
class FakeLLM:
|
||||||
"""模拟 LLM 兼容接口:记录请求体,按策略返回"保留/删除"。
|
"""模拟 LLM 兼容接口:记录请求体,按策略返回类别词。
|
||||||
|
|
||||||
支持两种策略:contents(按队列顺序,用于单次直调 _judge_target 的
|
支持两种策略:contents(按队列顺序,用于单次直调 _judge_category 的
|
||||||
确定性测试)或 decision_fn(按请求体内容决策,用于并发 invoke 测试,
|
确定性测试)或 decision_fn(按请求体内容决策,用于并发 invoke 测试,
|
||||||
保证任何线程执行顺序下判定结果都确定)。
|
保证任何线程执行顺序下判定结果都确定)。
|
||||||
"""
|
"""
|
||||||
@@ -70,12 +90,17 @@ def _patch_llm(monkeypatch, contents: list[str] | None = None, decision_fn=None)
|
|||||||
|
|
||||||
|
|
||||||
def _decision_by_target(body) -> str:
|
def _decision_by_target(body) -> str:
|
||||||
"""按目标字幕内容决策:含"答:"判为删除,其余保留(与 _SRT 的噪声对应)。"""
|
"""按目标字幕内容决策:含"答:"判为 garbage,其余为 dialogue。"""
|
||||||
target = next(
|
target = next(
|
||||||
line for line in body["messages"][1]["content"].splitlines()
|
line for line in body["messages"][1]["content"].splitlines()
|
||||||
if line.startswith("【目标】")
|
if line.startswith("【目标】")
|
||||||
)
|
)
|
||||||
return "删除" if "答:" in target else "保留"
|
return "garbage" if "答:" in target else "dialogue"
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# SRT 解析 / 序列化
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
|
||||||
def test_parse_srt_multiline_and_last_block() -> None:
|
def test_parse_srt_multiline_and_last_block() -> None:
|
||||||
@@ -105,14 +130,49 @@ def test_serialize_srt_renumbers() -> None:
|
|||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
def test_judge_target_window_and_keep(monkeypatch) -> None:
|
# ---------------------------------------------------------------------------
|
||||||
"""窗口只含纯文本(无时间戳)、目标带标记;模型答"保留"则返回 False。"""
|
# 确定性规则层
|
||||||
from nodes.llm_filter import _judge_target
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
|
||||||
|
def test_rule_verdict_deletes_garbage_patterns() -> None:
|
||||||
|
"""规则层直接删除:横线装饰、HTML/水印 token、URL/邮箱、单双 ASCII 字符。"""
|
||||||
|
tokens = {"html", "background", "___"}
|
||||||
|
for text in (
|
||||||
|
"---", "------", "------------------", "--- ---", "___", "= = =",
|
||||||
|
"…", "・", "HTML", "html", "Background", "background", "___",
|
||||||
|
"https://example.com/x", "www.example.com", "a@b.com",
|
||||||
|
"A", "V", "1", "DQ", "P4",
|
||||||
|
):
|
||||||
|
assert _rule_verdict(text, tokens) is True, text
|
||||||
|
# 空文本/纯空白也删除。
|
||||||
|
assert _rule_verdict(" ", tokens) is True
|
||||||
|
|
||||||
|
|
||||||
|
def test_rule_verdict_passes_cjk_and_dialogue() -> None:
|
||||||
|
"""规则层不误伤:CJK 单字、正常对话、内容性短句交给 LLM 判断(None)。"""
|
||||||
|
tokens = {"html"}
|
||||||
|
for text in ("嗯", "好", "谢谢你 松井小姐", "不这么做的话 可没法胜任患者的对象", "好舒服"):
|
||||||
|
assert _rule_verdict(text, tokens) is None, text
|
||||||
|
|
||||||
|
|
||||||
|
def test_rule_verdict_custom_overlay_tokens() -> None:
|
||||||
|
"""overlay_tokens 参数化:自定义 token 同样直接删除。"""
|
||||||
|
assert _rule_verdict("Cleaning", {"cleaning"}) is True
|
||||||
|
assert _rule_verdict("Cleaning", set()) is None
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# LLM 分类判断
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
|
||||||
|
def test_judge_category_window_and_dialogue(monkeypatch) -> None:
|
||||||
|
"""窗口只含纯文本(无时间戳)、目标带标记;模型答 dialogue 则保留。"""
|
||||||
entries = parse_srt(_SRT)
|
entries = parse_srt(_SRT)
|
||||||
fake = _patch_llm(monkeypatch, ["保留"])
|
fake = _patch_llm(monkeypatch, ["dialogue"])
|
||||||
# context_size=1,目标为第 2 条(index=1):窗口 0..2 共 3 行,目标在中间。
|
# context_size=1,目标为第 2 条(index=1):窗口 0..2 共 3 行,目标在中间。
|
||||||
assert _judge_target(entries, 1, context_size=1, params={}) is False
|
assert _judge_category(entries, 1, context_size=1, params={}) == "dialogue"
|
||||||
body = fake.bodies[0]
|
body = fake.bodies[0]
|
||||||
lines = body["messages"][1]["content"].splitlines()
|
lines = body["messages"][1]["content"].splitlines()
|
||||||
assert len(lines) == 3
|
assert len(lines) == 3
|
||||||
@@ -125,32 +185,71 @@ def test_judge_target_window_and_keep(monkeypatch) -> None:
|
|||||||
assert body["max_tokens"] == 16
|
assert body["max_tokens"] == 16
|
||||||
|
|
||||||
|
|
||||||
def test_judge_target_delete(monkeypatch) -> None:
|
def test_judge_category_delete_classes(monkeypatch) -> None:
|
||||||
"""模型答"删除"时返回 True(判定该条无意义)。"""
|
"""模型答 garbage/overlay/noise 时返回对应类别(删除类)。"""
|
||||||
from nodes.llm_filter import _judge_target
|
|
||||||
|
|
||||||
entries = parse_srt(_SRT)
|
entries = parse_srt(_SRT)
|
||||||
|
for cat in ("garbage", "overlay", "noise"):
|
||||||
|
_patch_llm(monkeypatch, [cat])
|
||||||
|
assert _judge_category(entries, 2, context_size=10, params={}) == cat
|
||||||
|
|
||||||
|
|
||||||
|
def test_judge_category_repeat_and_unknown_kept(monkeypatch) -> None:
|
||||||
|
"""repeat/dialogue 返回保留类;未识别输出(空/乱码/旧式"保留")回退 dialogue。"""
|
||||||
|
entries = parse_srt(_SRT)
|
||||||
|
for answer in ("repeat", "dialogue", "", "???", "保留"):
|
||||||
|
_patch_llm(monkeypatch, [answer])
|
||||||
|
got = _judge_category(entries, 2, context_size=10, params={})
|
||||||
|
assert got in ("repeat", "dialogue"), (answer, got)
|
||||||
|
# 旧式"删除"回答兼容为垃圾类。
|
||||||
_patch_llm(monkeypatch, ["删除"])
|
_patch_llm(monkeypatch, ["删除"])
|
||||||
assert _judge_target(entries, 2, context_size=10, params={}) is True
|
assert _judge_category(entries, 2, context_size=10, params={}) == "garbage"
|
||||||
|
|
||||||
|
|
||||||
def test_judge_target_model_and_auth(monkeypatch) -> None:
|
def test_judge_category_model_and_auth(monkeypatch) -> None:
|
||||||
"""模型名从参数取;配置 API Key 时附带 Bearer 鉴权头。"""
|
"""模型名从参数取;配置 API Key 时附带 Bearer 鉴权头。"""
|
||||||
from nodes.llm_filter import _judge_target
|
|
||||||
|
|
||||||
entries = parse_srt(_SRT)
|
entries = parse_srt(_SRT)
|
||||||
monkeypatch.setenv("LLM_API_KEY", "sk-test")
|
monkeypatch.setenv("LLM_API_KEY", "sk-test")
|
||||||
fake = _patch_llm(monkeypatch, ["保留"])
|
fake = _patch_llm(monkeypatch, ["dialogue"])
|
||||||
assert _judge_target(entries, 0, context_size=10, params={"model": "m/1"}) is False
|
assert _judge_category(entries, 0, context_size=10, params={"model": "m/1"}) == "dialogue"
|
||||||
assert fake.bodies[0]["model"] == "m/1"
|
assert fake.bodies[0]["model"] == "m/1"
|
||||||
assert fake.headers[0]["Authorization"] == "Bearer sk-test"
|
assert fake.headers[0]["Authorization"] == "Bearer sk-test"
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# 去重键与删除判定
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
|
||||||
|
def test_dedup_key_normalizes_whitespace_and_case() -> None:
|
||||||
|
"""去重键:去掉全部空白并统一大小写,空格差异视为同一文本。"""
|
||||||
|
assert _dedup_key("不这么做的话 可没法胜任患者的对象") == _dedup_key(
|
||||||
|
"不这么做的话可没法胜任患者的对象"
|
||||||
|
)
|
||||||
|
assert _dedup_key("HTML") == _dedup_key("html")
|
||||||
|
assert _dedup_key("你好 世界") != _dedup_key("你好世界2")
|
||||||
|
|
||||||
|
|
||||||
|
def test_should_delete_long_text_noise_protected() -> None:
|
||||||
|
"""长文本保护:≥min_keep_len 时 noise 不删,garbage/overlay 仍删;短文本三类都删。"""
|
||||||
|
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
|
||||||
|
assert _should_delete("overlay", long_text, min_keep_len=12) is True
|
||||||
|
assert _should_delete("noise", "答:无意义杂项", min_keep_len=12) is True
|
||||||
|
assert _should_delete("repeat", long_text, min_keep_len=12) is False
|
||||||
|
assert _should_delete("dialogue", long_text, min_keep_len=12) is False
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# invoke 全链路
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
|
||||||
def test_invoke_filters_and_renumbers(monkeypatch, tmp_path) -> None:
|
def test_invoke_filters_and_renumbers(monkeypatch, tmp_path) -> None:
|
||||||
"""全链路(并发):按 LLM 判定删除无意义条,保留条重新编号输出。"""
|
"""全链路(并发):按 LLM 类别判定删除无意义条,保留条重新编号输出。"""
|
||||||
srt = tmp_path / "in.srt"
|
srt = tmp_path / "in.srt"
|
||||||
srt.write_text(_SRT, encoding="utf-8")
|
srt.write_text(_SRT, encoding="utf-8")
|
||||||
# 内容决策:目标字幕含"答:"判删除,其余保留(任何线程顺序下结果确定)。
|
# 内容决策:目标字幕含"答:"判 garbage,其余 dialogue(任何线程顺序下结果确定)。
|
||||||
_patch_llm(monkeypatch, decision_fn=_decision_by_target)
|
_patch_llm(monkeypatch, decision_fn=_decision_by_target)
|
||||||
response = invoke(
|
response = invoke(
|
||||||
InvokeRequest(
|
InvokeRequest(
|
||||||
@@ -172,6 +271,79 @@ def test_invoke_filters_and_renumbers(monkeypatch, tmp_path) -> None:
|
|||||||
assert "00:00:13,000 --> 00:00:16,000\n第二句正常字幕\n" in out
|
assert "00:00:13,000 --> 00:00:16,000\n第二句正常字幕\n" in out
|
||||||
|
|
||||||
|
|
||||||
|
def test_invoke_rules_skip_llm(monkeypatch, tmp_path) -> None:
|
||||||
|
"""规则层命中时直接删除且不调用 LLM:横线/HTML/单字符全部清理。"""
|
||||||
|
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\nHTML\n\n"
|
||||||
|
"3\n00:00:09,000 --> 00:00:12,000\nV\n\n"
|
||||||
|
"4\n00:00:13,000 --> 00:00:16,000\n正常对话\n",
|
||||||
|
encoding="utf-8",
|
||||||
|
)
|
||||||
|
fake = _patch_llm(monkeypatch, ["dialogue"])
|
||||||
|
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
|
||||||
|
assert response.outputs["kept"] == 1
|
||||||
|
assert response.outputs["removed"] == 3
|
||||||
|
assert len(fake.bodies) == 1 # 仅"正常对话"需 LLM,其余全走规则。
|
||||||
|
|
||||||
|
|
||||||
|
def test_invoke_dedup_single_llm_call(monkeypatch, tmp_path) -> None:
|
||||||
|
"""去重:相同文本(含空格变体)只调一次 LLM,判定结果一致。"""
|
||||||
|
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\n别的台词\n",
|
||||||
|
encoding="utf-8",
|
||||||
|
)
|
||||||
|
fake = _patch_llm(monkeypatch, decision_fn=lambda body: "repeat")
|
||||||
|
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
|
||||||
|
# 2 个唯一文本(好舒服/别的台词)→ 恰好 2 次 LLM 调用。
|
||||||
|
assert len(fake.bodies) == 2
|
||||||
|
# repeat 类保留 → 4 条全部保留。
|
||||||
|
assert response.outputs["kept"] == 4
|
||||||
|
assert response.outputs["removed"] == 0
|
||||||
|
|
||||||
|
|
||||||
|
def test_invoke_dedupe_disabled(monkeypatch, tmp_path) -> None:
|
||||||
|
"""dedupe=0 关闭去重:每个条目都调一次 LLM。"""
|
||||||
|
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",
|
||||||
|
encoding="utf-8",
|
||||||
|
)
|
||||||
|
fake = _patch_llm(monkeypatch, decision_fn=lambda body: "dialogue")
|
||||||
|
response = invoke(
|
||||||
|
InvokeRequest(
|
||||||
|
run_id="r", node_instance_id="",
|
||||||
|
inputs={"srt_uri": str(srt)},
|
||||||
|
params={"dedupe": "0"},
|
||||||
|
output_dir=str(tmp_path / "out"),
|
||||||
|
)
|
||||||
|
)
|
||||||
|
assert response.status == "completed", response.error
|
||||||
|
assert len(fake.bodies) == 2
|
||||||
|
|
||||||
|
|
||||||
def test_invoke_context_size_param(monkeypatch, tmp_path) -> None:
|
def test_invoke_context_size_param(monkeypatch, tmp_path) -> None:
|
||||||
"""context_size 参数生效:窗口大小=2×context_size+1(两端截断除外)。"""
|
"""context_size 参数生效:窗口大小=2×context_size+1(两端截断除外)。"""
|
||||||
srt = tmp_path / "in.srt"
|
srt = tmp_path / "in.srt"
|
||||||
@@ -195,6 +367,52 @@ def test_invoke_context_size_param(monkeypatch, tmp_path) -> None:
|
|||||||
assert len(window) == 3
|
assert len(window) == 3
|
||||||
|
|
||||||
|
|
||||||
|
def test_invoke_overlay_tokens_param(monkeypatch, tmp_path) -> None:
|
||||||
|
"""overlay_tokens 参数:自定义 token 走规则层删除,不调用 LLM。"""
|
||||||
|
srt = tmp_path / "in.srt"
|
||||||
|
srt.write_text(
|
||||||
|
"1\n00:00:01,000 --> 00:00:04,000\nCleaning\n\n"
|
||||||
|
"2\n00:00:05,000 --> 00:00:08,000\n正常对话\n",
|
||||||
|
encoding="utf-8",
|
||||||
|
)
|
||||||
|
fake = _patch_llm(monkeypatch, decision_fn=lambda body: "dialogue")
|
||||||
|
response = invoke(
|
||||||
|
InvokeRequest(
|
||||||
|
run_id="r", node_instance_id="",
|
||||||
|
inputs={"srt_uri": str(srt)},
|
||||||
|
params={"overlay_tokens": ["cleaning"]},
|
||||||
|
output_dir=str(tmp_path / "out"),
|
||||||
|
)
|
||||||
|
)
|
||||||
|
assert response.status == "completed", response.error
|
||||||
|
assert response.outputs["removed"] == 1
|
||||||
|
assert response.outputs["kept"] == 1
|
||||||
|
assert len(fake.bodies) == 1 # 只判正常对话。
|
||||||
|
|
||||||
|
|
||||||
|
def test_invoke_overlay_tokens_json_string(monkeypatch, tmp_path) -> None:
|
||||||
|
"""overlay_tokens 以 JSON 字符串形式传入(工作流参数常见形态)同样生效。"""
|
||||||
|
srt = tmp_path / "in.srt"
|
||||||
|
srt.write_text(
|
||||||
|
"1\n00:00:01,000 --> 00:00:04,000\nXXLogo\n\n"
|
||||||
|
"2\n00:00:05,000 --> 00:00:08,000\n正常对话\n",
|
||||||
|
encoding="utf-8",
|
||||||
|
)
|
||||||
|
fake = _patch_llm(monkeypatch, ["dialogue"])
|
||||||
|
response = invoke(
|
||||||
|
InvokeRequest(
|
||||||
|
run_id="r", node_instance_id="",
|
||||||
|
inputs={"srt_uri": str(srt)},
|
||||||
|
params={"overlay_tokens": '["xxlogo", "xx"]'},
|
||||||
|
output_dir=str(tmp_path / "out"),
|
||||||
|
)
|
||||||
|
)
|
||||||
|
assert response.status == "completed", response.error
|
||||||
|
assert response.outputs["removed"] == 1
|
||||||
|
assert response.outputs["kept"] == 1
|
||||||
|
assert len(fake.bodies) == 1
|
||||||
|
|
||||||
|
|
||||||
def test_invoke_missing_input(tmp_path) -> None:
|
def test_invoke_missing_input(tmp_path) -> None:
|
||||||
"""缺少 srt_uri 时返回失败。"""
|
"""缺少 srt_uri 时返回失败。"""
|
||||||
response = invoke(
|
response = invoke(
|
||||||
@@ -255,3 +473,79 @@ def test_invoke_empty_srt(monkeypatch, tmp_path) -> None:
|
|||||||
assert response.outputs["kept"] == 0
|
assert response.outputs["kept"] == 0
|
||||||
assert response.outputs["removed"] == 0
|
assert response.outputs["removed"] == 0
|
||||||
assert fake.bodies == []
|
assert fake.bodies == []
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# 真实任务留存数据回归(testdata/ocr_srt_run_ac7f480a3ccb.srt)
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
|
||||||
|
def test_real_run_rules_and_dialogue_regression(monkeypatch, tmp_path) -> None:
|
||||||
|
"""真实数据回归:规则层清理垃圾、对话保留、时间轴单调、去重生效。
|
||||||
|
|
||||||
|
夹具为 run_ac7f480a3ccb(单线程 v4 跑完整 2 小时视频)的 OCR 输出 1666 条。
|
||||||
|
假 LLM 对非规则条目一律答 dialogue:验证规则层删掉全部垃圾(横线/水印/
|
||||||
|
单字符),真实对话全部保留,且相同文本只调一次 LLM(去重)。
|
||||||
|
"""
|
||||||
|
srt = REAL_SRT
|
||||||
|
if not srt.is_file():
|
||||||
|
pytest.skip("缺少 testdata/ocr_srt_run_ac7f480a3ccb.srt,跳过回归测试")
|
||||||
|
fake = _patch_llm(monkeypatch, decision_fn=lambda body: "dialogue")
|
||||||
|
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
|
||||||
|
|
||||||
|
entries = parse_srt(srt.read_text(encoding="utf-8"))
|
||||||
|
out = Path(response.outputs["srt_uri"]).read_text(encoding="utf-8")
|
||||||
|
|
||||||
|
# 规则层删除量 = 夹具中直接命中规则文本的条数(真实数据动态计算)。
|
||||||
|
# 规则层删除量 = 夹具中直接命中规则文本的条数(与 invoke 默认 token 集一致)。
|
||||||
|
from nodes.llm_filter import DEFAULT_OVERLAY_TOKENS
|
||||||
|
|
||||||
|
tokens = set(DEFAULT_OVERLAY_TOKENS)
|
||||||
|
rule_removed = sum(
|
||||||
|
1 for e in entries if _rule_verdict(e["text"], tokens) is True
|
||||||
|
)
|
||||||
|
assert response.outputs["removed"] == rule_removed
|
||||||
|
assert response.outputs["kept"] == len(entries) - rule_removed
|
||||||
|
|
||||||
|
# 垃圾全部清出输出(横线、HTML、单字符都不再出现;只检查文本字段,
|
||||||
|
# 排除 SRT 的序号行与时间轴行)。
|
||||||
|
for entry in parse_srt(out):
|
||||||
|
assert _rule_verdict(entry["text"], tokens) is not True, entry["text"]
|
||||||
|
# 真实对话保留(修复前被误删的自我介绍、请求对话、呻吟内容行)。
|
||||||
|
# 注意:片头标题卡(淫魔病院…)真实 LLM 判为 overlay(烧录标题),
|
||||||
|
# 不在必保留列表内;本测试假 LLM 一律答 dialogue,仅验证机制正确。
|
||||||
|
for kept_line in (
|
||||||
|
"谢谢你 松井小姐",
|
||||||
|
"我姓泷本 请多多指教",
|
||||||
|
"不这么做的话 可没法胜任患者的对象",
|
||||||
|
"你要看着北冈小姐的脸 你们俩相互看看嘛",
|
||||||
|
"好舒服",
|
||||||
|
):
|
||||||
|
assert kept_line in out, kept_line
|
||||||
|
|
||||||
|
# 时间轴单调递增(输出顺序即时间顺序)。
|
||||||
|
times = []
|
||||||
|
for m in __import__("re").finditer(
|
||||||
|
r"(\d{2}:\d{2}:\d{2},\d{3})\s*-->", out
|
||||||
|
):
|
||||||
|
t = m.group(1).replace(",", ".")
|
||||||
|
h, mi, s = t.split(":")
|
||||||
|
times.append(int(h) * 3600 + int(mi) * 60 + float(s))
|
||||||
|
assert all(a < b for a, b in zip(times, times[1:]))
|
||||||
|
|
||||||
|
# 去重:LLM 调用数 == 非规则条目中的唯一文本数(远小于条目数)。
|
||||||
|
unique_llm = len({
|
||||||
|
_dedup_key(e["text"])
|
||||||
|
for e in entries
|
||||||
|
if _rule_verdict(e["text"], tokens) is None
|
||||||
|
})
|
||||||
|
assert len(fake.bodies) == unique_llm
|
||||||
|
assert len(fake.bodies) < len(entries) # 去重确实省调用。
|
||||||
|
|||||||
@@ -305,6 +305,26 @@ def test_frame_extract_one_second_exact_frames(tmp_path) -> None:
|
|||||||
assert len(manifest) == 10
|
assert len(manifest) == 10
|
||||||
|
|
||||||
|
|
||||||
|
def test_frame_files_read_order_matches_frame_number(tmp_path) -> None:
|
||||||
|
"""真实任务留存数据:帧文件按帧号数值排序读取,而非字典序。
|
||||||
|
|
||||||
|
回归用例:ffmpeg 的 %04d 编号在超过 9999 帧后自动扩为 5 位
|
||||||
|
(frame_10000.png 等),此时 sorted() 默认字典序会把 5 位编号排在
|
||||||
|
4 位编号之前(如 frame_10009 < frame_1009),导致帧号回退、manifest
|
||||||
|
时间与图像错位。testdata/frames_boundary/ 是 2026-08 真实任务
|
||||||
|
run_339ec7ee437f(14236 帧 / 2 小时视频)中跨越该边界的真实帧文件。
|
||||||
|
"""
|
||||||
|
from nodes.frame_extract import _sorted_frame_files
|
||||||
|
|
||||||
|
boundary = TESTDATA / "frames_boundary"
|
||||||
|
files = _sorted_frame_files(boundary)
|
||||||
|
# 从文件名解析帧号:读取顺序必须等于帧号数值递增序(无回退)。
|
||||||
|
nums = [int(p.stem.split("_", 1)[1]) for p in files]
|
||||||
|
assert nums == sorted(nums)
|
||||||
|
# 边界关键对:5 位编号必须排在 4 位编号之后,禁止字典序错位。
|
||||||
|
assert nums.index(10000) > nums.index(9999)
|
||||||
|
assert nums.index(10009) > nums.index(1009)
|
||||||
|
|
||||||
# ---------------------------------------------------------------------------
|
# ---------------------------------------------------------------------------
|
||||||
# subtitle-ocr:OCR 循环 + 长度上限 + 合并 + SRT 组装
|
# subtitle-ocr:OCR 循环 + 长度上限 + 合并 + SRT 组装
|
||||||
# ---------------------------------------------------------------------------
|
# ---------------------------------------------------------------------------
|
||||||
|
|||||||
@@ -0,0 +1,215 @@
|
|||||||
|
"""多线程下字幕顺序正确性测试(模拟真实 API 返回,全量真实数据)。
|
||||||
|
|
||||||
|
目标:验证 subtitle-ocr 在**多线程**执行时能否正确处理字幕顺序。
|
||||||
|
|
||||||
|
- 不走真实 vlm-ocr(Ollama)网络调用:registry.invoke 被替换为
|
||||||
|
FakeVlmOcrApi,按 image_uri 文件名中的帧号,直接从测试数据(真实任务
|
||||||
|
run_ac7f480a3ccb 的**全量**逐帧 OCR 结果)取该帧文本返回,模拟真实
|
||||||
|
API 返回结构;输入可以是文件名(frames_manifest / image_uri);
|
||||||
|
- **真实 OCR 延迟模拟**:真实 vlm-ocr 每次调用延迟不可预判——大部分帧快、
|
||||||
|
少量帧明显慢(复杂画面/模型排队,真实约 0.1s~5s,测试按比例缩放)。
|
||||||
|
假 API 用种子化随机生成同样的快/慢分布(约 8% 慢帧),制造真实波动下
|
||||||
|
的乱序完成,对线程池的保序能力施加最贴近真实情况的压力;
|
||||||
|
- 核心断言:多线程(4/16 线程)产出的 SRT 与**用户确认过的精确结果**逐字节
|
||||||
|
一致——该结果正是真实任务 run_ac7f480a3ccb 以**单线程**(workflow v4
|
||||||
|
pool 1/1)运行产出并经用户确认的,因此逐字节一致即证明"多线程 == 单线程";
|
||||||
|
同时断言并发真实发生、完成顺序确实乱序、全量时间轴严格递增、每条字幕
|
||||||
|
与其起始时刻帧的文本对齐。
|
||||||
|
|
||||||
|
全量真实数据(testdata/,常驻夹具,真实任务 run_ac7f480a3ccb 全部 14236 帧):
|
||||||
|
- frames_manifest_full.json:完整帧清单(14236 条,image_uri 改为文件名);
|
||||||
|
- ocr_frames_full.json:{帧号: 该帧 OCR 文本}(位置 p ↔ 帧 p+1);
|
||||||
|
- ocr_srt_run_ac7f480a3ccb.srt:该任务单线程运行产出、用户确认过的精确结果(1666 条)。
|
||||||
|
"""
|
||||||
|
|
||||||
|
import json
|
||||||
|
import random
|
||||||
|
import re
|
||||||
|
import threading
|
||||||
|
import time
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
from wov_sdk.models import InvokeRequest, InvokeResponse
|
||||||
|
|
||||||
|
WORKSPACE = Path(__file__).resolve().parent.parent
|
||||||
|
TESTDATA = WORKSPACE / "testdata"
|
||||||
|
FULL_MANIFEST = TESTDATA / "frames_manifest_full.json"
|
||||||
|
FULL_OCR_TEXTS = TESTDATA / "ocr_frames_full.json"
|
||||||
|
CONFIRMED_SRT = TESTDATA / "ocr_srt_run_ac7f480a3ccb.srt"
|
||||||
|
TOTAL_FRAMES = 14236
|
||||||
|
|
||||||
|
# 帧号解析:image_uri 文件名形如 frame_0411.png。
|
||||||
|
_FRAME_RE = re.compile(r".*frame_(\d+)\.png")
|
||||||
|
|
||||||
|
|
||||||
|
def _frame_no(uri: str) -> int:
|
||||||
|
"""从 image_uri 文件名解析帧号。"""
|
||||||
|
return int(_FRAME_RE.match(Path(uri).name).group(1))
|
||||||
|
|
||||||
|
|
||||||
|
def _ts_to_seconds(ts: str) -> float:
|
||||||
|
"""SRT 时间戳(HH:MM:SS,mmm)转秒。"""
|
||||||
|
h, m, s = ts.replace(",", ".").split(":")
|
||||||
|
return int(h) * 3600 + int(m) * 60 + float(s)
|
||||||
|
|
||||||
|
|
||||||
|
class FakeVlmOcrApi:
|
||||||
|
"""模拟真实 vlm-ocr API:不发起真实网络调用,从测试数据返回该帧 OCR 文本。
|
||||||
|
|
||||||
|
- 输入是 image_uri 文件名,解析帧号后从全量真实 OCR 结果({帧号: 文本})
|
||||||
|
取该帧文本返回,模拟真实 API 返回结构(status=completed / outputs.text);
|
||||||
|
- **可变延迟模拟真实 OCR**:每次调用独立随机,大部分帧快
|
||||||
|
(0.2~2ms,对应真实约 0.1~1s),约 slow_ratio 比例的帧明显慢
|
||||||
|
(6~18ms,对应真实约 3~9s,如复杂画面/模型排队)。延迟不可预判,
|
||||||
|
对线程池的乱序恢复能力施加与真实情况一致的随机压力;
|
||||||
|
- 线程安全地记录并发峰值、完成顺序与每次延迟,供断言"多线程确实发生、
|
||||||
|
乱序完成、且存在明显慢帧"。
|
||||||
|
"""
|
||||||
|
|
||||||
|
def __init__(self, texts_by_frame: dict[int, str], seed: int = 20260817,
|
||||||
|
fast_ms: float = 1.0, slow_ms: float = 12.0,
|
||||||
|
slow_ratio: float = 0.08) -> None:
|
||||||
|
self._texts_by_frame = texts_by_frame
|
||||||
|
self._fast_ms = fast_ms
|
||||||
|
self._slow_ms = slow_ms
|
||||||
|
self._slow_ratio = slow_ratio
|
||||||
|
# 种子化随机源:延迟波动可复现(固定种子 → 测试确定性,不 flaky)。
|
||||||
|
self._rng = random.Random(seed)
|
||||||
|
self._lock = threading.Lock()
|
||||||
|
self.active = 0
|
||||||
|
self.max_active = 0
|
||||||
|
# 完成顺序(帧号):用于断言乱序完成确实发生。
|
||||||
|
self.completed_frames: list[int] = []
|
||||||
|
# 每次调用的实际延迟(毫秒):用于断言快/慢分布确实发生。
|
||||||
|
self.delays_ms: list[float] = []
|
||||||
|
|
||||||
|
def __call__(self, node_id: str, request: InvokeRequest) -> InvokeResponse:
|
||||||
|
frame_no = _frame_no(request.inputs["image_uri"])
|
||||||
|
with self._lock:
|
||||||
|
self.active += 1
|
||||||
|
self.max_active = max(self.max_active, self.active)
|
||||||
|
try:
|
||||||
|
# 真实 OCR 延迟:慢帧比例固定,具体哪帧慢由随机决定(不可预判)。
|
||||||
|
if self._rng.random() < self._slow_ratio:
|
||||||
|
delay = self._slow_ms * (0.5 + self._rng.random())
|
||||||
|
else:
|
||||||
|
delay = self._fast_ms * (0.2 + self._rng.random() * 1.8)
|
||||||
|
self.delays_ms.append(delay)
|
||||||
|
time.sleep(delay / 1000.0)
|
||||||
|
text = self._texts_by_frame[frame_no]
|
||||||
|
finally:
|
||||||
|
with self._lock:
|
||||||
|
self.active -= 1
|
||||||
|
self.completed_frames.append(frame_no)
|
||||||
|
# 与真实 vlm-ocr 节点一致的返回结构。
|
||||||
|
return InvokeResponse(status="completed", outputs={"text": text})
|
||||||
|
|
||||||
|
|
||||||
|
def _run_ocr(monkeypatch, manifest_path: Path, texts_by_frame: dict[int, str],
|
||||||
|
pool_min: int, pool_max: int, output_dir: Path, seed: int,
|
||||||
|
) -> tuple[Path, FakeVlmOcrApi]:
|
||||||
|
"""用给定线程配置运行 subtitle-ocr,返回 (产物路径, 假 API 实例)。"""
|
||||||
|
from nodes.subtitle_ocr import invoke as ocr_invoke
|
||||||
|
|
||||||
|
fake = FakeVlmOcrApi(texts_by_frame, seed=seed)
|
||||||
|
monkeypatch.setattr("wov_app.registry.invoke", fake)
|
||||||
|
response = ocr_invoke(
|
||||||
|
InvokeRequest(
|
||||||
|
run_id="order_test",
|
||||||
|
node_instance_id="",
|
||||||
|
inputs={"frames_manifest": str(manifest_path)},
|
||||||
|
params={"pool_min_workers": pool_min, "pool_max_workers": pool_max},
|
||||||
|
output_dir=str(output_dir),
|
||||||
|
)
|
||||||
|
)
|
||||||
|
assert response.status == "completed", response.error
|
||||||
|
return Path(response.outputs["srt_uri"]), fake
|
||||||
|
|
||||||
|
|
||||||
|
def _load_full_data() -> tuple[list[dict], dict[int, str]]:
|
||||||
|
"""加载全量夹具:manifest 与 {帧号: 文本}。"""
|
||||||
|
manifest = json.loads(FULL_MANIFEST.read_text(encoding="utf-8"))
|
||||||
|
texts_by_frame = {
|
||||||
|
int(key): value for key, value in json.loads(FULL_OCR_TEXTS.read_text(encoding="utf-8")).items()
|
||||||
|
}
|
||||||
|
return manifest, texts_by_frame
|
||||||
|
|
||||||
|
|
||||||
|
def _assert_alignment(srt_text: str, manifest: list[dict], texts_by_frame: dict[int, str]) -> None:
|
||||||
|
"""核心对齐断言:每条字幕的起始时刻对应的帧,其 OCR 文本必须就是本条字幕文本。
|
||||||
|
|
||||||
|
这正是"多线程下顺序正确"的最终验证:无论线程如何并发/乱序完成,
|
||||||
|
每条字幕贴的时刻必须是它真实来源帧的时刻。
|
||||||
|
"""
|
||||||
|
time_text = {
|
||||||
|
round(float(entry["time"]), 3): texts_by_frame[_frame_no(entry["image_uri"])]
|
||||||
|
for entry in manifest
|
||||||
|
}
|
||||||
|
blocks = re.findall(
|
||||||
|
r"(\d{2}:\d{2}:\d{2},\d{3})\s*-->\s*(\d{2}:\d{2}:\d{2},\d{3})\s*\n(.*?)(?=\n\s*\d+\s*\n|\Z)",
|
||||||
|
srt_text, re.DOTALL,
|
||||||
|
)
|
||||||
|
times = []
|
||||||
|
for start, _end, text in blocks:
|
||||||
|
start_s = _ts_to_seconds(start)
|
||||||
|
# 帧时间与 SRT 时间戳间允许 ±2ms 容差:format_timestamp 用 int 截断,
|
||||||
|
# 浮点 256.258 会以 256.25799.. 截断为 256,257(真实运行同样行为)。
|
||||||
|
best = min(time_text, key=lambda t: abs(t - start_s))
|
||||||
|
assert abs(best - start_s) <= 0.002, f"字幕起始时刻 {start_s}s 无对应帧"
|
||||||
|
assert time_text[best] == text.strip(), f"时刻 {start_s}s 的文本与帧不一致"
|
||||||
|
times.append(start_s)
|
||||||
|
assert all(a < b for a, b in zip(times, times[1:])), "时间轴必须严格递增"
|
||||||
|
|
||||||
|
|
||||||
|
class TestSubtitleOcrOrderUnderThreading:
|
||||||
|
"""多线程下字幕顺序正确性测试类(全量 14236 帧真实数据 + 真实可变延迟)。
|
||||||
|
|
||||||
|
确认结果(testdata/ocr_srt_run_ac7f480a3ccb.srt)是真实任务 run_ac7f480a3ccb
|
||||||
|
以**单线程**(workflow v4 pool 1/1)运行产出、经用户确认的精确结果:
|
||||||
|
- 多线程(4/16 线程)+ 真实可变延迟下重放全量数据,产物必须与它逐字节一致;
|
||||||
|
- 同时必须真实发生"多线程 + 乱序完成 + 明显慢帧",而顺序仍正确。
|
||||||
|
数据全部为常驻夹具,测试始终执行(不依赖 gitignored 数据)。
|
||||||
|
"""
|
||||||
|
|
||||||
|
def test_full_real_data_variable_latency_keeps_order(self, monkeypatch, tmp_path) -> None:
|
||||||
|
"""全量真实数据 + 真实可变延迟:多线程产物与单线程确认结果逐字节一致。"""
|
||||||
|
manifest, texts_by_frame = _load_full_data()
|
||||||
|
confirmed = CONFIRMED_SRT.read_text(encoding="utf-8")
|
||||||
|
|
||||||
|
outputs: dict[tuple, str] = {}
|
||||||
|
fakes: dict[tuple, FakeVlmOcrApi] = {}
|
||||||
|
for pool_min, pool_max in ((4, 4), (16, 16)):
|
||||||
|
out_dir = tmp_path / f"p{pool_min}-{pool_max}"
|
||||||
|
srt_path, fake = _run_ocr(
|
||||||
|
monkeypatch, FULL_MANIFEST, texts_by_frame,
|
||||||
|
pool_min=pool_min, pool_max=pool_max,
|
||||||
|
output_dir=out_dir, seed=20260817,
|
||||||
|
)
|
||||||
|
outputs[(pool_min, pool_max)] = srt_path.read_text(encoding="utf-8")
|
||||||
|
fakes[(pool_min, pool_max)] = fake
|
||||||
|
|
||||||
|
# ① 多线程产物与用户确认过的精确结果(真实单线程运行)逐字节一致。
|
||||||
|
assert outputs[(4, 4)] == confirmed, "4 线程产物与确认结果不一致"
|
||||||
|
assert outputs[(16, 16)] == confirmed, "16 线程产物与确认结果不一致"
|
||||||
|
assert outputs[(4, 4)] == outputs[(16, 16)]
|
||||||
|
|
||||||
|
# ② 多线程确实发生(并发峰值>1),且可变延迟下完成顺序乱序(靠后帧
|
||||||
|
# 先完成/慢帧滞后),但输出仍与确认结果一致——线程池正确恢复了顺序。
|
||||||
|
for (pool_min, pool_max), fake in fakes.items():
|
||||||
|
assert fake.max_active > 1, f"{pool_max} 线程配置下应真实并发"
|
||||||
|
assert fake.completed_frames != sorted(fake.completed_frames), \
|
||||||
|
f"{pool_max} 线程下可变延迟应产生乱序完成"
|
||||||
|
|
||||||
|
# ③ 延迟模拟符合真实 OCR:存在明显慢帧(>3 倍快帧均值),也含极快帧。
|
||||||
|
for (_, pool_max), fake in fakes.items():
|
||||||
|
delays = fake.delays_ms
|
||||||
|
assert len(delays) == TOTAL_FRAMES, "每帧都应产生一次调用延迟"
|
||||||
|
assert max(delays) > 3.0, f"{pool_max} 线程下应存在明显慢帧(真实 OCR 延迟波动)"
|
||||||
|
assert min(delays) < 1.0, "应存在快帧(大部分帧响应快)"
|
||||||
|
|
||||||
|
# ④ 全量时间轴严格递增 + 每条字幕与其起始时刻帧的文本对齐。
|
||||||
|
_assert_alignment(confirmed, manifest, texts_by_frame)
|
||||||
|
|
||||||
|
# ⑤ 已知真实内容存在于结果中(确认结果的代表性条目)。
|
||||||
|
for line in ("北冈小姐", "这是特别病房患者的病历表", "应该已经察觉到 至今为止的一切了吧"):
|
||||||
|
assert line in confirmed, line
|
||||||
@@ -2,10 +2,10 @@
|
|||||||
"id": "ocr-subtitle",
|
"id": "ocr-subtitle",
|
||||||
"name": "字幕OCR提取",
|
"name": "字幕OCR提取",
|
||||||
"description": "抽帧并 OCR 提取视频烧录字幕,经 LLM 过滤无意义内容后生成带时间轴的 SRT 基准数据。",
|
"description": "抽帧并 OCR 提取视频烧录字幕,经 LLM 过滤无意义内容后生成带时间轴的 SRT 基准数据。",
|
||||||
"version": 3,
|
"version": 4,
|
||||||
"definition": {
|
"definition": {
|
||||||
"name": "字幕OCR提取",
|
"name": "字幕OCR提取",
|
||||||
"version": 3,
|
"version": 4,
|
||||||
"nodes": [
|
"nodes": [
|
||||||
{
|
{
|
||||||
"id": "extract",
|
"id": "extract",
|
||||||
@@ -27,7 +27,9 @@
|
|||||||
"id": "ocr",
|
"id": "ocr",
|
||||||
"node_type": "subtitle-ocr",
|
"node_type": "subtitle-ocr",
|
||||||
"params": {
|
"params": {
|
||||||
"prompt": "提取图像中的文字,不要描述图片中的内容"
|
"prompt": "提取图像中的文字,不要描述图片中的内容",
|
||||||
|
"pool_min_workers": 1,
|
||||||
|
"pool_max_workers": 1
|
||||||
},
|
},
|
||||||
"inputs": {
|
"inputs": {
|
||||||
"frames_manifest": "extract.frames_manifest"
|
"frames_manifest": "extract.frames_manifest"
|
||||||
@@ -36,7 +38,10 @@
|
|||||||
{
|
{
|
||||||
"id": "filter",
|
"id": "filter",
|
||||||
"node_type": "llm-filter",
|
"node_type": "llm-filter",
|
||||||
"params": {},
|
"params": {
|
||||||
|
"pool_min_workers": 1,
|
||||||
|
"pool_max_workers": 1
|
||||||
|
},
|
||||||
"inputs": {
|
"inputs": {
|
||||||
"srt_uri": "ocr.srt_uri"
|
"srt_uri": "ocr.srt_uri"
|
||||||
}
|
}
|
||||||
|
|||||||