import json
import os
import sys
from typing import Any, Callable, List, Optional
from lib.clients.base import BaseClient
from lib.domain.torrent import TorrentStream
from lib.utils.general.utils import Indexer
from lib.utils.kodi.utils import kodilog
_BYTES_PER_GB = 1073741824
class ExternalScraperClient(BaseClient):
"""Generic client for external scraper modules.
Accepts any ``xbmc.python.module`` addon that exposes a ``sources()``
function returning provider tuples. The module is a soft dependency:
if the addon is not installed or fails to import, ``initialized`` is
``False`` and searches return an empty list.
The data dict passed to providers:
* Movies: ``{imdb, title, aliases, year}``
* TV: ``{imdb, tvdb, tvshowtitle, title, aliases, year, season, episode}``
"""
def __init__(
self,
module_id: str,
module_display_name: str,
notification: Optional[Callable],
) -> None:
super().__init__(None, notification)
self.module_id = module_id
self.module_name = module_id.split(".")[-1]
self.module_display_name = module_display_name
self._sources_func = None
self._providers = None
self.initialized = self._init_providers()
# ------------------------------------------------------------------
# Initialisation
# ------------------------------------------------------------------
def _init_providers(self) -> bool:
"""Locate the addon and import its provider list.
.. note::
``sys.path`` is **not** restored after import. Providers
perform runtime imports inside their ``sources()`` methods,
so the library path must remain on ``sys.path`` for the
lifetime of the client.
"""
addon_path = self._resolve_addon_path()
if not addon_path:
kodilog(f"ExternalScraper ({self.module_id}): addon not found")
return False
lib_path = os.path.join(addon_path, "lib")
if not os.path.isdir(lib_path):
kodilog(f"ExternalScraper ({self.module_id}): lib dir not found at {lib_path}")
return False
if lib_path not in sys.path:
sys.path.insert(0, lib_path)
try:
# Import the top-level package
module = __import__(self.module_name, fromlist=["sources"])
self._sources_func = module.sources
self._providers = self._sources_func(specified_folders=["torrents"])
provider_count = len(self._providers) if self._providers else 0
kodilog(f"ExternalScraper ({self.module_id}): loaded {provider_count} providers")
return bool(self._providers)
except Exception as exc:
kodilog(f"ExternalScraper ({self.module_id}): failed to import providers: {exc}")
return False
def _resolve_addon_path(self) -> Optional[str]:
"""Locate the addon path via xbmcaddon or JSON-RPC."""
# 1) xbmcaddon (fastest)
try:
import xbmcaddon
addon = xbmcaddon.Addon(self.module_id)
return addon.getAddonInfo("path")
except Exception:
pass
# 2) JSON-RPC fallback
try:
import xbmc
response = xbmc.executeJSONRPC(
json.dumps(
{
"jsonrpc": "2.0",
"method": "Addons.GetAddonDetails",
"params": {
"addonid": self.module_id,
"properties": ["path"],
},
"id": 1,
}
)
)
data = json.loads(response)
return data.get("result", {}).get("addon", {}).get("path")
except Exception as exc:
kodilog(f"ExternalScraper ({self.module_id}): JSON-RPC fallback failed: {exc}")
return None
# ------------------------------------------------------------------
# Search
# ------------------------------------------------------------------
def search(
self,
tmdb_id: str,
query: str,
mode: str,
media_type: str,
season: Optional[int],
episode: Optional[int],
imdb_id: str = "",
tvdb_id: str = "",
year: str = "",
aliases: Optional[List[str]] = None,
**kwargs: Any,
) -> List[TorrentStream]:
kodilog(
f"[ExternalScraper ({self.module_id})] search() called: "
f"query={query}, mode={mode}, imdb={imdb_id}, year={year}, aliases={aliases}, "
f"providers={len(self._providers) if self._providers else 0}"
)
if not self.initialized or not self._providers:
kodilog(f"[ExternalScraper ({self.module_id})] not initialised, returning empty")
return []
is_tv = mode == "tv" or media_type == "tv"
data = self._build_data(
query=query,
mode=mode,
season=season,
episode=episode,
imdb_id=imdb_id,
tvdb_id=tvdb_id,
year=year,
aliases=aliases,
)
all_results: List[TorrentStream] = []
for provider_name, source_class in self._providers:
# Filter providers by media-type capability
if is_tv:
if hasattr(source_class, "hasEpisodes") and not source_class.hasEpisodes:
continue
else:
if hasattr(source_class, "hasMovies") and not source_class.hasMovies:
continue
try:
provider_instance = source_class()
# Single episode / movie results
results = provider_instance.sources(data, {})
if results:
mapped = self._map_results(results, provider_name)
all_results.extend(mapped)
# Pack results for TV
if is_tv and hasattr(source_class, "pack_capable") and source_class.pack_capable:
try:
pack_results = provider_instance.sources_packs(data, {})
if pack_results:
mapped_packs = self._map_results(
pack_results, provider_name, is_pack=True
)
all_results.extend(mapped_packs)
except Exception as exc:
kodilog(
f"ExternalScraper ({self.module_id}): "
f"provider '{provider_name}' packs error: {exc}"
)
except Exception as exc:
kodilog(
f"ExternalScraper ({self.module_id}): provider '{provider_name}' error: {exc}"
)
kodilog(f"[ExternalScraper ({self.module_id})] returning {len(all_results)} results")
return all_results
# ------------------------------------------------------------------
# Data mapping helpers
# ------------------------------------------------------------------
@staticmethod
def _build_data(
query: str,
mode: str,
season: Optional[int],
episode: Optional[int],
imdb_id: str = "",
tvdb_id: str = "",
year: str = "",
aliases: Optional[List[str]] = None,
) -> dict:
"""Build the data dict that external providers expect.
Matches the convention:
* Movies: ``{imdb, title, aliases, year}``
* TV: ``{imdb, tvdb, tvshowtitle, title, aliases, year, season, episode}``
"""
is_tv = mode in ("tv", "episode")
data: dict = {
"aliases": aliases or [],
"year": str(year) if year else "",
}
if imdb_id:
data["imdb"] = imdb_id
if is_tv:
data["tvshowtitle"] = query
data["title"] = ""
data["season"] = str(season) if season else ""
data["episode"] = str(episode) if episode else ""
if tvdb_id:
data["tvdb"] = tvdb_id
else:
data["title"] = query
return data
def _map_results(
self,
results: list,
provider_name: str,
is_pack: bool = False,
) -> List[TorrentStream]:
"""Map external provider result dicts to Jacktook TorrentStream."""
mapped: List[TorrentStream] = []
for item in results:
try:
info_hash = (item.get("hash") or "").lower()
url = item.get("url", "") or ""
# Some legacy providers put the hash in url as "magnet:HASH"
# without the ?xt=urn:btih: prefix and leave hash empty.
if url.startswith("magnet:") and not url.startswith("magnet:?"):
legacy_hash = url[7:].strip()
if legacy_hash:
info_hash = legacy_hash.lower()
url = f"magnet:?xt=urn:btih:{info_hash}"
if info_hash and not url.startswith("magnet:?"):
url = f"magnet:?xt=urn:btih:{info_hash}"
raw_size = item.get("size", 0)
try:
size_bytes = int(float(raw_size) * _BYTES_PER_GB) if raw_size else 0
except (ValueError, TypeError):
size_bytes = 0
package = item.get("package")
result_is_pack = is_pack or package in ("season", "show")
try:
seeders = int(item.get("seeders", 0) or 0)
except (ValueError, TypeError):
seeders = 0
quality = item.get("quality", "") or "N/A"
title = item.get("name", "") or ""
mapped.append(
TorrentStream(
title=title,
type="torrent",
indexer=Indexer.EXTERNAL_SCRAPER,
subindexer=provider_name,
addonInstanceLabel=self.module_display_name,
infoHash=info_hash,
url=url,
size=size_bytes,
seeders=seeders,
quality=quality,
provider=provider_name,
isPack=result_is_pack,
publishDate="",
)
)
except Exception as exc:
kodilog(f"ExternalScraper: failed to map result: {exc}")
continue
return mapped
# ------------------------------------------------------------------
# BaseClient contract
# ------------------------------------------------------------------
def parse_response(self, res) -> List[TorrentStream]:
"""Not used — ExternalScraperClient does not make HTTP requests."""
return []