cat-shark
a8fe133aa4
解决转写漏句（有人说话但没识别出来）问题：silero VAD 对呻吟/轻语/BGM 混叠声学切段能力天然不足，把真话当非语音剔除（实测 savr-1054 全片仅 召回 115 条）。新增 decode_full 参数（默认 false 保持 VAD 现状）： - decode_full=true 时强制无 VAD 整段解码 + 跳过自动 VAD 分析，救回被 剔除的弱语音（savr-1054 全片 115 条 → 340 条） - 副作用是长时寒暄套话幻觉（おやすみなさい/ご視聴ありがとうございま した 等），whisper 转录后连带时间戳整条删除（clean_japanese_ha lllucinations），不留下 '-' 占位污染下游（占位会渲染进 ASS 成减号） - llm-translate 翻译后同样整条删除中文长时寒暄幻觉（clean_srt_text） - 短时（≤15s）相同词可能是剧情真实道晚安，保留（15s 阈值实测校准） - subtitle_cleanup 由 '-' 占位式改为整条删除式 + 剩余重编号，新增 JAPANESE_HALLUCINATION_TOKENS 词表 新增工作流 learn-translate（学习资料转译+翻译字幕）示范 decode_full 用法，并确立参数标注约定：params._note_<参数名> 存放设定理由与正反例、 _node_help 放节点参数手册（_ 前缀说明键，节点执行时忽略，零运行影响）。 调研记录见 docs/调研-whisper漏句与decode_full验证.md（A/B 实验、结论 修正与 5 个待决问题）。
114 lines
4.4 KiB
Python
114 lines
4.4 KiB
Python
"""种子数据测试。
|
||
|
||
验证从 workflows/*.json 数据文件加载默认工作流、幂等性，以及
|
||
开启种子与调度器后的应用生命周期。工作流定义来自数据文件而非代码。
|
||
"""
|
||
|
||
from pathlib import Path
|
||
|
||
from fastapi.testclient import TestClient
|
||
|
||
from wov_app.db import Database
|
||
from wov_app.main import app
|
||
from wov_app.seed import seed_default_workflows
|
||
|
||
# 单体根目录：tests/ 的上一级。
|
||
WORKSPACE = Path(__file__).resolve().parent.parent
|
||
|
||
|
||
def test_seed_default_workflows_idempotent(tmp_path) -> None:
|
||
"""验证从数据文件加载 demo/zh-direct/ocr-subtitle/learn-translate 工作流且重复调用幂等。"""
|
||
db = Database(tmp_path / "wov.db")
|
||
created = seed_default_workflows(db)
|
||
assert created == 4
|
||
assert db.get_workflow("demo") is not None
|
||
assert db.get_workflow("zh-direct") is not None
|
||
|
||
# demo：asr 显式声明 model_path 与长音频参数，模型选择完全数据化。
|
||
demo = db.get_latest_workflow_version("demo")["definition"]
|
||
assert demo["name"] == "视频字幕生成"
|
||
demo_asr = next(node for node in demo["nodes"] if node["id"] == "asr")
|
||
assert demo_asr["params"]["model_path"] == "faster-whisper-large-v3"
|
||
assert demo_asr["params"]["condition_on_previous_text"] is False
|
||
|
||
# zh-direct：使用中文直出模型并开启翻译任务。
|
||
zh = db.get_latest_workflow_version("zh-direct")["definition"]
|
||
assert zh["name"] == "中文直出字幕"
|
||
zh_asr = next(node for node in zh["nodes"] if node["id"] == "asr")
|
||
assert zh_asr["params"]["model_path"] == "whisper-large-v2-translate-zh-v0.2-st-ct2"
|
||
assert zh_asr["params"]["task"] == "translate"
|
||
|
||
# learn-translate：应用本次 decode_full 修复的新工作流。
|
||
learn = db.get_latest_workflow_version("learn-translate")["definition"]
|
||
learn_asr = next(node for node in learn["nodes"] if node["id"] == "asr")
|
||
assert learn_asr["params"]["decode_full"] is True
|
||
|
||
# 再次调用不重复创建。
|
||
assert seed_default_workflows(db) == 0
|
||
assert len(db.list_workflow_versions("demo")) == 1
|
||
|
||
|
||
def test_seed_custom_dir_and_empty(tmp_path) -> None:
|
||
"""验证自定义数据目录的加载与空目录返回 0。"""
|
||
db = Database(tmp_path / "wov.db")
|
||
custom = tmp_path / "workflows"
|
||
custom.mkdir()
|
||
(custom / "a.json").write_text(
|
||
"""
|
||
{
|
||
"id": "flow-a",
|
||
"name": "Flow A",
|
||
"description": "custom",
|
||
"version": 1,
|
||
"definition": {
|
||
"name": "Flow A",
|
||
"version": 1,
|
||
"nodes": [{"id": "step", "node_type": "echo"}],
|
||
"edges": [],
|
||
"entry_inputs": {},
|
||
"final_outputs": {}
|
||
}
|
||
}
|
||
""",
|
||
encoding="utf-8",
|
||
)
|
||
assert seed_default_workflows(db, custom) == 1
|
||
assert db.get_workflow("flow-a") is not None
|
||
# 空目录返回 0。
|
||
empty = tmp_path / "empty"
|
||
empty.mkdir()
|
||
assert seed_default_workflows(db, empty) == 0
|
||
# 已存在的工作流被跳过。
|
||
assert seed_default_workflows(db, custom) == 0
|
||
|
||
|
||
def test_lifespan_with_seed_and_scheduler(monkeypatch) -> None:
|
||
"""验证启用自动种子与调度器后应用正常启动，demo 与中文直出应用均可见。"""
|
||
monkeypatch.setenv("WOV_AUTO_SEED", "1")
|
||
monkeypatch.setenv("WOV_SCHEDULER_ENABLED", "1")
|
||
with TestClient(app) as client:
|
||
apps = client.get("/api/apps")
|
||
assert apps.status_code == 200
|
||
assert any(item["id"] == "demo" for item in apps.json())
|
||
assert any(item["id"] == "zh-direct" for item in apps.json())
|
||
|
||
|
||
def test_seed_workflows_have_chunk_seconds(tmp_path) -> None:
|
||
"""验证内置工作流的 asr 节点均显式声明分块参数。"""
|
||
db = Database(tmp_path / "wov.db")
|
||
seed_default_workflows(db)
|
||
for workflow_id in ("demo", "zh-direct"):
|
||
definition = db.get_latest_workflow_version(workflow_id)["definition"]
|
||
asr = next(node for node in definition["nodes"] if node["id"] == "asr")
|
||
assert asr["params"]["chunk_seconds"] == 60
|
||
|
||
|
||
def test_seed_workflows_vad_filter_off(tmp_path) -> None:
|
||
"""验证内置工作流 asr 显式开启 VAD。"""
|
||
db = Database(tmp_path / "wov.db")
|
||
seed_default_workflows(db)
|
||
for workflow_id in ("demo", "zh-direct"):
|
||
definition = db.get_latest_workflow_version(workflow_id)["definition"]
|
||
asr = next(node for node in definition["nodes"] if node["id"] == "asr")
|
||
assert asr["params"]["vad_filter"] is True
|