"""Юнит-тесты grabber-ai (запуск: .venv/Scripts/python tests/run_tests.py). Проверяют ключевые части без обращения к GPU: * поиск спана автора (точно / алиас / инициалы / кавычки / разделители); * канонизация имён студий; * снятие префикса строки дорожки; * сегментацию страниц на дорожки; * HTML -> текст; * BIO-выравнивание спана по offsets токенизатора. """ from __future__ import annotations import sys from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parents[1])) from src.html_extract import (RE_LINE_PREFIX, _split_entries, # noqa: E402 extract_segments, html_to_text) from src.studio_norm import canonical, find_studio_span # noqa: E402 PASS = 0 FAIL = 0 def check(name: str, cond: bool, extra: str = "") -> None: global PASS, FAIL if cond: PASS += 1 print(f" ok {name}") else: FAIL += 1 print(f"FAIL {name} {extra}") # ---------------------------------------------------------------- спаны print("== find_studio_span ==") txt = "Профессиональный (многоголосый закадровый) | HDRezka Studio 18+" s, e = find_studio_span(txt, "HDrezka") check("HDrezka найден", s is not None and "HDRezka" in txt[s:e], f"span={(s, e)}") s, e = find_studio_span("Авторский (одноголосый закадровый) - Ю. Сербин", "Юрий Сербин") check("инициалы Ю. Сербин", "Сербин" in "Авторский (одноголосый закадровый) - Ю. Сербин"[s:e] if s is not None else False) s, e = find_studio_span("Авторский (одноголосый закадровый) Гаврилов", "Андрей Гаврилов") check("фамилия Гаврилов", s is not None and e is not None and s > 0) s, e = find_studio_span('Профессиональный (многоголосый закадровый) Студия «Кравец», LostFilm', "Студия Кравец, LostFilm") check("кавычки в тексте", s is not None) s, e = find_studio_span("Профессиональный (многоголосый закадровый) - SDI Media | TV 1000", "SDI Media TV 1000") check("разделитель |", s is not None) check("нет автора -> None", find_studio_span("Профессиональный (дублированный)", "") is None) check("нет вхождения -> None", find_studio_span("Профессиональный (дублированный)", "HDrezka") is None) # ---------------------------------------------------------------- канонизация print("== canonical ==") check("HDRezka Studio -> HDrezka", canonical("HDRezka Studio") == "HDrezka") check("hdrezka -> HDrezka", canonical("hdrezka") == "HDrezka") check("Read Head Sound -> Red Head Sound", canonical("Read Head Sound") == "Red Head Sound") check("Ultradox Studio -> Ultradox", canonical("Ultradox Studio") == "Ultradox") check("Jaskier ненормативная лексика -> Jaskier", canonical("Jaskier ненормативная лексика") == "Jaskier") check("официальный -> ''", canonical("официальный") == "") # ---------------------------------------------------------------- префиксы print("== RE_LINE_PREFIX ==") for line, body in [ ("Перевод: Профессиональный (дублированный)", "Профессиональный (дублированный)"), ("Перевод 2: Любительский (одноголосый закадровый)", "Любительский (одноголосый закадровый)"), ("Перевод #1: Дублированный", "Дублированный"), ("Озвучивание: Дублированное | Official", "Дублированное | Official"), ("Аудио 1: Opus, 6 ch, 221 Kbps | Русский AVO (Дмитрий Пучков)", "Opus, 6 ch, 221 Kbps | Русский AVO (Дмитрий Пучков)"), ("Перевод:Профессиональный (дублированный)", "Профессиональный (дублированный)"), ]: check(f"префикс «{line[:20]}…»", RE_LINE_PREFIX.sub("", line) == body) # ---------------------------------------------------------------- сегментация print("== extract_segments ==") segs = extract_segments( "Описание релиза\n" "Перевод 1: Профессиональный (дублированный)\n" "Перевод 2: Профессиональный (двухголосый закадровый) tycoon\n" "Субтитры: русские, английские\n") check("две дорожки", len(segs) == 2, f"got {segs}") segs = extract_segments("Перевод 1: Профессиональный (дублированный) 2. Авторский (одноголосый закадровый) Гоблин") check("нумерация 1./2.", len(segs) == 2, f"got {segs}") segs = extract_segments("Перевод: Одноголосый, многоголосый закадровый") check("запятая-перечисление", len(segs) == 2, f"got {segs}") segs = extract_segments("Перевод: Профессиональный (многоголосый, закадровый)") check("скобки-уточнение не режутся", len(segs) == 1, f"got {segs}") segs = extract_segments("Перевод:\nПрофессиональный (дублированный) Невафильм") check("значение на следующей строке", segs == ["Профессиональный (дублированный) Невафильм"], f"got {segs}") segs = extract_segments("Перевод: Профессиональный (двухголосый закадровый) + Любительский (одноголосый закадровый)") check("разделитель +", len(segs) == 2, f"got {segs}") segs = extract_segments("Аудио 1: Opus, 6 ch, 221 Kbps | Русский AVO (Дмитрий Пучков)\n" "Аудио 2: Opus, 6 ch, 199 Kbps | Русский DVO (Кубик в Кубе)") check("nnm аудио-дорожки", len(segs) == 2, f"got {segs}") segs = extract_segments("Аудио: Русский (AC3, 6 ch, 640 Кбит/с), Английский (AC3)") check("codec-строка без типа пропущена", len(segs) == 0, f"got {segs}") check("_split_entries одиночная", _split_entries("Дублированный") == ["Дублированный"]) # ---------------------------------------------------------------- html print("== html_to_text ==") html = ("
меню форума
" '
Перевод: Профессиональный (дублированный) Невафильм
' "") text = html_to_text(html) check("селектор .post_body", "Невафильм" in text and "меню форума" not in text) html2 = "

hello

" check("fallback на весь текст", "hello" in html_to_text(html2)) print("== read_input / mojibake ==") import tempfile from src.html_extract import read_input with tempfile.TemporaryDirectory() as td: p = Path(td) / "mojibake.html" p.write_bytes("Перевод: Дублированный".encode("utf-8").decode("cp1251", errors="ignore").encode("utf-8")) check("mojibake восстановлен", "Дублированный" in read_input(p)) p2 = Path(td) / "cp1251.txt" p2.write_bytes("Перевод: Дублированный".encode("cp1251")) check("cp1251 прочитан", "Дублированный" in read_input(p2)) # ---------------------------------------------------------------- BIO print("== BIO-выравнивание (требует токенизатор) ==") try: from transformers import AutoTokenizer from src.train import make_collate tok = AutoTokenizer.from_pretrained("models/best") collate = make_collate(tok, {"проф. дубляж": 0}) text = "Авторский (одноголосый закадровый) - Сергей Визгунов" s0 = text.index("Сергей") enc, types, bio, ages, offsets = collate( [{"text": text, "translation_type": "проф. дубляж", "span": [s0, s0 + len("Сергей Визгунов")], "is_18plus": False}]) marked = {i for i, lab in enumerate(bio[0].tolist()) if lab >= 1} span_toks = [tok.convert_ids_to_tokens(int(enc["input_ids"][0, i])) for i in sorted(marked)] joined = "".join(t.lstrip("#") for t in span_toks).lower() # у ruBert-base токенизатор strips accents: «й» может стать «и» check("токены спана = ФИ", "визгунов" in joined and joined.startswith("сер"), f"tokens={span_toks}") check("первый токен B", bio[0, min(marked)].item() == 1 if marked else False) except Exception as e: # noqa: BLE001 print(f" пропущено (нет модели models/best): {e}") # ---------------------------------------------------------------- релиз / ссылки print("== release / fetch ==") from src.fetch import decode_bytes, detect_tracker # noqa: E402 from src.release import aggregate, title_segments # noqa: E402 check("tracker rutracker", detect_tracker("https://rutracker.org/forum/viewtopic.php?t=1") == "rutracker") check("tracker kinozal", detect_tracker("https://kinozal.tv/details.php?id=1") == "kinozal") check("tracker nnm", detect_tracker("https://nnmclub.to/forum/viewtopic.php?t=1") == "nnm") for bad in ("http://localhost:8000/", "http://127.0.0.1/", "http://169.254.169.254/", "https://rutracker.org.evil.com/", "file:///etc/passwd", "ftp://rutracker.org/"): try: detect_tracker(bad) check(f"SSRF отклонён: {bad}", False) except ValueError: check(f"SSRF отклонён: {bad}", True) check("decode cp1251", decode_bytes("Перевод".encode("cp1251")) == "Перевод") check("decode charset", decode_bytes("Перевод".encode("cp1251"), "text/html; charset=windows-1251") == "Перевод") check("title: ПМ (студия)", title_segments("Гангстерленд / 2025 / ПМ (LostFilm) / WEB-DLRip") == ["Профессиональный (многоголосый закадровый) | LostFilm"]) check("title: ДБ + список ПМ без студий", title_segments("Х / 2026 / ДБ (Red Head Sound), 5 x ПМ, ПД, ЛМ, СТ / 4K") == ["Профессиональный (дублированный) | Red Head Sound"]) check("title: пусто", title_segments("Просто фильм / 2020 / WEB-DL") == []) def _t(typ, st, age=False): return {"segment": f"{typ}|{st}", "translation_type": typ, "studio": st, "is_18plus": age} agg = aggregate([_t("проф. дубляж", "MovieDalen"), _t("проф. дубляж", "moviedalen", True), _t("проф. многоголосый", "LostFilm")]) check("aggregate: дедуп", len(agg["tracks"]) == 2, str(agg)) check("aggregate: 18+ = OR", agg["is_18plus"] is True) check("aggregate: студии", [x["studio"] for x in agg["studios"]] == ["MovieDalen", "LostFilm"]) check("aggregate: пусто", aggregate([])["is_18plus"] is False) from src.fetch import blocked_reason # noqa: E402 from src.release import merge_title_tracks, title_from_text # noqa: E402 check("заглушка Cloudflare", blocked_reason("Один момент…") is not None) check("нормальная страница", blocked_reason("XПеревод: A") is None) _page = [dict(_t("проф. двухголосый", None), studio_span=None, studio_confidence=None), dict(_t("проф. дубляж", "MovieDalen"), studio_span="MovieDalen", studio_confidence=1)] _title = [dict(_t("проф. двухголосый", "Red Head Sound"), studio_span="Red Head Sound", studio_confidence=0.9), dict(_t("проф. дубляж", "MovieDalen"), studio_span="x", studio_confidence=1), dict(_t("проф. многоголосый", "LostFilm"), studio_span="LostFilm", studio_confidence=1)] merge_title_tracks(_page, _title) check("title: студия вписана в дорожку без студии", _page[0]["studio"] == "Red Head Sound" and _page[0]["source"] == "page+title") check("title: дубль студии пропущен, новая добавлена", len(_page) == 3 and _page[2]["source"] == "title") from src.fetch import inline_tabs, kinozal_tabs # noqa: E402 from src.release import codec_segments, merge_extra_tracks, same_studio # noqa: E402 _kz = ('' '
Перевод: Профессиональный многоголосый
') check("kinozal: вкладки без «Техданных» и скриншотов", kinozal_tabs(_kz) == [("7", 0, "Релиз")]) _kz_text = html_to_text(inline_tabs(_kz, [("Релиз", "Аудио 1: AC3 - Русский (MVO, LostFilm)
")])) check("kinozal: вкладка «Релиз» попадает в текст", "Перевод: Профессиональный многоголосый" in _kz_text and "(MVO, LostFilm)" in _kz_text) check("codec_segments: русская дорожка с типом, без иноязычной", codec_segments("Аудио 3: AC3, 2.0 ch - Русский (MVO, HDrezka Studio)\n" "Аудио 6: E-AC3, 5.1 ch, 640 Kbps - English Dubbing\n" "Перевод: Профессиональный") == ["AC3, 2.0 ch - Русский (MVO, HDrezka Studio)"]) check("same_studio: варианты написания", same_studio("DVD", "DVD Магия") and same_studio("Михаил Яроцкий", "М.Яроцкий/Kyberpunk") and not same_studio("LostFilm", "HDrezka")) _page2 = [dict(_t("проф. многоголосый", None), studio_span=None, studio_confidence=None)] merge_extra_tracks(_page2, [ dict(_t("проф. многоголосый", "HDrezka"), studio_span="HDrezka Studio", studio_confidence=1), dict(_t("проф. многоголосый", "LostFilm"), studio_span="LostFilm", studio_confidence=1), dict(_t("одноголосый", None), studio_span=None, studio_confidence=None)], "codec", match=same_studio) check("codec: студия вписана, новая добавлена, без студии — нет", [(x["studio"], x["source"]) for x in _page2] == [("HDrezka", "page+codec"), ("LostFilm", "codec")]) _page3 = [dict(_t("проф. многоголосый", None), studio_span=None, studio_confidence=None)] merge_extra_tracks(_page3, [ dict(_t("неизвестно", "LostFilm"), studio_span="LostFilm", studio_confidence=1), dict(_t("неизвестно", "TVShows"), studio_span="TVShows", studio_confidence=1)], "codec", match=same_studio, adopt_type=True) check("codec: «Аудио Rus: … | LostFilm» получает тип дорожки страницы", [(x["translation_type"], x["studio"]) for x in _page3] == [("проф. многоголосый", "LostFilm"), ("проф. многоголосый", "TVShows")]) from src.release import codec_ok, model_segments # noqa: E402 from src.segmenter import kinozal_text, label_page # noqa: E402 class _FakeSeg: def select_lines(self, text): return text.splitlines() _main, _codec = model_segments(_FakeSeg(), "Перевод: Профессиональный многоголосый\n" "Аудио 3: AC3 - Русский (MVO, LostFilm)\n" "Аудио 6: AC3 - Казахский (Dub, TK Qazaqstan)") check("model_segments: основные и кодек-строки раздельно, иноязычная отброшена", _main == ["Профессиональный многоголосый"] and _codec == ["AC3 - Русский (MVO, LostFilm)"]) check("codec_ok: «English Dubbing» и «Original» — не русская дорожка", not codec_ok("Аудио 6: E-AC3 - English Dubbing") and not codec_ok("Аудио 5: E-AC3 - Original (французский)") and codec_ok("Аудио Rus: AC3, 2 ch | LostFilm")) check("label_page: raw — 1, неразмеченная кодек-строка — -1, сводка «Аудио:» — 0", label_page(["Перевод: Дублированный", "Аудио 2: AC3 - Русский (MVO, X)", "Аудио: Русский, английский (AC3)", "Размер: 6 ГБ"], ["Перевод: Дублированный"]) == [1, -1, 0, 0]) import json # noqa: E402 import tempfile as _tmp # noqa: E402 with _tmp.TemporaryDirectory() as _d: Path(_d, "123.json").write_text(json.dumps({"h1": "X / Y / 2026 / ПМ (LostFilm)", "tech_tab": "Перевод: Профессиональный многоголосый", "release_tab": "Аудио Rus: AC3 | LostFilm"}), encoding="utf-8") check("kinozal_text: заголовок + Техданные + Релиз из raw JSON", kinozal_text("kz-123", Path(_d)).splitlines() == ["X / Y / 2026 / ПМ (LostFilm)", "Перевод: Профессиональный многоголосый", "Аудио Rus: AC3 | LostFilm"] and kinozal_text("kz-404", Path(_d)) is None) _dump = ("Меню\nГангстерленд / MobLand / 2025 / ПМ (LostFilm) / WEB\n" "Гангстерленд / 2026 / ДБ (X) / WEB-DL") check("title_from_text: первая строка", "LostFilm" in title_from_text(_dump) and "ДБ" not in title_from_text(_dump)) from src.predict import refine_span # noqa: E402 check("«Авторский, одноголосый - X» не режется", _split_entries("Авторский, одноголосый - Ю. Сербин") == ["Авторский, одноголосый - Ю. Сербин"]) check("перечисление голосностей режется", len(_split_entries("Одноголосый, многоголосый закадровый")) == 2) _t4 = "Russian: 3/2 (L,C,R,l,r) + LFE ch |Дубляж" check("span: раскладка каналов -> None", refine_span(_t4, (14, 23)) is None) _t5 = "Русский MVO HDRezka Studio" check("span: расширение до границ слова", _t5[slice(*refine_span(_t5, (12, 14)))] == "HDRezka") check("span: один символ -> None", refine_span("М. Чадов", (0, 1)) is None) import json as _json # noqa: E402 import threading # noqa: E402 import urllib.error # noqa: E402 from http.server import BaseHTTPRequestHandler, HTTPServer # noqa: E402 import src.fetch as _fetch # noqa: E402 class _FakeFlare(BaseHTTPRequestHandler): def do_POST(self): req = _json.loads(self.rfile.read(int(self.headers["Content-Length"]))) ok = req.get("cmd") == "request.get" and "rutracker.org" in req.get("url", "") body = _json.dumps({"status": "ok", "solution": {"response": "Перевод: Дублированный | MovieDalen"}} if ok else {"status": "error", "message": "bad"}).encode() self.send_response(200) self.end_headers() self.wfile.write(body) def log_message(self, *a): pass _srv = HTTPServer(("127.0.0.1", 0), _FakeFlare) threading.Thread(target=_srv.serve_forever, daemon=True).start() _flare = f"http://127.0.0.1:{_srv.server_port}" def _deny(url, timeout, cookie): raise urllib.error.HTTPError(url, 403, "Forbidden", {}, None) _orig = _fetch._fetch_direct _fetch._fetch_direct = _deny try: h, tr = _fetch.fetch_html("https://rutracker.org/forum/viewtopic.php?t=1", flaresolverr=_flare) check("FlareSolverr: fallback при 403", "MovieDalen" in h and tr == "rutracker") try: _fetch.fetch_html("https://rutracker.org/forum/viewtopic.php?t=1") check("без FlareSolverr 403 пробрасывается", False) except urllib.error.HTTPError: check("без FlareSolverr 403 пробрасывается", True) finally: _fetch._fetch_direct = _orig _srv.shutdown() check("rutracker: форма логина в шапке не блокирует тему", blocked_reason('
x
') is None) check("rutracker: голая страница логина блокируется", blocked_reason('') is not None) _rt = ('
Перевод 1: Профессиональный ' '(дублированный) Продубляж
Перевод 2: Профессиональный ' '(многоголосый закадровый) AlexFilm
Субтитры: Русские
') check("html: студия-ссылка внутри строки не рвёт дорожку", extract_segments(html_to_text(_rt)) == ["Профессиональный (дублированный) Продубляж", "Профессиональный (многоголосый закадровый) AlexFilm"]) check("«Дубляж, Продубляж» не режется", len(_split_entries("Russian AC3 |Дубляж, Продубляж|")) == 1) _agg2 = aggregate([_t("проф. дубляж", "Продубляж"), _t("проф. дубляж", None), _t("проф. двухголосый", None)]) check("aggregate: дорожка без студии того же типа схлопывается", [x["translation_type"] for x in _agg2["tracks"]] == ["проф. дубляж", "проф. двухголосый"]) print("== строки «Аудио»: язык, студии, оригинал ==") from src.predict import guard_type from src.release import original_info from src.trackinfo import audio_lines_all, match_audio _au = audio_lines_all( "Аудио 1: Russian E-AC3 / 5.1 / 48 kHz / 768 kbps |Дубляж, Продубляж|\n" "Аудио 4: Ukrainian E-AC3 / 5.1 / 48 kHz / 256 kbps |Дубляж, Le Doyen|\n" "Аудио 5: English E-AC3+Atmos / 5.1 / 48 kHz / 768 kbps |Original|\n") check("аудио: «|Дубляж, Le Doyen|» -> студия Le Doyen без слова-типа", _au[1]["names"] == ["Le Doyen"], str(_au[1]["names"])) check("аудио: язык Ukrainian -> UKR", _au[1]["lang"] == "UKR") check("аудио: раскладка 5.1 -> 6 каналов", _au[1]["channels"] == "6") check("аудио: дорожка студии находит свою строку", match_audio("Профессиональный (дублированный) Le Doyen", _au) == [_au[1]]) _orig = original_info(_au) check("оригинал: ENG, 768 kbps, 6 каналов", _orig and _orig["lang"] == "ENG" and _orig["bitrate_kbps"] == 768 and _orig["channels"] == 6, str(_orig)) check("оригинал: нет строки Original -> None", original_info(_au[:2]) is None) _T = ["проф. дубляж", "проф. многоголосый", "люб. многоголосый", "многоголосый"] check("тип: «Любительский (многоголосый …) Дубляжная» не дубляж", guard_type("проф. дубляж", "Любительский (многоголосый закадровый) Дубляжная", "Дубляжная", _T) == "люб. многоголосый") check("тип: настоящий дубляж не трогается", guard_type("проф. дубляж", "Профессиональный (дублированный) Продубляж", "Продубляж", _T) == "проф. дубляж") print("== серии по студиям ==") from src.episodes import format_ranges as _fr, parse_episodes as _pe # noqa: E402 from src.release import series_info as _si # noqa: E402 check("episodes: диапазоны, перечисления и «из N»", [(_fr(_pe(s)["ranges"]), _pe(s)["total"]) for s in ("русский (MVO, Lostfilm) серии 1-2", "серии 1, 3-5 из 12", "с 1 по 4 серии", "1-3 серии", "S01E05-E08", "AC3, 5.1 ch, 640 Kbps", "Перевод: Профессиональный (многоголосый)")] == [("1-2", None), ("1,3-5", 12), ("1-4", None), ("1-3", None), ("5-8", None), ("", None), ("", None)]) check("episodes: соседние диапазоны склеиваются", _fr(_pe("серии 1-3, 4-6")["ranges"]) == "1-6") def _tk(typ, studio, eps=None): r = _pe(eps or "")["ranges"] return {**_t(typ, studio), "episode_ranges": r, "episodes": _fr(r) or None} _ag = aggregate([_tk("неизвестно", "LostFilm"), _tk("проф. многоголосый", "LostFilm", "серии 1-2"), _tk("проф. многоголосый", "LostFilm", "серии 3-4"), _tk("проф. дубляж", "Red Head Sound", "серия 1")]) check("aggregate: «неизвестно» сливается с известным типом, серии объединяются", [(x["translation_type"], x["studio"], x["episodes"]) for x in _ag["tracks"]] == [("проф. многоголосый", "LostFilm", "1-4"), ("проф. дубляж", "Red Head Sound", "1")], str(_ag["tracks"])) _s = _si("Тёмная материя / Dark Matter (2026) (сезон 2, серии 1-9 из 10)", _ag["tracks"]) check("series_info: сезон, серии, total и «студия -> серии»", (_s["season"], _s["episodes"], _s["total"]) == ("2", "1-9", 10) and _s["by_studio"] == {"LostFilm": "1-4", "Red Head Sound": "1"}, str(_s)) print("== сборщик датасета (tools/collect.py) ==") import csv as _csv # noqa: E402 import json as _json # noqa: E402 import json # noqa: E402 import tempfile as _tmp # noqa: E402 from pathlib import Path as _P # noqa: E402 sys.path.insert(0, str(_P(__file__).resolve().parents[1] / "tools")) import collect as _col # noqa: E402 _lab = {"rutracker": {"100", "200"}, "nnm": {"200", "300"}, "kinozal": set()} check("page_stem: размеченный id другого источника не используется bare", _col.page_stem("rutracker", "200", _lab) == "rt-200" and _col.page_stem("nnm", "200", _lab) == "nnm-200") check("page_stem: у размеченной раздачи прежний id, у новой — префикс", _col.page_stem("rutracker", "100", _lab) == "100" and _col.page_stem("kinozal", "5", _lab) == "kz-5") check("login_state: выход > гость > неизвестно", _col.login_state({"out": True, "guest": True}) == "in" and _col.login_state({"guest": True}) == "guest" and _col.login_state({}) == "unknown") _hv = _col.hv check("ошибка сервера: 503/502/Cloudflare определяются по заголовку и тексту, обычные страницы нет", all(_hv.RE_ERROR_TITLE.search(t) for t in ("503 Service Temporarily Unavailable", "Error 502", "Attention Required! | Cloudflare", "504 Gateway Time-out", "Ошибка 503")) and all(_hv.RE_ERROR_HEAD.search(t) for t in ("error code: 1015 you are being rate limited", "web server is returning an unknown error", "ошибка 503 сервис недоступен")) and not any(_hv.RE_ERROR_TITLE.search(t) for t in ("Зарубежное кино (HD Video) [стр. 2] :: RuTracker.org", "Фильм 5030 (2026)", "Раздачи :: Кинозал.МЕ"))) def _retry_run(results, retries=3): seq, hooks, slept = list(results), [], [] old = (_hv._goto_once, _hv.time.sleep, _hv.RETRY_HOOK) try: _hv._goto_once = lambda *a, **k: seq.pop(0) _hv.time.sleep = lambda s: slept.append(s) _hv.RETRY_HOOK = lambda url, a, n, why: hooks.append((a, n, why)) res = _hv.goto(None, "http://x", retries=retries) finally: _hv._goto_once, _hv.time.sleep, _hv.RETRY_HOOK = old return res, hooks, slept, seq _e = {"ok": False, "server_error": True, "reason": "503"} _r, _h, _s, _left = _retry_run([_e, _e, {"ok": True}]) check("retry: 503 дважды, потом успех — 2 повтора, паузы 3 и 8 с, хук получил попытки", _r["ok"] and _h == [(1, 3, "503"), (2, 3, "503")] and _s == [3.0, 8.0] and not _left) _r, _h, _s, _left = _retry_run([_e, _e, _e, _e]) check("retry: упорная ошибка — ровно 3 повтора и отказ", not _r["ok"] and len(_h) == 3 and not _left and _s == [3.0, 8.0, 20.0]) _r, _h, _s, _left = _retry_run([{"ok": False, "timeout": True, "reason": "таймаут"}, {"ok": False, "timeout": True, "reason": "таймаут"}]) check("retry: таймаут повторяется один раз", not _r["ok"] and len(_h) == 1 and not _left) _r, _h, _s, _left = _retry_run([{"ok": False, "dead": True}, {"ok": True}]) check("retry: удалённая раздача не повторяется", _r.get("dead") and not _h and len(_left) == 1) _sd = _P(_tmp.mkdtemp()) _old_raw = _col.RAW try: _col.RAW = _sd check("settings: по умолчанию 3 вкладки; значение сохраняется и ограничивается 1–4", _col.settings()["tabs"] == 3) _col.save_settings({**_col.settings(), "tabs": 9, "limit": 50}) _s = _col.settings() check("settings: сохранённые вкладки (9 -> 4), лимит и переменная DUB_TABS", _s["tabs"] == 4 and _s["limit"] == 50 and _col.os.environ["DUB_TABS"] == "4" and _col.clamp_tabs("x") == 3 and _col.clamp_tabs(0) == 1) finally: _col.RAW = _old_raw _col.os.environ["DUB_TABS"] = "3" check("is_dead: «нет раздачи» — удалена, обычная ошибка загрузки — нет", _col.is_dead({"ok": False, "error": "не открылась: нет раздачи: Торрент трекер"}) and not _col.is_dead({"ok": False, "error": "не открылась: Just a moment"}) and not _col.is_dead({"ok": True})) _dd = _P(_tmp.mkdtemp()) _old_dead = _col.DEAD try: _col.DEAD = _dd / "dead.json" _col.mark_dead("kinozal", "18364907") _col.mark_dead("kinozal", "5") _col.mark_dead("nnm", "7") check("mark_dead: id запоминаются по трекерам и без дублей", _col.load_dead("kinozal") == {"18364907", "5"} and _col.load_dead("nnm") == {"7"} and _col.load_dead("rutracker") == set()) finally: _col.DEAD = _old_dead check("has_body: пустой пост — не раздача", not _col.has_body("nnm", {"post": " "}) and _col.has_body("kinozal", {"release_tab": "Аудио"})) import discover as _dc # noqa: E402 def _picked(site, name): return bool(_dc.INCLUDE[site].search(name)) and not _dc.EXCLUDE.search(name) check("discover: rutracker берёт зарубежное кино, сериалы и годовые разделы", all(_picked("rutracker", n) for n in ["Зарубежное кино (UHD Video)", "Зарубежные сериалы (HD Video)", "Фильмы 2021-2025", "Новинки и сериалы в стадии показа (HD Video)", "Классика мирового кинематографа"])) check("discover: rutracker не берёт наше кино, русские сериалы, музыку и Apple", not any(_picked("rutracker", n) for n in ["Наше кино (HD Video)", "Русские сериалы (HD Video)", "Зарубежный Rock", "Фильмы HD для Apple TV", "Азиатские сериалы"])) check("discover: аниме-музыка (OST, Lossless, AMV, саундтреки) исключена, само аниме берётся", not any(_picked(site, n) for site in ("nnm", "rutracker") for n in ["Аниме OST (Lossless)", "Аниме OST", "Саундтреки к аниме", "Аниме AMV и клипы", "Anime Music (MP3)"]) and all(_picked("nnm", n) for n in ["Аниме (HD, FHD)", "Аниме (SD)"])) check("discover: nnm берёт форматы и годы, но не коллекции и оригиналы", all(_picked("nnm", n) for n in ["Зарубежные Фильмы (UHD)", "Зарубежные Новинки (SD, DVD)", "Зарубежная Классика (DVD)", "Старые Зарубежные Фильмы (SD)", "Зарубежные Мультсериалы"]) and not any(_picked("nnm", n) for n in ["Зарубежное кино (коллекции / *логии)", "Фильмы в оригинале (SD, DVD)", "Отечественные Фильмы (HD, FHD, UHD)", "Зарубежное видео для устройств Apple"])) check("discover: мультфильмы, мультсериалы и аниме берутся, русские мультфильмы нет", all(_picked("rutracker", n) for n in ["Мультфильмы (HD Video)", "Мультсериалы (SD Video)", "Аниме (HD Video)", "Онгоинги (HD Video)"]) and all(_picked("nnm", n) for n in ["Зарубежные Мультфильмы 21-го века (SD)", "Аниме с озвучкой (HD)", "Мультфильмы (3D)"]) and not any(_picked(s, n) for s, n in [("nnm", "Отечественные Мультфильмы 20-го века (SD)"), ("rutracker", "Наши мультфильмы"), ("nnm", "Мультимедиа и Графика")])) _kz = [x["ref"] for x in _dc.kinozal_grid()] check("discover: kinozal — общие списки фильмы/сериалы/мульты/аниме и сетка без 4K до 2012", all(any(f"c={c}&v=0&d=0" in u for u in _kz) for c in ("1002", "46", "21", "20")) and not any("v=7" in u and int(u.split("d=")[1].split("&")[0]) < 2012 for u in _kz) and len(_kz) == len(set(_kz))) _calls = [] _orig = (_dc.candidate_forums, _dc.probe, _dc.time.sleep) try: _dc.candidate_forums = lambda cdp, site: [{"ref": "1", "name": "Зарубежное кино (HD Video)"}, {"ref": "2", "name": "Зарубежное кино (DVD Video)"}, {"ref": "3", "name": "Зарубежные сериалы (UHD Video)"}, {"ref": "4", "name": "Фильмы 2026"}] _dc.probe = lambda cdp, site, ref: (_calls.append(ref), {"2": 50, "3": 2, "4": 0}.get(ref, 0))[1] _dc.time.sleep = lambda s: None _saved = {} _res = _dc.discover_forums(None, "rutracker", probes={"1": 47}, on_probe=lambda r, n: _saved.update({r: n})) finally: _dc.candidate_forums, _dc.probe, _dc.time.sleep = _orig _st = {e["ref"]: (e["status"], e["keep"]) for e in _res} check("discover: раздел из кэша проверок повторно не открывается", "1" not in _calls and _saved.get("1") is None and _st["1"] == ("ok", True), str(_calls)) check("discover: статусы ok / мало / не проверен (оставляется), прогресс-колбэки", _st["2"] == ("ok", True) and _st["3"] == ("мало", False) and _st["4"] == ("?", True) and _saved == {"2": 50, "3": 2, "4": 0}, str((_st, _saved))) import ui as _ui # noqa: E402 check("ui: таблица найденных разделов строится", _ui.sections_table("rutracker", _res) is not None and _ui.sections_table("kinozal", [{"ref": "https://kinozal.me/browse.php?s=&g=0&c=46&v=3&d=2020&w=0&t=0&f=0", "name": "2020, HD", "topics": None, "keep": True, "status": "ok"}]) is not None) import retrain as _rt # noqa: E402 _cmp = {"rules": {"nnm": (300, .84, .93), "rutracker": (250, .86, .94), "kinozal": (12, .5, .6)}, "old": {"nnm": (300, .64, .96), "rutracker": (250, .90, .95)}, "good": {"nnm": (300, .85, .95), "rutracker": (250, .91, .95)}, "badrec": {"nnm": (300, .90, .80), "rutracker": (250, .91, .95)}} check("verdict: precision выше и recall держится -> лучше", _rt.verdict(_cmp, "old", "good")[0]) check("verdict: провал recall на трекере -> не лучше", not _rt.verdict(_cmp, "old", "badrec")[0]) check("compare: разбор строки e2e", _rt.RE_CELL.findall("model: nnm n=301 P=0.637 R=0.961; kinozal n=12 P=0.5 R=0.75") == [("nnm", "301", "0.637", "0.961"), ("kinozal", "12", "0.5", "0.75")]) import meta as _mt # noqa: E402 _nnm = _mt.extract("nnm", {"id": "1", "title": ( "Хрустальное озеро / Crystal Lake (2026) WEB-DL [H.265/2160p] [4K, HDR10, DV 8.1, 10-bit] " "(сезон 1, серии 1-8 из 8) Novamedia, HDRezka"), "post": ( "Производство: США, A24, Peacock\nЖанр: ужасы, триллер, драма\n" "Режиссер: Логан Джордж, Селин Хелд\nАктеры: Линда Карделлини, Уильям Кэтлетт\n" "Возраст: 18+ (зрителям, достигшим 18 лет)\nРелиз от: kupec\n" "Продолжительность: ~00:56:00 серия\nКачество видео: WEB-DL\n" "Язык озвучки: Русский, Английский\nСубтитры: Русские (Forced), Английские\n" "Видео: HEVC/H.265, 3840x2160, ~15.0 Mbps\n" "Аудио 1: AC3, 6 ch, 384 Kbps - Русский (DUB, Novamedia) серии 1-8\n" "Аудио 2: AC3, 2 ch, 384 Kbps - Русский (MVO, HDRezka Studio 18+) серии 1\n" "Аудио 3: E-AC3 JOC, 6 ch, 640 Kbps - Английский\n")}) _t = _json.loads(_nnm["audio_tracks"]) check("meta nnm: сезон, серии, разрешение, кодек, HDR, возраст", (_nnm["season"], _nnm["episodes"], _nnm["episodes_total"], _nnm["resolution"], _nnm["video_codec"], _nnm["age_limit"], _nnm["is_series"], _nnm["kind"]) == ("1", "1-8", "8", "2160p", "HEVC", "18", True, "series") and "HDR10" in _nnm["hdr"] and "DV" in _nnm["hdr"], str({k: _nnm[k] for k in ("season", "episodes", "hdr", "kind")})) check("meta nnm: дорожки — язык, кодек, каналы, битрейт, тип, серии, 18+", len(_t) == 3 and (_t[0]["lang"], _t[0]["codec"], _t[0]["channels"], _t[0]["bitrate_kbps"], _t[0]["tag"], _t[0]["episodes"]) == ("RUS", "AC3", "6", 384, "DUB", "1-8") and _t[1]["is_18plus"] and _t[2]["lang"] == "ENG" and _t[2]["codec"].startswith("E-AC3"), str(_t)) check("meta nnm: субтитры, forced, языки озвучки", _nnm["sub_langs"] == "RUS+ENG" and _nnm["sub_forced"] and _nnm["audio_langs"] == "RUS+ENG") _rt = _mt.extract("rutracker", {"id": "2", "title": ( "Отрочество / Boyhood (Ричард Линклейтер) [2014, США, драма, BDRip 720p] Dub (НТВ+) + Sub Rus"), "post": ( "Страна: США\nЖанр: Драма\nГод выпуска: 2014\nПродолжительность: 02:45:16\n" "Оригинальная аудиодорожка: английский\nКонтейнер: MKV\n" "Видео: x264@L4.1, 1280x692, 10.7 Mbps, 23.976 fps.\n" "Аудио 1: 48 kHz, AC3, 3/2 (L,C,R,l,r) + LFE ch, 384.00 kbps\n" "Аудио 2: 48 kHz, DTS, 3/2 (L,C,R,l,r) + LFE ch, 1510 kbps avg | Одноголосый закадровый, А. Дасевич\n" "Аудио 3: 48 kHz, DTS, 2/0 ch, 1510 kbps | Original\n")}) _tr = _json.loads(_rt["audio_tracks"]) check("meta rutracker: год, контейнер, 720p/AVC, длительность, оригинал", (_rt["year"], _rt["container"], _rt["resolution"], _rt["video_codec"], _rt["duration_min"], _rt["orig_audio"]) == ("2014", "MKV", "720p", "AVC", 165, "ENG"), str(_rt)) check("meta rutracker: схема каналов 3/2+LFE = 6, тип VO, Original", _tr[0]["channels"] == "6" and _tr[1]["tag"] == "AVO" and _tr[2]["tag"] == "ORIG", str(_tr)) _kz = _mt.extract("kinozal", {"id": "3", "h1": "Гран-при / Grand Prix / 1966 / ДБ, ПМ / BDRip (1080p)", "release_tab": ("Жанр: Драма\nКачество: BDRip (720p)\n" "Видео: MPEG-4 AVC, 5200 Кбит/с, 1280x580, 23.976 кадр/с\n" "Аудио: Русский (AC3, 2 ch, 256 Кбит/с), (DTS, 6 ch, 1509 Кбит/с), " "английский (DTS, 6 ch, 1509 Кбит/с)\nРазмер: 10.66 ГБ\n" "Субтитры: Русские, английские")}) _tk = _json.loads(_kz["audio_tracks"]) check("meta kinozal: однострочное «Аудио» разбирается на дорожки с языками", [(x["lang"], x["codec"], x["channels"], x["bitrate_kbps"]) for x in _tk] == [("RUS", "AC3", "2", 256), ("RUS", "DTS", "6", 1509), ("ENG", "DTS", "6", 1509)], str(_tk)) check("meta kinozal: размер, качество, год, тип по разделу", _kz["size_gb"] == 10.66 and _kz["year"] == "1966" and _kz["resolution"] == "720p" and _mt.kind_of("Аниме (HD Video)", "", "", False) == "anime" and _mt.kind_of("Зарубежные Мультфильмы", "", "", False) == "cartoon") _d = _P(_tmp.mkdtemp()) _old = (_col.RAW, _col.PAGES, _col.LABELS, _col.POSTONLY, _col.META) try: _col.RAW, _col.PAGES, _col.LABELS, _col.POSTONLY, _col.META = ( _d / "raw", _d / "pages", _d / "labels.csv", _d / "po.txt", _d / "meta.csv") _col.LABELS.write_text("", encoding="utf-8") (_col.RAW / "nnm").mkdir(parents=True) _post = ("Год выпуска: 2020\nАудио 1: Opus, 6 ch, 221 Kbps | Русский AVO (Дмитрий Пучков)\n" "Аудио 2: AC3, 6 ch, 448 Kbps | Русский MVO (LostFilm)\n") (_col.RAW / "nnm" / "777.json").write_text(_json.dumps( {"id": "777", "source": "nnm", "ok": True, "title": "Фильм (2020) BDRip", "post": _post}, ensure_ascii=False), encoding="utf-8") (_col.RAW / "nnm" / "777.html").write_text( f"

Фильм (2020) BDRip

" f"{_post.replace(chr(10), '
')}
", encoding="utf-8") _st = _col.cmd_build() check("build: страница раздачи nnm записана с префиксом", (_col.PAGES / "nnm-777.txt").exists(), str(_st)) with _col.LABELS.open(encoding="utf-8", newline="") as _f: _rows = list(_csv.reader(_f)) check("build: метки добавлены с release_id=nnm-777", len(_rows) >= 3 and _rows[0][1] == "release_id" and all(r[1] == "nnm-777" for r in _rows[1:]), str(_rows[:2])) _before = _col.LABELS.read_text(encoding="utf-8") _calls_x = [] _orig_x = _col.metamod.extract _col.metamod.extract = lambda *a, **k: (_calls_x.append(1), _orig_x(*a, **k))[1] try: _st2 = _col.cmd_build() check("build идемпотентен: повторный запуск не дублирует метки", _col.LABELS.read_text(encoding="utf-8") == _before) check("build инкрементален: неизменённые страницы и meta не разбираются повторно", _st2["unchanged"] == 1 and _st2["meta_reused"] == 1 and not _calls_x, str(_st2)) (_col.RAW / "nnm" / "777.json").touch() # файл изменился -> разбирается заново _st3 = _col.cmd_build() check("build: изменённый файл разбирается заново, остальное из кэша", _st3["unchanged"] == 0 and len(_calls_x) == 1, str(_st3)) _col.cmd_build(rewrite=True) check("build: rewrite=True игнорирует кэши", len(_calls_x) == 2) finally: _col.metamod.extract = _orig_x check("build: meta.csv содержит релиз с префиксным id", "nnm-777" in _col.META.read_text(encoding="utf-8-sig")) _good = _col.LABELS.read_text(encoding="utf-8") _col.LABELS.write_text(_good.replace("LostFilm", "СТАРАЯ_РАЗМЕТКА"), encoding="utf-8") _st4 = _col.cmd_build(rewrite=True) check("build без relabel не трогает размеченные релизы", "СТАРАЯ_РАЗМЕТКА" in _col.LABELS.read_text(encoding="utf-8")) _st5 = _col.cmd_build(rewrite=True, relabel=True) _now = _col.LABELS.read_text(encoding="utf-8-sig") check("relabel: метки пересчитаны из сохранённого HTML, дублей нет", "СТАРАЯ_РАЗМЕТКА" not in _now and _now.count("nnm-777") == _good.count("nnm-777"), str(_st5)) check("relabel: копия прежних меток сохранена", bool(list(_d.glob("labels.bak_*.csv")))) finally: _col.RAW, _col.PAGES, _col.LABELS, _col.POSTONLY, _col.META = _old print("== 18+ из строк «Аудио» ==") _h = open("data/samples/rt-topic-6894333.html", encoding="utf-8").read() _pt = html_to_text(_h) _res = {"tracks": [{"segment": "Профессиональный (многоголосый закадровый) HDRezka", "studio": "HDrezka", "translation_type": "проф. многоголосый", "is_18plus": False, "source": "page"}, {"segment": "Профессиональный (многоголосый закадровый) LostFilm", "studio": "LostFilm", "translation_type": "проф. многоголосый", "is_18plus": False, "source": "page"}]} from src.audio18 import apply_audio_18 # noqa: E402 _res = apply_audio_18(_res, _pt) _hd = [t for t in _res["tracks"] if t["studio"] == "HDrezka"] check("HDrezka: две дорожки, обычная и 18+", sorted(t["is_18plus"] for t in _hd) == [False, True], str(_hd)) check("LostFilm без 18+", not any(t["is_18plus"] for t in _res["tracks"] if t["studio"] == "LostFilm")) check("общий флаг 18+ у релиза", _res["is_18plus"] is True) import sys as _s, os as _o # noqa: E402 _s.path.insert(0, _o.path.join("tools", "harvest")) import harvest_core as _hc # noqa: E402 _rows, _ = _hc.rt_parse({"id": "6894333", "title": "Джек Ричер / Reacher / Сезон 4 [2026, WEB-DL 1080p]", "post": "\n".join(l for l in _pt.splitlines() if l.startswith(("Перевод", "Аудио")))}) _hr = [r for r in _rows if r["studio"] == "HDrezka"] check("метки rt_parse: HDrezka обычная + 18+", sorted(int(r["age_18"]) for r in _hr) == [0, 1], str(_hr)) check("метки rt_parse: кодек и битрейт из строки «Аудио»", all(r["codec"] == "AC3" and r["kbps"] == "192" for r in _hr), str(_hr)) check("метки rt_parse: 5 студий + 1 доп. 18+", len(_rows) == 6, str(len(_rows))) _sf = __import__("src.release", fromlist=["x"]).series_info("Сериал / Сезон 1 / Серии 1-8 из 8 [2026]", [ {"studio": "LostFilm", "episodes": "1-4", "episode_ranges": [[1, 4]]}, {"studio": "HDrezka", "episodes": None, "episode_ranges": []}]) check("серии не указаны -> full_release", _sf["full_release"] == ["HDrezka"], str(_sf)) import json as _js2 # noqa: E402 sys.path.insert(0, "tools") import meta as _meta # noqa: E402 _mm = _meta.extract("rutracker", {"id": "6894333", "post": _pt[:6000], "title": "Джек Ричер / Reacher / Сезон 4 / Серии 1-8 из 8 [2026, WEB-DL 1080p]"}, page_text=_pt) _at = _js2.loads(_mm["audio_tracks"]) check("meta: 7 аудиодорожек rutracker-вида «Аудио RUS: …»", _mm["n_audio"] == 7, str(_mm["n_audio"])) check("meta: HDrezka обычная и 18+, кодек AC3, тип MVO из «Перевод»", sorted(a["is_18plus"] for a in _at if a["studio"] == "HDrezka") == [False, True] and all(a["codec"] == "AC3" and a["tag"] == "MVO" for a in _at if a["studio"])) print("== разнообразная выборка ==") import random as _rnd # noqa: E402 import pathlib as _pl, tempfile as _tf # noqa: E402 import diversity as _dv # noqa: E402 check("год из скобок заголовка", _dv.year_of("Матрица / The Matrix (1999) BDRip 1080p") == 1999 and _dv.year_of("Сериал / Season 2 [2021, WEB-DL 2160p]") == 2021 and _dv.year_of("Без года") is None) check("корзина: десятилетие до 2010, дальше по годам", _dv.year_band(1994) == "1990-е" and _dv.year_band(2023) == "2023" and _dv.year_band(None) == "?") check("качество и сериал в заголовке", _dv.bucket("Сериал (2021) Сезон 1 WEB-DL 2160p") == ("2021", "4K", "сериал") and _dv.bucket("Фильм (2005) DVDRip")[1] == "SD") _items = [{"id": str(i), "text": f"Фильм {i} / Film ({2024}) WEB-DL 1080p"} for i in range(8)] + [{"id": "old", "text": "Старое кино / Old (1975) DVDRip"}] _cnt = _dv.counts_from_meta(_pl.Path("нет_такого_файла.csv")) _cnt[_dv.bucket(_items[0]["text"])] = 500 # свежие 1080p уже в избытке _got = _dv.pick(_items, 1, _cnt, _rnd.Random(1)) check("pick: берётся тема из недопредставленной корзины (старое кино)", _got[0]["id"] == "old", str(_got)) _vis: set[int] = set() _rng = _rnd.Random(3) _order = [] for _ in range(40): _nx = _dv.next_pages(40, _vis, 1, _rng) _order += _nx _vis |= set(_nx) check("next_pages: все 40 страниц ровно по разу, не подряд", sorted(_order) == list(range(40)) and _order[:5] != [0, 1, 2, 3, 4], str(_order[:8])) check("next_pages: когда всё посещено — пусто", _dv.next_pages(5, set(range(5)), 1, _rng) == []) _tmp_meta = _pl.Path(_tf.mkdtemp()) / "meta.csv" _tmp_meta.write_text(chr(10).join(['source,title,year,quality,is_series', 'nnm,A,2020,WEB-DL (2160p),False', 'nnm,B,2020,WEB-DL (2160p),False', 'rutracker,C,1999,DVD,True']) + chr(10), encoding="utf-8") _c2 = _dv.counts_from_meta(_tmp_meta, "nnm") check("counts_from_meta: считает корзины нужного трекера", _c2[("2020", "4K", "фильм")] == 2 and sum(_c2.values()) == 2, str(_c2)) # раздел вразброс: страницы с годами, глубина 20, берутся разные страницы и недопредставленные годы _calls = [] _pages = {p: [{"id": f"{p}{k}", "text": f"Фильм {p}{k} / F ({2025 - p}) WEB-DL 1080p"} for k in range(5)] for p in range(20)} _o_li, _o_hi, _o_depth, _o_un = _col.list_items, _col.harvest_ids, _col.hv.list_depth, _col.update_state _col.list_items = lambda cdp, site, ref, page: (True, _pages.get(page, [])) _col.hv.list_depth = lambda cdp: 20 _col.update_state = lambda *a, **k: None _o_ls = _col.load_state _col.load_state = lambda: {} # состояние реального сбора не должно влиять на тест _col.harvest_ids = lambda cdps, site, ids, limit, delay, stop, section="": (_calls.append(list(ids)), len(ids))[1] try: _col.ui.get_stats("rutracker").begin(30) _n = _col.collect_ref_balanced("rutracker", "313", None, type("T", (), {"cdps": [None]})(), set(), _dv.Counter(), 12, 100, 0.0, __import__("threading").Event(), 3, _rnd.Random(5)) _years = {int(i[0] if len(i) == 2 else i[:2]) for c in _calls for i in c} check("balanced: набран лимит раздела по 3 раздачи со страницы", _n == 12 and all(len(c) <= 3 for c in _calls), str(_calls)) check("balanced: взяты разные страницы, а не первые подряд", len({c[0][:-1] for c in _calls}) == len(_calls) and {c[0][:-1] for c in _calls} != {"0", "1", "2", "3"}, str(_calls)) finally: _col.list_items, _col.harvest_ids, _col.hv.list_depth, _col.update_state = _o_li, _o_hi, _o_depth, _o_un _col.load_state = _o_ls print("== читаемый HTML ==") _old3 = _col.RAW _rt3 = _pl.Path(_tf.mkdtemp()) try: _col.RAW = _rt3 (_rt3 / "nnm").mkdir() _hp = _rt3 / "nnm" / "1.html" _hp.write_text("ТПривет", encoding="utf-8") _mt = _hp.stat().st_mtime_ns _r1 = _col.cmd_repair_html(with_css=False) _txt = _hp.read_text(encoding="utf-8") check("repair-html: добавлены UTF-8 и base, содержимое цело", '' in _txt and " люб. дубляж", _tt("Любительский (псевдодублированный)")[0] == "люб. дубляж") check("тип: обычный дубляж остаётся проф.", _tt("Профессиональный (дублированный) Продубляж")[0] == "проф. дубляж") check("тип: «Дубляжная» — имя студии, а не дубляж", _tt("Любительский (многоголосый закадровый) Дубляжная")[0] == "люб. многоголосый") check("тип: слово «дубляж» отдельно — дубляж", _tt("Дубляж, HDRezka")[0] == "проф. дубляж") _base = {"source": "kinozal", "release_id": "1", "ru_title": "x", "year": "", "is_series": False, "studio": "(не указана)", "translation_type": "проф. многоголосый", "professional": "проф.", "role_raw": "", "age_18": 0, "studio_source": "release", "note": ""} _sp = _hc2.split_multi_rows([{**_base, "raw": "Перевод: Профессиональный (двухголосый закадровый), авторский (одноголосый закадровый)"}]) check("раскладка: две дорожки из одной строки", [r["translation_type"] for r in _sp] == ["проф. двухголосый", "авт. одноголосый"], str([(r["translation_type"], r["raw"]) for r in _sp])) _one = _hc2.split_multi_rows([{**_base, "raw": "Перевод: Многоголосый, профессиональный - LostFilm"}]) check("раскладка: одна дорожка не режется", len(_one) == 1, str(len(_one))) _tech = _hc2.split_multi_rows([{**_base, "studio_source": "tech", "role_raw": "Многоголосый", "raw": "Перевод: A, B"}]) check("раскладка: nnm tech-строка не режется", len(_tech) == 1) _ao = _hc2.audio_only_rows({"id": "2"}, ( "Аудио 1: Russian E-AC3 / 5.1 / 48 kHz / 768 kbps |MVO, AlexFilm|\n" "Аудио 2: Ukrainian E-AC3 / 5.1 / 48 kHz / 256 kbps |Дубляж, Le Doyen|\n" "Аудио 3: Russian DTS, 3/2 (L,C,R,l,r) + LFE ch, 640 kbps |Дубляж, Blu-ray CEE|\n" "Аудио 4: English E-AC3 / 5.1 / 48 kHz / 768 kbps |Original|\n"), [{**_base, "studio": "AlexFilm", "raw": "Профессиональный (многоголосый закадровый) AlexFilm"}]) check("строки из «Аудио»: украинский дубляж добавлен, оригинал/источник/уже известные — нет", [(r["studio"], r["lang"], r["translation_type"]) for r in _ao] == [("Le Doyen", "UKR", "проф. дубляж")], str([(r["studio"], r["lang"], r["translation_type"]) for r in _ao])) from src.train import class_weights # noqa: E402 _cw = class_weights([{"translation_type": t} for t in ["a"] * 1000 + ["b"] * 100 + ["c"] * 4], {"a": 0, "b": 1, "c": 2}) check("веса классов: редкие тяжелее, потолок 4", _cw[0] < _cw[1] < _cw[2] <= 4.0, str(_cw)) print("== зависания сбора ==") import json as _js5, threading as _th5, time as _tm5 # noqa: E402 class _FakeCDP: def __init__(self, state): self.state, self.calls, self.aborted, self.reconnected = state, 0, 0, 0 def send(self, *a, **k): return {} def eval(self, expr, await_promise=True): self.calls += 1 return _js5.dumps(self.state) if "JSON.stringify" in expr else "короткая страница" def abort(self): self.aborted += 1 def reconnect(self): self.reconnected += 1 _t0 = _tm5.time() _res = _hc._goto_once(_FakeCDP({"t": "Тема перенесена", "r": "complete", "len": 300}), "http://x", min_len=900, timeout=45) check("короткая загруженная страница принимается за ~5 с, а не ждёт таймаут 45 с", _res.get("ok") and _res.get("short") and _tm5.time() - _t0 < 15, f"{_res} {_tm5.time() - _t0:.1f}") _res2 = _hc._goto_once(_FakeCDP({"t": "Just a moment...", "r": "complete", "len": 300}), "http://x", min_len=900, timeout=7) check("короткая страница проверки Cloudflare за готовую не считается", _res2.get("timeout"), str(_res2)) _old5 = (_col.fetch_release, _col.HARD_LIMIT_S, _col.SLOW_S, _col.mark_dead) try: _col.HARD_LIMIT_S, _col.SLOW_S = 0.6, 0.3 _col.fetch_release = lambda cdp, site, rid: (_tm5.sleep(1.2), {"ok": False, "error": "x"})[1] _fc = _FakeCDP({}) _col.ui.get_stats("rutracker").begin(5) _col.harvest_ids([_fc], "rutracker", ["111"], 5, 0.0, _th5.Event()) check("сторож: зависшая раздача — вызов оборван, вкладка переподключена", _fc.aborted == 1 and _fc.reconnected == 1, f"abort={_fc.aborted} reconnect={_fc.reconnected}") _col.HARD_LIMIT_S = 30 _fc2 = _FakeCDP({}) _col.harvest_ids([_fc2], "rutracker", ["112"], 5, 0.0, _th5.Event()) check("сторож не трогает нормальную (медленную, но завершившуюся) раздачу", _fc2.reconnected == 0) finally: _col.fetch_release, _col.HARD_LIMIT_S, _col.SLOW_S, _col.mark_dead = _old5 print("== параллельная переиндексация ==") import subprocess as _sp6 # noqa: E402 _pc = _sp6.run([sys.executable, "tests/par_check.py"], capture_output=True, text=True, encoding="utf-8", env={**__import__("os").environ, "PYTHONUTF8": "1"}) _line = next((l for l in _pc.stdout.splitlines() if l.startswith("RESULT ")), "") _pr6 = _json.loads(_line[7:]) if _line else {} check("параллельная сборка (процессы) даёт те же метки, страницы и meta, что и последовательная", _pr6.get("same") is True and _pr6.get("pages") == 50 and _pr6.get("labeled") == [50, 50], (_pc.stderr or _pc.stdout)[-300:]) check("worker_count: значение по умолчанию в пределах 1–12", 1 <= _col.worker_count() <= 12) print("== обучение по ссылкам ==") check("ссылка rutracker", _col.parse_link("https://rutracker.org/forum/viewtopic.php?t=6894333") == ("rutracker", "6894333")) check("ссылка nnm", _col.parse_link("https://nnmclub.to/forum/viewtopic.php?p=1&t=77") == ("nnm", "77")) check("ссылка kinozal", _col.parse_link("https://kinozal.me/details.php?id=2155882") == ("kinozal", "2155882")) check("не ссылка на раздачу", _col.parse_link("https://rutracker.org/forum/viewforum.php?f=313") is None) import tempfile as _tf, pathlib as _pl # noqa: E402 _old2 = (_col.PAGES, _col.LABELS, _col.MANUAL, _col.RAW) _tmp = _pl.Path(_tf.mkdtemp()) try: _col.PAGES, _col.LABELS, _col.MANUAL, _col.RAW = _tmp / "pages", _tmp / "labels.csv", _tmp / "manual_ids.txt", _tmp / "raw" _rows2 = [dict(_r, age_18=0) for _r in _rows[:2]] _stem = _col.store_manual("rutracker", {"id": "6894333"}, _rows2, "текст страницы") check("ручные метки: страница и id записаны", (_col.PAGES / f"{_stem}.txt").exists() and _stem in _col.MANUAL.read_text(encoding="utf-8").split()) _lab = _col.read_labels() check("ручные метки: строки помечены manual", len(_lab) == 2 and all(r["studio_source"] == "manual" for r in _lab)) _col.store_manual("rutracker", {"id": "6894333"}, _rows2[:1], "текст") check("повторная запись заменяет строки релиза", len(_col.read_labels()) == 1) from src.manual import manual_ids # noqa: E402 check("manual_ids читает файл", manual_ids(_col.MANUAL) == {_stem}) finally: _col.PAGES, _col.LABELS, _col.MANUAL, _col.RAW = _old2 print("== реестр моделей и сравнение (src/runs.py) ==") import json as _json # noqa: E402 import subprocess as _sp7 # noqa: E402 import tempfile as _tf7 # noqa: E402 from src import runs as _runs # noqa: E402 from src.train import DEFAULT_MODELS, MODEL_CONFIGS # noqa: E402 check("plain_text: снимает хвост из поля tail", _runs.plain_text({"text": "Дубляж [RUS AC3 6ch]", "tail": "[RUS AC3 6ch]"}) == "Дубляж") check("plain_text: пустой tail — текст как есть", _runs.plain_text({"text": "Дубляж [X]", "tail": ""}) == "Дубляж [X]") check("plain_text: старый датасет без tail — по шаблону хвоста", _runs.plain_text({"text": "Дубляж [RUS AC3 6ch 448kbps 18+]"}) == "Дубляж" and _runs.plain_text({"text": "Дубляж [? AC3]"}) == "Дубляж" and _runs.plain_text({"text": "Озвучка [Лостфильм]"}) == "Озвучка [Лостфильм]") _ra = [{"text": "a", "translation_type": "x", "span": None, "is_18plus": False}, {"text": "b", "translation_type": "y", "span": [0, 1], "is_18plus": True, "lang": "RUS"}] check("отпечаток выборки не зависит от порядка и меняется с метками", _runs.fingerprint(_ra) == _runs.fingerprint(_ra[::-1]) and _runs.fingerprint(_ra) != _runs.fingerprint([_ra[0], {**_ra[1], "translation_type": "z"}])) _lo, _hi = _runs.wilson(94, 100) check("Уилсон: интервал вокруг доли", _lo < 0.94 < _hi and 0.85 < _lo and _hi < 0.98 and _runs.wilson(0, 0) == (0.0, 0.0)) check("знаковый критерий: 30/5 значимо, 12/10 — шум, 0/0 — 1", _runs.sign_test(30, 5) < 0.001 and _runs.sign_test(12, 10) > 0.5 and _runs.sign_test(0, 0) == 1.0) check("парное сравнение считает починенные и сломанные", _runs.paired([1, 0, 1, 0, 1], [1, 1, 0, 0, 1])["fixed"] == 1 and _runs.paired([1, 0, 1, 0, 1], [1, 1, 0, 0, 1])["broken"] == 1) import retrain as _rt7 # noqa: E402 check("профили меню совпадают с профилями обучения", set(MODEL_CONFIGS) == set(_rt7.PROFILES)) check("python -m src.train по умолчанию — tiny2 и base, крупные профили только по запросу", DEFAULT_MODELS == ["tiny2", "base"] and {"large", "modern"} <= set(MODEL_CONFIGS)) with _tf7.TemporaryDirectory() as _td7: _m = Path(_td7) _old7 = (_runs.MODELS, _runs.RUNS) _runs.MODELS, _runs.RUNS = _m, _m / "runs" try: def _mk(d: Path, backbone="x/bert", cfg=None, input_names=None): d.mkdir(parents=True) (d / "model.pt").write_bytes(b"w" * 10) (d / "dub_config.json").write_text(_json.dumps({"backbone": backbone}), encoding="utf-8") (d / "train_config.json").write_text(_json.dumps(cfg or {"name": d.name, "uid": d.name}), encoding="utf-8") (d / "tokenizer_config.json").write_text(_json.dumps( {"model_input_names": input_names or ["input_ids", "token_type_ids", "attention_mask"]}), encoding="utf-8") _mk(_m / "best", cfg={"name": "base"}) _mk(_m / "base", cfg={"name": "base"}) # копия best — в списке не нужна _mk(_m / "runs" / "20260101_000000_large", backbone="x/large") _mk(_m / "runs" / "20260102_000000_modern", backbone="x/modern", input_names=["input_ids", "attention_mask"]) _names = [p.name for p in _runs.candidates()] check("кандидаты: best первой, копия best не дублируется", _names[0] == "best" and "base" not in _names and len(_names) == 3, str(_names)) check("удалять можно запуск, но не best", _runs.deletable(_m / "runs" / "20260101_000000_large") and not _runs.deletable(_m / "best")) _w = _runs.warnings_for(_m / "runs" / "20260102_000000_modern") check("предупреждения: другая основа и нет token_type_ids", any("другая основа" in x for x in _w) and any("token_type_ids" in x for x in _w), str(_w)) _arch = _runs.promote(_m / "runs" / "20260101_000000_large") check("promote: best заменена, прежняя в best_backup_*", _arch and (_m / _arch / "dub_config.json").exists() and _runs.describe(_m / "best")["backbone"] == "x/large" and not (_m / ".best_new").exists()) check("promote: исходный запуск остаётся (копирование, не перенос)", (_m / "runs" / "20260101_000000_large" / "model.pt").exists()) check("promote той же модели ничего не делает", _runs.promote(_m / "best") is None) _runs.delete(_m / _arch) check("delete убрал архивную копию", not (_m / _arch).exists()) try: _runs.delete(_m / "best") check("delete не трогает best", False) except ValueError: check("delete не трогает best", (_m / "best").exists()) finally: _runs.MODELS, _runs.RUNS = _old7 _ds7 = Path("data/processed/dataset.json") if _ds7.exists(): # короткий настоящий прогон: tiny2, 1 эпоха, подвыборка; запуск лежит в своём каталоге, best/reports не тронуты import random as _rnd7 # noqa: E402 with _tf7.TemporaryDirectory() as _td7: _t = Path(_td7) _all = _json.loads(_ds7.read_text(encoding="utf-8")) _tr = [r for r in _all if r["split"] == "train"] _te = [r for r in _all if r["split"] == "test"] _rnd7.Random(1).shuffle(_tr) _rnd7.Random(1).shuffle(_te) (_t / "dataset.json").write_text(_json.dumps(_tr[:300] + _te[:100], ensure_ascii=False), encoding="utf-8") (_t / "label_maps.json").write_text(Path("data/processed/label_maps.json").read_text(encoding="utf-8"), encoding="utf-8") _best_before = (Path("models/best/model.pt").stat().st_mtime if Path("models/best/model.pt").exists() else None) _sum_before = Path("reports/summary.md").read_bytes() if Path("reports/summary.md").exists() else None _p = _sp7.run([sys.executable, "-m", "src.train", "--runs", "--models", "tiny2", "--folds", "0", "--epochs-override", "1", "--data", str(_t / "dataset.json"), "--runs-dir", str(_t / "runs"), "--tag", "_t"], capture_output=True, text=True, encoding="utf-8", errors="replace") _dirs = list((_t / "runs").glob("*_tiny2_t")) check("обучение --runs: запуск создан со всеми файлами", _p.returncode == 0 and len(_dirs) == 1 and all((_dirs[0] / f).exists() for f in ("model.pt", "run.json", "eval.json", "label_maps.json", "confusion_tiny2.png")), (_p.stderr or "")[-300:]) check("обучение --runs не трогает models/best и reports/summary.md", (_best_before is None or Path("models/best/model.pt").stat().st_mtime == _best_before) and (_sum_before is None or Path("reports/summary.md").read_bytes() == _sum_before)) if _dirs: _rows7 = [r for r in _json.loads((_t / "dataset.json").read_text(encoding="utf-8")) if r["split"] == "test"] _fp7 = _runs.fingerprint(_rows7) _stored = _runs.load_eval(_dirs[0], _fp7) _fresh = _runs.eval_checkpoint(_dirs[0], _rows7) check("оценка чекпоинта воспроизводит метрики обучения построчно", _stored is not None and _stored["per_row"] == _fresh["per_row"] and len(_fresh["per_row"]) == len(_rows7)) _cmp = _runs.compare([_dirs[0]], _rows7) check("compare: модель без baseline, доверительный интервал считается", _cmp[0]["vs"] is None and _cmp[0]["ci"][0] <= _cmp[0]["ci"][1]) else: print(" пропущено (нет data/processed/dataset.json)") print("== Docker: сервис без пакетов обучения, токен, настройки ==") _code = chr(10).join(["import sys", "for m in ('sklearn','matplotlib','pandas','onnx','onnxruntime'): sys.modules[m] = None", "import src.service, src.predict, src.release, src.segmenter", "print('IMPORT_OK')"]) _r8 = _sp7.run([sys.executable, "-c", _code], capture_output=True, text=True, encoding="utf-8", errors="replace") check("сервис импортируется без sklearn/matplotlib/pandas/onnx", "IMPORT_OK" in _r8.stdout, (_r8.stderr or "")[-300:]) from src import spans as _spans # noqa: E402 from src import train as _train8 # noqa: E402 check("decode_span и MAX_LEN живут в src.spans, train их реэкспортирует", _train8.decode_span is _spans.decode_span and _train8.MAX_LEN == _spans.MAX_LEN == 128) from fastapi.testclient import TestClient # noqa: E402 from src.service import create_app # noqa: E402 class _Stub: model_dir, device, types, segmenter = Path("models/best"), "cpu", ["x"], None def predict_text(self, text): return {"translation_type": "x", "text": text} _c = TestClient(create_app(_Stub(), api_token="tok")) check("токен: /health открыт, /predict без токена — 401", _c.get("/health").status_code == 200 and _c.post("/predict", json={"text": "a"}).status_code == 401 and _c.post("/predict", json={"text": "a"}, headers={"Authorization": "Bearer bad"}).status_code == 401) check("токен: Bearer и X-API-Key принимаются", _c.post("/predict", json={"text": "a"}, headers={"Authorization": "Bearer tok"}).status_code == 200 and _c.post("/predict", json={"text": "a"}, headers={"X-API-Key": "tok"}).status_code == 200) check("без DUB_API_TOKEN /predict открыт, /health показывает segmenter и auth", TestClient(create_app(_Stub())).post("/predict", json={"text": "a"}).status_code == 200 and _c.get("/health").json()["auth"] is True and _c.get("/health").json()["segmenter"] == "rules") _lf = _c.post("/predict", json={"url": "https://www.lostfilm.tv/series/X/season_1/episode_2/"}, headers={"X-API-Key": "tok"}) _lfj = _lf.json() if _lf.status_code == 200 else {} check("ссылка lostfilm.tv: сразу MVO LostFilm без загрузки страницы", _lfj.get("tracker") == "lostfilm" and _lfj["studios"][0]["studio"] == "LostFilm" and _lfj["studios"][0]["translation_type"] == "проф. многоголосый" and _lfj["tracks"][0]["lang"] == "RUS", str(_lf.status_code)) from src.fetch import is_lostfilm # noqa: E402 check("is_lostfilm: lostfilm.tv и поддомены да, rutracker и подделка в пути нет", is_lostfilm("https://lostfilm.tv/x") and is_lostfilm("https://www.lostfilm.tv/x") and not is_lostfilm("https://rutracker.org/forum/viewtopic.php?t=1&u=lostfilm.tv") and not is_lostfilm("https://evil.com/lostfilm.tv")) import src.service as _svc8 # noqa: E402 _fetch_calls: list = [] _orig_fetch = _svc8.fetch_html _svc8.fetch_html = lambda url, **kw: (_fetch_calls.append(kw), ("", "rutracker"))[1] try: _svc8.predict_release = lambda *a, **k: {"tracks": []} TestClient(create_app(_Stub(), fetch_timeout=77)).post("/predict", json={"url": "https://rutracker.org/forum/viewtopic.php?t=1"}) check("DUB_FETCH_TIMEOUT доходит до загрузки страницы", _fetch_calls and _fetch_calls[0].get("timeout") == 77, str(_fetch_calls)) finally: _svc8.fetch_html = _orig_fetch import importlib # noqa: E402 importlib.reload(_svc8) check("env_float: мусор и пусто — значение по умолчанию", _svc8.env_float("DUB_NOPE_XYZ", 5.0) == 5.0) for _f in ("Dockerfile", ".dockerignore", "docker-compose.yml", ".env.example", ".gitea/workflows/docker.yml", "requirements-service.txt", "docs/docker.md", "docker-models/models/.gitkeep"): check(f"есть {_f}", Path(_f).exists()) check("в requirements-service.txt нет пакетов обучения", not any(x in Path("requirements-service.txt").read_text(encoding="utf-8") for x in ("scikit", "pandas", "matplotlib", "onnx"))) try: import yaml as _yaml # noqa: E402 for _f in ("docker-compose.yml", ".gitea/workflows/docker.yml"): _yaml.safe_load(Path(_f).read_text(encoding="utf-8")) check("YAML compose и workflow разбираются", True) except ImportError: print(" пропущено: нет PyYAML") print("== сегментатор (модель) ==") try: from src.segmenter import LineSegmenter, is_test_release # noqa: E402 _seg = LineSegmenter("models/segmenter", "cpu") _fx = open("data/samples/rt-topic-6870900.html", encoding="utf-8").read() _lines = _seg.select_lines(html_to_text(_fx)) check("сегментатор: найдены 4 строки «Перевод N» реальной страницы", sum(l.startswith("Перевод") for l in _lines) == 4, str(_lines[:5])) check("сегментатор: «Субтитры» и «Оригинальная аудиодорожка» не дорожки", not any(l.startswith(("Субтитры", "Оригинальная")) for l in _lines)) check("разбиение страниц детерминировано", is_test_release("6870900") == is_test_release("6870900")) check("сегментатор: строка из знаков препинания (>160 токенов) не роняет", len(_seg.score(["|;," * 100, "Перевод: Профессиональный (дублированный)"])) == 2) except Exception as e: # noqa: BLE001 print(f" пропущено (нет models/segmenter): {e}") print(f"\nитог: {PASS} ok, {FAIL} failed") sys.exit(1 if FAIL else 0)