cat-shark
4746e0363f
为视频生成 VR 双眼字幕的单体实现：FastAPI 后端、调度器与全部节点 （提音/转写/翻译/ASS/抽帧/OCR/LLM 过滤）在单进程内运行。 - 节点协议（wov_sdk 数据模型）与分布式版保持一致，预留回退桥梁 - 工作流即数据：DAG 存于 workflows/*.json，模型/链路改动只改数据 - 调度器：拓扑顺序执行、断点续跑（产物重建）、任务暂停/继续 - 抽帧按帧间隔（select 按帧号精确取帧），VLM OCR 与 LLM 过滤使用 自适应线程池弹性并发，并打印数据处理速度进度日志 - 100% 行覆盖率（pytest --cov-fail-under=100）
66 lines
2.5 KiB
Python
66 lines
2.5 KiB
Python
"""VLM OCR 节点真实集成测试。
|
|
|
|
复用 testdata/test_real_hav_sub.png（真实视频字幕截图，一次性入库，避免
|
|
每次测试生成）。调用本地 Ollama 服务（192.168.123.70:11434）的真实
|
|
glm-ocr 模型做 OCR。Ollama 服务或测试资产缺失时自动跳过；可用时必须执行。
|
|
"""
|
|
import urllib.request
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from nodes.vlm import invoke
|
|
from wov_sdk.models import InvokeRequest
|
|
|
|
OLLAMA_HOST = "http://192.168.123.70:11434"
|
|
MODEL = "glm-ocr:latest"
|
|
# 测试图片（真实视频字幕帧）上应识别出的字幕文本。
|
|
EXPECTED_TEXT = "还有没有什么困扰 或者奇怪的地方吗"
|
|
|
|
# 单体根目录：tests/ 的上一级。
|
|
WORKSPACE = Path(__file__).resolve().parent.parent
|
|
TEST_IMAGE = WORKSPACE / "testdata" / "test_real_hav_sub.png"
|
|
|
|
|
|
def _ollama_reachable() -> bool:
|
|
"""探测 Ollama 服务与目标模型是否可用。"""
|
|
try:
|
|
req = urllib.request.Request(
|
|
f"{OLLAMA_HOST}/api/show",
|
|
data=b'{"model": "%s"}' % MODEL.encode(),
|
|
headers={"Content-Type": "application/json"},
|
|
method="POST",
|
|
)
|
|
with urllib.request.urlopen(req, timeout=5) as resp:
|
|
return resp.status == 200
|
|
except (urllib.error.URLError, OSError):
|
|
return False
|
|
|
|
|
|
@pytest.mark.integration
|
|
def test_vlm_ocr_real_model(tmp_path) -> None:
|
|
"""复用真实字幕截图 + 真实 glm-ocr：应识别出关键字幕文本并清洗围栏垃圾。"""
|
|
if not _ollama_reachable():
|
|
pytest.skip("Ollama 服务或 glm-ocr 模型不可用，跳过真实模型集成测试")
|
|
if not TEST_IMAGE.is_file():
|
|
pytest.skip("缺少 testdata/test_real_hav_sub.png 测试资产，跳过集成测试")
|
|
|
|
response = invoke(
|
|
InvokeRequest(
|
|
run_id="vlm_integration",
|
|
node_instance_id="",
|
|
inputs={"image_uri": str(TEST_IMAGE)},
|
|
params={"model": MODEL, "ollama_host": OLLAMA_HOST},
|
|
output_dir=str(tmp_path / "out"),
|
|
)
|
|
)
|
|
assert response.status == "completed", response.error
|
|
text = response.outputs["text"]
|
|
print(text)
|
|
# 关键字幕文本应被识别出来。注意 glm-ocr 在此图上存在已知重复循环 bug：
|
|
# 识别出正确文本后可能继续循环输出，因此用"包含"断言而非全等，
|
|
# 下游 subtitle-ocr 的 max_result_chars 守卫会拦截超长输出。
|
|
assert EXPECTED_TEXT in text
|
|
# 围栏垃圾（```）不应出现在输出里。
|
|
assert "```" not in text
|