#!/usr/bin/env python3 """Quick web search UI for Ultimate Non-Debrid scrapers. Usage: python3 search_web.py [port] Then open http://localhost:8800 in a browser.""" import os, sys, json, html, re, urllib.request, urllib.parse from http.server import HTTPServer, BaseHTTPRequestHandler HOST = '0.0.0.0' PORT = int(sys.argv[1]) if len(sys.argv) > 1 else 8800 # Add addon scraper to path sys.path.insert(0, os.path.expanduser("~/addon-audit/build/ultimate-nondebrid/resources/lib")) # ── TorrServer connection (optional) ── TS_HOST = '127.0.0.1' TS_PORT = 8090 TS_USER = '' TS_PASS = '' # ── Jackett connection (optional) ── JACKETT_HOST = 'http://192.168.1.50' JACKETT_PORT = '9117' JACKETT_API = '' def search_jackett(query, media_type='movie', season='', episode=''): if not JACKETT_API: return None q = urllib.parse.quote(query) if media_type == 'episode' and season and episode: url = f"{JACKETT_HOST}:{JACKETT_PORT}/api/v2.0/indexers/all/results/torznab/api?apikey={JACKETT_API}&t=tvsearch&q={q}&season={season}&ep={episode}" else: url = f"{JACKETT_HOST}:{JACKETT_PORT}/api/v2.0/indexers/all/results/torznab/api?apikey={JACKETT_API}&t=movie&q={q}" try: resp = urllib.request.urlopen(url, timeout=15) xml = resp.read().decode() return parse_torznab(xml) except: return None def search_dmm_torznab(query, media_type='movie', season='', episode=''): host = 'https://zilean.elfhosted.com/torznab' q = urllib.parse.quote(query) try: if media_type == 'episode' and season and episode: url = f"{host}/tv?q={q}&season={season}&episode={episode}" else: url = f"{host}/movie?q={q}" except: url = f"{host}/movie?q={q}" try: resp = urllib.request.urlopen(url, timeout=10) return parse_torznab(resp.read().decode()) except: return None def parse_torznab(xml_text): results = [] import xml.etree.ElementTree as ET NS = {'t': 'http://torznab.com/schemas/2015/feed'} try: root = ET.fromstring(xml_text) for item in root.findall('.//item'): ih = '' for attr in item.findall('.//t:attr', NS): if attr.get('name') == 'infohash': ih = attr.get('value', '') if not ih: guid = item.findtext('guid', '') m = re.search(r'btih:([a-fA-F0-9]{40})', guid) if m: ih = m.group(1).lower() if not ih: continue title = item.findtext('title', 'Unknown') size_el = item.find('.//t:attr[@name="size"]', NS) size = int(float(size_el.get('value', 0))) if size_el is not None else 0 seeds_el = item.find('.//t:attr[@name="seeders"]', NS) seeds = int(seeds_el.get('value', 0)) if seeds_el is not None else 0 results.append({'title': title, 'hash': ih, 'size': size, 'seeders': seeds}) except: pass return results def detect_quality(name): n = name.lower() if any(x in n for x in ['2160', '4k', 'uhd']): return '🎬4K' if '1080' in n: return '📺1080p' if '720' in n: return '📺720p' if any(x in n for x in ['cam', 'ts.', 'tc.', 'tele']): return '📀SD' return '📀SD' def fmt_size(bytes): if not bytes: return '?' gb = bytes / 1073741824 if gb >= 1: return f'{gb:.2f} GB' mb = bytes / 1048576 return f'{mb:.0f} MB' PAGE = ''' Ultimate Search

🔎 Ultimate Search

