""" Resolver Module for resolving supplied source information into an object that can be played through Player module """ import importlib import sys from urllib import parse import requests import xbmcgui import xbmcvfs from resources.lib.common.thread_pool import ThreadPool from resources.lib.debrid.all_debrid import AllDebrid from resources.lib.debrid.premiumize import Premiumize from resources.lib.debrid.real_debrid import RealDebrid from resources.lib.modules.exceptions import FileIdentification from resources.lib.modules.exceptions import ResolverFailure from resources.lib.modules.exceptions import UnexpectedResponse from resources.lib.modules.globals import g from resources.lib.modules.resolver.torrent_resolvers import AllDebridResolver from resources.lib.modules.resolver.torrent_resolvers import PremiumizeResolver from resources.lib.modules.resolver.torrent_resolvers import RealDebridResolver class Resolver: """ Handles resolving of identified sources to a playable format to supply to Player module """ torrent_resolve_failure_style = None def __init__(self): self.torrent_resolve_failure_style = g.get_int_setting('general.resolvefailurehandling') sys.path.append(g.ADDON_USERDATA_PATH) self.return_data = None self.resolvers = { "all_debrid": AllDebridResolver, "premiumize": PremiumizeResolver, "real_debrid": RealDebridResolver, } def resolve_multiple_until_valid_link(self, sources, item_information, pack_select=False, silent=False): """ Resolves all supplied sources until an identified link is found :param sources: List of sources to resolve :param item_information: Metadata on item intended to be played :param pack_select: Set to True to force manual file selection :return: streamable URL or dictionary of adaptive source information """ stream_link = None release_title = None for source in sources: try: stream_link, release_title = self.resolve_single_source(source, item_information, pack_select, silent) if stream_link: break except Exception: g.log_stacktrace() continue return stream_link, release_title def resolve_single_source(self, source, item_information, pack_select=False, silent=False): """ Resolves source to a streamable object :param source: Item to attempt to resolve :param item_information: Metadata on item intended to be played :param pack_select: Set to True to force manual file selection :return: streamable URL or dictionary of adaptive source information """ stream_link = None try: if source["type"] == "adaptive": stream_link = source elif source["type"] == "direct": stream_link = source["url"] elif source["type"] == "torrent": stream_link = self._resolve_debrid_source( self.resolvers[source["debrid_provider"]], source, item_information, pack_select, ) if ( not stream_link and self.torrent_resolve_failure_style == 1 and not pack_select and not silent and xbmcgui.Dialog().yesno(g.ADDON_NAME, g.get_language_string(30490)) ): stream_link = self._resolve_debrid_source( self.resolvers[source["debrid_provider"]], source, item_information, True, ) elif source["type"] in ["hoster", "cloud"]: stream_link = self._resolve_hoster_or_cloud(source, item_information) if stream_link: return stream_link, source['release_title'] g.log(f"Failed to resolve source: {source}", "error") return None, None except ResolverFailure as e: g.log(f'Failed to resolve source: {e}') return None, None @staticmethod def _handle_provider_imports_resolving(source): provider = source["provider_imports"] provider_module = importlib.import_module(f"{provider[0]}.{provider[1]}") if hasattr(provider_module, "source"): provider_module = provider_module.source() source["url"] = provider_module.resolve(source["url"]) return source def _handle_debrid_hoster_resolving(self, source, item_information): stream_link = self._resolve_debrid_source( self.resolvers[source["debrid_provider"]], source, item_information, False ) if not stream_link: return try: requests.head(stream_link, timeout=3) return stream_link except requests.exceptions.RequestException as e: g.log(e, 'error') g.log("Head Request failed link likely dead, skipping", 'error') return def _resolve_hoster_or_cloud(self, source, item_information): stream_link = None if not source.get("url", False): return if source["type"] == "cloud" and source["debrid_provider"] == "premiumize": selected_file = Premiumize().item_details(source["url"]) return ( selected_file["stream_link"] if g.get_bool_setting("premiumize.transcoded") else selected_file["link"] ) if "provider_imports" in source: source = self._handle_provider_imports_resolving(source) if "debrid_provider" in source: stream_link = self._handle_debrid_hoster_resolving(source, item_information) elif source["url"].startswith("http"): stream_link = self._test_direct_url(source) elif xbmcvfs.exists(source["url"]): stream_link = source["url"] if stream_link is None: return if stream_link.endswith(".rar"): return return stream_link @staticmethod def _test_direct_url(source): try: ext = source["url"].split("?")[0] ext = ext.split("&")[0] ext = ext.split("|")[0] ext = ext.rsplit(".")[-1] ext = ext.replace("/", "").lower() if ext == "rar": raise TypeError("Incorrect file format - rar file provided") try: headers = source["url"].rsplit("|", 1)[1] except IndexError: headers = "" headers = parse.quote_plus(headers).replace("%3D", "=") if " " in headers else headers headers = dict(parse.parse_qsl(headers)) live_check = requests.head(source["url"], headers=headers, timeout=10) if live_check.status_code != 200: g.log("Head Request failed link likely dead, skipping") return stream_link = source["url"] except (IndexError, KeyError): stream_link = None return stream_link @staticmethod def _resolve_debrid_source(api, source, item_information, pack_select=False): stream_link = None api = api() if source["type"] == "torrent": try: stream_link = api.resolve_magnet(item_information, source, pack_select) except (UnexpectedResponse, FileIdentification) as e: g.log(e, "error") return None except Exception as e: g.log(f"Failing Magnet: {source['magnet']}") raise ResolverFailure(source) from e elif source["type"] in ["hoster", "cloud"]: try: stream_link = api.resolve_stream_url({"link": source["url"]}) except (UnexpectedResponse, FileIdentification) as e: g.log(e, "error") raise ResolverFailure(source) from e return stream_link @staticmethod def get_hoster_list(): """ Fetche :return: """ thread_pool = ThreadPool() hosters = {"premium": {}, "free": []} try: if g.get_bool_setting("premiumize.enabled") and g.get_bool_setting("premiumize.hosters"): thread_pool.put(Premiumize().get_hosters, hosters) if g.get_bool_setting("realdebrid.enabled") and g.get_bool_setting("rd.hosters"): thread_pool.put(RealDebrid().get_hosters, hosters) if g.get_bool_setting("alldebrid.enabled") and g.get_bool_setting("alldebrid.hosters"): thread_pool.put(AllDebrid().get_hosters, hosters) thread_pool.wait_completion() except ValueError: g.log_stacktrace() xbmcgui.Dialog().notification(g.ADDON_NAME, g.get_language_string(30485)) return hosters return hosters