#!/usr/bin/env python """Интерактивный сборщик датасета grabber-ai (rutracker / nnm-club / kinozal). Запуск без аргументов открывает меню: .venv\\Scripts\\python tools\\collect.py То же по командам (для скриптов): python tools/collect.py windows [--sites rutracker,nnm,kinozal] python tools/collect.py collect --site rutracker --refs 313,2366 --pages 3 --limit 100 python tools/collect.py backfill --site nnm --limit 200 python tools/collect.py build python tools/collect.py import-legacy --dir <каталог старого harvest> python tools/collect.py status Как это работает ---------------- * Chrome запускается ОДИН (общий профиль и cookie), по окну на трекер. Вы логинитесь в каждом окне сами; пароли и cookie скрипт не видит и не хранит. Профиль лежит вне репозитория: %LOCALAPPDATA%\\dub-ai-harvest\\chrome-profile (другой каталог: переменная DUB_CHROME_PROFILE). * Каждая страница сохраняется ЦЕЛИКОМ (outerHTML после загрузки вкладок kinozal) в data/raw/<трекер>/.html + .json. Это тот же вход, который модель увидит в grabber, а не только текст поста. * `build` превращает сырые страницы в data/pages/*.txt (через src.html_extract) и добавляет слабые метки в data/labels.csv (парсеры из tools/harvest). * id страниц: у уже размеченных раздач остаётся прежний (bare) id, у новых — с префиксом (rt-123, nnm-123, kz-123): так id разных трекеров не пересекаются и разбиение train/test у старых страниц не меняется. """ from __future__ import annotations import argparse import csv import json import os import random import re import sys import threading import time from collections import Counter, deque from pathlib import Path ROOT = Path(__file__).resolve().parents[1] DATA = ROOT / "data" RAW = DATA / "raw" PAGES = DATA / "pages" LABELS = DATA / "labels.csv" POSTONLY = DATA / "pages_postonly.txt" STATE = RAW / "_state.json" _home = Path(os.environ.get("LOCALAPPDATA") or Path.home()) / "dub-ai-harvest" os.environ.setdefault("DUB_HARVEST_OUT", str(_home)) os.environ.setdefault("DUB_CHROME_PROFILE", str(_home / "chrome-profile")) # свой порт: не подключаться к чужому Chrome с отладкой на 9222 (окна открылись бы там) os.environ.setdefault("DUB_CHROME_PORT", "9333") sys.path.insert(0, str(ROOT)) sys.path.insert(0, str(ROOT / "tools")) sys.path.insert(0, str(ROOT / "tools" / "harvest")) import harvest_core as hv # noqa: E402 import meta as metamod # noqa: E402 import ui # noqa: E402 META = DATA / "meta.csv" DEFAULTS = {"tabs": 3, "delay": 1.0, "limit": 100, "pages": 2, "mode": "balanced", "per_page": 6, "workers": 0} def clamp_tabs(value) -> int: try: return max(1, min(4, int(value))) except (TypeError, ValueError): return DEFAULTS["tabs"] def settings() -> dict: """Настройки сбора (запоминаются между запусками): вкладок, пауза, лимит, страниц списка.""" try: saved = json.loads((RAW / "_settings.json").read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): saved = {} cfg = {**DEFAULTS, **saved} cfg["tabs"] = clamp_tabs(cfg["tabs"]) return cfg def save_settings(cfg: dict) -> None: cfg = {**cfg, "tabs": clamp_tabs(cfg.get("tabs"))} RAW.mkdir(parents=True, exist_ok=True) (RAW / "_settings.json").write_text(json.dumps(cfg, ensure_ascii=False, indent=2), encoding="utf-8") os.environ["DUB_TABS"] = str(cfg["tabs"]) SITES = { "rutracker": {"prefix": "rt", "host": "rutracker.org", "home": "https://rutracker.org/forum/index.php", "refs": "313,2366", "what": "id разделов (viewforum.php?f=…)"}, "nnm": {"prefix": "nnm", "host": "nnmclub.to", "home": "https://nnmclub.to/forum/index.php", "refs": "227,768", "what": "id разделов (viewforum.php?f=…)"}, "kinozal": {"prefix": "kz", "host": "kinozal.me", "home": "https://kinozal.me/", "refs": "https://kinozal.me/browse.php", "what": "адреса списков (browse.php?… / top.php)"}, } def say(msg: str = "", tag: str = "") -> None: from rich.markup import escape prefix = f"[{ui.SITE_STYLE.get(tag, 'dim')}]{tag}[/] " if tag else "" ui.console.print(prefix + escape(msg)) # ---------------------------------------------------------------- метки / id def read_labels() -> list[dict]: if not LABELS.exists(): return [] with LABELS.open(encoding="utf-8-sig", newline="") as f: return list(csv.DictReader(f)) def migrate_labels() -> bool: """labels.csv со старой шапкой (без lang/audio_raw) -> новая шапка. cmd_build дописывает строки через DictWriter(FIELDS): под старой шапкой они сдвинули бы все колонки. Файл переписывается один раз, прежние данные не меняются; копия — labels.bak_schema.csv. """ if not LABELS.exists() or LABELS.stat().st_size == 0: return False with LABELS.open(encoding="utf-8-sig", newline="") as f: reader = csv.DictReader(f) head, rows = reader.fieldnames or [], list(reader) if head == hv.FIELDS: return False import shutil bak = LABELS.with_name("labels.bak_schema.csv") if not bak.exists(): shutil.copy2(LABELS, bak) with LABELS.open("w", encoding="utf-8-sig", newline="") as f: w = csv.DictWriter(f, fieldnames=hv.FIELDS, extrasaction="ignore", restval="") w.writeheader() w.writerows(rows) return True def labeled_ids(rows: list[dict] | None = None) -> dict[str, set[str]]: """{источник: {release_id}} по текущему data/labels.csv.""" out: dict[str, set[str]] = {s: set() for s in SITES} for r in rows if rows is not None else read_labels(): out.setdefault(r["source"], set()).add(r["release_id"]) return out def page_stem(source: str, rid: str, labeled: dict[str, set[str]]) -> str: """Имя страницы = release_id в labels.csv. Размеченные раньше раздачи сохраняют свой bare id (их пары page/label и разбиение train/test не меняются), если этот id не занят другим источником. Все новые получают префикс источника. """ if rid in labeled.get(source, ()): taken = any(rid in ids for s, ids in labeled.items() if s != source) if not taken: return rid return f"{SITES[source]['prefix']}-{rid}" def real_id(site: str, release_id: str) -> str: """release_id из labels.csv -> id раздачи на трекере (без нашего префикса).""" pre = SITES[site]["prefix"] + "-" return release_id[len(pre):] if release_id.startswith(pre) else release_id def labeled_real_ids(site: str, labeled: dict[str, set[str]] | None = None) -> set[str]: labeled = labeled if labeled is not None else labeled_ids() return {real_id(site, i) for i in labeled.get(site, ())} def stored_ids(site: str) -> set[str]: d = RAW / site if not d.exists(): return set() return {p.stem for p in d.glob("*.json") if not p.name.endswith(".legacy.json")} # ---------------------------------------------------------------- браузер def open_window(url: str) -> dict: """Новое ОКНО Chrome (а не вкладка) через браузерный CDP-эндпоинт.""" ver = hv.http_json("/json/version") browser = hv.CDP(ver["webSocketDebuggerUrl"]) try: tid = browser.send("Target.createTarget", {"url": url, "newWindow": True})["targetId"] finally: browser.close() for _ in range(40): for t in hv.http_json("/json/list"): if t.get("id") == tid: return t time.sleep(0.25) raise RuntimeError("окно не появилось") def activate(target_id: str) -> None: """Вынести окно на передний план.""" ver = hv.http_json("/json/version") browser = hv.CDP(ver["webSocketDebuggerUrl"]) try: browser.send("Target.activateTarget", {"targetId": target_id}) except RuntimeError: pass finally: browser.close() def site_target(site: str, create: bool = True) -> dict | None: host = SITES[site]["host"] for t in hv.http_json("/json/list"): if t.get("type") == "page" and host in (t.get("url") or ""): return t return open_window(SITES[site]["home"]) if create else None def connect_site(site: str): if not hv.chrome_start(SITES[site]["home"]): raise RuntimeError("не удалось запустить Chrome") target = site_target(site) cdp = hv.CDP(target["webSocketDebuggerUrl"]) cdp.send("Page.enable") cdp.send("Runtime.enable") return cdp LOGIN_JS = """JSON.stringify((() => { const q = s => !!document.querySelector(s); const head = document.body ? document.body.innerText.slice(0, 4000) : ''; return { out: q('a[href*="logout"], #logged-in-username'), guest: q('form[action*="login"]') || /Гость!|Зарегистрируйтесь/.test(head), title: document.title }; })())""" def login_state(info: dict) -> str: """'in' — вошли, 'guest' — гость, 'unknown' — не удалось определить.""" if info.get("out"): return "in" if info.get("guest"): return "guest" return "unknown" def check_login(site: str) -> str: cdp = connect_site(site) try: nav = hv.goto(cdp, SITES[site]["home"], min_len=300, timeout=60) if not nav["ok"]: return "unknown" return login_state(json.loads(cdp.eval(LOGIN_JS) or "{}")) finally: cdp.close() def cmd_windows(sites: list[str], wait: bool = True) -> None: if not hv.chrome_start(SITES[sites[0]]["home"]): say("Chrome не запустился (проверьте путь в harvest_core.CHROME_PATHS)") return say(f"Chrome: порт {hv.CHROME_PORT}, профиль {hv.CHROME_PROFILE}") for s in sites: activate(site_target(s)["id"]) say(f"окно {s}: {SITES[s]['home']}") if wait: say("\nОткрылись окна Chrome (если их не видно — проверьте панель задач).") say("Пароль вводится ТОЛЬКО в этих окнах браузера, не в консоли.") input("Войдите в аккаунт в каждом окне (Cloudflare тоже пройдите там) " "и нажмите Enter здесь… ") names = {"in": "вошли", "guest": "гость — войдите", "unknown": "не определено"} for s in sites: try: st = check_login(s) except Exception as exc: # noqa: BLE001 st = "unknown" say(f"{exc}", s) say(f"{s:10s}: {names[st]}") # ---------------------------------------------------------------- сбор def fetch_release(cdp, site: str, rid: str) -> dict: if site == "rutracker": rec = hv.rt_topic(cdp, rid) elif site == "nnm": rec = hv.nnm_topic(cdp, rid) else: rec = hv.kz_release(cdp, rid) if rec.get("ok"): if site == "kinozal": if not (rec.get("release_tab") or "").strip(): # вкладка не успела загрузиться rec["release_tab"] = hv.kz_read_tab(cdp, rid, 0, timeout=10) if not (rec.get("tech_tab") or "").strip(): rec["tech_tab"] = hv.kz_read_tab(cdp, rid, 100, timeout=10) hv.kz_read_tab(cdp, rid, 0) # вернуть вкладку «Описание» перед снимком HTML rec["_html"] = snapshot_html(cdp) return rec # Снимок страницы «как на сайте»: с кодировкой, адресом (картинки грузятся с сайта) и стилями. # Таблицы стилей выносятся в data/raw/_css/<хеш>.css (одна на все страницы), в HTML — ссылка. SNAPSHOT_JS = (Path(__file__).resolve().parent / "snapshot.js").read_text(encoding="utf-8") CSS_DIR_NAME = "_css" def store_css(text: str) -> str: """Таблица стилей -> data/raw/_css/<хеш>.css; возвращает имя файла.""" import hashlib name = hashlib.sha1(text.encode("utf-8")).hexdigest()[:12] + ".css" d = RAW / CSS_DIR_NAME d.mkdir(parents=True, exist_ok=True) if not (d / name).exists(): (d / name).write_text(text, encoding="utf-8") return name def snapshot_html(cdp) -> str: """Читаемый снимок открытой страницы: UTF-8, , стили файлами. Пусто, если не получилось.""" try: snap = json.loads(cdp.eval(SNAPSHOT_JS) or "{}") except (json.JSONDecodeError, TypeError): snap = {} html = snap.get("html") or "" for i, css in enumerate(snap.get("css") or []): html = html.replace(f"@@CSS{i}@@", f"../{CSS_DIR_NAME}/{store_css(css)}") return html or (cdp.eval("document.documentElement.outerHTML") or "") def cmd_repair_html(with_css: bool = True) -> dict: """Старые сохранённые страницы -> читаемые: кодировка UTF-8, и стили сайта. Время изменения файлов не меняется, поэтому кэши «уже разобрано» остаются в силе. Стили берутся один раз с живой страницы каждого трекера (нужен Chrome и вход). """ import re as _re stats = {"fixed": 0, "already": 0, "css_sites": 0} bases = {"rutracker": "https://rutracker.org/forum/", "nnm": "https://nnmclub.to/forum/", "kinozal": "https://kinozal.me/"} for site in SITES: d = RAW / site files = [p for p in sorted(d.glob("*.html"))] if d.exists() else [] if not files: continue links = "" if with_css: try: cdp = connect_site(site) try: hv.goto(cdp, SITES[site]["home"], min_len=300, timeout=60) snap = json.loads(cdp.eval(SNAPSHOT_JS) or "{}") finally: cdp.close() links = "".join(f'' for c in snap.get("css") or []) stats["css_sites"] += bool(links) except Exception as exc: # noqa: BLE001 say(f"стили {site} не получены: {exc}", site) prog = ui.progress() with prog: task = prog.add_task(f"{site}: правка HTML", total=len(files)) for hp in files: prog.advance(task) text = hp.read_text(encoding="utf-8", errors="replace") if _re.search(r"(?i)]+charset", text[:3000]): stats["already"] += 1 continue add = f'{links}' new, n = _re.subn(r"(?i)(]*>)", lambda m: m.group(1) + add, text, count=1) if not n: new = f"{add}{text}" st = hp.stat() hp.write_text(new, encoding="utf-8") os.utime(hp, ns=(st.st_atime_ns, st.st_mtime_ns)) stats["fixed"] += 1 say("repair-html: " + ", ".join(f"{k}={v}" for k, v in stats.items())) return stats def has_body(site: str, rec: dict) -> bool: if site == "kinozal": return bool((rec.get("release_tab") or "").strip()) return len((rec.get("post") or "").strip()) >= 30 def save_release(site: str, rec: dict) -> None: d = RAW / site d.mkdir(parents=True, exist_ok=True) rid = str(rec["id"]) html = rec.pop("_html", "") (d / f"{rid}.html").write_text(html, encoding="utf-8") rec["fetched"] = hv.stamp() # json пишется последним: он же признак «сохранено целиком» (d / f"{rid}.json").write_text(json.dumps(rec, ensure_ascii=False), encoding="utf-8") _state_lock = threading.Lock() def load_state() -> dict: try: return json.loads(STATE.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): return {} def update_state(site: str, ref: str, **fields) -> None: with _state_lock: st = load_state() st.setdefault(site, {}).setdefault(ref, {}).update(fields, updated=hv.stamp()) RAW.mkdir(parents=True, exist_ok=True) STATE.write_text(json.dumps(st, ensure_ascii=False, indent=2), encoding="utf-8") SECTIONS = RAW / "_sections.json" def load_sections() -> dict: try: return json.loads(SECTIONS.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): return {} def save_sections(cache: dict) -> None: RAW.mkdir(parents=True, exist_ok=True) tmp = SECTIONS.with_suffix(".tmp") tmp.write_text(json.dumps(cache, ensure_ascii=False, indent=2), encoding="utf-8") tmp.replace(SECTIONS) class quiet_harvest: """Глушит построчный лог harvest_core (в панели с прогрессом он только мешает).""" def __enter__(self): self._old, hv.log = hv.log, (lambda *_a, **_k: None) def __exit__(self, *exc): hv.log = self._old def ensure_sections(sites: list[str], force: bool = False) -> dict: """Находит разделы с зарубежными фильмами/сериалами (первый запуск) и кэширует. Кэш: data/raw/_sections.json. Повторно ищем только с force=True. """ import discover from rich.progress import (BarColumn, MofNCompleteColumn, Progress, SpinnerColumn, TextColumn, TimeElapsedColumn) cache = load_sections() for s in sites: created = cache.get("created", {}).get(s) age = (time.time() - time.mktime(time.strptime(created, "%Y-%m-%d %H:%M:%S"))) / 86400 \ if created else None fresh = (cache.get("version", {}).get(s) == discover.RULES_VERSION and age is not None and age < discover.SECTIONS_TTL_DAYS) if cache.get(s) and fresh and not force: ui.console.print(f"[dim]Разделы {s}: из кэша, {len(cache[s])} шт., " f"обновлено {created} (повторный поиск: «Дополнительно» → 7)[/]") continue old = cache.get(s) or [] probes = cache.setdefault("probes", {}).setdefault(s, {}) entries: list[dict] = [] with quiet_harvest(), Progress( SpinnerColumn(), TextColumn("[bold]{task.description}"), BarColumn(), MofNCompleteColumn(), TimeElapsedColumn(), TextColumn("[dim]{task.fields[cur]}"), console=ui.console, transient=True) as prog: task = prog.add_task(f"Поиск разделов: {s}", total=None, cur="подключение") try: cdp = connect_site(s) try: if s == "kinozal": prog.update(task, cur="проверяю адреса списков") entries = [{**x, "topics": None, "keep": True, "status": "ok"} for x in discover.kinozal_plan(cdp, lambda *_a: None)] else: def on_probe(ref, n, probes=probes): probes[ref] = n save_sections(cache) # сохраняем по ходу: прерванный поиск не пропадает entries = discover.discover_forums( cdp, s, on_stage=lambda t: prog.update(task, cur=t), on_step=lambda i, n, name: prog.update(task, total=n, completed=i, cur=name[:60]), probes=probes, on_probe=on_probe) finally: cdp.close() except Exception as exc: # noqa: BLE001 ui.console.print(f"[red]{s}: поиск разделов не удался: {exc}[/]") continue found = [{"ref": e["ref"], "name": e["name"], "topics": e["topics"]} for e in entries if e["keep"]] ui.console.print(ui.sections_table(s, entries)) if old and len(found) < 0.6 * len(old) and cache.get("version", {}).get(s) == discover.RULES_VERSION: ui.console.print(f"[yellow]{s}: найдено {len(found)} против {len(old)} — похоже на " f"неполную загрузку, прежний список сохранён[/]") continue if found: cache[s] = found cache.setdefault("created", {})[s] = hv.stamp() cache.setdefault("version", {})[s] = discover.RULES_VERSION save_sections(cache) return cache def section_refs(site: str) -> list[str]: return [x["ref"] for x in load_sections().get(site, [])] def choose_refs(site: str) -> str: found = section_refs(site) if found: if ask(f"{site}: найдено разделов {len(found)}. Использовать? (Y/n)", "y").lower() != "n": return ",".join(found) return ask(f"{site}: {SITES[site]['what']}", SITES[site]["refs"]) def list_ids(cdp, site: str, ref: str, page: int) -> tuple[bool, list[str]]: if site == "kinozal": return hv.kz_browse_ids(cdp, hv.kz_page_url(ref, page)) fn = hv.rt_forum_page if site == "rutracker" else hv.nnm_forum_page ok, items = fn(cdp, ref, page) return ok, [it["id"] for it in items] def list_items(cdp, site: str, ref: str, page: int) -> tuple[bool, list[dict]]: """Страница списка раздела -> (открылась, [{id, text}]) с заголовками тем.""" if site == "kinozal": return hv.kz_browse_items(cdp, hv.kz_page_url(ref, page)) return (hv.rt_forum_page if site == "rutracker" else hv.nnm_forum_page)(cdp, ref, page) DEAD = RAW / "_dead.json" def load_dead(site: str) -> set[str]: """id удалённых/несуществующих раздач: повторно не открываем.""" try: return set(json.loads(DEAD.read_text(encoding="utf-8")).get(site, [])) except (OSError, json.JSONDecodeError): return set() def mark_dead(site: str, rid: str) -> None: with _state_lock: try: data = json.loads(DEAD.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): data = {} data[site] = sorted(set(data.get(site, [])) | {rid}) RAW.mkdir(parents=True, exist_ok=True) DEAD.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8") def is_dead(rec: dict) -> bool: return not rec.get("ok") and "нет раздачи" in (rec.get("error") or "").lower() def _on_retry(url: str, attempt: int, retries: int, reason: str) -> None: """Страница вернула ошибку сервера/Cloudflare: покажем повтор в панели сбора.""" for site, info in SITES.items(): if info["host"] in url: ui.get_stats(site).state = f"повтор {attempt}/{retries}: {reason}" return hv.RETRY_HOOK = _on_retry def section_name(site: str, ref: str) -> str: for x in load_sections().get(site, []): if x["ref"] == ref: return x["name"] return ref def stats_start(site: str, limit: int) -> "ui.Stats": """Один трекер может пройти два этапа подряд (дозагрузка и новые): лимиты складываются.""" st = ui.get_stats(site) if not st.started or st.finished: st.begin(limit) else: st.limit += limit st.finished = 0.0 return st class Tabs: """Несколько вкладок одного трекера в общем Chrome (cookie и вход общие). Первая — окно, где вы входили; остальные открываются фоном и закрываются в конце. Число вкладок: переменная DUB_TABS (по умолчанию 3). """ def __init__(self, site: str, n: int | None = None): n = max(1, int(n or os.environ.get("DUB_TABS", 3))) self.main = connect_site(site) self.cdps = [self.main] self.extra: list[str] = [] for _ in range(n - 1): try: tab = hv.http_json("/json/new?about:blank", method="PUT") cdp = hv.CDP(tab["webSocketDebuggerUrl"]) cdp.send("Page.enable") cdp.send("Runtime.enable") except Exception: # noqa: BLE001 break # не вышло — работаем с тем, что есть self.cdps.append(cdp) self.extra.append(tab["id"]) def close(self) -> None: for cdp in self.cdps: cdp.close() for tid in self.extra: try: hv.http_json(f"/json/close/{tid}") except Exception: # noqa: BLE001 pass HARD_LIMIT_S = 150 # одна раздача дольше этого — считаем соединение зависшим SLOW_S = 40 # дольше этого — помечаем в списке ошибок def harvest_ids(cdps, site: str, ids: list[str], limit: int, delay: float, stop: threading.Event, section: str = "") -> int: """Скачивает раздачи параллельно: по рабочему на вкладку. Возвращает число сохранённых.""" cdps = list(cdps) if isinstance(cdps, (list, tuple)) else [cdps] st = ui.get_stats(site) todo = deque(ids) lock = threading.Lock() done = [0] st.state = "скачиваю" def worker(cdp) -> None: while not stop.is_set(): with lock: if done[0] >= limit or not todo: return rid = todo.popleft() done[0] += 1 # слот под лимит занят заранее st.state = f"скачиваю {rid}" saved = False t_start = time.time() guard = threading.Timer(HARD_LIMIT_S, cdp.abort) # сторож: зависший вызов обрывается guard.daemon = True guard.start() try: rec = fetch_release(cdp, site, rid) if is_dead(rec): # удалена: запоминаем и идём дальше mark_dead(site, rid) st.skipped += 1 elif rec.get("ok") and has_body(site, rec): rec["section_ref"] = section save_release(site, rec) st.ok(f"{rid} {rec.get('title') or rec.get('h1') or rec.get('page_title') or ''}") saved = True else: st.fail(f"{rid}: нет текста раздачи (нужен вход? {rec.get('error', '')})") except Exception as exc: # noqa: BLE001 st.fail(f"{rid}: {exc}") finally: guard.cancel() if time.time() - t_start > HARD_LIMIT_S: # сторож сработал: соединение мертво st.fail(f"{rid}: завис на {time.time() - t_start:.0f} с, вкладка переподключена") try: cdp.reconnect() except Exception as exc: # noqa: BLE001 st.fail(f"переподключение не удалось: {exc}") return elif time.time() - t_start > SLOW_S: st.fail(f"{rid}: медленно, {time.time() - t_start:.0f} с") # видно, на каких страницах if not saved: with lock: done[0] -= 1 # неудача не съедает лимит time.sleep(delay + random.random() * 0.5) helpers = [threading.Thread(target=worker, args=(c,), daemon=True) for c in cdps[1:]] for t in helpers: t.start() worker(cdps[0]) for t in helpers: t.join() return done[0] def collect_ref_balanced(site: str, ref: str, cdp, tabs: "Tabs", have: set, counts, ref_quota: int, limit_left: int, delay: float, stop: threading.Event, per_page: int, rng: random.Random) -> int: """Раздел вразброс: страницы по всей глубине, на странице — темы из недопредставленных корзин (год × качество × фильм/сериал). Возвращает число сохранённых раздач.""" import diversity as dv stat = ui.get_stats(site) st = load_state().get(site, {}).get(ref, {}) visited = set(st.get("visited", [])) depth = int(st.get("depth") or 0) saved, fails = 0, 0 for _ in range(300): if stop.is_set() or saved >= min(ref_quota, limit_left) or fails >= 3: break page = 0 if not depth else (next_pg[0] if (next_pg := dv.next_pages(depth, visited, 1, rng)) else None) if page is None: # раздел обойден целиком: начинаем новый круг update_state(site, ref, visited=[], finished=True) break stat.state = f"список, стр.{page + 1}" + (f"/{depth}" if depth else "") ok, items = list_items(cdp, site, ref, page) if not ok: fails += 1 stat.fail(f"{stat.section}: стр.{page + 1} не открылась") visited.add(page) continue visited.add(page) seen_depth = hv.list_depth(cdp) if not items: # дальше конца раздела depth = max(1, page) if page else 1 update_state(site, ref, visited=sorted(v for v in visited if v < depth), depth=depth) continue depth = max(depth, seen_depth, page + 1) fresh = [it for it in items if it["id"] not in have] want = min(per_page, ref_quota - saved, limit_left - saved) chosen = dv.pick(fresh, want, counts, rng) n = harvest_ids(tabs.cdps, site, [it["id"] for it in chosen], want, delay, stop, section=ref) have |= stored_ids(site) saved += n update_state(site, ref, visited=sorted(visited), depth=depth, next_page=0) return saved def run_collect(site: str, refs: list[str], pages: int, limit: int, delay: float, stop: threading.Event, from_page: int | None = None) -> int: """Новые раздачи из списков разделов; уже размеченные и сохранённые пропускаются. Лимит делится между разделами поровну, чтобы в выборку попали все форматы и годы. """ have = stored_ids(site) | labeled_real_ids(site) | load_dead(site) stat = stats_start(site, limit) stat.state = "подключение" tabs = Tabs(site) cdp = tabs.main total = 0 quota = max(1, -(-limit // max(len(refs), 1))) cfg = settings() balanced = cfg.get("mode") == "balanced" and from_page is None if balanced: import diversity as dv counts = dv.counts_from_meta(META, site) rng = random.Random() try: for ref in refs: if balanced: if stop.is_set() or total >= limit: break stat.section = section_name(site, ref) ref_quota = quota * 3 if "v=0&d=0" in ref else quota total += collect_ref_balanced(site, ref, cdp, tabs, have, counts, ref_quota, limit - total, delay, stop, int(cfg.get("per_page", 6)), rng) continue st = load_state().get(site, {}).get(ref, {}) # раздел уже дочитан до конца: новые раздачи появляются сверху, начинаем со стр.1 start = (0 if st.get("finished") else st.get("next_page", 0)) if from_page is None else from_page if st.get("finished"): update_state(site, ref, finished=False, next_page=0) ref_total = 0 stat.section = section_name(site, ref) # общие списки kinozal (все фильмы / сериалы / мульты / аниме) весят втрое больше сетки ref_quota = quota * 3 if "v=0&d=0" in ref else quota for k in range(pages): if stop.is_set() or total >= limit or ref_total >= ref_quota: break page = start + k stat.state = f"список, стр.{page + 1}" ok, ids = list_ids(cdp, site, ref, page) if not ok: stat.fail(f"{stat.section}: стр.{page + 1} не открылась") break if not ids: update_state(site, ref, finished=True) break fresh = [i for i in ids if i not in have] n = harvest_ids(tabs.cdps, site, fresh, min(limit - total, ref_quota - ref_total), delay, stop, section=ref) have |= stored_ids(site) total += n ref_total += n complete = n == len(fresh) and not stop.is_set() update_state(site, ref, next_page=page + 1 if complete else page) if not complete: break finally: tabs.close() stat.done("остановлено" if stop.is_set() else "готово") return total def backfill_ids(site: str, missing_pages_only: bool = True) -> list[str]: """Размеченные раздачи, у которых нет ни сохранённой страницы, ни data/pages.""" have = stored_ids(site) | load_dead(site) labeled = labeled_ids() ids = [] for release_id in sorted(labeled.get(site, set())): rid = real_id(site, release_id) if rid in have or (missing_pages_only and (PAGES / f"{release_id}.txt").exists()): continue ids.append(rid) random.Random(7).shuffle(ids) # ограниченный лимит -> разнообразная выборка return ids def run_backfill(site: str, limit: int, delay: float, stop: threading.Event) -> int: ids = backfill_ids(site) stat = stats_start(site, min(limit, len(ids))) stat.section = f"дозагрузка размеченных (осталось {len(ids)})" if not ids: stat.done() return 0 stat.state = "подключение" tabs = Tabs(site) try: n = harvest_ids(tabs.cdps, site, ids, limit, delay, stop, section="backfill") finally: tabs.close() stat.done("остановлено" if stop.is_set() else "готово") return n def run_threads(jobs: dict, stop: threading.Event) -> None: """jobs: {site: callable}. По потоку на трекер (у каждого своё окно и свой CDP).""" ui.STATS.clear() sites = list(jobs) threads = [threading.Thread(target=fn, name=s, daemon=True) for s, fn in jobs.items()] for t in threads: t.start() with ui.live(sites): try: while any(t.is_alive() for t in threads): time.sleep(0.25) except KeyboardInterrupt: stop.set() for s in sites: ui.get_stats(s).state = "останавливаюсь…" for t in threads: t.join(timeout=90) ui.console.print(ui.dashboard(sites)) def safe(site: str, fn): def run(): try: fn() except Exception as exc: # noqa: BLE001 say(f"ошибка: {exc}", site) return run # ---------------------------------------------------------------- сборка pages/labels PAR_MIN = 40 # меньше задач — быстрее в одном процессе, чем поднимать пул def worker_count() -> int: """Процессов для переиндексации: настройка `workers`, иначе число ядер минус два (не больше 12).""" try: n = int(settings().get("workers") or 0) except (TypeError, ValueError): n = 0 return n if n > 0 else max(1, min(12, (os.cpu_count() or 2) - 2)) class Runner: """Пул процессов на время переиндексации; для малых задач и workers=1 считает на месте.""" def __init__(self, workers: int | None = None) -> None: self.workers = workers or worker_count() self.pool = None def map(self, fn, tasks: list, chunk: int = 8): """Результаты в порядке задач (по мере готовности).""" if self.workers <= 1 or len(tasks) < PAR_MIN: for t in tasks: yield fn(t) return if self.pool is None: from concurrent.futures import ProcessPoolExecutor self.pool = ProcessPoolExecutor(max_workers=self.workers) yield from self.pool.map(fn, tasks, chunksize=chunk) def close(self) -> None: if self.pool is not None: self.pool.shutdown(wait=True, cancel_futures=True) self.pool = None def cmd_build(rewrite: bool = False, relabel: bool = False, runner: "Runner | None" = None) -> dict: """data/raw -> data/pages/*.txt и новые строки в data/labels.csv. relabel: метки уже размеченных релизов, у которых сохранён HTML, строятся заново текущими парсерами (старые строки заменяются; ручные метки и релизы без HTML не трогаются; если парсер ничего не нашёл — остаются прежние). Разбор идёт в нескольких процессах (настройка workers, по умолчанию ядра − 2). """ import rebuild_workers as rw if migrate_labels(): say("labels.csv: добавлены колонки lang и audio_raw (копия: labels.bak_schema.csv)") own_runner = runner is None runner = runner or Runner() rows = read_labels() labeled = labeled_ids(rows) PAGES.mkdir(parents=True, exist_ok=True) new_rows: list[dict] = [] replaced: set[tuple[str, str]] = set() # (источник, release_id) с новыми метками manual = set(MANUAL.read_text(encoding="utf-8").split()) if MANUAL.exists() else set() stats = {"pages": 0, "labeled": 0, "skipped_no_labels": 0, "unchanged": 0} postonly = set(POSTONLY.read_text(encoding="utf-8").split()) if POSTONLY.exists() else set() built = {} if rewrite else load_json(RAW / "_built.json") # что уже разобрано: «подпись» файлов + имя страницы t0 = time.time() per: dict[str, dict] = {} prog = ui.progress() prog.start() try: for site in SITES: d = RAW / site if not d.exists(): continue done = built.setdefault(site, {}) files = [jp for jp in sorted(d.glob("*.json")) if not jp.name.endswith(".legacy.json") and jp.with_suffix(".html").exists()] ps = per[site] = {"files": len(files), "unchanged": 0, "pages": 0, "relabeled": 0, "new": 0, "no_labels": 0, "t": time.time()} task_id = prog.add_task(f"{site}: разбор HTML", total=len(files)) # проход 1 (быстрый, на месте): что изменилось и что с ним делать jobs: list[dict] = [] for jp in files: hp = jp.with_suffix(".html") sig = file_sig(jp, hp) prev = done.get(jp.stem) if prev and prev["sig"] == sig and ( prev["stem"] is None or (PAGES / f"{prev['stem']}.txt").exists()): stats["unchanged"] += 1 # файлы и парсер не менялись ps["unchanged"] += 1 prog.advance(task_id) continue rid = jp.stem stem = page_stem(site, rid, labeled) known = rid in labeled.get(site, ()) or stem in labeled.get(site, ()) page = PAGES / f"{stem}.txt" # полная страница заменяет урезанную (текст поста) из импорта старого harvest jobs.append({"site": site, "json": str(jp), "html": str(hp), "sig": sig, "stem": stem, "known": known, "parse": not known or (relabel and stem not in manual), "page": str(page) if (rewrite or not page.exists() or stem in postonly) else None}) # проход 2: разбор HTML и запись страниц — параллельно for job, res in zip(jobs, runner.map(rw.build_one, jobs)): prog.advance(task_id) rid, stem = Path(job["json"]).stem, job["stem"] done[rid] = {"sig": job["sig"], "stem": stem} if res["error"]: say(f"{site} {rid}: разбор не удался: {res['error']}") if res["skipped"]: stats["skipped_no_labels"] += 1 ps["no_labels"] += 1 done[rid]["stem"] = None continue if res["error"] and not job["known"]: continue if res["wrote"]: stats["pages"] += 1 ps["pages"] += 1 if stem in postonly: postonly.discard(stem) stats["upgraded"] = stats.get("upgraded", 0) + 1 labs = res["labs"] for r in labs: r["release_id"] = stem new_rows.append(r) if labs and job["known"]: replaced.add((site, stem)) ps["relabeled"] += 1 elif labs: ps["new"] += 1 if labs: stats["labeled"] += 1 finally: prog.stop() if replaced: import shutil backup = LABELS.with_name(f"labels.bak_{time.strftime('%Y%m%d_%H%M%S')}.csv") shutil.copy2(LABELS, backup) old = [r for r in rows if (r["source"], r["release_id"]) not in replaced] stats["relabeled"], stats["rows_before"] = len(replaced), len(rows) with LABELS.open("w", encoding="utf-8-sig", newline="") as f: w = csv.DictWriter(f, fieldnames=hv.FIELDS, extrasaction="ignore") w.writeheader() w.writerows(old) stats["backup"] = backup.name if new_rows: fresh_file = not LABELS.exists() or LABELS.stat().st_size == 0 with LABELS.open("a", encoding="utf-8-sig" if fresh_file else "utf-8", newline="") as f: w = csv.DictWriter(f, fieldnames=hv.FIELDS, extrasaction="ignore") if fresh_file: w.writeheader() w.writerows(new_rows) stats["new_label_rows"] = len(new_rows) if POSTONLY.exists(): POSTONLY.write_text("\n".join(sorted(postonly)) + ("\n" if postonly else ""), encoding="utf-8") save_json(RAW / "_built.json", built) try: m = cmd_meta(rewrite, runner) finally: if own_runner: runner.close() stats["meta_rows"], stats["meta_reused"] = m["rows"], m["reused"] ui.console.print(ui.summary_table("Итоги переиндексации", [ "трекер", "файлов HTML", "без изменений", "страниц записано", "метки пересчитаны", "новые релизы", "без меток", "сек"], [[s_, p_["files"], p_["unchanged"], p_["pages"], p_["relabeled"], p_["new"], p_["no_labels"], f"{time.time() - p_['t']:.0f}"] for s_, p_ in per.items()] + [["meta.csv", m["rows"] + 0, f"из кэша {m['reused']}", "", "", "", "", f"{time.time() - t0:.0f} всего, процессов {runner.workers}"]])) say("build: " + ", ".join(f"{k}={v}" for k, v in stats.items())) return stats # Версия разбора: поднимите при правке парсеров (harvest_core, html_to_text) — тогда # кэш «уже разобрано» сбросится сам и данные пересоберутся один раз. PARSE_VERSION = 2 def load_json(path: Path) -> dict: try: return json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): return {} def save_json(path: Path, data: dict) -> None: path.parent.mkdir(parents=True, exist_ok=True) tmp = path.with_suffix(".tmp") tmp.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8") tmp.replace(path) def file_sig(*paths: Path) -> str: """Подпись входа: время и размер файлов + версия разбора.""" parts = [f"{p.stat().st_mtime_ns}:{p.stat().st_size}" for p in paths if p.exists()] return "|".join(parts) + f"|v{PARSE_VERSION}" def cmd_meta(rewrite: bool = False, runner: "Runner | None" = None) -> dict: """data/raw (и импорт старого harvest) -> data/meta.csv: одна строка на релиз. Строки кэшируются (data/raw/_meta_cache.json) по подписи файлов, версии извлекателя и названия раздела: разбирается только новое и изменённое. Текст страницы берётся из data/pages (уже разобран при сборке), а не из HTML заново; извлечение идёт в нескольких процессах. """ import rebuild_workers as rw own_runner = runner is None runner = runner or Runner() labeled = labeled_ids() cache = {} if rewrite else load_json(RAW / "_meta_cache.json") fresh_cache: dict[str, dict] = {} out: list[dict] = [] reused = 0 prog = ui.progress() prog.start() try: for site in SITES: d = RAW / site if not d.exists(): continue files: dict[str, Path] = {} for jp in sorted(d.glob("*.json"), key=lambda p: not p.name.endswith(".legacy.json")): files[jp.name.split(".")[0]] = jp # legacy идут первыми, полные их заменят task_id = prog.add_task(f"{site}: meta.csv", total=len(files)) jobs: list[dict] = [] for key, jp in files.items(): hp = jp.with_suffix(".html") ck = f"{site}:{key}" sig = f"{file_sig(jp, hp)}|m{metamod.META_VERSION}" hit = cache.get(ck) if hit and hit["sig"] == sig: rec_ref = hit["ref"] name = section_name(site, rec_ref) if rec_ref and rec_ref != "backfill" else "" if hit["section"] == name: # название раздела не менялось out.append(hit["row"]) fresh_cache[ck] = hit reused += 1 prog.advance(task_id) continue rec = json.loads(jp.read_text(encoding="utf-8")) rid = str(rec["id"]) ref = rec.get("section_ref", "") name = section_name(site, ref) if ref and ref != "backfill" else "" stem = page_stem(site, rid, labeled) jobs.append({"site": site, "rec": rec, "name": name, "html": str(hp), "ck": ck, "sig": sig, "ref": ref, "stem": stem, "page": str(PAGES / f"{stem}.txt") if hp.exists() else None}) for job, row in zip(jobs, runner.map(rw.meta_one, jobs)): prog.advance(task_id) row["release_id"] = job["stem"] out.append(row) fresh_cache[job["ck"]] = {"sig": job["sig"], "ref": job["ref"], "section": job["name"], "row": row} finally: prog.stop() if own_runner: runner.close() save_json(RAW / "_meta_cache.json", fresh_cache) out.sort(key=lambda r: (r["source"], r["release_id"])) META.parent.mkdir(parents=True, exist_ok=True) with META.open("w", encoding="utf-8-sig", newline="") as f: w = csv.DictWriter(f, fieldnames=metamod.FIELDS, extrasaction="ignore") w.writeheader() w.writerows(out) return {"rows": len(out), "reused": reused} def legacy_text(site: str, rec: dict) -> str: """Текст страницы из записи старого harvest (только то, что там сохранено).""" if site == "kinozal": full = rec.get("page_text") or "" tab = (rec.get("release_tab") or "").strip() tech = (rec.get("tech_tab") or "").strip() parts = [full] + [t for t in (tab, tech) if t and t not in full] return "\n\n".join(p for p in parts if p) return f"{rec.get('title') or ''}\n\n{rec.get('post') or ''}" def cmd_import_legacy(src_dir: str) -> dict: """Страницы для уже размеченных раздач из старых jsonl (без браузера). Это урезанный вход — только текст поста (kinozal: ещё и вкладки). Такие страницы перечислены в data/pages_postonly.txt, чтобы оценку на них можно было отделять от полных страниц. """ base = Path(src_dir) labeled = labeled_ids() PAGES.mkdir(parents=True, exist_ok=True) added: list[str] = [] for site in SITES: for cand in (base / "raw" / f"{site}.jsonl", base / f"{site}.jsonl"): if cand.exists(): break else: say(f"{site}: нет {site}.jsonl в {base}") continue n = 0 for line in cand.read_text(encoding="utf-8").splitlines(): if not line.strip(): continue rec = json.loads(line) rid = str(rec.get("id")) if not rec.get("ok", True) or rid not in labeled.get(site, ()): continue stem = page_stem(site, rid, labeled) page = PAGES / f"{stem}.txt" text = legacy_text(site, rec).strip() keep = {k: rec[k] for k in ("id", "source", "page_title", "title", "post", "h1", "release_tab", "tech_tab") if k in rec} if rec.get("page_text"): keep["page_text"] = rec["page_text"][:8000] (RAW / site).mkdir(parents=True, exist_ok=True) (RAW / site / f"{rid}.legacy.json").write_text( json.dumps(keep, ensure_ascii=False), encoding="utf-8") if page.exists() or len(text) < 40: continue page.write_text(text, encoding="utf-8") added.append(stem) n += 1 say(f"{site}: страниц добавлено {n}") if added: prev = POSTONLY.read_text(encoding="utf-8").split() if POSTONLY.exists() else [] POSTONLY.write_text("\n".join(sorted(set(prev) | set(added))) + "\n", encoding="utf-8") return {"added": len(added)} # ---------------------------------------------------------------- статус def read_meta() -> list[dict]: if not META.exists(): return [] with META.open(encoding="utf-8-sig", newline="") as f: return list(csv.DictReader(f)) def coverage_rows() -> list[dict]: labeled = labeled_ids() pages = {p.stem for p in PAGES.glob("*.txt")} if PAGES.exists() else set() postonly = set(POSTONLY.read_text(encoding="utf-8").split()) if POSTONLY.exists() else set() meta_cnt = Counter(r["source"] for r in read_meta()) rows = [] for site in SITES: stems = {page_stem(site, i, labeled) for i in labeled.get(site, set())} with_page = stems & pages rows.append({"site": site, "labeled": len(stems), "pages": len(with_page), "postonly": len(with_page & postonly), "raw": len(stored_ids(site)), "meta": meta_cnt.get(site, 0), "missing": len(stems - pages)}) return rows def cmd_status(detailed: bool = True) -> None: """Покрытие датасета, состав по трекерам (тип, разрешение, кодек, годы, языки), прогресс.""" from rich.columns import Columns from rich.panel import Panel from rich.table import Table ui.console.print(ui.coverage_table(coverage_rows())) if not detailed: return meta = read_meta() if meta: for site in SITES: m = [r for r in meta if r["source"] == site] if not m: continue years = Counter((r["year"][:3] + "0-е") if r["year"] else "—" for r in m) langs = Counter(x for r in m for x in r["audio_langs"].split("+") if x) blocks = [ ui.distribution("Тип", Counter(r["kind"] for r in m)), ui.distribution("Разрешение", Counter(r["resolution"] or "—" for r in m)), ui.distribution("Кодек видео", Counter(r["video_codec"] or "—" for r in m)), ui.distribution("Годы", Counter(dict(sorted(years.items(), reverse=True)))), ui.distribution("Языки дорожек", langs), ] ui.console.print(Panel(Columns(blocks, equal=True, expand=True), title=f"[bold]{site}[/] · {len(m)} релизов", border_style=ui.SITE_STYLE[site])) else: say("meta.csv пока нет: запустите import-legacy или «Собрать pages и labels».") rows = [(s, ref, v) for s, refs in load_state().items() for ref, v in refs.items()] if rows: t = Table(title="Прогресс по разделам", title_justify="left") for c in ("Трекер", "Раздел", "След. стр.", "Статус"): t.add_column(c) for s, ref, v in rows: t.add_row(s, section_name(s, ref)[:70], str(v.get("next_page", 0) + 1), "список закончился" if v.get("finished") else "идёт") ui.console.print(t) # ---------------------------------------------------------------- меню def ask(prompt: str, default: str = "") -> str: return ui.ask(prompt, default).strip() def ask_sites(default: str = "rutracker,nnm,kinozal") -> list[str]: raw = ask("Трекеры через запятую (rutracker, nnm, kinozal)", default) sites = [s.strip() for s in raw.split(",") if s.strip() in SITES] return sites or list(SITES) def wizard_collect() -> None: """Пункт 1: окна -> вход -> сбор -> страницы и метки -> data/processed.""" stop = threading.Event() sites = ask_sites() cmd_windows(sites) say("\nЧто собирать?") say(" n новые раздачи из списков разделов") say(" b страницы для уже размеченных раздач (nnm, kinozal)") say(" nb и то и другое") mode = ask("Режим", "nb") cfg = settings() limit = int(ask("Максимум страниц на трекер за запуск", str(cfg["limit"]))) tabs = clamp_tabs(ask("Вкладок на трекер, параллельно (1–4; больше — быстрее, но выше риск лимитов)", str(cfg["tabs"]))) delay = float(ask("Пауза между раздачами на одну вкладку, с", str(cfg["delay"]))) pages = int(ask("Страниц списка на раздел (для режима n)", str(cfg["pages"]))) if "n" in mode else 0 save_settings({**cfg, "limit": limit, "tabs": tabs, "delay": delay, "pages": pages or cfg["pages"]}) # запоминаем на следующий запуск jobs = {} if "n" in mode: ensure_sections(sites) for s in sites: refs = choose_refs(s) if "n" in mode else "" def job(s=s, refs=refs): if "b" in mode: run_backfill(s, limit, delay, stop) if "n" in mode and not stop.is_set(): run_collect(s, [r.strip() for r in refs.split(",") if r.strip()], pages, limit, delay, stop) jobs[s] = safe(s, job) run_threads(jobs, stop) say("\nСобираю data/pages и data/labels.csv…") cmd_build() say("Пересобираю data/processed…") import retrain if not retrain.prepare(): say("prepare_data завершился с ошибкой") return cmd_status() say("\nДатасет собран. Дальше — пункт 2 (переобучить и протестировать).") # ---------------------------------------------------------------- обучение по ссылкам MANUAL = DATA / "manual_ids.txt" MODELS_TMP = ROOT / "models" / ".compare.json" RE_LINK = { "rutracker": re.compile(r"rutracker\.[a-z]+/forum/viewtopic\.php\?(?:[^\s#]*&)?t=(\d+)", re.I), "nnm": re.compile(r"nnmclub\.[a-z]+/forum/viewtopic\.php\?(?:[^\s#]*&)?t=(\d+)", re.I), "kinozal": re.compile(r"kinozal\.[a-z]+/details\.php\?(?:[^\s#]*&)?id=(\d+)", re.I), } PARSERS = {"kinozal": hv.kz_parse, "rutracker": hv.rt_parse, "nnm": hv.nnm_parse} def parse_link(url: str) -> tuple[str, str] | None: """Ссылка на раздачу -> (трекер, id раздачи).""" for site, rx in RE_LINK.items(): m = rx.search(url or "") if m: return site, m.group(1) return None def store_manual(site: str, rec: dict, rows: list[dict], page_text: str) -> str: """Записывает проверенные вручную метки релиза. Прежние строки этого релиза в labels.csv заменяются, страница кладётся в data/pages, id попадает в data/manual_ids.txt (всегда train, усиление). """ migrate_labels() labeled = labeled_ids() stem = page_stem(site, str(rec["id"]), labeled) keep = [r for r in read_labels() if not (r["source"] == site and r["release_id"] == stem)] for r in rows: r["release_id"] = stem r["studio_source"] = "manual" PAGES.mkdir(parents=True, exist_ok=True) (PAGES / f"{stem}.txt").write_text(page_text, encoding="utf-8") with LABELS.open("w", encoding="utf-8-sig", newline="") as f: w = csv.DictWriter(f, fieldnames=hv.FIELDS, extrasaction="ignore") w.writeheader() w.writerows(keep + rows) ids = set(MANUAL.read_text(encoding="utf-8").split()) if MANUAL.exists() else set() MANUAL.write_text("\n".join(sorted(ids | {stem})) + "\n", encoding="utf-8") return stem def _rows_table(rows: list[dict]): from rich import box from rich.table import Table t = Table(box=box.SIMPLE_HEAD, expand=True, pad_edge=False) for col in ("#", "студия", "тип", "18+", "кодек", "кбит/с", "строка"): t.add_column(col, overflow="ellipsis", no_wrap=col != "строка") for i, r in enumerate(rows, 1): t.add_row(str(i), r["studio"], r["translation_type"], "[red]18+[/red]" if int(r["age_18"] or 0) else "", r["codec"], r["kbps"], (r["raw"] or "")[:70]) return t def edit_rows(rows: list[dict]) -> list[dict]: """Ручная правка меток: «2 18», «3 d», «1 t проф. двухголосый», «1 s HDrezka».""" while True: ui.console.print(_rows_table(rows)) say("Правка: «N 18» — переключить 18+, «N d» — удалить, «N t <тип>», «N s <студия>»; " "Enter — сохранить") cmd = ask("Правка", "").strip() if not cmd: return rows parts = cmd.split(None, 2) try: r = rows[int(parts[0]) - 1] op = parts[1] except (ValueError, IndexError): say("не понял; пример: «2 18»") continue arg = parts[2] if len(parts) > 2 else "" if op == "18": now = not int(r["age_18"] or 0) r["age_18"] = int(now) r["note"] = hv.mark_18(r["note"], True) if now else (r["note"] or "").replace("18+", "").strip() elif op == "d": rows.remove(r) elif op == "t" and arg: r["translation_type"] = arg elif op == "s" and arg: r["studio"] = arg else: say("не понял; пример: «2 18»") def wizard_links() -> None: """Обучение по конкретным ссылкам: страница -> метки -> ручная правка -> labels.csv.""" from src.html_extract import html_to_text say("Вставьте ссылки на раздачи (rutracker / nnm-club / kinozal), по одной в строке.") say("Пустая строка — закончить. Страницы берутся через ваш Chrome (нужен вход).") links: list[tuple[str, str]] = [] while True: line = input("> ").strip() if not line: break got = parse_link(line) if got is None: say(f"не похоже на ссылку на раздачу: {line}") elif got not in links: links.append(got) if not links: return saved: list[str] = [] for site in SITES: todo = [rid for s, rid in links if s == site] if not todo: continue cdp = connect_site(site) try: for rid in todo: say(f"скачиваю {rid}…", site) rec = fetch_release(cdp, site, rid) if not rec.get("ok") or not has_body(site, rec): say(f"{rid}: страница не загрузилась ({rec.get('reason', 'пусто')})", site) continue html = rec.pop("_html", "") rows, _ = PARSERS[site](rec) say(f"{rec.get('page_title') or rec.get('title') or rid}", site) rows = edit_rows(rows) if rows else [] if not rows: say("меток нет — релиз не сохранён", site) continue save_release(site, {**rec, "_html": html}) saved.append(store_manual(site, rec, rows, html_to_text(html))) finally: cdp.close() if saved: cmd_build() say(f"\nДобавлено релизов: {len(saved)} ({', '.join(saved)}). Они всегда идут в train и " "повторяются при обучении; модель переучите пунктом «Переобучить».") if ask("Переобучить сейчас? (y/N)", "n").lower() == "y": wizard_retrain() class Stages: """План шагов с нумерацией, временем каждого шага и итоговой таблицей.""" def __init__(self, titles: list[str]) -> None: self.titles, self.done, self.t0 = titles, [], time.time() self.cur: tuple[int, float] | None = None def plan(self) -> None: ui.console.print("[bold]План:[/bold]") for i, t in enumerate(self.titles, 1): ui.console.print(f" [dim]{i}.[/dim] {t}") ui.console.print() def start(self, title: str) -> None: self.finish() i = len(self.done) + 1 ui.console.rule(f"[bold]шаг {i}/{len(self.titles)}[/bold] {title} " f"[dim](всего прошло {ui.fmt_time(time.time() - self.t0)})[/dim]") self.cur = (i, time.time(), title) def finish(self, ok: bool = True) -> None: if self.cur: self.done.append((self.cur[2], time.time() - self.cur[1], ok)) self.cur = None def summary(self) -> None: self.finish(ok=False if self.cur else True) ui.console.print(ui.summary_table("Итоги по шагам", ["шаг", "время", "результат"], [ [t, ui.fmt_time(sec), "ok" if ok else "прервано/ошибка"] for t, sec, ok in self.done] + [["всего", ui.fmt_time(time.time() - self.t0), ""]])) def dataset_stats() -> None: """Состав данных перед обучением: размечено, страницы, 18+, ручные метки.""" rows = read_labels() manual = set(MANUAL.read_text(encoding="utf-8").split()) if MANUAL.exists() else set() by = Counter(r["source"] for r in rows) adult = Counter(r["source"] for r in rows if str(r.get("age_18", "")).strip() in ("1", "True")) rel = {(r["source"], r["release_id"]) for r in rows} relc = Counter(s_ for s_, _ in rel) ui.console.print(ui.summary_table("Данные для обучения", ["трекер", "релизов", "строк-дорожек", "из них 18+"], [ [s_, relc.get(s_, 0), by.get(s_, 0), adult.get(s_, 0)] for s_ in SITES] + [["всего", len(rel), len(rows), sum(adult.values())], ["ручные (пункт 6)", len(manual), "", ""]])) def ask_profiles() -> tuple[list[str], int]: """Какие основы обучать и нужна ли кросс-валидация.""" import retrain say("\nОсновы для обучения (каждая — отдельный запуск в models/runs/, models/best не трогается):") for name, text in retrain.PROFILES.items(): say(f" {name:7s} {text}") raw = ask("Профили через запятую", "base") profiles = [p.strip() for p in raw.split(",") if p.strip() in retrain.PROFILES] or ["base"] say("Кросс-валидация (5 доп. прогонов) нужна только для оценки устойчивости, на итоговую модель не влияет.") folds = 5 if ask("С кросс-валидацией? Дольше примерно в 6 раз (y/N)", "n").lower() == "y" else 0 return profiles, folds def wizard_retrain() -> None: """Пункт 2: (переиндексация) -> prepare -> обучение -> тесты -> сравнение -> подмена весов.""" import retrain say("Данные: сохранённый HTML трекеров можно разобрать заново без захода на сайты.") reindex = ask("Переиндексировать data/raw (страницы, метки, meta) текущими парсерами? (y/N)", "n").lower() == "y" say("\nЧто переобучать?") say(" 1 сегментатор страниц (минуты, GPU)") say(" 2 сегментатор + основные модели (тип, автор, 18+, язык; выбор основы и сравнение с прежней лучшей)") what = ask("Выбор", "1") profiles, folds = ask_profiles() if what == "2" else ([], 0) do_prepare = ask("Пересобрать data/processed из data/labels.csv? (Y/n)", "y").lower() != "n" train_title = f"Обучение моделей: {', '.join(profiles)}" + (" (+ кросс-валидация)" if folds else "") titles = ((["Переиндексация data/raw (страницы, метки, meta)"] if reindex else []) + (["prepare_data: data/labels.csv → data/processed"] if do_prepare else []) + ([train_title] if what == "2" else []) + ["Обучение сегментатора (models/segmenter_v2)", "Тесты проекта", "Сравнение по трекерам (правила / старые веса / новые)"]) ui.clear() st = Stages(titles) st.plan() try: if reindex: st.start(titles[0]) r = cmd_build(rewrite=True, relabel=True) if r.get("relabeled"): say(f"Метки пересчитаны у {r['relabeled']} релизов, копия прежнего файла: data/{r['backup']}") dataset_stats() if do_prepare: st.start("prepare_data: data/labels.csv → data/processed") if not retrain.prepare(): say("prepare_data завершился с ошибкой") return if what == "2": st.start(train_title) if not retrain.train_models(profiles, folds): say("обучение основных моделей завершилось с ошибкой") return new = "models/segmenter_v2" st.start("Обучение сегментатора (models/segmenter_v2)") if not retrain.train_segmenter(new): say("обучение сегментатора завершилось с ошибкой") return st.start("Тесты проекта") tests_ok = retrain.run_tests() dirs = (["models/segmenter"] if (ROOT / "models" / "segmenter").exists() else []) + [new] st.start("Сравнение по трекерам (правила / старые веса / новые)") try: results = retrain.compare(dirs) except RuntimeError as exc: say(str(exc)) return st.finish() if len(dirs) == 2: better, why = retrain.verdict(results, dirs[0], dirs[1]) say(f"\nНовые веса {'лучше' if better else 'НЕ лучше'} старых: {why}") say(f"Тесты проекта: {'ok' if tests_ok else 'ЕСТЬ ПАДЕНИЯ'}") default = "y" if better and tests_ok else "n" else: default = "y" finally: st.summary() if ask("Заменить models/segmenter новыми весами? (y/N)", default).lower() == "y": archive = retrain.promote(new) say(f"Готово. Старые веса: models/{archive}" if archive else "Готово.") else: say(f"Оставлено как есть. Новые веса: {new}") if what == "2": say("\nТеперь выберите основную модель.") wizard_models() def _pct(x: float | None) -> str: return "—" if x is None else f"{x * 100:.1f}" def wizard_models() -> None: """Пункт 7: сравнить все модели на ОДНОЙ текущей тестовой выборке и выбрать, какая станет models/best.""" import retrain from src import runs if not runs.candidates(): say("Нет обученных моделей в models/.") return if not runs.DATASET.exists(): say("Нет data/processed/dataset.json: сначала prepare_data (пункт 2).") return say("Все модели оцениваются заново на текущих тестовых строках (в том входном формате, на котором они " "обучены), поэтому цифры сравнимы. Результат кэшируется.\n") while True: data = retrain.evaluate_models(MODELS_TMP) if not data: say("оценка не удалась") return items = data["items"] rows = [] for i, it in enumerate(items, 1): m, lo, hi = it["metrics"], it["ci"][0], it["ci"][1] tag = (it["name"] + (f" ({it['profile']})" if it["profile"] and it["profile"] != it["name"] else "") + (" *" if it.get("leak_risk") else "")) rows.append([i, tag, it["backbone"].split("/")[-1], it["mb"], f"{_pct(m['joint_exact'])} [{_pct(lo)}–{_pct(hi)}]", _pct(m["type_macro_f1"]), _pct(m["span_exact"]), _pct(m.get("age_f1")), _pct(m.get("lang_accuracy")), runs.verdict(it)]) ui.console.print(ui.summary_table( f"Тестовая выборка: {data['n']} строк (отпечаток {data['fingerprint']}); числа в %", ["№", "модель", "основа", "МБ", "joint exact [95%]", "type F1", "span", "18+ F1", "язык", "против models/best"], rows)) base = next((it for it in items if it["name"] == "best"), None) good = [(it["metrics"]["joint_exact"], i) for i, it in enumerate(items, 1) if it.get("vs") and it["vs"]["p"] <= 0.05 and it["vs"]["fixed"] > it["vs"]["broken"]] if good: say(f"Рекомендую №{max(good)[1]}: значимо лучше текущей на тех же строках.") else: say("Ни одна модель не лучше текущей значимо — менять best нет оснований." if base else "models/best нет: выберите любую.") if any(it.get("leak_risk") for it in items): say("* модель обучена на другом разбиении данных: часть текущих тестовых строк она могла видеть " "при обучении, её цифры завышены, и новая модель сравнивается с ней в невыгодных условиях.") say("joint exact — строка распознана целиком (тип + автор + 18+). Разница в пределах доверительного " "интервала — это шум теста, а не улучшение.") raw = ask("№ модели для models/best (dN — удалить запуск, r — пересчитать, Enter — выйти)", "").lower() if not raw: return if raw == "r": retrain.evaluate_models(MODELS_TMP, refresh=True) continue delete = raw.startswith("d") try: it = items[int(raw.lstrip("d")) - 1] except (ValueError, IndexError): say("нет такого номера") continue if delete: if not runs.deletable(it["dir"]): say("Удалять можно только запуски (models/runs) и best_backup_*.") elif ask(f"Удалить {it['dir']} ({it['mb']} МБ)? (y/N)", "n").lower() == "y": runs.delete(it["dir"]) say("удалено") continue if it["name"] == "best": say("Это и есть текущая models/best.") continue for w in runs.warnings_for(it["dir"]): say("⚠ " + w) if ask(f"Сделать {it['name']} моделью models/best? Прежняя уйдёт в models/best_backup_* (y/N)", "n").lower() == "y": archive = runs.promote(it["dir"]) say(f"Готово. Прежняя модель: models/{archive}. Перезапустите сервис, чтобы он взял новую." if archive else "Готово.") def banner_lines() -> list[str]: rows = coverage_rows() labeled = sum(r["labeled"] for r in rows) pages = sum(r["pages"] for r in rows) models = [d for d in ("best", "segmenter", "segmenter_v2") if (ROOT / "models" / d).exists()] cfg = settings() return [f"Размечено релизов: {labeled} страниц: {pages} meta: {sum(r['meta'] for r in rows)}", f"Модели в models/: {', '.join(models) or 'нет'}", f"Сбор: вкладок на трекер {cfg['tabs']} пауза {cfg['delay']} с лимит {cfg['limit']}", f"Chrome: порт {hv.CHROME_PORT} профиль: {hv.CHROME_PROFILE}"] def settings_screen() -> None: """Пункт «Настройки»: число вкладок (потоков) на трекер и параметры сбора.""" cfg = settings() ui.console.print("[bold]Настройки сбора[/bold] [dim](Enter — оставить текущее)[/dim]\n") ui.console.print("Вкладок на трекер — сколько страниц качается одновременно. " "[dim]1 — медленно и безопасно; 3–4 — в 2–3 раза быстрее, но чаще лимиты и Cloudflare.[/dim]") cfg["tabs"] = clamp_tabs(ask("Вкладок на трекер (1–4)", str(cfg["tabs"]))) cfg["delay"] = float(ask("Пауза между раздачами на одну вкладку, с", str(cfg["delay"]))) cfg["limit"] = int(ask("Лимит страниц на трекер за запуск", str(cfg["limit"]))) ui.console.print("\nРежим выборки: [bold]balanced[/bold] — страницы вразброс по всему разделу, в каждой " "берутся темы из недопредставленных годов и качеств; [bold]sequential[/bold] — подряд " "от новых к старым.") cfg["mode"] = "sequential" if ask("Режим (balanced / sequential)", cfg["mode"]).lower().startswith("s") else "balanced" cfg["per_page"] = max(1, int(ask("Раздач с одной страницы списка (режим balanced)", str(cfg["per_page"])))) cfg["pages"] = int(ask("Страниц списка на раздел (режим sequential)", str(cfg["pages"]))) cfg["workers"] = max(0, int(ask("Процессов для переиндексации (0 — авто: ядра минус 2)", str(cfg["workers"])))) save_settings(cfg) say(f"Сохранено: вкладок {cfg['tabs']}, пауза {cfg['delay']} с, лимит {cfg['limit']}, " f"режим {cfg['mode']}, раздач со страницы {cfg['per_page']}, страниц списка {cfg['pages']}") def menu() -> None: while True: ui.clear() ui.console.print(ui.banner(banner_lines())) ui.console.print(ui.menu_table([ ("1", "Собрать датасет [dim](окна → вход → поиск разделов → сбор → pages/labels/meta)[/dim]"), ("2", "Переобучить и протестировать [dim](обучение → тесты → сравнение → подмена весов)[/dim]"), ("3", "Статистика датасета [dim](по трекерам: тип, качество, годы, языки)[/dim]"), ("4", "Дополнительно [dim](отдельные шаги сбора, импорт, поиск разделов)[/dim]"), ("5", "Настройки [dim](вкладок на трекер, пауза, лимиты)[/dim]"), ("6", "Обучение по ссылкам [dim](добавить раздачи вручную, поправить метки, переобучить)[/dim]"), ("7", "Модели [dim](сравнить обученные на одной выборке, выбрать лучшую, откатиться)[/dim]"), ("0", "Выход")])) choice = ask("Пункт", "").lower() if choice in ("0", "q", ""): return ui.clear() # чистый экран под выбранный режим try: if choice == "1": wizard_collect() elif choice == "2": wizard_retrain() elif choice == "3": cmd_status() elif choice == "4": advanced_menu() continue elif choice == "5": settings_screen() elif choice == "6": wizard_links() elif choice == "7": wizard_models() else: continue except KeyboardInterrupt: say("\nпрервано") except Exception as exc: # noqa: BLE001 say(f"ошибка: {exc}") ui.pause() # дать прочитать результат до очистки def advanced_menu() -> None: stop = threading.Event() while True: ui.clear() ui.console.print(ui.menu_table([ ("1", "Открыть окна браузера и войти в аккаунты"), ("2", "Собрать новые раздачи из списков"), ("3", "Дособрать страницы для уже размеченных раздач"), ("4", "Собрать data/pages, data/labels.csv и data/meta.csv из data/raw"), ("5", "Импортировать старый harvest [dim](быстро, без браузера)[/dim]"), ("6", "Статус"), ("7", "Найти разделы с зарубежными фильмами и сериалами заново"), ("0", "Назад")])) choice = ask("Пункт", "").lower() if choice in ("0", "q", ""): return ui.clear() stop.clear() try: if choice == "1": cmd_windows(ask_sites()) elif choice == "2": sites = ask_sites() pages = int(ask("Страниц списка на раздел", "2")) limit = int(ask("Максимум раздач на трекер", "50")) delay = float(ask("Пауза между раздачами, с", "2")) jobs = {} for s in sites: refs = ask(f"{s}: {SITES[s]['what']}", SITES[s]["refs"]) jobs[s] = safe(s, lambda s=s, refs=refs: run_collect( s, [r.strip() for r in refs.split(",") if r.strip()], pages, limit, delay, stop)) run_threads(jobs, stop) elif choice == "3": sites = ask_sites("kinozal,nnm") limit = int(ask("Максимум страниц на трекер", "100")) delay = float(ask("Пауза между раздачами, с", "2")) run_threads({s: safe(s, lambda s=s: run_backfill(s, limit, delay, stop)) for s in sites}, stop) elif choice == "4": cmd_build(ask("Пересобрать всё заново, игнорируя кэш? (y/N)", "n").lower() == "y") elif choice == "5": default = os.environ.get("DUB_LEGACY_DIR", "") cmd_import_legacy(ask("Каталог старого harvest (содержит raw/*.jsonl)", default)) elif choice == "6": cmd_status() elif choice == "7": ensure_sections(ask_sites(), force=True) else: continue except KeyboardInterrupt: say("\nпрервано") except Exception as exc: # noqa: BLE001 say(f"ошибка: {exc}") ui.pause() def main() -> int: try: sys.stdout.reconfigure(encoding="utf-8", errors="replace") except (AttributeError, ValueError): pass os.environ["DUB_TABS"] = str(settings()["tabs"]) # число вкладок из сохранённых настроек ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) sub = ap.add_subparsers(dest="cmd") p = sub.add_parser("windows") p.add_argument("--sites", default=",".join(SITES)) p = sub.add_parser("collect") p.add_argument("--site", required=True, choices=list(SITES)) p.add_argument("--refs", help="разделы/адреса списков через запятую") p.add_argument("--pages", type=int, default=2) p.add_argument("--limit", type=int, default=50) p.add_argument("--delay", type=float, default=2.0) p.add_argument("--from-page", type=int) p.add_argument("--tabs", type=int, default=3, help="вкладок на трекер (параллельно)") p = sub.add_parser("backfill") p.add_argument("--site", required=True, choices=list(SITES)) p.add_argument("--limit", type=int, default=100) p.add_argument("--delay", type=float, default=2.0) p.add_argument("--tabs", type=int, default=3, help="вкладок на трекер (параллельно)") p = sub.add_parser("repair-html", help="сделать старые сохранённые страницы читаемыми (UTF-8, стили)") p.add_argument("--no-css", action="store_true", help="только кодировка и адрес, без браузера") p = sub.add_parser("build") p.add_argument("--rewrite", action="store_true") p.add_argument("--relabel", action="store_true", help="пересчитать метки размеченных релизов из сохранённого HTML (с бэкапом labels.csv)") p = sub.add_parser("import-legacy") p.add_argument("--dir", required=True) sub.add_parser("status") sub.add_parser("meta") p = sub.add_parser("discover") p.add_argument("--sites", default=",".join(SITES)) p.add_argument("--force", action="store_true", help="искать заново, игнорируя кэш") args = ap.parse_args() stop = threading.Event() if args.cmd is None: menu() elif args.cmd == "discover": ensure_sections([s for s in args.sites.split(",") if s in SITES], args.force) elif args.cmd == "windows": cmd_windows([s for s in args.sites.split(",") if s in SITES]) elif args.cmd in ("collect", "backfill") and args.tabs: os.environ["DUB_TABS"] = str(args.tabs) if args.cmd == "collect": found = ",".join(section_refs(args.site)) refs = [r.strip() for r in (args.refs or found or SITES[args.site]["refs"]).split(",") if r.strip()] run_threads({args.site: safe(args.site, lambda: run_collect( args.site, refs, args.pages, args.limit, args.delay, stop, args.from_page))}, stop) elif args.cmd == "backfill": run_threads({args.site: safe(args.site, lambda: run_backfill( args.site, args.limit, args.delay, stop))}, stop) elif args.cmd == "repair-html": cmd_repair_html(not args.no_css) elif args.cmd == "build": cmd_build(args.rewrite, args.relabel) elif args.cmd == "import-legacy": cmd_import_legacy(args.dir) elif args.cmd == "status": cmd_status() elif args.cmd == "meta": say(f"meta: {cmd_meta()['rows']} строк -> {META.relative_to(ROOT)}") return 0 if __name__ == "__main__": sys.exit(main())