107 lines
4.3 KiB
Python
107 lines
4.3 KiB
Python
"""smoke: transcribe_audio / core.asr_lfasr 的纯本地部分(不联讯飞)。
|
|
|
|
验证:
|
|
1. asr_lfasr.parse_transcript:json_1best 为串(lattice)与对象(lattice2)两形态、
|
|
空文本段跳过、rl 角色提取;
|
|
2. asr_lfasr.format_transcript:多说话人按连续同角色合并「【说话人N】」段落,
|
|
单说话人纯文本;
|
|
3. TranscribeAudioTool 入参错误路径(路径不存在 / 扩展名不对 / 视频提示 / 空文件 /
|
|
越界 / 坏 language),全部在上传前早返,不触网。
|
|
|
|
跑法: .venv/Scripts/python.exe scripts/smoke_transcribe_audio.py
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import sys
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
|
|
|
from core.asr_lfasr import format_transcript, parse_transcript # noqa: E402
|
|
from tools.transcribe_audio import TranscribeAudioTool # noqa: E402
|
|
|
|
FAIL = 0
|
|
|
|
|
|
def check(name: str, ok: bool, detail: str = "") -> None:
|
|
global FAIL
|
|
tag = "PASS" if ok else "FAIL"
|
|
print(f"[{tag}] {name}" + (f" -- {detail}" if detail else ""))
|
|
if not ok:
|
|
FAIL = 1
|
|
|
|
|
|
def seg(text: str, rl: str = "0") -> dict:
|
|
"""构造一个 lattice 项(json_1best 为再编码 JSON 串,官方返回形态)。"""
|
|
st = {"rl": rl, "rt": [{"ws": [{"cw": [{"w": ch}]} for ch in text]}]}
|
|
return {"json_1best": json.dumps({"st": st}, ensure_ascii=False)}
|
|
|
|
|
|
def main() -> int:
|
|
# ---- 1. parse_transcript ----
|
|
order = json.dumps({"lattice": [seg("今天开会", "1"), seg("好的", "2"), seg("", "1")]},
|
|
ensure_ascii=False)
|
|
segs = parse_transcript(order)
|
|
check("parse: 2 segments (empty skipped)", len(segs) == 2, str(segs))
|
|
check("parse: roles extracted", segs == [("1", "今天开会"), ("2", "好的")], str(segs))
|
|
|
|
# lattice2 形态:json_1best 已是对象
|
|
obj_item = json.loads(seg("测试", "0")["json_1best"])
|
|
segs2 = parse_transcript(json.dumps({"lattice2": [{"json_1best": obj_item}]},
|
|
ensure_ascii=False))
|
|
check("parse: lattice2 object form", segs2 == [("", "测试")], str(segs2))
|
|
|
|
# ---- 2. format_transcript ----
|
|
multi = [("1", "大家好。"), ("1", "开始吧。"), ("2", "收到。"), ("1", "散会。")]
|
|
out = format_transcript(multi)
|
|
check("format: multi-speaker merged",
|
|
out == "【说话人1】大家好。开始吧。\n\n【说话人2】收到。\n\n【说话人1】散会。",
|
|
repr(out))
|
|
single = [("1", "自言"), ("1", "自语")]
|
|
check("format: single speaker plain", format_transcript(single) == "自言自语")
|
|
check("format: no-role plain", format_transcript([("", "甲"), ("", "乙")]) == "甲乙")
|
|
|
|
# ---- 3. tool 早返错误路径(不触网) ----
|
|
with tempfile.TemporaryDirectory(prefix="zcbot-smoke-ta-") as td:
|
|
root = Path(td)
|
|
wd = root / "taskdir"
|
|
wd.mkdir()
|
|
tool = TranscribeAudioTool(working_dir=wd, base_dir=wd, user_root=root)
|
|
|
|
r = tool.execute(audio="inbound/nope.m4a")
|
|
check("missing file -> [Error]", r.startswith("[Error]") and "找不到" in r, r[:60])
|
|
|
|
(wd / "notes.docx").write_bytes(b"PK\x03\x04fake")
|
|
r = tool.execute(audio="notes.docx")
|
|
check("non-audio ext -> [Error]", r.startswith("[Error]") and "扩展名" in r, r[:60])
|
|
|
|
(wd / "clip.mp4").write_bytes(b"\x00\x00\x00\x18ftyp")
|
|
r = tool.execute(audio="clip.mp4")
|
|
check("video ext -> hint to extract audio",
|
|
r.startswith("[Error]") and "音轨" in r, r[:80])
|
|
|
|
(wd / "empty.mp3").write_bytes(b"")
|
|
r = tool.execute(audio="empty.mp3")
|
|
check("empty file -> [Error]", r.startswith("[Error]") and "空" in r, r[:60])
|
|
|
|
outside = root.parent / "zcbot-smoke-outside.mp3"
|
|
try:
|
|
outside.write_bytes(b"xx")
|
|
r = tool.execute(audio=str(outside))
|
|
check("out-of-root abs path -> [Error]",
|
|
r.startswith("[Error]") and "越界" in r, r[:60])
|
|
finally:
|
|
outside.unlink(missing_ok=True)
|
|
|
|
r = tool.execute(audio="a.mp3", language="fr")
|
|
check("bad language -> [Error]", r.startswith("[Error]") and "language" in r, r[:60])
|
|
|
|
print("RESULT:", "FAIL" if FAIL else "ALL PASS")
|
|
return FAIL
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|