169 lines
8.2 KiB
Python
169 lines
8.2 KiB
Python
"""E2E-проверка инференса (запуск: .venv/Scripts/python tests/e2e_check.py).
|
|
|
|
1. Дорожки, извлечённые из сохранённых страниц (data/pages/*.txt) и HTML-проб
|
|
(data/samples/*.html), сравниваются с метками датасета по парам
|
|
(тип, канонический автор) на уровне релиза (precision/recall).
|
|
2. Smoke-тест на формулировках, которых нет в датасете дословно.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import sys
|
|
from collections import defaultdict
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
|
|
|
from src.predict import DubPredictor # noqa: E402
|
|
from src.html_extract import read_input # noqa: E402
|
|
from src.release import predict_release # noqa: E402
|
|
from src.studio_norm import canonical # noqa: E402
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
def release_check(predictor: DubPredictor, rows: list[dict], use_title: bool = True) -> dict:
|
|
"""E2E на текстовых дампах страниц: пары (тип, автор) по релизам."""
|
|
by_release: dict[tuple[str, str], list[dict]] = defaultdict(list)
|
|
for r in rows:
|
|
by_release[(r["source"], r["release_id"])].append(r)
|
|
|
|
id2src: dict[str, list[tuple[str, str]]] = defaultdict(list)
|
|
for (src, rid) in by_release:
|
|
id2src[rid].append((src, rid))
|
|
|
|
prec_sum = rec_sum = n = 0
|
|
per_src: dict[str, list[list[float]]] = defaultdict(list)
|
|
misses = []
|
|
pages = sorted((ROOT / "data" / "pages").glob("*.txt"))
|
|
for page in pages:
|
|
cands = id2src.get(page.stem, [])
|
|
if len(cands) != 1:
|
|
continue # нет меток или идентификатор неоднозначен между источниками
|
|
src, rid = cands[0]
|
|
gt_rows = by_release[(src, rid)]
|
|
gold = {(r["translation_type"], r["studio"] or None) for r in gt_rows}
|
|
try:
|
|
if use_title: # путь релиза: описание + студии из заголовка раздачи
|
|
tracks = predict_release(predictor, text=read_input(page))["tracks"]
|
|
else:
|
|
tracks = predictor.predict_file(page)
|
|
except Exception as e: # noqa: BLE001
|
|
misses.append(f"{page.name}: ошибка {e}")
|
|
continue
|
|
pred = {(t["translation_type"], t["studio"]) for t in tracks}
|
|
tp = len(gold & pred)
|
|
p = tp / len(pred) if pred else 0.0
|
|
r = tp / len(gold) if gold else 0.0
|
|
prec_sum += p
|
|
rec_sum += r
|
|
n += 1
|
|
per_src[src].append([p, r])
|
|
if p < 1 or r < 1:
|
|
misses.append(f"[{src}] {page.name}: gold={sorted(map(str, gold))} "
|
|
f"pred={sorted(map(str, pred))}")
|
|
stats = {"n_releases": n,
|
|
"mean_precision": prec_sum / max(n, 1),
|
|
"mean_recall": rec_sum / max(n, 1),
|
|
"per_source": {s: {"n": len(v),
|
|
"precision": sum(x[0] for x in v) / len(v),
|
|
"recall": sum(x[1] for x in v) / len(v)}
|
|
for s, v in per_src.items()},
|
|
"misses": misses[:8]}
|
|
return stats
|
|
|
|
|
|
# (текст, ожидаемый тип, ожидаемый автор-подстрока или None, ожидаемый 18+)
|
|
SMOKE = [
|
|
("Профессиональный (многоголосый закадровый) | HDRezka Studio 18+",
|
|
"проф. многоголосый", "HDrezka", True),
|
|
("Перевод: Дублированный | MovieDalen", "проф. дубляж", "MovieDalen", False),
|
|
("Авторская одноголосая озвучка — Дмитрий Пучков",
|
|
"авт. одноголосый", "Пучков", False),
|
|
("Аудио: русский DVO (СВ Студия)", "проф. двухголосый", "СВ Студия", False),
|
|
("перевод: любительский многоголосый, NewStudio",
|
|
"люб. многоголосый", "NewStudio", False),
|
|
("Одноголосый закадровый (Василий Горчаков)",
|
|
"одноголосый", "Горчаков", False),
|
|
("Закадровый одноголосый перевод: Юрий Немахов",
|
|
"одноголосый", "Немахов", False),
|
|
("Многоголосое озвучивание TVShows, 18+", "многоголосый", "TVShows", True),
|
|
("Озвучено студией LostFilm (многоголосое закадровое)",
|
|
"многоголосый", "LostFilm", False),
|
|
("Русский DUB (Кинопоиск)", "проф. дубляж", "Кинопоиск", False),
|
|
]
|
|
|
|
|
|
def smoke_check(predictor: DubPredictor) -> dict:
|
|
ok_type = ok_studio = ok_age = 0
|
|
fails = []
|
|
for text, want_type, want_studio, want_age in SMOKE:
|
|
r = predictor.predict_text(text)
|
|
t_ok = r["translation_type"] == want_type
|
|
s_ok = bool((want_studio is None and r["studio"] is None)
|
|
or (r["studio"] and want_studio.lower() in r["studio"].lower()))
|
|
a_ok = r["is_18plus"] == want_age
|
|
ok_type += t_ok
|
|
ok_studio += s_ok
|
|
ok_age += a_ok
|
|
if not (t_ok and s_ok and a_ok):
|
|
fails.append(f"«{text[:50]}» -> {r['translation_type']}, "
|
|
f"studio={r['studio']}, 18+={r['is_18plus']} "
|
|
f"(ожидалось {want_type}, {want_studio}, {want_age})")
|
|
return {"n": len(SMOKE), "type_ok": ok_type, "studio_ok": ok_studio,
|
|
"age_ok": ok_age, "fails": fails}
|
|
|
|
|
|
def main() -> None:
|
|
rows = json.loads((ROOT / "data" / "processed" / "dataset.json")
|
|
.read_text(encoding="utf-8"))
|
|
predictor = DubPredictor()
|
|
|
|
print("== e2e: страницы против меток (только описание, baseline) ==")
|
|
base = release_check(predictor, rows, use_title=False)
|
|
print(f"релизов: {base['n_releases']}, precision {base['mean_precision']:.3f}, "
|
|
f"recall {base['mean_recall']:.3f}; "
|
|
+ "; ".join(f"{s}: P={v['precision']:.3f} R={v['recall']:.3f}"
|
|
for s, v in base["per_source"].items()))
|
|
print("\n== e2e: страницы против меток (описание + заголовок, predict_release) ==")
|
|
rel = release_check(predictor, rows)
|
|
print(f"релизов проверено: {rel['n_releases']}, "
|
|
f"mean precision {rel['mean_precision']:.3f}, "
|
|
f"mean recall {rel['mean_recall']:.3f}")
|
|
for s, v in rel.get("per_source", {}).items():
|
|
print(f" [{s}] n={v['n']} precision={v['precision']:.3f} "
|
|
f"recall={v['recall']:.3f}")
|
|
for m in rel["misses"]:
|
|
print(f" ! {m}")
|
|
|
|
print("\n== e2e: HTML-пробы (проверка кодировок/структуры) ==")
|
|
for f in sorted((ROOT / "data" / "samples").glob("*.html")):
|
|
try:
|
|
tracks = predictor.predict_file(f)
|
|
names = ", ".join(f"{t['translation_type']}/{t['studio'] or '—'}"
|
|
for t in tracks[:4])
|
|
print(f" {f.name}: {len(tracks)} дорожек: {names}")
|
|
except Exception as e: # noqa: BLE001
|
|
print(f" {f.name}: ошибка {e}")
|
|
|
|
print("\n== smoke: формулировки вне датасета ==")
|
|
sm = smoke_check(predictor)
|
|
print(f"тип {sm['type_ok']}/{sm['n']}, автор {sm['studio_ok']}/{sm['n']}, "
|
|
f"18+ {sm['age_ok']}/{sm['n']}")
|
|
for f in sm["fails"]:
|
|
print(f" ! {f}")
|
|
|
|
rt = rel.get("per_source", {}).get("rutracker", {})
|
|
# жёсткий критерий: тип на OOD-формулировках и полнота на rutracker-страницах;
|
|
# studio на придуманных форматах — диагностика (см. README «Ограничения»)
|
|
hard_fail = (sm["type_ok"] < sm["n"] - 2
|
|
or rel["n_releases"] == 0
|
|
or (rt and rt["recall"] < 0.75))
|
|
print("\nитог:", "OK" if not hard_fail else "ЕСТЬ ПРОБЛЕМЫ")
|
|
sys.exit(1 if hard_fail else 0)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|