# -*- coding: utf-8 -*- # TorrServer-based P2P torrent streaming scraper # Searches for torrents via DMM API (free) or optional Jackett, plays via TorrServer import json import requests from modules import kodi_utils, source_utils from modules.torrserver_api import TorrServerAPI from modules.utils import clean_file_name, normalize internal_results, get_file_info, release_info_format = source_utils.internal_results, source_utils.get_file_info, source_utils.release_info_format set_property, notification = kodi_utils.set_property, kodi_utils.notification int_window_prop = 'ultimate_nondebrid.internal_results.%s' DMM_API = 'https://zilean.elfhosted.com' DMM_TORZNAB = 'https://zilean.elfhosted.com/torznab' USER_AGENT = 'UltimateNonDebrid/1.0' class source: def __init__(self): self.scrape_provider = 'torrserver' self.sources = [] def results(self, info): try: self.sources = [] media_type = info.get('media_type', 'movie') title = info.get('title', '') year = info.get('year', '') season = info.get('season', '') episode = info.get('episode', '') # 1. Try Jackett first if configured (most reliable) hash_results = self._search_jackett(title, year, season, episode, media_type) # 2. Fallback to DMM Torznab (self-hosted Zilean) if not hash_results: hash_results = self._search_dmm_torznab(title, year, season, episode, media_type) if not hash_results: return internal_results(self.scrape_provider, self.sources) # Build source items for item in hash_results: try: quality, details = get_file_info(url=item.get('name', '')) source_item = { 'name': item.get('name', title), 'display_name': item.get('name', title), 'quality': quality, 'size': item.get('size', 0), 'size_label': '%.2f GB' % (float(item.get('size', 0)) / 1073741824) if item.get('size') else 'Unknown', 'extraInfo': details, 'url': 'magnet:?xt=urn:btih:' + item['hash'], 'hash': item['hash'], 'id': item['hash'], 'source': self.scrape_provider, 'scrape_provider': self.scrape_provider, 'torrent': True, 'direct': False, } self.sources.append(source_item) except: pass except: pass return internal_results(self.scrape_provider, self.sources) def _search_jackett(self, title, year, season, episode, media_type): """Search via Jackett API (user must run their own Jackett instance)""" from caches.settings_cache import get_setting host = get_setting('ultimate_nondebrid.jackett_host', '') apikey = get_setting('ultimate_nondebrid.jackett_apikey', '') port = get_setting('ultimate_nondebrid.jackett_port', '9117') if not host or not apikey: return None try: query = title if media_type == 'movie': url = '%s:%s/api/v2.0/indexers/all/results/torznab/api?apikey=%s&t=movie&q=%s' % ( host, port, apikey, requests.utils.quote(query)) else: url = '%s:%s/api/v2.0/indexers/all/results/torznab/api?apikey=%s&t=tvsearch&q=%s&season=%s&ep=%s' % ( host, port, apikey, requests.utils.quote(query), season, episode) resp = requests.get(url, timeout=15, headers={'User-Agent': USER_AGENT}) results = self._parse_torznab(resp.text) return results except: return None def _search_dmm_torznab(self, title, year, season, episode, media_type): """Search via DMM Torznab endpoint (requires self-hosted Zilean)""" try: from caches.settings_cache import get_setting dmm_host = get_setting('ultimate_nondebrid.dmm_host', DMM_TORZNAB) query = title if media_type == 'episode': url = '%s/tv?q=%s&season=%s&episode=%s' % (dmm_host, requests.utils.quote(query), season, episode) else: url = '%s/movie?q=%s' % (dmm_host, requests.utils.quote(query)) resp = requests.get(url, timeout=10, headers={'User-Agent': USER_AGENT}) if resp.status_code != 200: return None return self._parse_torznab(resp.text) except: return None def _search_dmm_legacy(self, title, year, season, episode, media_type): """Search via DMM/Zilean API (free, public endpoint)""" try: query = title if media_type == 'episode': params = {'query': query, 'season': season, 'episode': episode} url = DMM_API + '/dmm/filtered' else: url = DMM_API + '/dmm/search' params = {'query': query} resp = requests.get(url, params=params, timeout=10, headers={'User-Agent': USER_AGENT}) if resp.status_code != 200: return None data = resp.json() results = [] for item in data if isinstance(data, list) else data.get('results', []): info_hash = item.get('infoHash', item.get('hash', '')) if info_hash: results.append({ 'hash': info_hash, 'name': item.get('filename', item.get('name', title)), 'size': item.get('size', 0), }) return results[:50] # Limit to 50 results except: return None def _parse_torznab(self, xml_text): """Parse Torznab XML response from Jackett""" import xml.etree.ElementTree as ET results = [] try: root = ET.fromstring(xml_text) for item in root.findall('.//item'): hash_elem = item.find('.//{http://torznab.com/schemas/2015/feed}attr[@name=\"infohash\"]') info_hash = '' if hash_elem is not None: info_hash = hash_elem.get('value', '') if not info_hash: guid = item.findtext('guid', '') if 'btih:' in guid: import re m = re.search(r'btih:([a-fA-F0-9]{40})', guid) if m: info_hash = m.group(1).lower() if info_hash: size = item.findtext('{http://torznab.com/schemas/2015/feed}attr[@name=\"size\"]', '0') results.append({ 'hash': info_hash, 'name': item.findtext('title', ''), 'size': int(float(size)) if size else 0, }) except: pass return results[:100] class resolver: """Resolves a torrent source to a playable URL via TorrServer""" @staticmethod def resolve(host, port, username, password, info_hash, file_id=0): try: ts = TorrServerAPI(host, port, username, password) if not ts.alive: notification('TorrServer not reachable at %s:%s' % (host, port), 3000) return None # Check if already added try: torrents = ts.torrents() if isinstance(torrents, list): for t in torrents: if t.get('hash', '').lower() == info_hash.lower(): # Already exists, get stream URL return ts.play_torrent(info_hash, file_id) except: pass # Add magnet to TorrServer magnet = 'magnet:?xt=urn:btih:' + info_hash try: ts.add_magnet(magnet) return ts.play_torrent(info_hash, file_id) except: pass return None except: return None