"""Zero-Cue Catalog Lookup & Structural Disambiguation Plugin for Armiger & Armiger Ultra. Resolves bare-title ("zero-cue") torrent releases that omit explicit season/episode syntax (e.g. `Title (2024) [1080p].mkv`) and disambiguates multi-item media bundles, interactive software archives, and structured instructional materials before/after neural classification. Supports both zero-deserialization `ZCB3` `mmap` binary catalogs (`armiger.zero_cue_binary.ZCB3MmapCatalog`) and JSON catalog fallback (`validated_zero_cue_catalog.json`), plus schema-validated JSON rule/threshold overrides (`plugin_overrides["zero_cue"]`) from `calibration.json`. """ from __future__ import annotations from collections.abc import Mapping import hashlib import hmac import json import logging import pathlib import re import unicodedata from typing import Any logger = logging.getLogger(__name__) try: from armiger.zero_cue_binary import ZCB3_MAGIC, ZCB3MmapCatalog except ImportError: from zero_cue_binary import ZCB3_MAGIC, ZCB3MmapCatalog # type: ignore try: from armiger.calibration_and_dht import MAX_CUSTOM_REGEX_INPUT_CHARS, safe_regex_search except ImportError: try: from calibration_and_dht import MAX_CUSTOM_REGEX_INPUT_CHARS, safe_regex_search # type: ignore[no-redef] except ImportError: MAX_CUSTOM_REGEX_INPUT_CHARS = 512 def safe_regex_search(pattern: Any, text: str, **kwargs: Any) -> Any: if not pattern: return False return pattern.search(text[:MAX_CUSTOM_REGEX_INPUT_CHARS]) if hasattr(pattern, "search") else re.search(pattern, text[:MAX_CUSTOM_REGEX_INPUT_CHARS]) try: from armiger.audio_language_guard import detect_localized_foreign_title_stem, split_and_score_title_slug except ImportError: try: from audio_language_guard import detect_localized_foreign_title_stem, split_and_score_title_slug # type: ignore[no-redef] except ImportError: def detect_localized_foreign_title_stem(text: str) -> dict[str, Any]: return {"is_foreign_stem": False, "language": None} def split_and_score_title_slug(text: str) -> dict[str, Any]: return {"tokens": [], "is_foreign": False} CATS = [ "movie", "tv_show", "xxx", "music", "comic", "game", "audiobook", "software", "ebook", "course", "unknown", ] IDX_MOVIE = 0 IDX_TV = 1 IDX_XXX = 2 IDX_MUSIC = 3 IDX_COMIC = 4 IDX_GAME = 5 IDX_AUDIOBOOK = 6 IDX_SOFTWARE = 7 IDX_EBOOK = 8 IDX_COURSE = 9 EXPLICIT_TV_RE = re.compile( r"(?i)(?:\bS\d{1,2}E\d{1,3}\b|\bS\d{1,2}\s*-\s*S\d{1,2}\b|\bSeason\s*(?:0?[1-9]|[1-9]\d)\b|\bStagione\s*\d+\b|\bSaison\s*\d+\b|" r"\bTemporada\s*\d+\b|\bStaffel\s*\d+\b|\bComplete\s+Series\b|\bTV\s*Series\b|\bMini\s*Series\b|" r"\bEpisode\s+[1-9]\d*\b|\bEp\.?\s*[1-9]\d*\b|\bEpiz[oó]da\b|\bЭпизод\s*\d+\b|第[一二三四五六七八九十\d]+[季集]|【[^】]*日剧】|更新至\d+[期集]|\s-\s(?:SP|NCOP|NCED)\d*\b|\b\d{3}\s*\[(?:480p|720p|1080p|2160p)\]|" r"\b[1-9]\d{2,3}\s*\((?:WEBRip|WEB-DL|BDRip|DVDRip|1080p|720p|480p)\b|" r"\bFull[\s\._]Event\b|\bPPV\b|\b1-\d{2}\s*\+\s*OVA\b|\btemp\.\d+\.completa\b|" r"\b\d{2}\s*-\s*\d{2,3}\s+VOSTFR\b|\s-\s\d{2,3}(?:v\d)?\s*(?:\[|\(|\b1080p\b|\b720p\b))" ) EXPLICIT_MOVIE_PACK_RE = re.compile( r"(?i)(?:\bMovies?\s*\d{1,2}\s*[-~]\s*\d{1,2}\b|\bMovies?\s+Collection\b|\bTrilogy\b|\bQuadrilogy\b|\bTetralogy\b|" r"\bAnthology\b|\b\d+\s+Movies?\s+Pack\b|\[\d-\d\]\s*\[m?1080p\]\s*\[\d{4}-\d{4}\]|\bThe\s+Movie\b|" r"\(\s*Movies?\s*\)|\[\s*Movies?\s*\]|" r"\bMovie\s+-\s+00\b|\s-\s*Movie\s*(?:\[|\(|V\d)|\bMovie\s+\d{1,2}\s*-|\bMovie:\s|" r"\bGekijouban\b|\bFilm\s+Collection\b|\bIl\s+Film\b|\bLe\s+Film\b|\bDer\s+Film\b|\bKolekcja\b|" r"\bTheatrical\s+Cut\b|\bDirector'?s\s+Cut\b|\bExtra\s+Edition\b)" ) GAME_ARCHIVE_RE = re.compile( r"(?i)(?:\bNTSC-U\b|\bNTSC-J\b|\[linux-x86\]|\[linux-x64\]|\[win64\]|" r"\bInteractive\s+Fiction\b|\bGOG\s+Edition\b|\bLinux\s+Native\b|\((?:Europe|USA|Japan|World)\)\.(?:zip|7z|rar|iso|chd)\b|" r"supertux|super-tux-kart|org\.supertuxkart|\bfreeciv|\bopenttd|\b0ad[-_\b]|\bminetest|\bfreedoom|\bwesnoth|" r"\bsauerbraten|\bxonotic|\bopenarena|\bwarsow|\barx-libertatis|\bopenglad|\bcrispy-doom|\bflightgear|" r"\bendless-sky|\bmegaglest|\bastromenace|\bspeed-dreams|\btorcs\b|\bvdrift|\bstuntrally|\bredeclipse|" r"\bassaultcube|\bteeworlds|\bddnet\b|\bopenra\b|\bwz2100|\bwarzone2100|\bfheroes2|\bvcmi\b|\bopenmw|" r"\bopenxcom|\bcorsixth|\bsimutrans|\blincity|\bunknown-horizons|\bwidelands|\bhedgewars|\bpingus\b|" r"\bfrogatto|\bflare-game|\bvalyriatear|\bnaev\b|\boolite\b|\bpioneerspacesim|\bvegastrike|\burquan|\buqm\b|" r"segasoft|gremlin[\s\._-]+interactive|crosscountrycanada|petepilotti|\.levelset\.)" ) _SOUNDTRACK_AUDIO_EXCLUSION_RE = re.compile( r"(?i)(?:\b(?:ost|soundtrack|original[\s._-]*score|score|flac|mp3|cbr|vbr|320kbps|lossless|discography)\b)" ) ADULT_SITERIP_RE = re.compile( r"(?i)(?:\b(?:xxx|nsfw|18\+|adult[\s._-]+only)\b.*\b(?:1080p|720p|2160p|mp4|mkv|siterip)\b|" r"\bsiterip\b.*\b(?:xxx|18\+|adult|nsfw)\b)" ) EXPLICIT_COMIC_RE = re.compile( r"(?i)(?:\b(?:cbz|cbr|cb7|fumetti)\b|^bd[\.\s_-]+fr\b)" ) COURSE_BOOK_RE = re.compile( r"(?i)(?:\bTestbuch\b|\bKursbuch\b|\bArbeitsbuch\b|\bOpenCourseWare\b|\bVideo\s+Course\b|" r"\bData\s+Science\s+Course\b|\bLanguage\s+Course\b|\bFull\s+Tutorial\b|\bLecture\s+Series\b|\bStudy\s+Guide\b)" ) YEAR_RE = re.compile(r"(? bool: if not tokens or not text: return False lowered = text.lower() for tok in tokens: t = str(tok).strip().lower() if not t: continue pat = r"(? str: """Compute deterministic 64-bit BLAKE2b / explicit-HMAC digest for a catalog lookup key. Offline catalog lookup IDs always use the canonical unkeyed BLAKE2b-64 digest (`person=b"armiger-zc-v2"`) unless an explicit `hmac_key` is passed by the caller; environment variables never implicitly alter offline catalog lookup IDs. """ if hmac_key is not None and hmac_key != b"" and hmac_key != "": k_bytes = hmac_key.encode("utf-8") if isinstance(hmac_key, str) else bytes(hmac_key) return hmac.new(k_bytes, key.encode("utf-8"), hashlib.sha256).hexdigest()[:16] return hashlib.blake2b(key.encode("utf-8"), digest_size=8, person=b"armiger-zc-v2").hexdigest() def norm_title(s: str) -> str: """Normalize a title string to NFKC lowercase alphanumeric tokens.""" if not s: return "" s = unicodedata.normalize("NFKC", s).casefold() s = re.sub(r"[\W_]+", " ", s, flags=re.UNICODE) return " ".join(s.split()) _AUTH_EPISODE_CUT_RE = re.compile( r"(?i)(?:\bS\d{1,2}(?:E\d{1,3}|\b)|\bSeason[\s._-]*\d+\b|\bComplete[\s._-]*Series\b|\s+-\s+\d{1,3}(?:\b|\s*[\[\(]))" ) def extract_authoritative_title(raw_name: str) -> tuple[str, int | None]: """Extract the authoritative clean title and release year without relaxed sequel-number stripping.""" s = raw_name.strip() if not s: return "", None s = re.sub(r"\.(?:mkv|mp4|avi|m2ts|ts|iso|zip|rar|tar\.bz2|epub|pdf|mp3|flac)$", "", s, flags=re.I) s_no_lead = re.sub(r"^(?:\s*[\[\(][^\]\)]{1,45}[\]\)]\s*|[_\s]+)+", "", s) base = s_no_lead if len(norm_title(s_no_lead)) >= 2 else s year_matches = [m for m in YEAR_RE.finditer(base) if m.start() >= 2] year = int(year_matches[0].group(0)) if year_matches else None cut_pos = len(base) if year_matches: cut_pos = min(cut_pos, year_matches[0].start()) m_ns = NOISE_CUT_RE.search(base) if m_ns and m_ns.start() >= 2: cut_pos = min(cut_pos, m_ns.start()) m_ep = _AUTH_EPISODE_CUT_RE.search(base) if m_ep and m_ep.start() >= 2: cut_pos = min(cut_pos, m_ep.start()) stem = base[:cut_pos] stem = re.split(r"[\[\(\|]", stem)[0] clean = norm_title(stem) if not clean: clean = norm_title(re.split(r"[\[\(\|]", base)[0]) return clean, year def parse_clean_candidates(raw_name: str, extra_spans: list[str] | None = None) -> tuple[list[tuple[str, int | None]], int | None]: """Extract ordered candidate `(clean_title, year)` pairs sorted longest/most-specific first.""" s = raw_name.strip() s = re.sub(r"\.(?:mkv|mp4|avi|m2ts|ts|iso|zip|rar|tar\.bz2|epub|pdf|mp3|flac)$", "", s, flags=re.I) s_no_lead = re.sub(r"^(?:\s*[\[\(][^\]\)]{1,45}[\]\)]\s*|[_\s]+)+", "", s) years = [int(m.group(0)) for m in YEAR_RE.finditer(s_no_lead)] year = years[0] if years else None raw_cands: list[tuple[str, int | None]] = [] def add(ct: str, cy: int | None) -> None: ct = ct.strip() if len(ct) >= 4: raw_cands.append((ct, cy)) for base in (s_no_lead, s): m_yr = YEAR_RE.search(base) if m_yr and m_yr.start() >= 3: prefix = base[:m_yr.start()] c1 = norm_title(prefix) add(c1, int(m_yr.group(0))) add(c1, None) for subp in re.split(r"(?i)\s+-\s+|\s+aka\s+", prefix): cs = norm_title(subp) if len(cs) >= 5: add(cs, int(m_yr.group(0))) m_ns = NOISE_CUT_RE.search(base) if m_ns and m_ns.start() >= 3: c2 = norm_title(base[:m_ns.start()]) add(c2, year) add(c2, None) c3_raw = re.split(r"[\[\(\|]", base)[0] c3 = norm_title(c3_raw) add(c3, year) add(c3, None) c4 = norm_title(re.sub(r"(?i)(?:\b(?:the\s+movie|movie|ova|oav|special)\b|\s+-\s+\d{1,3}|\s+\d{1,3})\s*$", "", c3_raw)) add(c4, year) add(c4, None) for grp in re.findall(r"\(([^()]{5,80})\)", s_no_lead) + s_no_lead.split("|")[1:]: if not NOISE_CUT_RE.search(grp) and not YEAR_RE.fullmatch(grp.strip()): for part in re.split(r"[/｜|]", grp): ca = norm_title(part) if len(ca) >= 6: add(ca, year) add(ca, None) if extra_spans: for sp in sorted(extra_spans, key=lambda x: len(x or ""), reverse=True): nsp = norm_title(sp) if len(nsp) >= 5: add(nsp, year) add(nsp, None) slug_res = split_and_score_title_slug(s_no_lead) if slug_res["found_concat_slug"] and len(slug_res["segmented_words"]) >= 2: segmented_title = norm_title(" ".join(slug_res["segmented_words"])) if len(segmented_title) >= 6: add(segmented_title, year) add(segmented_title, None) raw_cands.sort(key=lambda item: (item[1] is not None, len(item[0])), reverse=True) ordered: list[tuple[str, int | None]] = [] seen: set[tuple[str, int | None]] = set() for item in raw_cands: if item not in seen: seen.add(item) ordered.append(item) return ordered, year def lookup_zero_cue_catalog( raw_name: str, catalog: Mapping[str, Any], is_anime: bool = True, extra_spans: list[str] | None = None, ) -> tuple[str | None, str | None]: """Query the local validated catalog (either `ZCB3MmapCatalog` or Python `dict`) in strict specificity order: 1. `(title, year)` exact match in `title_year_map` (472,022 entries) 2. `title` match in `anime_map` when `is_anime=True` and stem length >= 8 (257,019 entries) 3. `title` match in unambiguous `title_only_map` (200,367 entries) """ cands, yr = parse_clean_candidates(raw_name, extra_spans=extra_spans) title_year_map = catalog.get("title_year_map", {}) anime_map = catalog.get("anime_map", {}) title_only_map = catalog.get("title_only_map", {}) for ct, cy in cands: if cy is not None and len(ct) >= 4: key = f"{ct}|{cy}" hkey = hash_catalog_key(key) hit = title_year_map.get(hkey) if hit is not None: return hit, "title_year" if isinstance(title_year_map, dict) and key in title_year_map: return title_year_map[key], "title_year" if is_anime: for ct, _ in cands: if len(ct) >= 8: hct = hash_catalog_key(ct) hit = anime_map.get(hct) if hit is not None: return hit, "anime_catalog" if isinstance(anime_map, dict) and ct in anime_map: return anime_map[ct], "anime_catalog" for ct, _ in cands: if len(ct) >= 14: hct = hash_catalog_key(ct) hit = title_only_map.get(hct) if hit is not None: return hit, "title_only" if isinstance(title_only_map, dict) and ct in title_only_map: return title_only_map[ct], "title_only" if cands and yr is None: ct0 = cands[0][0] if len(ct0) >= 7: hct0 = hash_catalog_key(ct0) hit = title_only_map.get(hct0) if hit is not None: return hit, "title_only" if isinstance(title_only_map, dict) and ct0 in title_only_map: return title_only_map[ct0], "title_only" return None, None _AUDIOBOOK_FILE_TREE_RE = re.compile( r"(?i)(?:_librivox\b|(?:^|[/\\\s._\-\[\(])chapter[\s._-]*\d{1,3}(?:[\s._-]+[a-z0-9_]+)?\.(?:mp3|m4b|ogg|flac|aac)\b|(?:^|[/\\\s._\-\[\(])part[\s._-]*\d{1,2}(?:[\s._-]+[a-z0-9_]+)?\.(?:m4b|mp3)\b|(?:^|[/\\\s._\-\[\(])\d{2,3}[\s._-]+chapter[\s._-]*\d+)" ) _COURSE_FILE_TREE_RE = re.compile( r"(?i)(?:(?:^|[/\\\s._\-\[\(])(?:module|lesson|lecture|section|week|unit)[\s._-]*\d{1,2}(?=$|[/\\\s._\-\]\)]).*\.(?:mp4|mkv|pdf|zip|srt|html)\b|(?:^|[/\\\s._\-\[\(])\d{1,2}[\s._-]+\d{1,2}[\s._-]+[a-z].*\.(?:mp4|srt)\b)" ) _COMIC_ALLOWED_COMPANION_EXTS = frozenset({ "cbr", "cbz", "cb7", "cbt", "cba", "pdf", "epub", "mobi", "azw3", "jpg", "jpeg", "png", "webp", "gif", "txt", "nfo", "sfv", "md5", "xml", "json", }) _AUDIOBOOK_ALLOWED_COMPANION_EXTS = _COMIC_ALLOWED_COMPANION_EXTS | frozenset({ "m4b", "mp3", "ogg", "flac", "m4a", "opus", "aac", "wav", "cue", "m3u", "pls", "log", }) _COURSE_ALLOWED_COMPANION_EXTS = _AUDIOBOOK_ALLOWED_COMPANION_EXTS | frozenset({ "mp4", "mkv", "avi", "mov", "webm", "ts", "srt", "vtt", "ass", "ssa", "sub", "html", "htm", "md", "csv", "py", "ipynb", "js", "css", "sql", "sh", "zip", "rar", "7z", }) def _is_primary_book_payload( raw_name: str, file_exts: set[str], target_exts: set[str], has_tv_cue: bool, has_movie_cue: bool, allowed_exts: frozenset[str] = _COMIC_ALLOWED_COMPANION_EXTS, ) -> bool: return bool( (file_exts & target_exts) and file_exts <= allowed_exts and not GAME_ARCHIVE_RE.search(raw_name) and not _V3_GAME_STRONG_RE.search(raw_name) and not _V3_SOFTWARE_RE.search(raw_name) and not has_tv_cue and not has_movie_cue ) def _extract_file_surface(files: list[Any] | tuple[Any, ...] | None) -> tuple[str, set[str], int]: """Return `(joined_paths_sample, lowercase_extensions_set, file_count)` from a torrent file list.""" if not files: return "", set(), 0 paths: list[str] = [] exts: set[str] = set() for f in files: p = str(f.get("path") or f.get("name") or "") if isinstance(f, Mapping) else str(f or "") p = p.strip() if not p: continue if len(paths) < 32: paths.append(p) if "." in p: ext = p.rsplit(".", 1)[-1].lower() if 1 <= len(ext) <= 6: exts.add(ext) return " | ".join(paths), exts, len(files) def apply_structural_and_zero_cue_disambiguation( raw_name: str, probs_category: list[float] | Any, p_adult: float, p_anime: float, catalog: Mapping[str, Any] | None = None, extra_spans: list[str] | None = None, plugin_rules: Mapping[str, Any] | None = None, files: list[Any] | tuple[Any, ...] | None = None, ) -> tuple[int, str]: """Apply deterministic structural rules (Stage 1) and Zero-Cue catalog lookup (Stage 2) to an 11-class category probability vector from Armiger or Armiger Ultra. """ rules = plugin_rules or {} p = [float(x) for x in probs_category] file_surface, file_exts, file_cnt = _extract_file_surface(files) zc_ceiling = float(rules.get("confidence_ceiling", rules.get("zero_cue_confidence_ceiling", 0.88))) anime_min_p = float(rules.get("anime_threshold", rules.get("anime_catalog_min_prob", 0.20))) adult_siterip_floor = float(rules.get("adult_siterip_floor", 0.92)) comic_archive_floor = float(rules.get("comic_archive_floor", 0.90)) movie_pack_sum_floor = float(rules.get("movie_pack_sum_floor", 0.45)) movie_pack_margin = float(rules.get("movie_pack_margin", 0.40)) tv_cue_adult_ceiling = float(rules.get("tv_cue_adult_ceiling", 0.65)) tv_cue_xxx_ceiling = float(rules.get("tv_cue_xxx_ceiling", 0.50)) tv_cue_margin = float(rules.get("tv_cue_margin", 0.40)) tv_cue_floor = float(rules.get("tv_cue_floor", 0.75)) game_archive_adult_ceiling = float(rules.get("game_archive_adult_ceiling", 0.70)) game_archive_floor = float(rules.get("game_archive_floor", 0.88)) course_book_floor = float(rules.get("course_book_floor", 0.88)) has_tv_cue = bool( EXPLICIT_TV_RE.search(raw_name) or _match_safe_tokens(raw_name, list(rules.get("extra_tv_tokens") or [])) or safe_regex_search(rules.get("explicit_tv_regex"), raw_name, field_name="zero_cue.explicit_tv_regex") ) has_movie_cue = bool( EXPLICIT_MOVIE_PACK_RE.search(raw_name) or _match_safe_tokens(raw_name, list(rules.get("extra_movie_tokens") or [])) or safe_regex_search(rules.get("explicit_movie_pack_regex"), raw_name, field_name="zero_cue.explicit_movie_pack_regex") ) if ADULT_SITERIP_RE.search(raw_name) or safe_regex_search(rules.get("adult_siterip_regex"), raw_name, field_name="zero_cue.adult_siterip_regex"): p[IDX_XXX] = max(p[IDX_XXX], adult_siterip_floor) return IDX_XXX, "structural_adult_siterip" top_i = int(max(range(len(p)), key=p.__getitem__)) has_primary_comic = _is_primary_book_payload(raw_name, file_exts, {"cbr", "cbz", "cb7"}, has_tv_cue, has_movie_cue) if ( EXPLICIT_COMIC_RE.search(raw_name) or has_primary_comic or safe_regex_search(rules.get("explicit_comic_regex"), raw_name, field_name="zero_cue.explicit_comic_regex") ) and not re.search(r"(?i)how\s+to\s+draw", raw_name): if top_i in (IDX_EBOOK, IDX_COMIC) or p[top_i] < comic_archive_floor: p[IDX_COMIC] = max(p[IDX_COMIC], comic_archive_floor) return IDX_COMIC, "structural_comic_archive" has_primary_ab = ( _is_primary_book_payload(raw_name, file_exts, {"m4b", "mp3", "ogg", "flac"}, has_tv_cue, has_movie_cue, _AUDIOBOOK_ALLOWED_COMPANION_EXTS) and ("m4b" in file_exts or (file_surface and _AUDIOBOOK_FILE_TREE_RE.search(file_surface))) and (top_i in (IDX_AUDIOBOOK, IDX_EBOOK, IDX_MUSIC) or p[top_i] < 0.90) ) if _AUDIOBOOK_FILE_TREE_RE.search(raw_name) or has_primary_ab: p[IDX_AUDIOBOOK] = max(p[IDX_AUDIOBOOK], 0.90) return IDX_AUDIOBOOK, "structural_audiobook_files" has_primary_course = ( bool(file_surface and _COURSE_FILE_TREE_RE.search(file_surface) and file_cnt >= 3) and _is_primary_book_payload(raw_name, file_exts, {"mp4", "mkv", "pdf", "zip", "srt", "html"}, has_tv_cue, has_movie_cue, _COURSE_ALLOWED_COMPANION_EXTS) and (top_i in (IDX_COURSE, IDX_EBOOK, IDX_TV, IDX_MOVIE) or p[top_i] < course_book_floor) ) if ( COURSE_BOOK_RE.search(raw_name) or _V3_COURSE_RE.search(raw_name) or has_primary_course or _match_safe_tokens(raw_name, list(rules.get("extra_course_tokens") or [])) or safe_regex_search(rules.get("course_book_regex"), raw_name, field_name="zero_cue.course_book_regex") ): p[IDX_COURSE] = max(p[IDX_COURSE], course_book_floor) return IDX_COURSE, "structural_course_book" if has_movie_cue and not has_tv_cue: if p[IDX_MOVIE] + p[IDX_TV] > movie_pack_sum_floor: p[IDX_MOVIE] = max(p[IDX_MOVIE], p[IDX_TV] + movie_pack_margin) return int(max(range(len(p)), key=p.__getitem__)), "structural_movie_pack" elif has_tv_cue and not has_movie_cue: if p[IDX_XXX] < tv_cue_xxx_ceiling and p_adult < tv_cue_adult_ceiling: p[IDX_TV] = max(p[IDX_TV], p[IDX_MOVIE] + tv_cue_margin, tv_cue_floor) return int(max(range(len(p)), key=p.__getitem__)), "structural_tv_cue" if ( (GAME_ARCHIVE_RE.search(raw_name) or _match_safe_tokens(raw_name, list(rules.get("extra_game_tokens") or [])) or safe_regex_search(rules.get("game_archive_regex"), raw_name, field_name="zero_cue.game_archive_regex")) and not _SOUNDTRACK_AUDIO_EXCLUSION_RE.search(raw_name) and p_adult < game_archive_adult_ceiling ): p[IDX_GAME] = max(p[IDX_GAME], game_archive_floor) return IDX_GAME, "structural_game_archive" pred_idx = int(max(range(len(p)), key=p.__getitem__)) if catalog is not None and pred_idx in (IDX_MOVIE, IDX_TV) and not has_tv_cue and not has_movie_cue: if max(p[IDX_MOVIE], p[IDX_TV]) < zc_ceiling: cat_hit, src_kind = lookup_zero_cue_catalog(raw_name, catalog, is_anime=bool(p_anime >= anime_min_p), extra_spans=extra_spans) if cat_hit in ("movie", "tv_show") and src_kind is not None: return (IDX_MOVIE if cat_hit == "movie" else IDX_TV), f"zero_cue_{src_kind}" return pred_idx, "neural_backbone" class ZeroCueCatalogPlugin: """High-level wrapper for loading `validated_zero_cue_catalog.zcb3` (via zero-deserialization `mmap`) with automatic fallback to `validated_zero_cue_catalog.json`. """ _DEFAULT_CATALOG_DIR: pathlib.Path = pathlib.Path(__file__).resolve().parent def __init__( self, catalog_path: str | pathlib.Path | Mapping[str, Any] | None = None, *, plugin_rules: Mapping[str, Any] | None = None, ) -> None: self.plugin_rules: dict[str, Any] = dict(plugin_rules or {}) self.catalog_mode: str = "empty" self.last_load_warning: str | None = None if isinstance(catalog_path, Mapping): self.catalog_path = None self.catalog: Mapping[str, Any] = catalog_path self.catalog_mode = "mmap_zcb3" if isinstance(catalog_path, ZCB3MmapCatalog) else "memory_dict" return if catalog_path is None: default_zcb3 = self._DEFAULT_CATALOG_DIR / "validated_zero_cue_catalog.zcb3" default_json = self._DEFAULT_CATALOG_DIR / "validated_zero_cue_catalog.json" catalog_path = default_zcb3 if default_zcb3.exists() else default_json self.catalog_path = pathlib.Path(catalog_path) self.catalog = self._load_catalog_with_fallback(self.catalog_path) def _load_catalog_with_fallback(self, path: pathlib.Path) -> Mapping[str, Any]: # 1. If path points to a .zcb3 / .zcb3.zlib / .zcb3.lzma file (or starts with b"ZCB3" magic) if path.exists(): suffix_lower = "".join(path.suffixes).lower() is_zcb3_ext = ".zcb3" in suffix_lower if not is_zcb3_ext: try: with path.open("rb") as f_peek: is_zcb3_ext = f_peek.read(4) == ZCB3_MAGIC except OSError: is_zcb3_ext = False if is_zcb3_ext: try: zcat = ZCB3MmapCatalog(path) self.catalog_mode = zcat.storage_mode return zcat except (ValueError, OSError) as exc: self.last_load_warning = f"zcb3_load_failed:{path.name}:{exc}" logger.warning( "Failed to load ZCB3 binary catalog %s (%s); attempting JSON fallback", path, exc, ) base_stem = path.name.split(".zcb3")[0] if ".zcb3" in path.name.lower() else path.stem sibling_json = path.with_name(f"{base_stem}.json") if sibling_json.exists() and sibling_json != path: self.catalog_mode = "json_fallback" return json.loads(sibling_json.read_text(encoding="utf-8")) raise # 2. If path points to .json, prefer a sibling .zcb3 file if present if path.suffix.lower() == ".json": sibling_zcb3 = path.with_suffix(".zcb3") if sibling_zcb3.exists(): try: zcat = ZCB3MmapCatalog(sibling_zcb3) self.catalog_mode = zcat.storage_mode return zcat except (ValueError, OSError) as exc: self.last_load_warning = f"zcb3_load_failed:{sibling_zcb3.name}:{exc}" logger.warning( "Failed to load sibling ZCB3 binary catalog %s (%s); falling back to JSON %s", sibling_zcb3, exc, path, ) # 3. Automatic fallback to JSON catalog self.catalog_mode = "json_fallback" return json.loads(path.read_text(encoding="utf-8")) # 4. If requested .zcb3 does not exist, fall back to sibling .json if path.suffix.lower() == ".zcb3": sibling_json = path.with_suffix(".json") if sibling_json.exists(): self.catalog_mode = "json_fallback" return json.loads(sibling_json.read_text(encoding="utf-8")) self.catalog_mode = "empty" return {} def close(self) -> None: if hasattr(self.catalog, "close"): self.catalog.close() def resolve( self, raw_name: str, probs_category: list[float] | Any, p_adult: float = 0.0, p_anime: float = 0.0, extra_spans: list[str] | None = None, plugin_rules: Mapping[str, Any] | None = None, files: list[Any] | tuple[Any, ...] | None = None, ) -> dict[str, Any]: eff_rules = plugin_rules if plugin_rules is not None else self.plugin_rules idx, source = apply_structural_and_zero_cue_disambiguation( raw_name=raw_name, probs_category=probs_category, p_adult=p_adult, p_anime=p_anime, catalog=self.catalog, extra_spans=extra_spans, plugin_rules=eff_rules, files=files, ) if source == "neural_backbone": idx_v3, src_v3, probs_v3 = resolve_category_v3( raw_name=raw_name, probs_category=probs_category, p_adult=p_adult, p_anime=p_anime, catalog=self.catalog, extra_spans=extra_spans, plugin_rules=eff_rules, files=files, ) if src_v3 != "v3_neural_calibrated" or eff_rules: return { "category_index": idx_v3, "category": CATS[idx_v3], "resolution_source": src_v3, "calibrated_probs": probs_v3, "catalog_mode": self.catalog_mode, } return {"category_index": idx, "category": CATS[idx], "resolution_source": source, "catalog_mode": self.catalog_mode} # --- Armiger v3.0 Structural Disambiguator & Prior-Corrected Logit Calibrator --- _V3_COMIC_RE = re.compile( r"(?i)(?:\.(?:cbr|cbz|cb7)$|\b(?:cbr|cbz|cb7)\b|\([0-9]{4}\)\s*\(digital\)|\bc2c\b.*\.(?:cbr|cbz|rar|zip)$)" ) _V3_AUDIOBOOK_RE = re.compile(r"(?i)(?:\.m4b$|\baudiobook\b|\bunabridged\b|\babridged\b|_librivox\b)") _V3_EBOOK_RE = re.compile(r"(?i)(?:\.(?:epub|mobi|azw3)$|\b(?:epub|mobi|azw3)\b)") _V3_GAME_STRONG_RE = re.compile( r"(?i)(?:\.(?:nsp|xci|cia|3ds|vpk|wbfs)$|\b(?:fitgirl|dodi[\s._-]*repack|elamigos|kaos[\s._-]*repack|" r"codex|skidrow|plaza|flt|rune|tenoke|razor1911|reloaded|cpy|empress|rg[\s._-]*mechanics)\b|" r"\bgog[\s._-]+(?:repack|edition|rip|v\d)\b|(?:^|[\s._\-\[\(])plus[\s._-]+\d+[\s._-]+trainer(?=$|[\s._\-\]\)])|" r"\bbuild[\s._-]+\d{5,9}\b|\bps[2345]\b.*\b(?:usa|eur|jpn|pkg|iso)\b|\bxbox\s*360\b|\bnintendo\s+switch\b|\brpg\s+maker\b)" ) _V3_COURSE_RE = re.compile( r"(?i)(?:(?:^|[\s._\-\[\(])(?:udemy|coursera|pluralsight|lynda|linkedin[\s._-]*learning|skillshare|masterclass|" r"the[\s._-]+great[\s._-]+courses|ttc[\s._-]+video|packt|oreilly|cbt[\s._-]+nuggets|ine[\s._-]+ccie|" r"laracasts|frontend[\s._-]+masters|egghead)(?=$|[\s._\-\]\)])|" r"(?:^|[\s._\-\[\(])(?:testbuch|kursbuch|arbeitsbuch|lehrbuch|übungsbuch)(?=$|[\s._\-\]\)])|" r"(?:^|[\s._\-\[\(])(?:ucberkeley|mit)[\s._-]+(?:webcast|opencourseware)(?=$|[\s._\-\]\)])|" r"(?:^|[\s._\-\[\(])(?:lecture[\s._-]+series|video[\s._-]+training|video[\s._-]+course|full[\s._-]+tutorial|study[\s._-]+guide|open\s*course\s*ware)(?=$|[\s._\-\]\)]))" ) _V3_SOFTWARE_RE = re.compile( r"(?i)(?:\b(?:macos|macosx|mac\s+os\s+x|win32|win64|x86_64|pre-activated|crack|keygen)\b.*\b(?:v?\d+\.\d+|20[12]\d)\b|" r"\b(?:multilingual|portable|bootable)\b.*\b(?:v\d+\.\d+\.\d+|x64|win64|macos|dmg|setup|installer)\b|" r"\b(?:adobe|autodesk|jetbrains|microsoft\s+office|windows\s+(?:10|11|server)|" r"vmware|parallels|ableton|fl\s+studio|native\s+instruments|kontakt|waves\s+complete)\b)" ) _V3_MUSIC_RE = re.compile( r"(?i)(?:\.(?:flac|alac|ape|wv|dsf|dff)$|\b(?:discography|discografia|ost|soundtrack|" r"lossless|24bit|16bit|44\.1khz|96khz|192khz|cd-rip|vinyl[\s._-]*rip|web-flac|mp3[\s._-]*320)\b)" ) _V3_CONCERT_OR_DOC_VIDEO_RE = re.compile( r"(?i)(?:^|[\s._\-\[\(])(?:philharmoniker|symphony[\s._-]+orchestra|live[\s._-]+in[\s._-]+[a-z]+|live[\s._-]+at[\s._-]+[a-z]+)(?=$|[\s._\-\]\)]).*\b(?:1080i|1080p|720p|bluray|mkv|ts)\b" ) _V3_ADULT_STUDIO_RE = re.compile( r"(?i)(?:\b(?:xxx|siterip|onlyfans|manyvids|brazzers|naughtyamerica|realitykings|mofos|" r"bangbros|vixen|tushy|blacked|deeper|slayed|metart|sexart|hegre|twistys|playboy|penthouse|" r"jav|fc2[\s._-]*ppv|uncensored[\s._-]*leak|1pondo|caribbeancom|heyzo|pacopacomama)\b|" r"\b[a-z]{2,5}-\d{3,5}\b.*\b(?:mp4|mkv|wmv|avi|1080p|720p|fhd|hd)\b)" ) def _boost_and_normalize_v3(p: list[float], idx: int, min_prob: float) -> list[float]: target = min(0.99, max(float(p[idx]), float(min_prob))) rem_budget = max(0.0, 1.0 - target) other_sum = sum(float(p[i]) for i in range(len(p)) if i != idx) out: list[float] = [] for i in range(len(p)): if i == idx: out.append(round(target, 6)) elif other_sum > 0.0: out.append(round(rem_budget * (float(p[i]) / other_sum), 6)) else: out.append(round(rem_budget / max(1, len(p) - 1), 6)) diff = round(1.0 - sum(out), 6) out[idx] = round(max(0.0, min(1.0, out[idx] + diff)), 6) return out DEFAULT_V3_CATEGORY_PRIOR_BOOST: tuple[float, ...] = ( 0.0, 0.0, 0.0, 0.05, 0.15, 0.12, 0.15, 0.12, 0.15, 0.18, -0.35 ) def calibrate_category_logits_v3( logits_or_probs: list[float] | Any, temperature: float = 0.85, prior_boost: list[float] | tuple[float, ...] | None = None, ) -> list[float]: """Apply prior-shift log-odds correction and temperature scaling to 11-class category probabilities.""" import math arr = [float(x) for x in logits_or_probs] s = sum(arr) is_prob = all(x >= 0.0 for x in arr) and (0.98 <= s <= 1.02) logits = [math.log(max(x, 1e-12)) for x in arr] if is_prob else list(arr) eff_boost = list(prior_boost) if prior_boost is not None and len(prior_boost) >= 11 else list(DEFAULT_V3_CATEGORY_PRIOR_BOOST) scaled = [(logits[i] + float(eff_boost[i])) / max(float(temperature), 0.1) for i in range(min(len(logits), 11))] m = max(scaled) exps = [math.exp(x - m) for x in scaled] z = sum(exps) or 1.0 out = [round(x / z, 6) for x in exps] if out: diff = round(1.0 - sum(out), 6) best_i = max(range(len(out)), key=out.__getitem__) out[best_i] = round(max(0.0, min(1.0, out[best_i] + diff)), 6) return out def resolve_category_v3( raw_name: str, probs_category: list[float] | Any, p_adult: float = 0.0, p_anime: float = 0.0, catalog: Mapping[str, Any] | None = None, extra_spans: list[str] | None = None, plugin_rules: Mapping[str, Any] | None = None, files: list[Any] | tuple[Any, ...] | None = None, ) -> tuple[int, str, list[float]]: """Armiger v3.0 11-class category resolver combining calibrated neural probabilities, structural/format rules, and zero-cue BLAKE2b-64 catalog lookup (`ZCB3` or JSON). """ rules = plugin_rules or {} file_surface, file_exts, file_cnt = _extract_file_surface(files) cat_temp = float(rules.get("category_temperature", 0.85)) cat_prior = rules.get("category_prior_boost") zc_ceiling = float(rules.get("zero_cue_confidence_ceiling", rules.get("confidence_ceiling", 0.93))) anime_min_p = float(rules.get("anime_catalog_min_prob", rules.get("anime_threshold", 0.15))) adult_siterip_floor = float(rules.get("adult_siterip_floor", 0.94)) comic_archive_floor = float(rules.get("comic_archive_floor", 0.92)) movie_pack_sum_floor = float(rules.get("movie_pack_sum_floor", 0.40)) movie_pack_margin = float(rules.get("movie_pack_margin", 0.40)) tv_cue_adult_ceiling = float(rules.get("tv_cue_adult_ceiling", 0.65)) tv_cue_xxx_ceiling = float(rules.get("tv_cue_xxx_ceiling", 0.50)) tv_cue_margin = float(rules.get("tv_cue_margin", 0.40)) tv_cue_floor = float(rules.get("tv_cue_floor", 0.85)) game_archive_adult_ceiling = float(rules.get("game_archive_adult_ceiling", 0.70)) game_archive_floor = float(rules.get("game_archive_floor", 0.90)) course_book_floor = float(rules.get("course_book_floor", 0.91)) p = calibrate_category_logits_v3(probs_category, temperature=cat_temp, prior_boost=cat_prior) s = raw_name.strip() has_tv_cue = bool( EXPLICIT_TV_RE.search(s) or _match_safe_tokens(s, list(rules.get("extra_tv_tokens") or [])) or safe_regex_search(rules.get("explicit_tv_regex"), s, field_name="zero_cue.explicit_tv_regex") ) has_movie_cue = bool( EXPLICIT_MOVIE_PACK_RE.search(s) or _match_safe_tokens(s, list(rules.get("extra_movie_tokens") or [])) or safe_regex_search(rules.get("explicit_movie_pack_regex"), s, field_name="zero_cue.explicit_movie_pack_regex") ) has_video_ext_or_codec = bool( re.search(r"(?i)(?:\.(?:mkv|mp4|avi|m2ts|ts|wmv|mov)$|\b(?:1080p|720p|2160p|480p|bluray|web-dl|webrip|bdrip|dvdrip|x264|x265|hevc|h264|h265|xvid)\b)", s) or (file_exts & {"mkv", "mp4", "avi", "m2ts", "ts", "wmv", "mov"}) ) if ( ADULT_SITERIP_RE.search(s) or safe_regex_search(rules.get("adult_siterip_regex"), s, field_name="zero_cue.adult_siterip_regex") or (_V3_ADULT_STUDIO_RE.search(s) and p_adult >= 0.25) ): return IDX_XXX, "v3_structural_adult", _boost_and_normalize_v3(p, IDX_XXX, adult_siterip_floor) top_i = int(max(range(len(p)), key=p.__getitem__)) has_primary_comic = _is_primary_book_payload(s, file_exts, {"cbr", "cbz", "cb7"}, has_tv_cue, has_movie_cue) and ( top_i in (IDX_EBOOK, IDX_COMIC) or p[top_i] < comic_archive_floor ) if ( _V3_COMIC_RE.search(s) or EXPLICIT_COMIC_RE.search(s) or has_primary_comic or safe_regex_search(rules.get("explicit_comic_regex"), s, field_name="zero_cue.explicit_comic_regex") ) and not re.search(r"(?i)how\s+to\s+draw", s): return IDX_COMIC, "v3_structural_comic", _boost_and_normalize_v3(p, IDX_COMIC, comic_archive_floor) has_primary_ab_v3 = ( _is_primary_book_payload(s, file_exts, {"m4b", "mp3", "ogg", "flac"}, has_tv_cue, has_movie_cue, _AUDIOBOOK_ALLOWED_COMPANION_EXTS) and ("m4b" in file_exts or (file_surface and _AUDIOBOOK_FILE_TREE_RE.search(file_surface))) and (top_i in (IDX_AUDIOBOOK, IDX_EBOOK, IDX_MUSIC) or p[top_i] < 0.92) ) if _V3_AUDIOBOOK_RE.search(s) or has_primary_ab_v3: return IDX_AUDIOBOOK, "v3_structural_audiobook", _boost_and_normalize_v3(p, IDX_AUDIOBOOK, 0.92) has_primary_course_v3 = ( bool(file_surface and _COURSE_FILE_TREE_RE.search(file_surface) and file_cnt >= 3) and _is_primary_book_payload(s, file_exts, {"mp4", "mkv", "pdf", "zip", "srt", "html"}, has_tv_cue, has_movie_cue, _COURSE_ALLOWED_COMPANION_EXTS) and (top_i in (IDX_COURSE, IDX_EBOOK, IDX_TV, IDX_MOVIE) or p[top_i] < course_book_floor) ) if ( _V3_COURSE_RE.search(s) or COURSE_BOOK_RE.search(s) or has_primary_course_v3 or _match_safe_tokens(s, list(rules.get("extra_course_tokens") or [])) or safe_regex_search(rules.get("course_book_regex"), s, field_name="zero_cue.course_book_regex") ): return IDX_COURSE, "v3_structural_course", _boost_and_normalize_v3(p, IDX_COURSE, course_book_floor) has_primary_ebook = _is_primary_book_payload(s, file_exts, {"epub", "mobi", "azw3"}, has_tv_cue, has_movie_cue) and ( top_i in (IDX_EBOOK, IDX_COMIC) or p[top_i] < 0.91 ) if (_V3_EBOOK_RE.search(s) or has_primary_ebook) and not _V3_COMIC_RE.search(s) and not _V3_AUDIOBOOK_RE.search(s): return IDX_EBOOK, "v3_structural_ebook", _boost_and_normalize_v3(p, IDX_EBOOK, 0.91) if ( GAME_ARCHIVE_RE.search(s) or _V3_GAME_STRONG_RE.search(s) or (file_exts & {"nsp", "xci", "cia", "3ds", "vpk", "wbfs"}) or _match_safe_tokens(s, list(rules.get("extra_game_tokens") or [])) or safe_regex_search(rules.get("game_archive_regex"), s, field_name="zero_cue.game_archive_regex") ) and p_adult < game_archive_adult_ceiling and not has_tv_cue: if not _SOUNDTRACK_AUDIO_EXCLUSION_RE.search(s) and not (file_exts and file_exts <= {"mp3", "flac", "ogg", "m4a", "wav", "cue", "log", "jpg", "png", "txt", "nfo"}): return IDX_GAME, "v3_structural_game", _boost_and_normalize_v3(p, IDX_GAME, game_archive_floor) if _V3_CONCERT_OR_DOC_VIDEO_RE.search(s) and not has_tv_cue: return IDX_MOVIE, "v3_structural_video_override", _boost_and_normalize_v3(p, IDX_MOVIE, 0.88) if _V3_MUSIC_RE.search(s) and not has_video_ext_or_codec and not has_tv_cue: return IDX_MUSIC, "v3_structural_music", _boost_and_normalize_v3(p, IDX_MUSIC, 0.90) if _V3_SOFTWARE_RE.search(s) and not has_video_ext_or_codec and not has_tv_cue and p_adult < 0.50: if not (file_exts and file_exts <= {"epub", "pdf", "mobi", "azw3", "mp3", "flac", "m4b", "txt", "nfo"}): return IDX_SOFTWARE, "v3_structural_software", _boost_and_normalize_v3(p, IDX_SOFTWARE, 0.89) if has_movie_cue and not has_tv_cue: if p[IDX_MOVIE] + p[IDX_TV] > movie_pack_sum_floor: floor_m = min(0.99, max(p[IDX_MOVIE], p[IDX_TV] + movie_pack_margin, 0.85)) return IDX_MOVIE, "v3_structural_movie_pack", _boost_and_normalize_v3(p, IDX_MOVIE, floor_m) elif has_tv_cue and not has_movie_cue: if p[IDX_XXX] < tv_cue_xxx_ceiling and p_adult < tv_cue_adult_ceiling: floor_t = min(0.99, max(p[IDX_TV], p[IDX_MOVIE] + tv_cue_margin, tv_cue_floor)) return IDX_TV, "v3_structural_tv_cue", _boost_and_normalize_v3(p, IDX_TV, floor_t) pred_idx = int(max(range(len(p)), key=p.__getitem__)) if catalog is not None and pred_idx in (IDX_MOVIE, IDX_TV) and not has_tv_cue and not has_movie_cue: if max(p[IDX_MOVIE], p[IDX_TV]) < zc_ceiling: cat_hit, src_kind = lookup_zero_cue_catalog( s, catalog, is_anime=bool(p_anime >= anime_min_p), extra_spans=extra_spans ) if cat_hit in ("movie", "tv_show") and src_kind is not None: idx_hit = IDX_MOVIE if cat_hit == "movie" else IDX_TV return idx_hit, f"v3_zero_cue_{src_kind}", _boost_and_normalize_v3(p, idx_hit, 0.90) return pred_idx, "v3_neural_calibrated", p