GET /v1/jobs/view - список последних задач со ссылками на разбор. Токен
подставляется в ссылки, иначе по каждой пришлось бы дописывать его руками.
Отдельный тест следит, чтобы маршрут не перехватывался /v1/jobs/{job_id}.
build/compare_reference.py считает WER относительно эталона, выписанного
на слух. Косвенные метрики ловят крупные сдвиги и слепы к систематическим
мелким потерям - так и пропустили обрыв концов фраз.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
86 lines
3.8 KiB
Python
86 lines
3.8 KiB
Python
#!/usr/bin/env python3
|
||
"""Сравнивает расшифровку с эталоном, выписанным человеком на слух.
|
||
|
||
До сих пор качество мерили косвенно: числом слов и совпадением моделей между
|
||
собой. Это ловит крупные сдвиги и слепо к систематическим мелким потерям -
|
||
именно так пропустили обрыв концов фраз. Эталон даёт прямую меру.
|
||
|
||
uv run python build/compare_reference.py эталон.txt расшифровка.json
|
||
"""
|
||
import argparse
|
||
import difflib
|
||
import json
|
||
import re
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
# Пунктуация и регистр к делу не относятся: сравниваем, какие слова сказаны.
|
||
WORD_RE = re.compile(r"[\w-]+", re.UNICODE)
|
||
|
||
|
||
def words(text: str) -> list[str]:
|
||
return [w.lower().replace("ё", "е") for w in WORD_RE.findall(text)]
|
||
|
||
|
||
def from_reference(path: Path) -> list[str]:
|
||
"""Читает эталон, отбрасывая пометки говорящих и тайм-коды."""
|
||
out = []
|
||
for line in path.read_text(encoding="utf-8").splitlines():
|
||
# Строки вида "МЕНЕДЖЕР: текст" или "[01:23] КЛИЕНТ: текст".
|
||
line = re.sub(r"^\s*\[?\d{1,2}:\d{2}\]?\s*", "", line)
|
||
line = re.sub(r"^\s*(МЕНЕДЖЕР|КЛИЕНТ|М|К|M|C|Г\d)\s*[:\-]\s*", "", line,
|
||
flags=re.IGNORECASE)
|
||
out.extend(words(line))
|
||
return out
|
||
|
||
|
||
def from_result(path: Path) -> list[str]:
|
||
data = json.loads(path.read_text(encoding="utf-8"))
|
||
result = data.get("result") or data
|
||
return words(" ".join(t["text"] for t in result.get("turns", [])))
|
||
|
||
|
||
def report(reference: list[str], actual: list[str], show: int) -> dict:
|
||
"""Считает ошибки по операциям редактирования: пропуски, лишние, замены."""
|
||
matcher = difflib.SequenceMatcher(a=reference, b=actual, autojunk=False)
|
||
missing: list[str] = []
|
||
extra: list[str] = []
|
||
swapped: list[tuple[str, str]] = []
|
||
for tag, i1, i2, j1, j2 in matcher.get_opcodes():
|
||
if tag == "delete":
|
||
missing.extend(reference[i1:i2])
|
||
elif tag == "insert":
|
||
extra.extend(actual[j1:j2])
|
||
elif tag == "replace":
|
||
swapped.extend(zip(reference[i1:i2], actual[j1:j2]))
|
||
# Хвосты неравной длины - это тоже пропуск или лишнее.
|
||
missing.extend(reference[i1 + (j2 - j1):i2])
|
||
extra.extend(actual[j1 + (i2 - i1):j2])
|
||
|
||
errors = len(missing) + len(extra) + len(swapped)
|
||
wer = errors / max(1, len(reference))
|
||
print(f"эталон {len(reference)} слов, распознано {len(actual)}")
|
||
print(f"пропущено {len(missing)}, лишних {len(extra)}, перепутано {len(swapped)}")
|
||
print(f"WER {wer * 100:.1f} % (доля верных слов {(1 - wer) * 100:.1f} %)\n")
|
||
if missing[:show]:
|
||
print("не распознано:", " ".join(missing[:show]))
|
||
if swapped[:show]:
|
||
print("услышано иначе:",
|
||
", ".join(f"{a}->{b}" for a, b in swapped[:show]))
|
||
return {"wer": wer, "missing": len(missing), "extra": len(extra),
|
||
"swapped": len(swapped), "reference": len(reference)}
|
||
|
||
|
||
def main() -> int:
|
||
ap = argparse.ArgumentParser()
|
||
ap.add_argument("reference", type=Path)
|
||
ap.add_argument("result", type=Path)
|
||
ap.add_argument("--show", type=int, default=40, help="сколько примеров печатать")
|
||
args = ap.parse_args()
|
||
report(from_reference(args.reference), from_result(args.result), args.show)
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|