cat-shark
8a715a8064
按"测试规则"重写 tests/：一个模块一个目录、用例按数据→过程→验证三段书写、 不保留全局 conftest.py、测试过程只调用真实生产代码。 结构（73 个文件、30 个模块目录、477 用例）： - tests/nodes/ 15 个模块目录（srt/whisper/ass/ffmpeg/frame_extract/vlm/ subtitle_ocr/llm/llm_filter/subtitle_cleanup/subtitle_correction/ proper_nouns/adaptive_pool/vad_profiler/echo）； - tests/app/ 11 个模块目录（db/scheduler/batch/maintenance/registry/seed/ storage/config/logging/main/routers 三组 API）； - tests/sdk/test_models、tests/web/test_crop、tests/shared（公共设施）。 测试数据随模块目录入库（tests/**/data/），删除根级 testdata/；.gitignore 的 data/ 改为 /data/，否则会连带忽略 tests/**/data/ 导致测试数据无法入库。 顺带发现并修复三个真实缺陷： - nodes/srt.py：相邻条目缺少空行时把下一条时间轴吞进正文（静默错位）， 改为正文行遇时间戳行即报错； - src/wov_app/scheduler.py：_file_size 只捕获 OSError，含 \x00 的产物 URI 抛 ValueError 导致任务误判失败，改为同时捕获； - nodes/subtitle_correction.py：生产代码依赖测试包解析 SRT， 改用生产模块 nodes/srt.py。 真实模型/服务集成测试按外部状态跳过：新增 tests/shared/gpu_memory.py （运行时探测显存、CUDA OOM 转跳过）与 tests/shared/llm_service.py （无 Key / 余额 / 限流转跳过）。全量 477 passed。
66 lines
2.3 KiB
Python
66 lines
2.3 KiB
Python
"""SRT 条目解析（纯函数，按秒返回，供参考字幕与产物字幕共用）。
|
|
|
|
与 `nodes/srt.py` 的区别：`nodes/srt.py` 保留原始毫秒字符串供生产链路
|
|
序列化；本模块把时间戳换算为**秒浮点数**，用于测试中的时间对齐计算与
|
|
篇幅统计，不参与生产输出。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
|
|
# 匹配 SRT 时间轴行（时间戳 --> 时间戳），作为条目边界。
|
|
_SRT_TIME_LINE_RE = re.compile(
|
|
r"(\d{2}:\d{2}:\d{2},\d{3})\s*-->\s*(\d{2}:\d{2}:\d{2},\d{3})",
|
|
)
|
|
|
|
|
|
def parse_srt_entries(text: str) -> list[dict]:
|
|
"""解析 SRT 为 [{start, end, text}]（秒为单位）。
|
|
|
|
按行分块：一个条目 = 序号行 + 时间轴行 + 若干文本行（可空）。即使文本为
|
|
空串（如翻译补空占位、空字幕）也计入一条，时间轴不丢失。正文多行用空格
|
|
连接，便于按关键词检索。
|
|
"""
|
|
entries: list[dict] = []
|
|
lines = text.splitlines()
|
|
index = 0
|
|
while index < len(lines):
|
|
line = lines[index].strip()
|
|
# 跳过序号行与空行，找时间轴行。
|
|
if not line or not _SRT_TIME_LINE_RE.search(line):
|
|
index += 1
|
|
continue
|
|
match = _SRT_TIME_LINE_RE.search(line)
|
|
start = _ts_to_seconds(match.group(1))
|
|
end = _ts_to_seconds(match.group(2))
|
|
index += 1
|
|
# 收集后续非序号、非时间轴的文本行（可空/多行），直到空行或序号行。
|
|
text_parts: list[str] = []
|
|
while index < len(lines):
|
|
nxt = lines[index].strip()
|
|
if not nxt:
|
|
break # 空行：条目结束
|
|
if _SRT_TIME_LINE_RE.search(nxt):
|
|
break # 下一个时间轴：条目结束
|
|
if nxt.isdigit():
|
|
break # 下一个序号：条目结束
|
|
text_parts.append(nxt)
|
|
index += 1
|
|
entries.append(
|
|
{
|
|
"start": start,
|
|
"end": end,
|
|
"text": " ".join(text_parts),
|
|
}
|
|
)
|
|
index += 1
|
|
return entries
|
|
|
|
|
|
def _ts_to_seconds(ts: str) -> float:
|
|
"""把 SRT 时间戳（HH:MM:SS,mmm）换算为秒。"""
|
|
hours, minutes, rest = ts.split(":")
|
|
seconds, millis = rest.split(",")
|
|
return int(hours) * 3600 + int(minutes) * 60 + int(seconds) + int(millis) / 1000
|