%s
TorrServer: %s TorrServer · Jackett
''' class Handler(BaseHTTPRequestHandler): def do_GET(self): parsed = urllib.parse.urlparse(self.path) params = urllib.parse.parse_qs(parsed.query) path = parsed.path if path == '/settings': return self.serve_settings() if path == '/save_settings': return self.save_settings(params) # Serve static assets inline query = params.get('q', [''])[0] media_type = params.get('type', ['movie'])[0] use_jackett = params.get('jackett', [''])[0] == '1' results_html = '
Enter a search query above
' if query: results_html = self.do_search(query, media_type, use_jackett) # Check TorrServer ts_alive = self.check_torrserver() ts_status = '● Connected' if ts_alive else '● Not running' mchecked = 'checked' if media_type == 'movie' else '' tchecked = 'checked' if media_type == 'tv' else '' jchecked = 'checked' if use_jackett else '' html = self.render_page(query, media_type, use_jackett, results_html, ts_alive) self.send_response(200) self.send_header('Content-Type', 'text/html; charset=utf-8') self.end_headers() self.wfile.write(html.encode()) def render_page(self, query, media_type, use_jackett, results_html, ts_alive): q = html.escape(query) m_ck = 'checked' if media_type == 'movie' else '' t_ck = 'checked' if media_type == 'tv' else '' j_ck = 'checked' if use_jackett else '' ts_s = '● Connected' if ts_alive else '● Not running' return f''' Ultimate Search

🔎 Ultimate Search

Settings · Clear
{results_html}
TorrServer: {ts_s} TorrServer · Jackett
''' def do_search(self, query, media_type, use_jackett): results = [] # Try Jackett first if enabled if use_jackett: r = search_jackett(query, media_type) if r: results.extend(r) # Try Torznab/DMM as fallback if not results: r = search_dmm_torznab(query, media_type) if r: results.extend(r) if not results: return '
No results found. Try a different query or enable Jackett.
' rows = [] for r in results[:100]: title = html.escape(r.get('title', 'Unknown')) ih = r.get('hash', '') qual = detect_quality(title) size = fmt_size(r.get('size', 0)) seeds = r.get('seeders', 0) magnet = f'magnet:?xt=urn:btih:{ih}&dn={urllib.parse.quote(title)}' ts_url = f'http://{TS_HOST}:{TS_PORT}/play?hash={ih}&file_id=0' if self.check_torrserver() else '' actions = f'Magnet' if ts_url: actions += f' · Stream' rows.append(f''' {title} {qual} {size} {seeds} {ih[:16]}... {actions} ''') table = f''' {''.join(rows)}
Title Quality Size Seeds Hash Action
{len(results)} results
''' return table def check_torrserver(self): try: resp = urllib.request.urlopen(f'http://{TS_HOST}:{TS_PORT}/echo', timeout=2) return resp.read() == b'uptime' except: return False def serve_settings(self): page = f''' Settings

⚙️ Settings

← Back to search

''' self.send_response(200) self.send_header('Content-Type', 'text/html; charset=utf-8') self.end_headers() self.wfile.write(page.encode()) def save_settings(self, params): global TS_HOST, TS_PORT, JACKETT_HOST, JACKETT_PORT, JACKETT_API TS_HOST = params.get('ts_host', [TS_HOST])[0] TS_PORT = params.get('ts_port', [str(TS_PORT)])[0] JACKETT_HOST = params.get('jackett_host', [JACKETT_HOST])[0] JACKETT_PORT = params.get('jackett_port', [JACKETT_PORT])[0] JACKETT_API = params.get('jackett_api', [JACKETT_API])[0] self.send_response(302) self.send_header('Location', '/') self.end_headers() def log_message(self, format, *args): print(f"[{self.client_address[0]}] {format % args}") if __name__ == '__main__': server = HTTPServer((HOST, PORT), Handler) print(f"\n 🌐 Ultimate Search Web UI") print(f" http://localhost:{PORT}") print(f" http://192.168.1.50:{PORT} (from network)") print(f"\n Press Ctrl+C to stop\n") try: server.serve_forever() except KeyboardInterrupt: server.shutdown()