"""Armiger Torznab Filtering Proxy & Pre-Grab Admission Gate for Prowlarr, Bitmagnet, and the *Arr Stack. Sits transparently between upstream Torznab indexers (`Prowlarr`, `Bitmagnet`, `DH-Waypoint`, `Jackett`) and downstream Torznab media managers (`Sonarr`, `Radarr`, `Lidarr`, `Readarr`). Every Torznab XML `` returned by an upstream search is evaluated in milliseconds by `ArmigerUltraPipeline.classify_batch` before `Sonarr` or `Radarr` can grab it: 1. Blocks disguised malware executables (`.mkv.exe`, `.scr`, `.vbs`, `.lnk`, `.ps1`, passworded archives). 2. Blocks fake-seeder malware honeypots (`SwarmSurvivalNet` `is_honeypot_trap == True`). 3. Blocks dead swarms (`viability_tier == "dead"` when `ARMIGER_REJECT_DEAD_SWARMS=true`). 4. Blocks unrequested foreign-language audio traps (`AudioLanguageGuard` cross-head check when `ARMIGER_REQUIRE_ENGLISH_AUDIO=true`, with optional anime/subtitle allowance via `ARMIGER_ALLOW_ENGLISH_SUBS=true`). 5. Enriches surviving `` nodes with `` and `` attributes. """ from __future__ import annotations from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer import json import os from pathlib import Path import sys from typing import Any, Dict, List, Optional, Tuple import urllib.parse import urllib.request import xml.etree.ElementTree as ET ROOT_DIR = Path(__file__).resolve().parents[2] if str(ROOT_DIR) not in sys.path: sys.path.insert(0, str(ROOT_DIR)) from armiger_dht_rsi import DHTRSIMode from armiger_ultra_pipeline import ArmigerUltraPipeline, ArmigerUltraPrediction TORZNAB_NS = "http://torznab.com/schemas/2015/feed" ET.register_namespace("torznab", TORZNAB_NS) class ArmigerTorznabFilter: """Filters and enriches Torznab RSS/XML feeds using ArmigerUltraPipeline.""" def __init__( self, pipeline: ArmigerUltraPipeline, *, require_english_audio: bool = False, allow_english_subs: bool = True, reject_dead_swarms: bool = True, allowed_categories: Optional[List[str]] = None, use_zero_cue_catalog: bool = True, ) -> None: self.pipeline = pipeline self.require_english_audio = require_english_audio self.allow_english_subs = allow_english_subs self.reject_dead_swarms = reject_dead_swarms self.allowed_categories = set(allowed_categories) if allowed_categories else None self.use_zero_cue_catalog = use_zero_cue_catalog @staticmethod def _extract_item_row(item_elem: ET.Element) -> Dict[str, Any]: title = (item_elem.findtext("title") or "").strip() size_text = item_elem.findtext("size") or "0" try: size_val = int(size_text) except ValueError: size_val = 0 attrs: Dict[str, str] = {} for attr in item_elem.findall(f"{{{TORZNAB_NS}}}attr"): name = (attr.attrib.get("name") or "").lower().strip() val = attr.attrib.get("value") or "" if name: attrs[name] = val seeders = int(attrs.get("seeders", "0") or 0) peers = int(attrs.get("peers", "0") or 0) leechers = max(peers - seeders, int(attrs.get("leechers", "0") or 0)) grabs = int(attrs["grabs"]) if "grabs" in attrs and attrs["grabs"].strip() else None infohash = attrs.get("infohash") or "" files_attr = attrs.get("files") or "" if files_attr: files = [{"path": p.strip(), "size": size_val} for p in files_attr.split(";") if p.strip()] else: tail = title.rsplit(".", 1)[-1].lower() if "." in title else "" known_exts = {"mkv", "mp4", "avi", "flac", "mp3", "m4b", "epub", "pdf", "cbz", "cbr", "iso", "exe", "scr", "vbs", "bat", "cmd", "lnk", "ps1", "msi"} if tail in known_exts: files = [{"path": title, "size": size_val}] else: low = title.lower() if any(k in low for k in ("flac", "24bit", "lossless")): synth_ext = ".flac" elif any(k in low for k in ("m4b", "audiobook", "unabridged")): synth_ext = ".m4b" elif any(k in low for k in ("epub", "ebook")): synth_ext = ".epub" elif any(k in low for k in ("cbz", "cbr")): synth_ext = ".cbz" else: synth_ext = ".mkv" files = [{"path": f"{title}{synth_ext}", "size": size_val}] return { "infohash": infohash, "name": title, "size": size_val, "files": files, "swarm": { "tracker_known": bool(seeders > 0 or leechers > 0 or "seeders" in attrs), "tracker_seeders": seeders, "tracker_leechers": leechers, "tracker_completed": grabs, }, } def should_admit(self, pred: ArmigerUltraPrediction) -> Tuple[bool, str]: """Evaluate whether a Torznab release should be admitted to Sonarr/Radarr/Lidarr/Readarr.""" if not pred.csam_safety["is_safe"]: return False, str(pred.csam_safety.get("reason") or "csam_or_malware_blocked") if pred.honeypot_trap["is_honeypot_trap"]: return False, "honeypot_trap_blocked" if pred.junk == "junk" or not pred.mondrian_gates["passed_all"]: return False, str(pred.mondrian_gates.get("safety_gate_reason") or "mondrian_gate_blocked") if self.reject_dead_swarms and pred.swarm_viability["viability_tier"] == "dead": return False, "dead_swarm_blocked" if self.allowed_categories and pred.category not in self.allowed_categories: return False, f"category_mismatch:{pred.category}" if self.require_english_audio and pred.lang == "non_english": has_subs = pred.english_subs == "has_eng_subs" if not (self.allow_english_subs and has_subs): return False, "non_english_audio_blocked" return True, "admitted" def filter_torznab_xml(self, xml_bytes: bytes) -> Tuple[bytes, Dict[str, Any]]: """Parse upstream Torznab XML, remove blocked `` nodes, and enrich surviving items.""" root = ET.fromstring(xml_bytes) channel = root.find("channel") if channel is None: return xml_bytes, {"total": 0, "admitted": 0, "blocked": 0, "blocked_reasons": {}} items = channel.findall("item") if not items: return xml_bytes, {"total": 0, "admitted": 0, "blocked": 0, "blocked_reasons": {}} rows = [self._extract_item_row(it) for it in items] predictions = self.pipeline.classify_batch( rows, use_zero_cue_catalog=self.use_zero_cue_catalog, ) admitted_count = 0 blocked_count = 0 blocked_reasons: Dict[str, int] = {} for item_elem, pred in zip(items, predictions): admit, reason = self.should_admit(pred) if not admit: channel.remove(item_elem) blocked_count += 1 blocked_reasons[reason] = blocked_reasons.get(reason, 0) + 1 continue admitted_count += 1 for attr_name, attr_val in ( ("armiger_category", pred.category), ("armiger_lang", pred.lang), ("armiger_viability", str(pred.swarm_viability["viability_tier"])), ("armiger_survival_prob", f"{pred.swarm_viability['survival_net_prob']:.4f}"), ): attr_el = ET.SubElement(item_elem, f"{{{TORZNAB_NS}}}attr") attr_el.set("name", attr_name) attr_el.set("value", attr_val) filtered_bytes = ET.tostring(root, encoding="utf-8", xml_declaration=True) stats = { "total": len(items), "admitted": admitted_count, "blocked": blocked_count, "blocked_reasons": blocked_reasons, } return filtered_bytes, stats def create_proxy_server(host: str = "localhost", port: int = 8097) -> ThreadingHTTPServer: model_dir = os.environ.get("ARMIGER_MODEL_DIR", str(ROOT_DIR)) variant = os.environ.get("ARMIGER_VARIANT", "int8") upstream_url = os.environ.get("UPSTREAM_TORZNAB_URL", "http://dh-waypoint:3333/torznab").rstrip("/") pipeline = ArmigerUltraPipeline.from_directory( model_dir, variant=variant, rsi_mode=DHTRSIMode(os.environ.get("ARMIGER_DHT_RSI_MODE", "pull_only")), preload_catalog=os.environ.get("ARMIGER_USE_ZERO_CUE_CATALOG", "true").lower() == "true", ) torznab_filter = ArmigerTorznabFilter( pipeline, require_english_audio=os.environ.get("ARMIGER_REQUIRE_ENGLISH_AUDIO", "false").lower() == "true", allow_english_subs=os.environ.get("ARMIGER_ALLOW_ENGLISH_SUBS", "true").lower() == "true", reject_dead_swarms=os.environ.get("ARMIGER_REJECT_DEAD_SWARMS", "true").lower() == "true", ) class ProxyHandler(BaseHTTPRequestHandler): def do_GET(self) -> None: if self.path == "/healthz": body = json.dumps({"status": "ok", "service": "armiger-torznab-proxy", "variant": variant}).encode("utf-8") self.send_response(200) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) return parsed = urllib.parse.urlsplit(self.path) params = urllib.parse.parse_qs(parsed.query) t_param = (params.get("t") or [""])[0].lower() target = f"{upstream_url}{parsed.path}" if parsed.query: target = f"{target}?{parsed.query}" req = urllib.request.Request(target, method="GET") with urllib.request.urlopen(req, timeout=15.0) as resp: raw_xml = resp.read() content_type = resp.headers.get("Content-Type", "application/rss+xml; charset=utf-8") if t_param in ("caps", ""): out_bytes = raw_xml stats = {"total": 0, "admitted": 0, "blocked": 0} else: out_bytes, stats = torznab_filter.filter_torznab_xml(raw_xml) self.send_response(200) self.send_header("Content-Type", content_type) self.send_header("X-Armiger-Total", str(stats["total"])) self.send_header("X-Armiger-Admitted", str(stats["admitted"])) self.send_header("X-Armiger-Blocked", str(stats["blocked"])) self.send_header("Content-Length", str(len(out_bytes))) self.end_headers() self.wfile.write(out_bytes) def log_message(self, fmt: str, *args: Any) -> None: return return ThreadingHTTPServer((host, port), ProxyHandler) if __name__ == "__main__": listen_host = os.environ.get("ARMIGER_PROXY_HOST", "localhost") listen_port = int(os.environ.get("ARMIGER_PROXY_PORT", "8097")) srv = create_proxy_server(listen_host, listen_port) print(f"Armiger Torznab proxy listening on {listen_host}:{listen_port}") srv.serve_forever()