"""Результат на уровне релиза: дорожки, студии, 18+. Публичный API: title_from_html(html) — заголовок раздачи (/<h1>) title_segments(title) — синтетические сегменты из заголовка («ПМ (LostFilm)» -> «Профессиональный (многоголосый закадровый) | LostFilm») aggregate(tracks) — дедуп дорожек + список студий + is_18plus predict_release(predictor, html=..., text=..., url=..., title=...) """ from __future__ import annotations import re from bs4 import BeautifulSoup from .audio18 import apply_audio_18 from .trackinfo import audio_lines_all, match_audio from .episodes import format_ranges, merge_ranges, parse_episodes from .html_extract import (RE_LINE_PREFIX, _split_entries, extract_segments, html_to_text) # аббревиатуры типа озвучки в заголовках раздач -> фраза, понятная модели TITLE_TYPES = { "ПМ": "Профессиональный (многоголосый закадровый)", "ПД": "Профессиональный (двухголосый закадровый)", "ПО": "Профессиональный (одноголосый закадровый)", "ЛМ": "Любительский (многоголосый закадровый)", "ЛД": "Любительский (двухголосый закадровый)", "ЛО": "Любительский (одноголосый закадровый)", "АП": "Авторский (одноголосый закадровый)", "ДБ": "Профессиональный (дублированный)", "Дубляж": "Профессиональный (дублированный)", } _ABBR = "|".join(sorted(map(re.escape, TITLE_TYPES), key=len, reverse=True)) # «ПМ (LostFilm)», «ДБ (Red Head Sound)»; студия обязательна — без неё нечего добавлять RE_TITLE_TRACK = re.compile(rf"(?<![\w])({_ABBR})\s*\(([^()]{{2,60}})\)") def title_from_html(html: str) -> str: soup = BeautifulSoup(html, "lxml") for sel in ("h1", "title"): node = soup.select_one(sel) if node and node.get_text(strip=True): return re.sub(r"\s+", " ", node.get_text(" ", strip=True)) return "" # заголовок раздачи в текстовом дампе: «Название / ... / 2025 / ... » RE_TITLE_LINE = re.compile(r"^.{3,300}?\s/\s.*\b(?:19|20)\d\d\b.*\s/\s.*$") def title_from_text(text: str) -> str: """Первая строка текстового дампа, похожая на заголовок раздачи. Дальше по странице идут списки других раздач с такими же строками, поэтому берётся только первое совпадение. """ for line in (text or "").splitlines(): line = line.strip() if RE_TITLE_LINE.match(line): return line return "" def title_segments(title: str) -> list[str]: """Дорожки, перечисленные в заголовке раздачи в формате «ТИП (студия)».""" out = [] for abbr, studio in RE_TITLE_TRACK.findall(title or ""): for name in re.split(r"\s*[,;]\s*", studio.strip()): if name: out.append(f"{TITLE_TYPES[abbr]} | {name}") return out UNKNOWN_TYPE = "неизвестно" def _key(t: dict) -> tuple: return (t["translation_type"], (t.get("studio") or t["segment"]).casefold()) def _compatible(a: dict, b: dict) -> bool: """Одна дорожка? Язык и 18+ из строки «Аудио» различают дорожки одной студии (русский и украинский дубляж; обычная и 18+); неизвестное значение подходит любому.""" la, lb = a.get("lang"), b.get("lang") if la and lb and la != lb: return False xa, xb = a.get("_audio18"), b.get("_audio18") return xa is None or xb is None or xa == xb def _absorb(host: dict, t: dict) -> None: host["is_18plus"] = host["is_18plus"] or t["is_18plus"] for key in ("lang", "audio", "_audio18"): if host.get(key) is None and t.get(key) is not None: host[key] = t[key] if t.get("episode_ranges"): both = merge_ranges(host.get("episode_ranges") or [], t["episode_ranges"]) host["episode_ranges"], host["episodes"] = both, format_ranges(both) def aggregate(tracks: list[dict]) -> dict: """Дорожки -> агрегат релиза. Дубли (тип + студия) схлопываются. Если одна и та же дорожка есть и в описании, и в заголовке, остаётся версия из описания (она идёт первой), 18+ объединяется по OR. """ groups: dict[tuple, list[dict]] = {} uniq: list[dict] = [] for t in tracks: host = next((m for m in groups.get(_key(t), []) if _compatible(m, t)), None) if host is not None: _absorb(host, t) else: new = dict(t) groups.setdefault(_key(t), []).append(new) uniq.append(new) # студия с типом «неизвестно» — та же дорожка, что и строка с известным типом # (список «Перевод:» плюс строка «Аудио … серии 1-5»): сливаем, серии переносим known = {} for t in uniq: if t.get("studio") and t["translation_type"] != UNKNOWN_TYPE: known.setdefault(t["studio"].casefold(), t) rest = [] for t in uniq: host = known.get((t.get("studio") or "").casefold()) if host is not None and host is not t and t["translation_type"] == UNKNOWN_TYPE: host["is_18plus"] = host["is_18plus"] or t["is_18plus"] if t.get("episode_ranges"): both = merge_ranges(host.get("episode_ranges") or [], t["episode_ranges"]) host["episode_ranges"], host["episodes"] = both, format_ranges(both) else: rest.append(t) uniq = rest # дорожка без студии, при наличии дорожки того же типа со студией, — как # правило, та же дорожка в виде кодек-строки («AC3 … | MVO, AlexFilm») typed = {t["translation_type"] for t in uniq if t.get("studio")} uniq = [t for t in uniq if t.get("studio") or t["translation_type"] not in typed] studios = [{"studio": t["studio"], "translation_type": t["translation_type"], "lang": t.get("lang"), "is_18plus": t["is_18plus"], "episodes": t.get("episodes")} for t in uniq if t.get("studio")] return { "tracks": uniq, "studios": studios, "translation_types": sorted({t["translation_type"] for t in uniq}), "is_18plus": any(t["is_18plus"] for t in uniq), } def original_info(lines: list[dict]) -> dict | None: """Оригинальная дорожка из строки «Аудио … |Original|» (без студии и комментариев); нет — None.""" for ln in lines: if ln["original"] and not ln["names"] and not ln["commentary"]: return {"lang": ln["lang"] or None, "codec": ln["codec"] or None, "channels": int(ln["channels"]) if ln["channels"] else None, "bitrate_kbps": int(ln["kbps"]) if ln["kbps"] else None, "raw": ln["raw"]} return None def lostfilm_release(title: str | None = None) -> dict: """Ссылка на lostfilm.tv: озвучка известна заранее — MVO LostFilm, страница не нужна.""" track = {"segment": "Профессиональный (многоголосый закадровый) LostFilm", "translation_type": "проф. многоголосый", "type_confidence": 1.0, "studio_span": "LostFilm", "studio": "LostFilm", "studio_confidence": 1.0, "is_18plus": False, "age_confidence": 1.0, "lang": "RUS", "lang_confidence": None, "lang_source": "url", "role": "перевод", "official": False, "audio": None, "source": "url", "episode_ranges": [], "episodes": None} result = aggregate([track]) result.update(original=None, series=series_info(title or "", result["tracks"]), title=title or None, segmenter="rules") return result RE_SEASON = re.compile(r"(?i)сезон[ыа]?\s*(\d+(?:\s*[-–]\s*\d+)?)|\bS(\d{1,2})\b(?!E)") def series_info(title: str, tracks: list[dict]) -> dict | None: """Сезон, выпущенные серии и «какая студия какие серии» (для Sonarr). Серии берутся из заголовка («сезон 1, серии 1-8 из 8»), озвучка по сериям — из строк дорожек. Дорожка без указания серий считается покрывающей весь релиз (`episodes: null`); такие студии перечислены в `full_release`. """ ep = parse_episodes(title or "") season = RE_SEASON.search(title or "") by_studio = {t["studio"]: t["episodes"] for t in tracks if t.get("studio") and t.get("episodes")} if not (ep["ranges"] or season or by_studio): return None known = merge_ranges(ep["ranges"], *[t.get("episode_ranges") or [] for t in tracks]) return {"season": re.sub(r"\s+", "", (season.group(1) or season.group(2))) if season else None, "episodes": format_ranges(ep["ranges"]) or format_ranges(known) or None, "total": ep["total"], "by_studio": by_studio, # у студии серии не указаны -> озвучены все серии, вышедшие в релизе "full_release": sorted({t["studio"] for t in tracks if t.get("studio") and not t.get("episodes")})} def _studio_tokens(name: str) -> set[str]: return {w for w in re.findall(r"\w+", name.casefold()) if len(w) >= 4} def same_studio(a: str | None, b: str | None) -> bool: """Одна студия в разном написании: «DVD» ~ «DVD Магия», «НТВ» ~ «НТВ+», «Михаил Яроцкий» ~ «М.Яроцкий/Kyberpunk» (общее слово ≥ 4 букв).""" a, b = (a or "").casefold().strip(), (b or "").casefold().strip() if not a or not b: return False return a in b or b in a or bool(_studio_tokens(a) & _studio_tokens(b)) def merge_extra_tracks(tracks: list[dict], extra: list[dict], source: str, match=lambda a, b: (a or "").casefold() == (b or "").casefold(), adopt_type: bool = False) -> None: """Добавляет дорожки второстепенного источника к дорожкам страницы (на месте). * такая студия уже есть (по `match`) -> пропуск; * есть дорожка того же типа без студии -> студия вписывается в неё (source=page+<source>): это та же дорожка; * иначе -> новая дорожка (source=<source>). Дорожки без студии не берутся: тип без студии страница уже даёт, а в кодек-строках он часто грубее («одноголосый» вместо «авт. одноголосый»). Дорожка с типом «неизвестно» («Аудио Rus: AC3 … | LostFilm») берётся только при `adopt_type` и если у страницы ровно один тип без студии («Перевод: Профессиональный многоголосый»): она получает этот тип. """ bare = {x["translation_type"] for x in tracks if not x.get("studio") and x["translation_type"] != UNKNOWN_TYPE} for t in extra: if not t.get("studio"): continue if t["translation_type"] == UNKNOWN_TYPE: if not (adopt_type and len(bare) == 1): continue t["translation_type"] = next(iter(bare)) if any(match(x.get("studio"), t["studio"]) for x in tracks): continue same = next((x for x in tracks if not x.get("studio") and x["translation_type"] == t["translation_type"]), None) if same is not None: same.update(studio=t["studio"], studio_span=t["studio_span"], studio_confidence=t["studio_confidence"], is_18plus=same["is_18plus"] or t["is_18plus"], source=f"page+{source}") else: t["source"] = source tracks.append(t) def merge_title_tracks(tracks: list[dict], title_tracks: list[dict]) -> None: """Дорожки из заголовка («ПМ (LostFilm)») -> к дорожкам страницы.""" merge_extra_tracks(tracks, title_tracks, "title") # кодек-строка: «Аудио 3: AC3 …», «Аудио Rus: …», «Аудиодорожка 1: …», «Audio #2: …» RE_AUDIO_PREFIX = re.compile(r"(?i)^\s*(?:Аудио\w*|Audio)\s*(?:№|#)?\s*(?:\d+|rus|ukr|eng)?\s*[:.]") # с типом озвучки: «Аудио 3: AC3 … (MVO, HDrezka Studio)»; «English Dubbing» — не тип RE_CODEC_TRACK = re.compile( r"(?i)^\s*(?:Аудио|Audio)\s*(?:№|#)?\s*\d*\s*:.*(?:\b(?:MVO|DVO|AVO|DUB|VO)\b" r"|дубл|многоголос|двухголос|одноголос|закадр|авторск)") RE_RU_TRACK = re.compile(r"(?i)русск|russian|\brus\b|украин|ukrain|\bukr\b") RE_FOREIGN_TRACK = re.compile( r"(?i)казах|english|англ|француз|немец|немецк|итальян|испан|японск|корейск|китайск" r"|\boriginal\b|оригинал|\beng\b") def codec_ok(line: str) -> bool: """Кодек-строка русской (украинской) дорожки: «Казахский (Dub, TK Qazaqstan)», «English Dubbing», «Original» — нет.""" if RE_RU_TRACK.search(line): return True return not RE_FOREIGN_TRACK.search(line) and bool(RE_CODEC_TRACK.match(line)) def _entries(line: str) -> list[str]: body = RE_LINE_PREFIX.sub("", line, count=1) return [e for e in (re.sub(r"\s+", " ", x).strip() for x in _split_entries(body)) if e] def codec_segments(page_text: str) -> list[str]: """Кодек-строки с типом озвучки («Аудио 2: AC3 … (MVO, LostFilm)») по правилу. Запасной путь к тому, что выбирает сегментатор (`model_segments`): старые веса сегментатора кодек-строки отбрасывают. Они нужны там, где студии есть только в них (kinozal: вкладка «Релиз» против «Перевод:» без студий в «Техданных»), и идут второстепенным источником, см. `merge_extra_tracks`. """ out = [] for line in (page_text or "").splitlines(): if RE_CODEC_TRACK.match(line) and codec_ok(line): out += _entries(line) return out def model_segments(segmenter, page_text: str) -> tuple[list[str], list[str]]: """Дорожки страницы: строки выбирает обученный сегментатор. Возвращает (основные, кодек): строки «Перевод/Озвучка …» и кодек-строки «Аудио N: AC3 … (MVO, LostFilm)». Кодек-строки обычно дублируют основные в другом написании, поэтому они второстепенный источник (`predict_release`); если основных нет (nnm: дорожки только в «Аудио N:») — они и есть дорожки. Нарезка внутри строки («Дублированный, многоголосый») в данных не размечена и пока делается правилами `_split_entries`. """ main, codec = [], [] for line in segmenter.select_lines(page_text): if not RE_AUDIO_PREFIX.match(line): main += _entries(line) elif codec_ok(line): codec += _entries(line) return main, codec def predict_release(predictor, *, html: str | None = None, text: str | None = None, title: str | None = None, segmenter: str = "model") -> dict: """Предсказание для релиза целиком (html или текст страницы). segmenter: "model" — строки-дорожки выбирает обученный сегментатор (если он загружен и нашёл хотя бы одну строку), иначе правила; "rules" — только правила. Кодек-строки («Аудио N: … (MVO, LostFilm)») и заголовок раздачи — дополнительные источники: студии, названные только там, добавляются как дорожки с пометкой source=codec / source=title. """ if (html is None) == (text is None): raise ValueError("нужно ровно одно из: html, text") page_text = html_to_text(html) if html is not None else text segs_page, segs_codec, used = [], [], "rules" seg_model = getattr(predictor, "segmenter", None) if segmenter == "model" and seg_model is not None: segs_page, segs_codec = model_segments(seg_model, page_text) if not segs_page: segs_page, segs_codec = segs_codec, [] used = "model" if segs_page else "rules" if not segs_page: segs_page = extract_segments(page_text) lines = audio_lines_all(page_text) texts, infos = [], [] for seg in segs_page: # дорожка + каждая подходящая строка «Аудио» (обычная и 18+) for info in (match_audio(seg, lines) or [None]): texts.append(seg) infos.append(info) tracks = predictor._predict_batch(texts, infos) if texts else [] for t in tracks: t["source"] = "page" ep = parse_episodes(t["segment"]) t["episode_ranges"] = ep["ranges"] t["episodes"] = format_ranges(ep["ranges"]) or None # кодек-строки — только если страница не назвала студию хотя бы у одной # дорожки (kinozal: «Перевод: многоголосый» + «Аудио 3: … (MVO, LostFilm)»); # иначе они дублируют «Перевод N:» в другом написании и дают ложные студии seen = set(segs_page) segs_codec = list(dict.fromkeys(x for x in segs_codec + codec_segments(page_text) if x not in seen)) if segs_codec and (not tracks or any(not t.get("studio") for t in tracks)): codec = predictor._predict_batch(segs_codec) for t in codec: ep = parse_episodes(t["segment"]) t["episode_ranges"] = ep["ranges"] t["episodes"] = format_ranges(ep["ranges"]) or None merge_extra_tracks(tracks, codec, "codec", match=same_studio, adopt_type=True) if title is None: title = title_from_html(html) if html is not None else title_from_text(text) segs = title_segments(title or "") if segs: merge_title_tracks(tracks, predictor._predict_batch(segs)) result = aggregate(tracks) result["original"] = original_info(lines) apply_audio_18(result, page_text) for t in result["tracks"]: t.pop("_audio18", None) result["series"] = series_info(title or "", result["tracks"]) result["title"] = title or None result["segmenter"] = used return result