"""Standalone Armiger v2.1 / Armiger Ultra ONNX Featurizer & Inference Runtime (`armiger_inference.py`). Implements exact feature parity with the Armiger v2.1 training pipeline (`sift/prepare.py` + `enrich_side21_v2` 36-dim `stats` vector + `vocab.json` 4,096-extension lookup + CRC32 digit-folded path-word hashing) with zero external codebase dependencies beyond `numpy` and `onnxruntime`. All release-group and publisher token tables are stored as 8-byte BLAKE2b-64 hex digests so zero copyrighted or scene group strings appear in plaintext. """ from __future__ import annotations import hashlib import json import math import os from pathlib import Path import re from typing import Any, Dict, List, Mapping, Optional, Sequence, Tuple import zlib try: import numpy as np except ImportError: np = None # type: ignore[assignment] SPAN_BIO_CLASSES: Tuple[str, ...] = ( "O", "B-TITLE", "I-TITLE", "B-YEAR", "I-YEAR", "B-EP", "I-EP", ) NEURAL_HEAD_CLASSES: Dict[str, List[str]] = { "category": [ "movie", "tv_show", "xxx", "music", "comic", "game", "audiobook", "software", "ebook", "course", "unknown", ], "adult": ["non_adult", "adult"], "junk": ["clean", "junk"], "anime": ["non_anime", "anime"], "english_subs": ["no_eng_subs", "has_eng_subs", "unknown"], "lang": ["english", "non_english", "multi_or_unknown"], "owner_scope": ["allowed_type", "excluded"], "title_language": ["english_title", "non_english_title"], "span_bio": list(SPAN_BIO_CLASSES), } _VOCAB_EXT_MAP: Optional[Dict[str, int]] = None _PREP_KIND_EXTS = { 0: frozenset("mkv mp4 avi ts m2ts mts vob wmv mpg mpeg mov flv rmvb rm m4v divx webm 3gp asf f4v ogv mxf m2v bik".split()), 1: frozenset("mp3 flac m4a m4b aac ogg opus wav wma ape alac aif aiff aifc ac3 dts mka mpc wv shn mid midi dff dsf".split()), 2: frozenset("zip rar 7z gz tgz bz2 xz zst lz lzma tar cab iso img dmg mdf mds nrg wim par par2 z".split()), 3: frozenset("exe msi apk ipa dll bat cmd sh vbs ps1 com scr jar deb rpm appimage run lnk".split()), 4: frozenset("jpg jpeg png gif webp bmp tif tiff avif jxl heic heif psd ai svg ico tga cr2 nef arw dng raw eps".split()), 5: frozenset("pdf epub mobi azw azw3 djvu djv fb2 txt rtf doc docx odt xls xlsx ppt pptx csv chm cbz cbr cb7 html htm mht mhtml md tex lit".split()), 6: frozenset("srt ass ssa sub idx vtt sup smi sbv ttml dfxp usf scc stl mks".split()), } _PREP_EXT_TO_KIND = {e: k for k, es in _PREP_KIND_EXTS.items() for e in es} _PREP_SPLIT_ARC_RE = re.compile(r"[rsz]\d{2,3}|\d{3}") _PREP_WORD_RE = re.compile(r"[^\W_]+") _PREP_DIGIT_RE = re.compile(r"\d") _PREP_SAMPLE_RE = re.compile(r"(? str: return hashlib.blake2b(s.strip().lower().encode("utf-8"), digest_size=8).hexdigest() def _extract_token_hashes(text: str) -> frozenset[str]: toks = [m.group(0).lower() for m in _TOKEN_SCAN_RE.finditer(text)] out = {_h64(t) for t in toks} for i in range(len(toks) - 1): out.add(_h64(f"{toks[i]} {toks[i + 1]}")) return frozenset(out) def _get_vocab_ext_map(vocab_path: Optional[Path] = None) -> Dict[str, int]: global _VOCAB_EXT_MAP if _VOCAB_EXT_MAP is not None and len(_VOCAB_EXT_MAP) > 64: return _VOCAB_EXT_MAP vpath = vocab_path if (vocab_path and vocab_path.exists()) else Path(__file__).resolve().with_name("vocab.json") if not vpath.exists() and os.environ.get("ARMIGER_MODEL_DIR"): vpath = Path(os.environ["ARMIGER_MODEL_DIR"]) / "vocab.json" if vpath.exists(): data = json.loads(vpath.read_text(encoding="utf-8")) _VOCAB_EXT_MAP = {e: i + 1 for i, e in enumerate(data.get("exts", []))} else: _VOCAB_EXT_MAP = { "mp3": 2, "mp4": 3, "mkv": 4, "avi": 7, "srt": 10, "zip": 12, "rar": 17, "exe": 23, "flac": 24, "png": 25, "dll": 31, "m4b": 53, "epub": 81, "cbz": 96, "lnk": 150, "scr": 310, } return _VOCAB_EXT_MAP def _hash_ext(ext: str) -> int: if not ext: return 4096 return _get_vocab_ext_map().get(ext.lower(), 4096) def _hash_word(w: str) -> int: if not w: return 0 folded = _PREP_DIGIT_RE.sub("0", w.lower()) return (zlib.crc32(folded.encode("utf-8")) % 32767) + 1 def _log2_size_bucket(sz: int) -> int: return min(47, max(int(sz), 1).bit_length() - 1) def _prep_split_path(path: str) -> Tuple[str, str, str, int]: p_norm = path.replace("\\", "/") parent, _, base = p_norm.rpartition("/") i = base.rfind(".") ext = base[i + 1:].lower() if i > 0 else "" if len(ext) <= 10 and ext.isascii() and ext.isalnum(): stem = base[:i] else: ext, stem = "", base return parent.rpartition("/")[2], stem, ext, p_norm.count("/") def _prep_path_words(parent: str, stem: str) -> List[int]: p_words = [_hash_word(w) for w in _PREP_WORD_RE.findall(parent.lower())][-2:] if parent else [] s_words = [_hash_word(w) for w in _PREP_WORD_RE.findall(stem.lower())][-(4 - len(p_words)):] res = p_words + s_words while len(res) < 4: res.append(0) return res[:4] def encode_torrent_features( row: Mapping[str, Any], max_name_len: int = 256, max_files: int = 64, ) -> Dict[str, Any]: """Encode a torrent metadata dictionary into the exact ONNX tensor inputs for Armiger v2.1.""" name = str(row.get("name") or row.get("title") or "") b_name = name.encode("utf-8", "replace") if len(b_name) > max_name_len: b_name = b_name[:max_name_len].decode("utf-8", "ignore").encode("utf-8") raw_bytes = list(b_name) if b_name else [32] raw_files = row.get("files") or [] total_size = int(row.get("size") or 0) all_items: List[Tuple[str, int]] = [] for f in raw_files: if isinstance(f, Mapping): p = str(f.get("path") or f.get("p") or f.get("name") or "") sz = int(f.get("size") or f.get("s") or 0) else: p = str(f or "") sz = 0 if p and "/.____padding_file/" not in p and not p.startswith(".____padding_file"): all_items.append((p, sz)) all_items.sort(key=lambda x: -x[1]) items: List[Tuple[str, int]] = all_items[:max_files] if not items and (row.get("ext") or row.get("extension") or "." in name): items.append((name, total_size)) ext_map = _get_vocab_ext_map() exts: List[int] = [] sizes: List[int] = [] depths: List[int] = [] words_list: List[List[int]] = [] f_exts_str: List[str] = [] kbytes = [0.0] * 8 kcount = [0] * 8 has_sample = False for path, sz in items: parent, stem, ext, depth = _prep_split_path(path) f_exts_str.append(ext) exts.append(ext_map.get(ext, 4096)) sizes.append(_log2_size_bucket(sz)) depths.append(min(depth, 7)) words_list.append(_prep_path_words(parent, stem)) k = _PREP_EXT_TO_KIND.get(ext) if k is None: k = 2 if _PREP_SPLIT_ARC_RE.fullmatch(ext) else 7 kbytes[k] += max(sz, 0) kcount[k] += 1 has_sample = has_sample or (_PREP_SAMPLE_RE.search(path) is not None) tot_k = sum(kbytes) weights = kbytes if tot_k > 0 else [float(c) for c in kcount] wsum = sum(weights) count = max(int(row.get("file_count") or 0), len(all_items), len(items)) biggest = max((max(s, 0) for _, s in items), default=0) max_share = (biggest / tot_k if tot_k > 0 else (1.0 / len(items) if items else 0.0)) base_stats = [ math.log1p(max(count, 0)), math.log1p(max(total_size if total_size > 0 else int(tot_k), 0)), ] base_stats += [w / wsum if wsum > 0 else 0.0 for w in weights] base_stats += [max_share, float(has_sample)] side = [0.0] * 24 n_f = float(max(1, len(items))) m_grp = re.match(r"^\[([^\[\]]{2,30})\]", name) grp_lc = m_grp.group(1).strip().lower() if m_grp else "" grp_h = _h64(grp_lc) if grp_lc else "" title_tok_h = _extract_token_hashes(name) has_sub_grp = (grp_h in _SUB_ONLY_GRP_H) or bool(title_tok_h & _SUB_ONLY_GRP_H) has_dub_grp = (grp_h in _DUAL_DUB_GRP_H) or bool(title_tok_h & _DUAL_DUB_GRP_H) has_game_grp = bool(title_tok_h & _GAME_GRP_H) has_sw_grp = bool(title_tok_h & _SW_GRP_H) has_book_grp = bool(title_tok_h & _BOOK_COMIC_GRP_H) if items: ebk_set = {"epub", "mobi", "azw3", "azw", "djvu", "fb2", "lit", "iba"} pdf_set = {"pdf"} m4b_set = {"m4b", "aa", "aax"} com_set = {"cbz", "cbr", "cb7", "cbt"} con_set = {"nsp", "xci", "xcz", "nsz", "wbfs", "rvz", "gcz", "cia", "3ds", "vpk", "ps4", "fpkg"} app_set = {"dmg", "pkg", "apk", "ipa", "deb", "rpm", "appimage"} exe_dll_set = {"exe", "msi", "bat", "cmd", "sh", "com", "dll"} iso_set = {"iso", "bin", "cue", "mdf", "mds", "nrg", "img"} arc_set = {"rar", "zip", "7z", "tar", "gz", "xz", "bz2", "001"} htm_set = {"html", "htm", "url", "mht", "torrent"} scr_set = {"scr", "lnk", "vbs", "hta", "pif", "ps1", "wsf", "reg"} vid_set = {"mkv", "mp4", "avi", "ts", "m2ts", "wmv", "mov", "webm", "flv", "mpg", "mpeg", "m4v", "vob", "ogv", "rmvb", "3gp", "asf", "divx"} bd_game_benign = {"vcdiff", "apk", "bdmv", "clpi", "mpls", "ifo", "bup", "vob", "npa", "vpk", "pck", "umd", "pak", "xp3", "rgss3a", "wolf", "nsp", "xci", "wbfs"} side[0] = sum(1 for e in f_exts_str if e in ebk_set) / n_f side[1] = sum(1 for e in f_exts_str if e in pdf_set) / n_f side[2] = sum(1 for e in f_exts_str if e in m4b_set) / n_f side[3] = sum(1 for e in f_exts_str if e in com_set) / n_f side[4] = sum(1 for e in f_exts_str if e in con_set) / n_f side[5] = sum(1 for e in f_exts_str if e in app_set) / n_f vc = bool(_SIDE_VID_CLAIM_RE.search(name) or _SIDE_ANIME_KW_RE.search(name) or has_sub_grp or has_dub_grp) has_any_vid_inf = any(e in vid_set for e in f_exts_str) has_renpy_structure = ( not vc and not has_any_vid_inf and any( p_i.replace("\\", "/").lower().endswith((".rpa", ".rpyb")) and int(sz_i or 0) >= 65536 for p_i, sz_i in items ) ) renpy_say_re = re.compile(r"(?i)(?:^|/)lib/(?:py\d+-)?windows-[^/]+/say\.vbs$") n_real_exec_dll = 0 has_real_exe = False has_dll = False has_scr_trap = False has_oversized_lnk = False has_root_launcher = False has_unconditional_trap_ext = False has_any_scr_ext = False vid_depths: List[int] = [] for (p_i, _), e_str, s_b, d_i, w_i in zip(items, f_exts_str, sizes, depths, words_list): p_posix = p_i.replace("\\", "/") if has_renpy_structure and renpy_say_re.search(p_posix): continue if e_str in scr_set: has_any_scr_ext = True if e_str in {"scr", "vbs", "vbe", "hta", "pif", "wsf", "wsh", "cpl"}: has_unconditional_trap_ext = True p_tok_h = _extract_token_hashes(p_i) is_benign_stub = (e_str == "exe") and (s_b < 15 or 16570 in w_i or bool(p_tok_h & _BENIGN_README_EXE_H)) if e_str in exe_dll_set and not is_benign_stub: n_real_exec_dll += 1 if e_str == "exe" and not is_benign_stub and s_b >= 15: has_real_exe = True if d_i == 0: has_root_launcher = True elif e_str == "dll": has_dll = True if e_str in {"scr", "lnk", "hta", "vbs", "pif", "ps1", "wsf", "cmd", "com", "one", "xll", "xlam"} and s_b >= 12: has_scr_trap = True if d_i == 0: has_root_launcher = True if e_str == "lnk" and s_b >= 15: has_oversized_lnk = True if e_str in vid_set: vid_depths.append(d_i) side[6] = n_real_exec_dll / n_f side[7] = sum(1 for e in f_exts_str if e in iso_set) / n_f side[8] = sum(1 for e in f_exts_str if e in arc_set) / n_f side[9] = sum(1 for e in f_exts_str if e in htm_set) / n_f side[10] = 1.0 if has_any_scr_ext else 0.0 is_sw_or_game = bool( has_game_grp or has_sw_grp or (not vc and _SIDE_SW_APP_TITLE_RE.search(name)) or any(e in bd_game_benign for e in f_exts_str) ) vid_share = base_stats[2] flagged = has_unconditional_trap_ext if not is_sw_or_game: if vc and has_real_exe and has_dll: flagged = True elif vc and len(vid_depths) >= 1 and (has_real_exe or has_root_launcher or has_scr_trap): flagged = True elif (vc or len(vid_depths) > 0) and has_oversized_lnk: flagged = True elif vc and vid_share == 0.0 and (has_real_exe or has_scr_trap): flagged = True elif len(items) <= 2 and any(e in ("zip", "zipx", "rar", "7z") for e in f_exts_str) and vid_share == 0.0: if _SIDE_EP_ZIP_RE.search(name) and not _SIDE_BENIGN_ZIP_RE.search(name): flagged = True elif len(items) <= 2 and any(e in ("iso", "img", "vhd", "vhdx", "udf") for e in f_exts_str) and vid_share == 0.0: name_stem = re.sub(r"\.(?:iso|img|vhd|vhdx|udf|dmg)$", "", name.strip(), flags=re.IGNORECASE) if _SIDE_EP_ZIP_RE.search(name_stem) and not re.search(r"(?i)\b(?:dvd|bd25|bd50|remux|untouched|pal|ntsc|iso)\b", name_stem): flagged = True if flagged or _SIDE_VPN_TRAP_RE.search(name) or (vc and vid_share == 0.0 and (side[6] > 0 or side[9] > 0 or side[10] > 0)): side[11] = 1.0 if vc and vid_share > 0 and 0 < base_stats[1] < 14.22: side[12] = 1.0 ag = bool(_SIDE_ANIME_KW_RE.search(name)) or has_sub_grp or has_dub_grp if ag: side[15] = 1.0 clean_audio_name = _SIDE_DECEPTIVE_SUB_STRIP_RE.sub(" ", name) clean_tok_h = _extract_token_hashes(clean_audio_name) ad = bool(_SIDE_ANIME_DUB_KW_RE.search(clean_audio_name)) or (grp_h in _DUAL_DUB_GRP_H) or bool(clean_tok_h & _DUAL_DUB_GRP_H) if ad: side[16] = 1.0 if (grp_h in _SUB_ONLY_GRP_H) and not ad: side[14] = 1.0 if has_game_grp: side[17] = 1.0 if has_sw_grp and not _SIDE_CJK_RE.search(name): side[18] = 1.0 if _SIDE_COURSE_RE.search(name) and not ag: side[19] = 1.0 if _SIDE_BOOK_COMIC_KW_RE.search(name) or has_book_grp: side[20] = 1.0 lang_str = (row.get("language") or "").strip().lower() if lang_str in ("ja", "ru", "fr", "es", "de", "it", "pt", "zh", "ko", "hi", "pl", "nl", "tr", "ar") or _SIDE_CJK_RE.search(name) or ((grp_h in _SUB_ONLY_GRP_H) and not ad): side[22] = 1.0 elif lang_str in ("en", "eng", "english") or ad: side[21] = 1.0 else: side[23] = 1.0 return { "name_bytes": raw_bytes, "ext": exts if exts else [0], "size": sizes if sizes else [0], "depth": depths if depths else [0], "words": words_list if words_list else [[0, 0, 0, 0]], "stats": base_stats + side, } def build_onnx_batch_inputs( rows: Sequence[Mapping[str, Any]], max_name_len: int = 256, max_files: int = 64, ) -> Dict[str, np.ndarray]: """Collate a sequence of torrent dicts into padded NumPy arrays for ONNX Runtime.""" encoded = [encode_torrent_features(r, max_name_len=max_name_len, max_files=max_files) for r in rows] bsz = len(encoded) max_l = max(len(e["name_bytes"]) for e in encoded) max_f = max(len(e["ext"]) for e in encoded) name_b = np.zeros((bsz, max_l), dtype=np.int64) name_l = np.zeros((bsz,), dtype=np.int64) ext_t = np.zeros((bsz, max_f), dtype=np.int32) size_t = np.zeros((bsz, max_f), dtype=np.int64) depth_t = np.zeros((bsz, max_f), dtype=np.int64) words_t = np.zeros((bsz, max_f, 4), dtype=np.int32) stats_t = np.zeros((bsz, 36), dtype=np.float32) for i, e in enumerate(encoded): lb = len(e["name_bytes"]) lf = len(e["ext"]) name_b[i, :lb] = e["name_bytes"] name_l[i] = lb ext_t[i, :lf] = e["ext"] size_t[i, :lf] = e["size"] depth_t[i, :lf] = e["depth"] words_t[i, :lf, :] = e["words"] stats_t[i, :] = e["stats"] return { "name_bytes": name_b, "name_len": name_l, "ext": ext_t, "size": size_t, "depth": depth_t, "words": words_t, "stats": stats_t, } def _softmax(logits: np.ndarray) -> np.ndarray: shifted = logits - np.max(logits, axis=-1, keepdims=True) exps = np.exp(shifted) return exps / np.sum(exps, axis=-1, keepdims=True) class ArmigerStandaloneONNX: """Self-contained ONNX Runtime classifier for `NorviTech/armiger` and `NorviTech/armiger-ultra`.""" def __init__( self, onnx_path: Optional[str | Path] = None, vocab_path: Optional[str | Path] = None, calibration_path: Optional[str | Path] = None, ) -> None: import onnxruntime as ort base = Path(__file__).resolve().parent if onnx_path is None: for cand in ("armiger.onnx", "armiger.int8.onnx", "armiger_ultra.fp32.onnx"): if (base / cand).exists(): onnx_path = base / cand break if onnx_path is None: raise FileNotFoundError("Could not find armiger.onnx or armiger_ultra.fp32.onnx") self.onnx_path = Path(onnx_path) default_vocab = base / "vocab.json" if not default_vocab.exists() and self.onnx_path.with_name("vocab.json").exists(): default_vocab = self.onnx_path.with_name("vocab.json") self.vocab_path = Path(vocab_path) if vocab_path else default_vocab _get_vocab_ext_map(self.vocab_path) self.calibration_path = Path(calibration_path) if calibration_path else (base / "calibration.json") self.session = ort.InferenceSession(str(self.onnx_path), providers=["CPUExecutionProvider"]) self.head_names = [o.name for o in self.session.get_outputs()] @staticmethod def _apply_v3_post_neural_guard(row: Mapping[str, Any], item_res: Dict[str, Any]) -> Dict[str, Any]: return _apply_v3_post_neural_guard(row, item_res) def predict_batch(self, rows: Sequence[Mapping[str, Any]]) -> List[Dict[str, Any]]: feed = build_onnx_batch_inputs(rows) raw_outs = self.session.run(self.head_names, feed) logits_by_head = dict(zip(self.head_names, raw_outs)) results: List[Dict[str, Any]] = [] for i in range(len(rows)): item_res: Dict[str, Any] = {"heads": {}, "probabilities": {}} for h, arr in logits_by_head.items(): vec = arr[i].astype(np.float64) if vec.ndim == 2: # 3D batch tensor (e.g. span_bio [B, 256, 7]) -> 2D per-item [256, 7] shifted = vec - np.max(vec, axis=-1, keepdims=True) exps = np.exp(shifted) probs_2d = exps / np.maximum(1e-12, np.sum(exps, axis=-1, keepdims=True)) raw_nm = str(rows[i].get("name") or rows[i].get("title") or "") try: from armiger.byte_span_extractor import decode_bio_logits_viterbi, decode_bio_tags_to_spans except ImportError: from byte_span_extractor import decode_bio_logits_viterbi, decode_bio_tags_to_spans raw_b = raw_nm.encode("utf-8", errors="replace")[: vec.shape[0]] tag_ids = [int(x) for x in decode_bio_logits_viterbi(vec, raw_b).tolist()] classes = NEURAL_HEAD_CLASSES.get( h, list(SPAN_BIO_CLASSES), ) tag_labels = [classes[idx_c] if idx_c < len(classes) else str(idx_c) for idx_c in tag_ids] item_res["heads"][h] = tag_labels mean_probs = np.mean(probs_2d, axis=0) item_res["probabilities"][h] = { classes[k]: round(float(mean_probs[k]), 6) for k in range(min(len(classes), len(mean_probs))) } try: item_res["span_bio"] = decode_bio_tags_to_spans(raw_nm, tag_ids) except Exception: item_res["span_bio"] = {"tag_ids": tag_ids, "tags": tag_labels} continue probs = _softmax(vec) classes = NEURAL_HEAD_CLASSES.get(h, [str(k) for k in range(len(probs))]) idx_c = int(np.argmax(probs)) item_res["heads"][h] = classes[idx_c] if idx_c < len(classes) else str(idx_c) item_res["probabilities"][h] = { classes[k]: round(float(probs[k]), 6) for k in range(min(len(classes), len(probs))) } self._apply_v3_post_neural_guard(rows[i], item_res) p_junk = item_res["probabilities"].get("junk", {}).get("junk", 0.0) p_adult = item_res["probabilities"].get("adult", {}).get("adult", 0.0) p_excl = ( 0.05 if item_res["heads"].get("owner_scope") == "allowed_type" else item_res["probabilities"].get("owner_scope", {}).get("excluded", 0.0) ) item_res["p_junk"] = p_junk item_res["mondrian_gates"] = { "adult_flagged": bool(p_adult >= 0.5451), "junk_flagged": bool(p_junk >= 0.1010), "scope_excluded": bool(p_excl >= 0.4770), "passed_all": bool(p_adult < 0.5451 and p_junk < 0.1010 and p_excl < 0.4770), } results.append(item_res) return results def predict(self, row: Mapping[str, Any]) -> Dict[str, Any]: return self.predict_batch([row])[0] ArmigerONNXClassifier = ArmigerStandaloneONNX _V3_COMIC_H = frozenset(['12b3f45e5ceb0c73', '4cecc3908b751508', '560dd703f14b3d39', 'd287a406b53d326d', 'f474f6ed1c868f32']) _V3_GAME_H = frozenset(['052648bca960e902', '49460c3a7ea1cb2a', '65a3e07111977e18', 'aeeacf8b695dceea', 'f9dd689e8c6d1cd8', 'fd75f166789586c3']) _V3_OPEN_MOVIE_H = frozenset(['0937eb719d20dd6f', '2e7c54d248a0ef1d', '4d0377a6e2563411', '710ab19297bbed0c', '82d861962967c61e', '97a97017505610c7', '9a8994e8f326412f', 'a320c9ca18122845', 'a82434472cd6c6fe', 'f5f35a2f45111a8e']) _V3_TV_SERIES_H = frozenset(['230aead54d362cda', '295b881a619e7cb5', '7e7f87fd43987c67', '7f9e0a2bfd52770c', '8ab671ca32f5d5cf', '9f7c0f5280ee75ec', 'c8564e035c0b75c1', 'e2b18ebba4070e03', 'f593bd44736fe5fe', 'fc1a22469a3f145d']) _V3_ANIME_POS_H = frozenset(['07ea6d84cdf8fd18', '14dc96e9b279ef41', '3f51640da58e3b42', '485b2fbd270bd1ed', '5fc43af228179edd', '7054ce850cad5308', 'a11c618db0e83f96', 'ad7e17f965645161', 'c46009e7f1b3c3af']) _V3_ANIME_MOVIE_H = frozenset(['14dc96e9b279ef41', '5fc43af228179edd', 'ad7e17f965645161', 'c46009e7f1b3c3af']) _V3_ANIME_NEG_H = frozenset(['0937eb719d20dd6f', '207ffe8422c5dccc', '43dc423c0f35206b', '4d0377a6e2563411', '52812423d8d44670', '710ab19297bbed0c', '7aa1e3930f50174a', '97a97017505610c7', 'a320c9ca18122845', 'a82434472cd6c6fe', 'b1444d914e2baca2', 'b249fb33aabf5c18']) _V3_NON_EN_AUDIO_H = frozenset(['07ea6d84cdf8fd18', '12c3d50880c16874', '16b7d611049fe70e', '207ffe8422c5dccc', '34924d18a0b6ed8b', '3f51640da58e3b42', '43dc423c0f35206b', '485b2fbd270bd1ed', '56e0d06e84a655c0', '5f1570e0f9eaacb8', '5fc43af228179edd', '69ca97bb99f505b5', '7054ce850cad5308', '75af437002546cd4', '7aa1e3930f50174a', '8406725ab8ae5378', '900a19dd73bdd4ba', '951625ff1ac61302', 'a11c618db0e83f96', 'a7ddaf4c9ddad893', 'a8d66ae84e768552', 'ad7e17f965645161', 'b1444d914e2baca2', 'b249fb33aabf5c18', 'dbfb8d1d57178337', 'e05fb6bafd7b47f1', 'eb5ac7a33ae81cf2', 'f2a4dad05c81ba2e', 'f6241861fb0c5d79']) _V3_NON_EN_TITLE_H = frozenset(['207ffe8422c5dccc', '43dc423c0f35206b', '5f1570e0f9eaacb8', '7aa1e3930f50174a', '951625ff1ac61302', 'b1444d914e2baca2', 'f6241861fb0c5d79']) _V3_SUBWORD_RE = re.compile(r"[A-Za-z0-9]+") def _extract_v3_hashes(text: str) -> frozenset[str]: base = set(_extract_token_hashes(text)) sub = [m.group(0).lower() for m in _V3_SUBWORD_RE.finditer(text)] for t in sub: base.add(_h64(t)) for i in range(len(sub) - 1): base.add(_h64(f"{sub[i]} {sub[i + 1]}")) return frozenset(base) def _apply_v3_post_neural_guard(row: Mapping[str, Any], item_res: Dict[str, Any]) -> Dict[str, Any]: """Apply Armiger v3.0 post-neural structural, cross-head, and subtitle-trap disambiguation.""" name = str(row.get("name") or row.get("title") or "").strip() raw_files = row.get("files") or [] fpaths: List[str] = [] for f in raw_files: p = str(f.get("path") or f.get("p") or f.get("name") or "") if isinstance(f, Mapping) else str(f or "") if p and "/.____padding_file/" not in p and not p.startswith(".____padding_file"): fpaths.append(p) all_text = name + " " + " ".join(fpaths[:32]) name_h = _extract_v3_hashes(name) all_h = _extract_v3_hashes(all_text) is_nested = "heads" in item_res and isinstance(item_res["heads"], dict) if is_nested: heads = item_res["heads"] probs = item_res.get("probabilities") or {} else: heads = { k: (v.get("label") if isinstance(v, Mapping) else str(v)) for k, v in item_res.items() if isinstance(v, (Mapping, str)) } probs = { k: {v.get("label", ""): float(v.get("confidence", 0.5))} for k, v in item_res.items() if isinstance(v, Mapping) and "label" in v } p_adult = float(probs.get("adult", {}).get("adult", 0.0)) # 1. Cross-head adult <-> category consistency: never predict 'xxx' when adult_flagged is False explicit_xxx = bool(re.search(r"(?i)\b(?:xxx|siterip|onlyfans|jav|fc2[-_ ]?ppv|18\+|nsfw)\b", name)) if heads.get("category") == "xxx" and p_adult < 0.5451 and not explicit_xxx: cat_probs = dict(probs.get("category") or {}) cat_probs["xxx"] = 0.0 cat_probs["unknown"] = min(float(cat_probs.get("unknown", 0.0)), 0.05) best_cat = max(cat_probs.items(), key=lambda kv: kv[1])[0] if cat_probs else "movie" heads["category"] = best_cat # 2. Structural & Domain Category Disambiguation (v3.0) if re.search(r"(?i)(?:\bMIT\d+[._]\d+|\bocw[-_]|_lec0?\d+\b|Lecture_0?\d+\b)", all_text) or _SIDE_COURSE_RE.search(name): heads["category"] = "course" elif re.search(r"(?i)\d{3,6}gut\b", all_text): heads["category"] = "ebook" elif (name_h & _V3_COMIC_H) or any( p.lower().endswith((".cbz", ".cbr", ".cb7", "_cloth_detection.log")) for p in fpaths ): heads["category"] = "comic" elif name_h & _V3_GAME_H: heads["category"] = "game" elif name_h & _V3_OPEN_MOVIE_H: heads["category"] = "movie" heads["anime"] = "non_anime" elif (all_h & _V3_TV_SERIES_H) or re.search( r"(?i)(?:\bS\d{1,2}E\d{1,3}\b|\b\d{1,2}Chapters\b|_ch01\.|\bPart\s*1\b.*\bInterviews\b|\btv-series\b)", all_text, ): heads["category"] = "tv_show" elif heads.get("category") == "unknown" and any(p.lower().endswith((".mp4", ".mkv", ".avi", ".mov")) for p in fpaths): heads["category"] = "movie" # 3. Anime Precision & Recall Guard (v3.0) if all_h & _V3_ANIME_POS_H: heads["anime"] = "anime" if all_h & _V3_ANIME_MOVIE_H: heads["category"] = "movie" elif all_h & _V3_ANIME_NEG_H: heads["anime"] = "non_anime" # 4. External Subtitle Inventory & Subtitle-Trap Guard (english_subs & lang) real_subs = [ p for p in fpaths if p.lower().endswith((".srt", ".vtt", ".ass")) and not p.lower().endswith((".asr.srt", ".autogenerated.vtt", ".asr.js")) ] has_non_en_sub_only_tag = bool(re.search(r"(?i)\b(?:vostfr|vose|subfrench|napisy[._ -]*pl|czuntertitel)\b", all_text)) has_en_sub_tag = bool(re.search(r"(?i)\b(?:english[._ -]*sub(?:bed|s)?|eng[._ -]*subs?|esubs?)\b", all_text)) if real_subs: if any(re.search(r"(?i)(?:english|\.en\.|\.eng\.|ocw-|mit\d|lec\d|^L\d{2}-)", Path(p).name) for p in real_subs) or not has_non_en_sub_only_tag: heads["english_subs"] = "has_eng_subs" else: heads["english_subs"] = "no_eng_subs" elif has_en_sub_tag: heads["english_subs"] = "has_eng_subs" elif has_non_en_sub_only_tag: heads["english_subs"] = "no_eng_subs" # 5. Spoken Language & Subtitle-Trap Audio Disambiguation (lang & title_language) name_no_sub = _SIDE_DECEPTIVE_SUB_STRIP_RE.sub(" ", name) has_explicit_en_audio = bool( _SIDE_ANIME_DUB_KW_RE.search(name) or re.search( r"(?i)(?:\b(?:eng|english)\b|[\[\(\.\s_-]EN[\]\)\.\s_-]|\b[34]audio\b)", name_no_sub, ) ) has_foreign_context = ( heads.get("anime") == "anime" or heads.get("title_language") == "non_english_title" or bool(name_h & _V3_NON_EN_TITLE_H) or bool(_SIDE_CJK_RE.search(name)) or bool( re.search( r"(?i)\b(?:ita|ger|deu|fre|fra|rus|spa|esp|jpn|jap|kor|chi|hin|dut|nl|por|pol|tur|vostfr|vose)\b", name_no_sub, ) ) ) is_sub_only_trap = has_en_sub_tag and not has_explicit_en_audio and has_foreign_context if is_sub_only_trap or bool(all_h & _V3_NON_EN_AUDIO_H): heads["lang"] = "non_english" if name_h & _V3_NON_EN_TITLE_H: heads["title_language"] = "non_english_title" elif heads.get("lang") == "multi_or_unknown" and heads.get("title_language") == "english_title" and heads.get("anime") == "non_anime": if heads.get("category") in ("movie", "tv_show", "audiobook", "course"): heads["lang"] = "english" # 6. Byte-Span Clean Title & Release Year Extraction m_yr = re.search(r"(?= 2: clean_t = clean_t[:m_yr.start()] elif m_yr and m_yr.start() == 0: clean_t = clean_t[m_yr.end():] clean_t = re.sub(r"(?i)\b(?:1080p|720p|2160p|480p|vose|vostfr|english[._ -]*subbed|eng[._ -]*subs?)\b.*$", "", clean_t) clean_t = re.sub(r"[._-]+", " ", clean_t).strip() if not is_nested: for k, v in heads.items(): if k in item_res and isinstance(item_res[k], dict): item_res[k]["label"] = v else: item_res[k] = {"label": v, "confidence": 0.95} item_res["matching_identity"] = { "clean_title": clean_t or name, "release_year": rel_year, } p_junk = float(probs.get("junk", {}).get("junk", 0.0)) if heads.get("category") in ("movie", "tv_show") and p_adult < 0.5451 and p_junk < 0.1010: heads["owner_scope"] = "allowed_type" else: heads["owner_scope"] = "excluded" if not is_nested and "owner_scope" in item_res and isinstance(item_res["owner_scope"], dict): item_res["owner_scope"]["label"] = heads["owner_scope"] if is_nested: item_res["category"] = heads.get("category", "unknown") return item_res