import re
from typing import Any, Dict, List, Optional
from lib.api.stremio.addon_manager import Addon, build_addon_instance_label
from lib.api.stremio.models import Meta, MetaPreview, Stream
from lib.clients.base import BaseClient, TorrentStream
from lib.clients.stremio.helpers import get_addon_display_name
from lib.utils.debrid.debrid_utils import process_external_cache
from lib.utils.general.utils import USER_AGENT_HEADER, IndexerType, info_hash_to_magnet
from lib.utils.kodi.settings import get_int_setting
from lib.utils.kodi.utils import convert_size_to_bytes, get_setting, kodilog
from lib.utils.localization.language_detection import find_languages_in_string
EXCLUDED_RD_ADDONS = ["org.nuvio.streams", "org.mycine.addon"]
class StremioAddonCatalogsClient(BaseClient):
def __init__(self, params: Dict[str, Any]) -> None:
super().__init__(None, None)
self.params = params
self.base_url = self.params["addon_url"]
def search(
self,
imdb_id: str,
mode: str,
media_type: str,
season: Optional[int],
episode: Optional[int],
dialog: Any,
) -> None:
pass
def parse_response(self, res: Any) -> List[TorrentStream]:
return []
def search_catalog(self, query: str) -> Optional[Dict[str, Any]]:
return self.get_catalog_info(search=query)
def get_catalog_info(self, **kwargs) -> Optional[Dict[str, Any]]:
extra_path = ""
for key, value in kwargs.items():
if value:
extra_path += f"/{key}={value}"
if not extra_path:
path_suffix = ".json"
else:
path_suffix = f"{extra_path}.json"
catalog_type = self.params.get("catalog_type")
catalog_id = self.params.get("catalog_id")
url = f"{self.base_url}/catalog/{catalog_type}/{catalog_id}{path_suffix}"
kodilog(f"Using Stremio addon catalog URL: {url}")
res = self.session.get(
url, headers=USER_AGENT_HEADER, timeout=get_int_setting("stremio_timeout")
)
if res.status_code != 200:
return
data = res.json()
if "metas" in data:
data["metas"] = [MetaPreview.from_dict(m) for m in data["metas"]]
return data
def get_meta_info(self) -> Optional[Dict[str, Any]]:
catalog_type = self.params.get("catalog_type")
meta_id = self.params.get("meta_id")
url = f"{self.base_url}/meta/{catalog_type}/{meta_id}.json"
kodilog(f"Using Stremio addon meta URL: {url}")
res = self.session.get(
url, headers=USER_AGENT_HEADER, timeout=get_int_setting("stremio_timeout")
)
if res.status_code != 200:
return
data = res.json()
if "meta" in data:
data["meta"] = Meta.from_dict(data["meta"])
return data
def get_stream_info(self) -> Optional[Dict[str, Any]]:
catalog_type = self.params.get("catalog_type")
meta_id = self.params.get("meta_id")
url = f"{self.base_url}/stream/{catalog_type}/{meta_id}.json"
kodilog(f"Using Stremio addon stream URL: {url}")
res = self.session.get(
url, headers=USER_AGENT_HEADER, timeout=get_int_setting("stremio_timeout")
)
if res.status_code != 200:
return
data = res.json()
if "streams" in data:
data["streams"] = [Stream.from_dict(s) for s in data["streams"]]
return data
class StremioAddonClient(BaseClient):
def __init__(self, addon: Addon) -> None:
super().__init__(None, None)
self.addon = addon
self.display_name = get_addon_display_name(addon)
self.instance_label = build_addon_instance_label(
{
"manifest": {"id": addon.manifest.id, "name": self.display_name},
"transportUrl": addon.transport_url,
"transportName": addon.transport_name,
}
)
self.indexer_name = (addon.manifest.name or addon.manifest.id).split(" ")[0]
def search(
self,
video_id: str,
mode: str,
media_type: str,
season: Optional[int],
episode: Optional[int],
) -> List[TorrentStream]:
try:
prefix = video_id.split(":")[0] if ":" in video_id else "tt"
if mode == "tv" or media_type == "tv":
if not self.addon.isSupported("stream", "series", prefix):
return []
if prefix == "kitsu":
url = f"{self.addon.url()}/stream/series/{video_id}:{episode}.json"
else:
url = f"{self.addon.url()}/stream/series/{video_id}:{season}:{episode}.json"
elif mode == "movies" or media_type == "movies":
if not self.addon.isSupported("stream", "movie", prefix):
return []
url = f"{self.addon.url()}/stream/movie/{video_id}.json"
else:
return []
kodilog("Using Stremio addon search URL: " + url)
res = self.session.get(
url,
headers=USER_AGENT_HEADER,
timeout=get_int_setting("stremio_timeout"),
)
if res.status_code != 200:
return []
response = self.parse_response(res)
kodilog(f"Stremio addon {self.display_name} returned {len(response)} results")
return response
except Exception as e:
self.handle_exception(f"Error in {self.display_name}: {e!s}")
return []
def parse_response(self, res: Any, is_external_cache: bool = False) -> List[TorrentStream]:
if not is_external_cache:
res_json = res.json()
streams = res_json.get("streams", [])
else:
pass
if hasattr(res, "json"):
data = res.json()
elif isinstance(res, dict):
data = res
else:
data = {}
results = []
streams = data.get("streams", [])
for item in streams:
stream = Stream.from_dict(item)
parsed = self.parse_torrent_description(stream.description or stream.title or "")
if is_external_cache:
match = re.search(r"\b\w{40}\b", stream.url or "")
info_hash = match.group() if match else item.get("infoHash")
url = ""
is_cached = True
else:
info_hash = stream.infoHash
url = stream.url
is_cached = bool(url)
stream_type = IndexerType.STREMIO_DEBRID if url else IndexerType.TORRENT
stream_provider = stream.get_provider() or parsed["provider"]
stream_subindexer = stream.get_sub_indexer(self.addon)
stream_size = stream.get_parsed_size() or item.get("sizebytes") or parsed["size"]
stream_seeders = item.get("seed", 0) or parsed["seeders"]
results.append(
TorrentStream(
title=stream.get_parsed_title(),
type=stream_type,
indexer=self.indexer_name,
subindexer=stream_subindexer,
addonKey=self.addon.key(),
addonName=self.display_name,
addonSourceName=self.addon.manifest.name or self.addon.manifest.id,
addonInstanceLabel=self.instance_label,
guid=info_hash_to_magnet(info_hash) if info_hash else "",
infoHash=info_hash if info_hash else "",
size=stream_size,
seeders=stream_seeders,
languages=parsed["languages"],
fullLanguages=parsed["languages"],
provider=stream_provider,
publishDate="",
peers=0,
url=url if url else "",
proxyHeaders=stream.get_proxy_headers(),
isCached=is_cached,
)
)
return results
def parse_torrent_description(self, desc: str) -> Dict[str, Any]:
if not desc:
return {
"size": 0,
"seeders": 0,
"provider": "",
"languages": [],
}
# Extract size
size_pattern = r"💾 ([\d.,]+ (?:GB|MB|Go|Mo))"
size_match = re.search(size_pattern, desc, re.IGNORECASE)
size = size_match.group(1) if size_match else None
if size:
size = convert_size_to_bytes(size)
# Extract seeders
seeders_pattern = r"[👥👤] (\d+)"
seeders_match = re.search(seeders_pattern, desc)
seeders = int(seeders_match.group(1)) if seeders_match else None
# Extract provider
provider_pattern = r"([🌐🔗⚙️])\s*([^🌐🔗⚙️]+)"
provider_match = re.findall(provider_pattern, desc)
words = [match[1].strip() for match in provider_match]
if words:
words = words[-1].splitlines()[0]
provider = words
return {
"size": size or 0,
"seeders": seeders or 0,
"provider": provider or "",
"languages": find_languages_in_string(desc),
}
def should_use_rd_cache(self) -> bool:
"""Checks if RD cache should be used for this addon."""
return bool(
get_setting("real_debrid_enabled")
and get_setting("real_debrid_cached_check")
and not get_setting("torrent_enable")
and not get_setting("stremio_loggedin")
and self.addon.manifest.id not in EXCLUDED_RD_ADDONS
)
def should_use_ad_cache(self) -> bool:
"""Checks if AD cache should be used for this addon."""
return bool(
get_setting("alldebrid_enabled")
and get_setting("alldebrid_cached_check")
and not get_setting("torrent_enable")
and not get_setting("stremio_loggedin")
and self.addon.manifest.id not in EXCLUDED_RD_ADDONS
)
def get_cached_results(
self,
imdb_id: str,
mode: str,
season: Optional[int],
episode: Optional[int],
debridType="",
debridToken="",
) -> List[TorrentStream]:
cached_results = process_external_cache(
data={
"imdb_id": imdb_id,
"season": season,
"episode": episode,
"mode": mode,
},
debrid=debridType,
token=str(debridToken),
url=self.addon.url(),
)
if not cached_results:
return []
return self.parse_response(cached_results, is_external_cache=True)