cat-shark
2b3a650612
- frame-extract 新增 _sorted_frame_files：ffmpeg %04d 编号超过 9999 帧后扩为 5 位，sorted() 字典序会把 5 位编号排在 4 位之前，导致 frames.json 时间与 图像错位（真实发生于 run_339ec7ee437f 的 14236 帧任务） - subtitle-ocr 提取 _merge_kept 供重组装复用 - 回归测试用真实任务留存数据（testdata/frames_boundary/），并新增真实 OCR 数据按正确时间轴重组装为 SRT 的集成测试（test_integration_reassemble_ocr）
98 lines
4.4 KiB
Python
98 lines
4.4 KiB
Python
"""真实任务 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
|