import contextlib
import hashlib
import json
import os
import re
import unicodedata
from collections import deque
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import datetime, timedelta
from email.utils import parsedate_to_datetime
from enum import Enum
from typing import Dict, List
from urllib.parse import unquote
from zipfile import ZipFile
import requests
import xbmc
from xbmc import getSupportedMedia
from xbmcgui import Dialog, DialogProgressBG
from xbmcplugin import addDirectoryItem, setContent, setPluginCategory
from lib.api.fanart.fanart import get_fanart
from lib.api.tmdbv3api.tmdb import TMDb
from lib.api.trakt.lists_cache import lists_cache
from lib.api.trakt.main_cache import main_cache
from lib.api.trakt.trakt_cache import trakt_cache, trakt_watched_cache
from lib.api.tvdbapi.tvdbapi import TVDBAPI
from lib.clients.base import TorrentStream
from lib.clients.stremio.constants import (
STREMIO_ADDONS_CATALOGS_KEY,
STREMIO_ADDONS_KEY,
STREMIO_TV_ADDONS_KEY,
STREMIO_USER_ADDONS,
)
from lib.clients.subtitle.utils import get_language_code
from lib.db.cached import cache
from lib.db.pickle_db import PickleDatabase
from lib.gui.qr_progress_dialog import QRProgressDialog
from lib.utils.debrid.qrcode_utils import make_qrcode
from lib.utils.general.processors import PostProcessBuilder, PreProcessBuilder
from lib.utils.kodi.settings import get_cache_expiration, is_cache_enabled
from lib.utils.kodi.utils import (
ADDON_HANDLE,
ADDON_PATH,
EPISODES_TYPE,
MOVIES_TYPE,
SEASONS_TYPE,
SHOWS_TYPE,
TITLES_TYPE,
build_url,
container_refresh,
copy2clip,
dialog_text,
get_jacktorr_setting,
get_setting,
get_setting_fresh,
kodilog,
make_list_item,
notification,
sleep,
translatePath,
translation,
)
from lib.vendor.torf._magnet import Magnet
pickle_db = PickleDatabase()
PROVIDER_COLOR_MIN_BRIGHTNESS = 128
URL_REGEX = r"^(?!\/)(rtmps?:\/\/|mms:\/\/|rtsp:\/\/|https?:\/\/|ftp:\/\/)?([^\/:]+:[^\/@]+@)?(www\.)?(?=[^\/:\s]+\.[^\/:\s]+)([^\/:\s]+\.[^\/:\s]+)(:\d+)?(\/[^#\s]*[\s\S]*)?(\?[^#\s]*)?(#.*)?$"
USER_AGENT_HEADER = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/121.0.0.0 Safari/537.36"
}
USER_AGENT_STRING = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/121.0.0.0 Safari/537.36"
TMDB_IMAGE_SIZES = {
"low": {"poster": "w185", "thumb": "w185", "profile": "w185", "fanart": "w300"},
"medium": {"poster": "w342", "thumb": "w342", "profile": "w342", "fanart": "w780"},
"high": {"poster": "w780", "thumb": "w780", "profile": "w780", "fanart": "w1280"},
"original": {
"poster": "original",
"thumb": "original",
"profile": "original",
"fanart": "original",
},
}
def get_image_size(image_type):
tier_map = {"0": "low", "1": "medium", "2": "high", "3": "original"}
tier = tier_map.get(get_setting_fresh("image_resolution_tier", "2"), "high")
return TMDB_IMAGE_SIZES.get(tier, {}).get(image_type, "")
MEDIA_FUSION_DEFAULT_KEY = "eJwBYACf_4hAkZJe85krAoD5hN50-2M0YuyGmgswr-cis3uap4FNnLMvSfOc4e1IcejWJmykujTnWAlQKRi9cct5k3IRqhu-wFBnDoe_QmwMjJI3FnQtFNp2u3jDo23THEEgKXHYqTMrLos="
dialog_update = {"count": -1, "percent": 50}
UNDESIRED_QUALITIES = (
"SD",
"CAM",
"TELE",
"SYNC",
"TS",
"HDTV",
"HDCAM",
"HDTS",
"HDTC",
"HDTVRip",
)
video_extensions = (
".001",
".3g2",
".3gp",
".asf",
".asx",
".avc",
".avi",
".avs",
".bdm",
".bdmv",
".bin",
".bivx",
".dat",
".divx",
".dv",
".dvr-ms",
".evo",
".f4v",
".fli",
".flv",
".h264",
".img",
".iso",
".m2t",
".m2ts",
".m2v",
".m3u8",
".m4v",
".mk3d",
".mkv",
".mov",
".mp4",
".mpeg",
".mpg",
".mpl",
".mpls",
".mts",
".nrg",
".nuv",
".ogm",
".ogv",
".pva",
".qt",
".rcv",
".rec",
".rmvb",
".sdp",
".svq3",
".tp",
".trp",
".ts",
".ty",
".udf",
".vc1",
".vdr",
".viv",
".vob",
".vp3",
".webm",
".wmv",
".xvid",
)
non_direct_exts = {
".zip",
".rar",
".001",
".002",
".strm",
".ifo",
".vob",
".bdat",
".pvr",
".dvr-ms",
".disc",
".nrg",
".img",
".bin",
".iso",
}
class Enum:
@classmethod
def values(cls):
return [value for name, value in vars(cls).items() if not name.startswith("_")]
class DebridType(Enum):
RD = "RealDebrid"
PM = "Premiumize"
TB = "Torbox"
OC = "Offcloud"
ED = "EasyDebrid"
DB = "Debrider"
AD = "AllDebrid"
class IndexerType(Enum):
TORRENT = "Torrent"
DIRECT = "Direct"
DEBRID = "Debrid"
STREMIO_DEBRID = "Stremio"
class Indexer(Enum):
PROWLARR = "Prowlarr"
STREMIO = "Stremio"
JACKETT = "Jackett"
TORRENTIO = "Torrentio"
PEERFLIX = "Peerflix"
MEDIAFUSION = "MediaFusion"
JACKGRAM = "Jackgram"
TELEGRAM = "Telegram"
ELHOSTED = "Elfhosted"
BURST = "Burst"
ZILEAN = "Zilean"
EASYNEWS = "Easynews"
EXTERNAL_SCRAPER = "External Scraper"
class SearchVariant(Enum):
DEFAULT = "default"
TITLE_YEAR = "title_year"
ORIGINAL_TITLE = "original_title"
ORIGINAL_TITLE_YEAR = "original_title_year"
class Players(Enum):
JACKTORR = "Jacktorr"
TORREST = "Torrest"
ELEMENTUM = "Elementum"
JACKGRAM = "Jackgram"
class Anime(Enum):
SEARCH = "Anime_Search"
POPULAR = "Anime_Popular"
POPULAR_RECENT = "Anime_Popular_Recent"
TRENDING = "Anime_Trending"
TRENDING_RECENT = "Anime_Trending_Recent"
AIRING = "Anime_On_The_Air"
MOST_WATCHED = "Anime_Most_Watched"
FAVORITED = "Anime_Favorited"
YEARS = "Anime_Years"
GENRES = "Anime_Genres"
TOP_RATED = "Anime_Top_Rated"
class Animation(Enum):
POPULAR = "Animation_Popular"
class Cartoons(Enum):
SEARCH = "Cartoons_Search"
POPULAR = "Cartoons_Popular"
POPULAR_RECENT = "Cartoons_Popular_Recent"
YEARS = "Cartoons_Years"
torrent_clients = [
Players.JACKTORR,
Players.TORREST,
Players.ELEMENTUM,
]
torrent_indexers = [
Indexer.PROWLARR,
Indexer.JACKETT,
Indexer.TORRENTIO,
Indexer.PEERFLIX,
Indexer.ELHOSTED,
Indexer.BURST,
Indexer.EXTERNAL_SCRAPER,
]
PROVIDER_COLORS = {
"debrid": "FF1E90FF", # dodger blue
"direct": "FFFFA500", # orange
"stremio": "FF8A2BE2", # blue-violet
"torrent": "FFFFD700", # gold
}
class DialogListener:
def __init__(self):
self._dialog = DialogProgressBG()
@property
def dialog(self):
return self._dialog
def __enter__(self):
return self
def __exit__(self, exc_type, exc_val, exc_tb):
with contextlib.suppress(BaseException):
self._dialog.close()
def is_debrid_activated():
return (
get_setting_fresh("real_debrid_enabled")
or get_setting_fresh("premiumize_enabled")
or get_setting_fresh("torbox_enabled")
or get_setting_fresh("offcloud_enabled")
or get_setting_fresh("debrider_enabled")
or get_setting_fresh("alldebrid_enabled")
)
def check_debrid_enabled(debrid_type):
if debrid_type == DebridType.RD:
return is_rd_enabled()
elif debrid_type == DebridType.PM:
return is_pm_enabled()
elif debrid_type == DebridType.TB:
return is_tb_enabled()
elif debrid_type == DebridType.OC:
return is_oc_enabled()
elif debrid_type == DebridType.DB:
return is_debrider_enabled()
elif debrid_type == DebridType.AD:
return is_ad_enabled()
else:
kodilog(f"Unknown debrid type: {debrid_type}", level=xbmc.LOGERROR)
def is_rd_enabled():
return get_setting_fresh("real_debrid_enabled")
def is_ad_enabled():
return get_setting_fresh("alldebrid_enabled")
def is_pm_enabled():
return get_setting_fresh("premiumize_enabled")
def is_debrider_enabled():
return get_setting_fresh("debrider_enabled")
def is_tb_enabled():
return get_setting_fresh("torbox_enabled")
def is_oc_enabled():
return get_setting_fresh("offcloud_enabled")
def is_ed_enabled():
return get_setting("easydebrid_enabled")
def build_list_item(label, icon="", poster_path=""):
item = make_list_item(label=label)
item.setArt(
{
"poster": poster_path,
"icon": os.path.join(ADDON_PATH, "resources", "img", icon),
"thumb": os.path.join(ADDON_PATH, "resources", "img", icon),
"fanart": os.path.join(ADDON_PATH, "fanart.png"),
}
)
return item
def make_listing(data):
title = data.get("title")
ids = data.get("ids", {})
tv_data = data.get("tv_data", {})
mode = data.get("mode", "")
mode = "episode" if mode == "tv" else "movies"
list_item = make_list_item(label=title)
list_item.setContentLookup(False)
list_item.setProperty("IsPlayable", "true")
data["episode"] = tv_data.get("episode", data.get("episode", ""))
data["season"] = tv_data.get("season", data.get("season", ""))
data["name"] = tv_data.get("name", "")
data["id"] = ids.get("tmdb_id")
data["imdb_id"] = ids.get("imdb_id")
set_media_infoTag(list_item, data=data, mode=mode, detailed=True)
return list_item
def set_media_infoTag(list_item, data, fanart_data=None, mode="video", detailed=False):
if fanart_data is None:
fanart_data = {}
mode = "movies" if mode in ("movie", "movies") else mode
info_tag = list_item.getVideoInfoTag()
_set_basic_info(info_tag, data)
_set_media_type(info_tag, mode)
_set_identification(info_tag, data)
_set_artwork(list_item, data, fanart_data)
_set_released_info(info_tag, data)
_set_trakt_ui_state(list_item, info_tag, data, mode)
if mode == "tv" or mode == "season" or mode == "episode":
_set_show_info(info_tag, data, mode)
if detailed:
_set_detailed_info(info_tag, data, mode)
def _set_basic_info(info_tag, data):
info_tag.setTitle(data.get("title", data.get("name", "")))
info_tag.setOriginalTitle(
data.get("original_title", data.get("original_name", data.get("title", "")))
)
info_tag.setPlot(truncate_text(data.get("overview") or data.get("biography", "No overview")))
def _set_media_type(info_tag, mode):
if mode == "movies":
info_tag.setMediaType("movie")
elif mode == "tv":
info_tag.setMediaType("tvshow")
elif mode == "season":
info_tag.setMediaType("season")
elif mode == "episode":
info_tag.setMediaType("episode")
else:
info_tag.setMediaType("video")
def _set_identification(info_tag, data):
if "imdb_id" in data:
info_tag.setIMDBNumber(data["imdb_id"])
if "id" in data:
unique_ids = {
"tmdb": str(data.get("id", "")),
"imdb": data.get("imdb_id", ""),
"tvdb": data.get("tvdb_id", ""),
}
unique_ids = {k: str(v) for k, v in unique_ids.items() if v}
if unique_ids:
info_tag.setUniqueIDs(unique_ids, "tmdb")
def _set_artwork(list_item, data, fanart_data):
set_listitem_artwork(list_item, data, fanart_data)
def _set_released_info(info_tag, data):
if "first_air_date" in data:
first_air_date = data["first_air_date"]
if first_air_date:
info_tag.setFirstAired(first_air_date)
info_tag.setYear(int(first_air_date.split("-")[0]))
elif "air_date" in data:
info_tag.setFirstAired(data["air_date"])
elif "release_date" in data:
release_date = data["release_date"]
if release_date:
info_tag.setPremiered(release_date)
info_tag.setYear(int(release_date.split("-")[0]))
if "runtime" in data:
runtime = data.get("runtime")
if runtime:
info_tag.setDuration(runtime * 60)
def _set_detailed_info(info_tag, data, mode):
# Rating
info_tag.setRating(
data.get("vote_average", data.get("rating", 0)),
votes=data.get("vote_count", data.get("votes", 0)),
)
info_tag.setUserRating(int(float(data.get("popularity", 0))))
# Classification
genres = data.get("genre_ids", data.get("genres", []))
final_genres = extract_genres(genres, mode)
info_tag.setGenres(final_genres)
# Countries
countries = list(data.get("origin_country", data.get("countries", [])))
if isinstance(countries, str):
countries = [countries]
info_tag.setCountries(countries)
# Cast & crew
info_tag.setCast(get_cast_and_crew(data))
def _get_valid_media_id(*raw_ids):
for raw_id in raw_ids:
if raw_id in (None, ""):
continue
return str(raw_id)
return ""
def _coerce_int(value):
try:
return int(value)
except (TypeError, ValueError):
return None
def _get_runtime_seconds(data):
runtime = data.get("runtime")
if runtime in (None, ""):
return None
try:
runtime_seconds = int(float(runtime) * 60)
except (TypeError, ValueError):
return None
if runtime_seconds <= 0:
return None
return runtime_seconds
def _is_valid_trakt_progress(progress):
try:
progress = float(progress)
except (TypeError, ValueError):
return False
return 0 < progress <= 100
def _set_trakt_progress_state(list_item, info_tag, data, progress):
if not _is_valid_trakt_progress(progress):
return
progress = float(progress)
list_item.setProperty("PercentPlayed", str(progress))
total = _get_runtime_seconds(data)
if total is None:
return
current = total * progress / 100
info_tag.setResumePoint(current, total)
def _set_trakt_ui_state(list_item, info_tag, data, mode):
normalized_mode = "movies" if mode in ("movie", "movies") else mode
if normalized_mode == "movies":
tmdb_id = _get_valid_media_id(data.get("tmdb_id"), data.get("id"))
if not tmdb_id:
return
is_watched = trakt_watched_cache.get_watched_status("movie", tmdb_id)
if is_watched:
info_tag.setPlaycount(1)
return
progress = trakt_watched_cache.get_progress("movie", tmdb_id)
_set_trakt_progress_state(list_item, info_tag, data, progress)
return
if normalized_mode != "episode":
return
tmdb_id = _get_valid_media_id(data.get("tmdb_id"))
season = _coerce_int(data.get("season"))
episode = _coerce_int(data.get("episode"))
if not tmdb_id or season is None or episode is None:
return
is_watched = trakt_watched_cache.get_watched_status("episode", tmdb_id, season, episode)
if is_watched:
info_tag.setPlaycount(1)
return
progress = trakt_watched_cache.get_progress("episode", tmdb_id, season, episode)
_set_trakt_progress_state(list_item, info_tag, data, progress)
def _set_show_info(info_tag, data, mode):
if "seasons" in data:
seasons = list(data["seasons"])
if isinstance(seasons, list):
named_seasons = [
(season.get("season_number", 0), season.get("name", "")) for season in seasons
]
info_tag.addSeasons(named_seasons)
else:
info_tag.addSeason(seasons.get("season_number", 0), seasons.get("name", ""))
if mode == "tv" or mode == "season":
info_tag.setTvShowTitle(data.get("title", data.get("name", "")))
season = _coerce_int(data.get("season", data.get("season_number", 0)))
if season is not None:
info_tag.setSeason(season)
elif mode == "episode":
info_tag.setTvShowTitle(data.get("title", data.get("name", "")))
season = _coerce_int(data.get("season", data.get("season_number", 0)))
episode = _coerce_int(data.get("episode", data.get("episode_number", 0)))
if season is not None:
info_tag.setSeason(season)
if episode is not None:
info_tag.setEpisode(episode)
def extract_genres(genres, media_type="movies"):
genre_list = []
path = "movie_genres" if media_type == "movies" else "show_genres"
from lib.clients.tmdb.utils.utils import tmdb_get
genre_response = tmdb_get(path=path)
if not genre_response or "genres" not in genre_response:
kodilog(f"Failed to fetch genres for {media_type}", level=xbmc.LOGDEBUG)
return genre_list
genre_mapping = {g["id"]: g["name"] for g in genre_response.get("genres", [])}
for g in genres:
if isinstance(g, dict) and "name" in g: # Case: { "id": 28, "name": "Action" }
genre_list.append(g["name"])
elif isinstance(g, int):
genre_list.append(genre_mapping.get(g, str(g)))
elif isinstance(g, str):
try:
genre_id = int(g)
genre_list.append(genre_mapping.get(genre_id, g))
except ValueError:
genre_list.append(g)
return genre_list
def get_cast_and_crew(data):
"""Return combined list of xbmc.Actor objects for cast and crew."""
cast_list = []
casts = extract_cast(data)
crew = extract_crew(data)
# Build actors
cast_list.extend(build_actor(c, is_cast=True) for c in casts)
cast_list.extend(build_actor(c, is_cast=False) for c in crew)
return cast_list
def extract_cast(data):
if hasattr(data, "to_dict"):
data = data.to_dict()
credits = data.get("credits", {})
if isinstance(credits, dict) and "cast" in credits:
return credits.get("cast", []) or []
casts = data.get("casts", {})
if isinstance(casts, dict) and "cast" in casts:
return casts.get("cast", []) or []
if "cast" in data:
return data.get("cast", []) or []
if "actors" in data:
return data.get("actors", []) or []
return []
def extract_crew(data):
if hasattr(data, "to_dict"):
data = data.to_dict()
credits = data.get("credits", {})
if isinstance(credits, dict) and "crew" in credits:
return credits.get("crew", []) or []
if "crew" in data:
return data.get("crew", []) or []
return []
def build_actor(member, is_cast=True):
"""Build an xbmc.Actor from a cast or crew member dict."""
if is_cast:
return xbmc.Actor(
name=member.get("name", "Unknown"),
role=member.get("character", "Unknown"),
thumbnail=(
f"http://image.tmdb.org/t/p/w185{member['profile_path']}"
if member.get("profile_path")
else ""
),
order=member.get("order", 0),
)
else:
return xbmc.Actor(
name=member.get("name", "Unknown"),
role=member.get("job", "Unknown"),
thumbnail=(
f"http://image.tmdb.org/t/p/w185{member['profile_path']}"
if member.get("profile_path")
else ""
),
order=0, # Crew usually has no ordering
)
def set_listitem_artwork(list_item, data, fanart_data):
thumb_sources = [
(data.get("poster_path"), get_image_size("thumb")),
(data.get("still_path"), get_image_size("thumb")),
]
poster_sources = [
(data.get("poster_path"), get_image_size("poster")),
(data.get("still_path"), get_image_size("poster")),
(data.get("profile_path"), get_image_size("profile")),
]
fanart_sources = [
(data.get("backdrop_path"), get_image_size("fanart")),
(data.get("still_path"), get_image_size("fanart")),
]
clear_logo = [(extract_tmdb_logo_url(data), "original")]
def first_valid(sources, fallback_key=""):
for path, size in sources:
url = tmdb_url(path, size)
if url:
return url
if fallback_key:
return data.get(fallback_key, "") or fanart_data.get(fallback_key, "")
return ""
list_item.setArt(
{
"thumb": first_valid(thumb_sources),
"poster": first_valid(poster_sources, "poster"),
"fanart": first_valid(fanart_sources, "fanart"),
"icon": first_valid(poster_sources),
"banner": first_valid(fanart_sources, "banner"),
"clearart": first_valid(fanart_sources, "clearart"),
"clearlogo": first_valid(clear_logo, "clearlogo"),
"tvshow.clearart": first_valid(fanart_sources, "clearart"),
"tvshow.clearlogo": first_valid(clear_logo, "clearlogo"),
"tvshow.landscape": first_valid(fanart_sources, "landscape"),
"tvshow.banner": first_valid(fanart_sources, "banner"),
}
)
def extract_tmdb_logo_url(data):
images = data.get("images", {}) or {}
logos = images.get("logos", []) or []
if logos:
file_path = logos[0].get("file_path")
if file_path:
return file_path
def tmdb_url(path, size):
return f"http://image.tmdb.org/t/p/{size}{path}" if path else ""
def get_rpdb_poster(imdb_id, api_key):
cache_key = f"rpdb_poster|{imdb_id}"
cached = cache.get(cache_key)
if cached:
return cached
try:
response = requests.get(
f"https://api.ratingposterdb.com/{api_key}/imdb/{imdb_id}?lang=en",
timeout=10,
)
if response.status_code == 200:
data = response.json()
poster_url = data.get("poster") or data.get("poster_large")
if poster_url:
cache.set(cache_key, poster_url, timedelta(hours=24))
return poster_url
except requests.RequestException as e:
kodilog(f"RPDB request error: {e}")
return None
def build_media_metadata(ids, mode: str) -> dict:
metadata = {
"poster": "",
"fanart": "",
"clearlogo": "",
"overview": "",
"clearart": "",
"keyart": "",
"banner": "",
"landscape": "",
"title": "",
"original_title": "",
"genres": [],
"countries": [],
"year": None,
"runtime": None,
"tmdb_id": 0,
"tvdb_id": 0,
"vote_average": 0,
"vote_count": 0,
"popularity": 0,
"cast": [],
}
tmdb_id = ids.get("tmdb_id", "")
tvdb_id = ids.get("tvdb_id", "")
imdb_id = ids.get("imdb_id", "")
if not tmdb_id and imdb_id:
from lib.clients.tmdb.utils.utils import tmdb_get
res = tmdb_get("find_by_imdb_id", imdb_id)
if mode == "movies" and getattr(res, "movie_results", []):
tmdb_id = str(res.movie_results[0]["id"])
ids["tmdb_id"] = tmdb_id
elif mode in ("tv", "episode", "season", "anime") and getattr(res, "tv_results", []):
tmdb_id = str(res.tv_results[0]["id"])
ids["tmdb_id"] = tmdb_id
metadata["tmdb_id"] = str(tmdb_id)
metadata["tvdb_id"] = str(tvdb_id)
tmdb_logo_path = None
# TMDB details
if tmdb_id:
from lib.clients.tmdb.utils.utils import get_tmdb_media_details
details = get_tmdb_media_details(tmdb_id, mode)
poster_path = getattr(details, "poster_path", "")
metadata["poster"] = tmdb_url(poster_path, get_image_size("poster")) if poster_path else ""
if imdb_id and get_setting_fresh("rpdb_enabled"):
rpdb_api_key = get_setting_fresh("rpdb_api_key", "")
if rpdb_api_key:
rpdb_poster = get_rpdb_poster(imdb_id, rpdb_api_key)
if rpdb_poster:
metadata["poster"] = rpdb_poster
metadata["overview"] = getattr(details, "overview", "")
metadata["title"] = getattr(details, "title", getattr(details, "name", ""))
metadata["original_title"] = getattr(
details, "original_title", getattr(details, "original_name", "")
)
metadata["year"] = (
int(str(getattr(details, "release_date", "0"))[:4])
if getattr(details, "release_date", None)
else None
)
metadata["runtime"] = getattr(details, "runtime", None)
metadata["vote_average"] = getattr(details, "vote_average", 0)
metadata["vote_count"] = getattr(details, "vote_count", 0)
metadata["popularity"] = getattr(details, "popularity", 0)
metadata["cast"] = extract_cast(details)
tmdb_logo_path = extract_tmdb_logo_url(details) or ""
if tmdb_logo_path:
metadata["clearlogo"] = tmdb_url(tmdb_logo_path, "original")
backdrop_path = getattr(details, "backdrop_path", "")
if backdrop_path:
metadata["fanart"] = tmdb_url(backdrop_path, "original")
# Fanart details
if tmdb_id or tvdb_id:
fanart_details = get_fanart_details(tvdb_id=tvdb_id, tmdb_id=tmdb_id, mode=mode)
if not tmdb_logo_path and fanart_details.get("clearlogo"):
metadata["clearlogo"] = fanart_details.get("clearlogo")
for key in ("fanart", "clearart", "keyart", "banner", "landscape"):
if fanart_details.get(key):
metadata[key] = fanart_details.get(key)
return metadata
def set_watched_file(data):
title = data.get("title", "")
is_torrent = data.get("is_torrent", False)
is_direct = data.get("type", "") == IndexerType.DIRECT
is_stremio = data.get("type", "") == IndexerType.STREMIO_DEBRID
if is_direct:
color = get_random_color("Direct", formatted=False)
title = f"[B][COLOR {color}][Direct][/COLOR][/B] - {title}"
elif is_stremio:
color = get_random_color("Stremio", formatted=False)
title = f"[B][COLOR {color}][Stremio Debrid][/COLOR][/B] - {title}"
elif is_torrent:
color = get_random_color("Torrent", formatted=False)
title = f"[B][COLOR {color}][Torrent][/COLOR][/B] - {title}"
else:
color = get_random_color("Cached", formatted=False)
title = f"[B][COLOR {color}][Cached][/COLOR][/B] - {title}"
data["timestamp"] = datetime.now().strftime("%a, %d %b %Y %I:%M %p")
pickle_db.set_item(key="jt:lfh", subkey=title, value=data)
def set_watched_title(title, ids, mode, tg_data="", media_type="", library_data=None):
payload = {
"timestamp": datetime.now().strftime("%a, %d %b %Y %I:%M %p"),
"ids": ids,
"mode": media_type if mode == "multi" else mode,
"tg_data": tg_data,
}
if library_data:
payload.update(library_data)
pickle_db.set_item(key="jt:lth", subkey=title, value=payload)
library_payload = {
"title": title,
"ids": ids,
"mode": media_type if mode == "multi" else mode,
"tg_data": tg_data,
"timestamp": payload["timestamp"],
}
if library_data:
library_payload.update(library_data)
add_to_library(library_payload)
def add_to_library(data):
title = data.get("title")
if not title:
return
pickle_db.set_item(key="jt:lib", subkey=title, value=data)
cache.delete("library_view|tv")
cache.delete("library_view|movies")
def remove_from_library(title):
pickle_db.delete_item(key="jt:lib", subkey=title)
cache.delete("library_view|tv")
cache.delete("library_view|movies")
def is_in_library(title):
return pickle_db.get_item("jt:lib", title) is not None
def get_fanart_details(tvdb_id="", tmdb_id="", mode="tv"):
identifier = "{}|{}".format("fanart.tv", tvdb_id if tvdb_id and tvdb_id != "None" else tmdb_id)
cached = cache.get(identifier)
if cached:
return cached
from lib.clients.tmdb.utils.utils import LANGUAGES
try:
language_index = int(get_setting("language", 18))
lang = LANGUAGES[language_index].split("-")[0].strip()
except Exception:
lang = "en"
if mode == "tv":
data = get_fanart(mode, lang, tvdb_id)
else:
data = get_fanart(mode, lang, tmdb_id)
if data:
hours = get_cache_expiration() if is_cache_enabled() else 0
cache.set(identifier, data, timedelta(hours=hours))
return data or {}
def get_cached_results(query, mode, media_type, episode, cache_scope=""):
if mode == "tv" or media_type == "tv" or mode == "anime":
return get_cached(query, params=(episode, "index", cache_scope))
return get_cached(query, params=("index", cache_scope))
def cache_results(results, query, mode, media_type, episode, cache_scope=""):
if mode == "tv" or media_type == "tv" or mode == "anime":
set_cached(results, query, params=(episode, "index", cache_scope))
else:
set_cached(results, query, params=("index", cache_scope))
def get_cached(path, params=None):
if params is None:
params = {}
identifier = f"{path}|{params}"
return cache.get(identifier)
def set_cached(data, path, params=None):
if params is None:
params = {}
identifier = f"{path}|{params}"
cache.set(
identifier,
data,
timedelta(hours=get_cache_expiration() if is_cache_enabled() else 0),
)
def db_get(name, func, path, params):
identifier = f"{path}|{params}"
data = cache.get(identifier)
if not data:
if name == "search_client":
data = func()
cache.set(
identifier,
data,
timedelta(hours=get_cache_expiration() if is_cache_enabled() else 0),
)
return data
def tvdb_get(path, params=None):
if params is None:
params = {}
identifier = f"{path}|{params}"
data = cache.get(identifier)
if data:
return data
if path == "get_imdb_id":
data = TVDBAPI().get_imdb_id(params)
cache.set(
identifier,
data,
timedelta(hours=get_cache_expiration() if is_cache_enabled() else 0),
)
return data
def set_pluging_category(heading: str):
setPluginCategory(ADDON_HANDLE, heading)
def set_content_type(mode, media_type="movies"):
if mode == "season":
setContent(ADDON_HANDLE, SEASONS_TYPE)
elif mode == "episode":
setContent(ADDON_HANDLE, EPISODES_TYPE)
elif mode == "files":
setContent(ADDON_HANDLE, "files")
elif mode in ("tv", "anime") or media_type == "tv":
setContent(ADDON_HANDLE, SHOWS_TYPE)
elif mode == "movies" or media_type == "movies":
setContent(ADDON_HANDLE, MOVIES_TYPE)
else:
setContent(ADDON_HANDLE, TITLES_TYPE)
def get_provider_color(provider_name, formatted=True):
key = provider_name.lower()
if key in PROVIDER_COLORS:
color = PROVIDER_COLORS[key]
else:
color = get_random_color(provider_name, formatted=False)
if formatted:
return f"[B][COLOR {color}]{provider_name}[/COLOR][/B]"
else:
return color
# This method was taken from script.elementum.jackett addon
def get_random_color(provider_name, formatted=True):
hash = hashlib.sha256(provider_name.encode("utf")).hexdigest()
colors = []
spec = 10
for i in range(0, 3):
offset = spec * i
rounded = round(int(hash[offset : offset + spec], 16) / int("F" * spec, 16) * 255)
colors.append(int(max(rounded, PROVIDER_COLOR_MIN_BRIGHTNESS)))
while (sum(colors) / 3) < PROVIDER_COLOR_MIN_BRIGHTNESS:
for i in range(0, 3):
colors[i] += 10
for i in range(0, 3):
colors[i] = f"{colors[i]:02x}"
color_format = "FF" + "".join(colors).upper()
if formatted:
return f"[B][COLOR {color_format}]{provider_name}[/COLOR][/B]"
else:
return color_format
def get_colored_languages(languages):
if not languages:
return ""
return " ".join(get_random_color(lang) for lang in languages)
def execute_thread_pool(results, func, *args, **kwargs):
thread_number = get_setting("thread_number", 8)
with ThreadPoolExecutor(max_workers=int(thread_number)) as executor:
futures = [executor.submit(func, res, *args, **kwargs) for res in results]
for future in as_completed(futures):
try:
future.result()
except Exception as e:
kodilog(f"Thread pool error: {e}")
def execute_thread_pool_collection(results, func, *args, **kwargs):
thread_number = get_setting("thread_number", 6)
collected = []
def wrapper(res):
try:
result = func(res, *args, **kwargs)
if result is not None:
collected.append(result)
except Exception as e:
kodilog(f"Thread pool error: {e}")
with ThreadPoolExecutor(max_workers=int(thread_number)) as executor:
executor.map(wrapper, results)
return collected
def paginate_list(data, page_size=10):
for i in range(0, len(data), page_size):
yield data[i : i + page_size]
def clear_cache_on_update():
clear_history_by_type(update=True)
def clear_all_cache():
cache.clean_all()
def clear_trakt_db_cache():
trakt_cache.clear_all()
trakt_watched_cache.clear_all()
lists_cache.delete_all_lists()
main_cache.delete_all()
notification(translation(30244))
def clear_tmdb_cache():
prefixes = [
"search_%",
"movie_%",
"tv_%",
"season_%",
"episode_%",
"discover_%",
"trending_%",
"popular_%",
"person_%",
"find_%",
"anime_%",
"collection_%",
"get_imdb_id%",
"latest_%",
"tmdb_ui_meta",
]
cache.delete("movie_genres|None")
cache.delete("show_genres|None")
for prefix in prefixes:
cache.delete_like(f"{prefix}|%")
TMDb().cache_clear()
notification(translation(30240))
def clear_database_cache():
clear_history_by_type("all")
def clear_stremio_cache():
cache.delete(STREMIO_ADDONS_KEY)
cache.delete(STREMIO_ADDONS_CATALOGS_KEY)
cache.delete(STREMIO_TV_ADDONS_KEY)
cache.delete(STREMIO_USER_ADDONS)
cache.delete_like("search_catalog%")
cache.delete_like("list_catalog%")
cache.delete_like("list_stremio_%")
notification(translation(30244))
def clear_debrid_cache():
cache.delete_like("%|%deb%")
cache.delete_like("real_debrid%")
cache.delete_like("premiumize%")
cache.delete_like("all_debrid%")
cache.delete_like("torbox%")
cache.delete_like("debrider%")
notification(translation(30244))
def clear_mdblist_cache():
cache.delete("get_user_lists|None")
cache.delete("top_mdbd_lists|None")
cache.delete_like("get_user_lists%")
cache.delete_like("get_list_items%")
cache.delete_like("top_mdbd_lists%")
cache.delete_like("search_lists%")
notification(translation(30244))
def clear_history_by_type(type="all", update=False):
if update:
msg = translation(90110)
else:
msg = translation(90111)
confirmed = Dialog().yesno(translation(90112), msg)
if confirmed:
keys = []
if type == "lth":
keys = ["jt:lth"]
elif type == "lfh":
keys = ["jt:lfh"]
else:
keys = ["jt:lth", "jt:lfh"]
for key in keys:
pickle_db.set_key(key, {})
container_refresh()
def limit_results(results):
limit = int(get_setting("indexers_total_results", 10))
return results[:limit]
def get_description_length():
return int(get_setting("indexers_desc_length", 0))
def truncate_text(text, max_length=None):
if max_length is None:
max_length = get_description_length()
if not max_length or max_length <= 0:
return text
if not text:
return text
if len(text) <= max_length:
return text
return text[:max_length].rstrip() + "..."
def remove_duplicate(results):
seen_values = []
result_dict = []
for res in results:
if res not in seen_values:
result_dict.append(res)
seen_values.append(res)
return result_dict
def unzip(zip_location, destination_location, destination_check):
try:
zipfile = ZipFile(zip_location)
zipfile.extractall(path=destination_location)
if os.path.exists(destination_check):
status = True
else:
status = False
except Exception:
status = False
return status
def pre_process(
results: List[TorrentStream],
mode: str,
episode_name: str,
episode: int,
season: int,
skip_episode_filter: bool = False,
) -> List[TorrentStream]:
builder = PreProcessBuilder(results).remove_duplicates()
builder.extract_codec_hdr()
builder.filter_by_codec()
builder.filter_by_hdr()
if mode == "tv" and not skip_episode_filter:
builder.filter_sources(episode_name, episode, season)
builder.filter_by_source()
builder.filter_by_quality()
if get_setting("filter_size_enabled"):
builder.filter_by_size()
results = builder.get_results()
return results
def post_process(results: List[TorrentStream], season: int = 0) -> List[TorrentStream]:
return (
PostProcessBuilder(results)
.check_season_pack(season)
.sort_results()
.limit_results()
.get_results()
)
def filter_debrid_episode(
files, episode_num: int, season_num: int, strict: bool = True
) -> List[Dict]:
str_season, str_episode = str(season_num), str(episode_num)
season_fill, episode_fill = str_season.zfill(2), str_episode.zfill(2)
ep_plus_1 = str(episode_num + 1).zfill(2)
ep_minus_1 = str(episode_num - 1).zfill(2)
def get_filename(file):
return file.get("path") or file.get("filename") or file.get("name") or file.get("n") or ""
# Normalize filenames for matching
def normalize_title(title):
title = unquote(title).replace("'", "")
title = re.sub(r"[^A-Za-z0-9-]+", ".", title)
return title.lower()
string_list = []
# SXXEYY variants and S2 - 11 / S02.11
for s, e in [
(season_fill, episode_fill),
(str_season, episode_fill),
(season_fill, str_episode),
(str_season, str_episode),
]:
string_list.append(rf"s{s}[.-]?e{e}")
string_list.append(rf"s{s}[.\s-]?{e}") # S2 - 11 or S02.11
# Season X Episode Y or SxE patterns
for s, e in [
(season_fill, episode_fill),
(str_season, episode_fill),
(season_fill, str_episode),
(str_season, str_episode),
]:
string_list.append(rf"(season[.-]?{s}[.-]?episode[.-]?{e})")
string_list.append(rf"{s}[x.]?{e}") # 2x11 or 02.11
# Episode ±1
string_list.append(rf"s{season_fill}e{ep_minus_1}[.-]?e{episode_fill}")
string_list.append(rf"s{season_fill}e{episode_fill}[.-]?e{ep_plus_1}")
# Episode-only patterns
string_list.append(rf"episode[.-]?{episode_fill}")
string_list.append(rf"[.-]ep[.-]?{episode_fill}")
string_list.append(r"cap\.")
# Optional strict negative lookahead to avoid false positives like 10x11 matching 1x01
if strict:
lookahead = rf"^(?=.*\.e?0*{episode_fill}\.)(?:(?!((?:s|season)[.-]?\d+[.-x]?(?:ep?|episode)[.-]?\d+)|\d+x\d+).)*$"
string_list.append(lookahead)
combined_pattern = "|".join(string_list)
try:
regex = re.compile(combined_pattern, re.IGNORECASE)
except re.error as e:
kodilog(f"Regex compilation failed: {e}")
return files
filtered = [f for f in files if regex.search(normalize_title(get_filename(f)))]
return filtered
def clean_auto_play_undesired(results: List[TorrentStream]) -> List[TorrentStream]:
return [
r
for r in results
if not getattr(r, "isPack", False)
and not any(u.lower() in r.title.lower() for u in UNDESIRED_QUALITIES)
]
def is_torrent_url(uri):
res = requests.head(uri, timeout=20, headers=USER_AGENT_HEADER)
return bool(res.status_code == 200 and res.headers.get("Content-Type") == "application/octet-stream")
def supported_video_extensions():
media_types = getSupportedMedia("video").split("|")
return [ext for ext in media_types if ext not in non_direct_exts]
def add_next_button(func_name, page=1, **kwargs):
list_item = make_list_item(label="Next")
list_item.setArt({"icon": os.path.join(ADDON_PATH, "resources", "img", "nextpage.png")})
page += 1
addDirectoryItem(
ADDON_HANDLE,
build_url(func_name, page=page, **kwargs),
list_item,
isFolder=True,
)
def is_video(s):
return s.lower().endswith(video_extensions)
def get_info_hash_from_magnet(magnet):
return Magnet.from_string(magnet).infohash
def is_magnet_link(link):
if link.startswith("magnet:?"):
return link
def is_url(url):
return bool(re.match(URL_REGEX, url))
def info_hash_to_magnet(info_hash):
return f"magnet:?xt=urn:btih:{info_hash}"
def get_state_string(state):
if 0 <= state <= 9:
return translation(30650 + state)
return translation(30660)
def get_service_host():
return get_jacktorr_setting("service_host")
def get_username():
return get_jacktorr_setting("service_login")
def get_password():
return get_jacktorr_setting("service_password")
def ssl_enabled():
return bool(get_jacktorr_setting("ssl_connection"))
def get_port():
return get_jacktorr_setting("service_port")
def unicode_flag_to_country_code(unicode_flag):
if len(unicode_flag) != 2:
return "Invalid flag Unicode"
first_letter = unicodedata.name(unicode_flag[0]).replace(
"REGIONAL INDICATOR SYMBOL LETTER ", ""
)
second_letter = unicodedata.name(unicode_flag[1]).replace(
"REGIONAL INDICATOR SYMBOL LETTER ", ""
)
country_code = first_letter.lower() + second_letter.lower()
return country_code
def debrid_dialog_update(type, total, dialog, lock):
with lock:
dialog_update["count"] += 1
dialog_update["percent"] += 2
dialog.update(
dialog_update.get("percent"),
f"Jacktook [COLOR FFFF6B00]Debrid-{type}[/COLOR]",
f"Checking: {dialog_update.get('count')}/{total}",
)
def get_public_ip():
public_ip = cache.get("public_ip")
if public_ip:
return public_ip
urls = ["https://ipconfig.io/ip", "https://api.ipify.org/"]
for url in urls:
try:
response = requests.get(url, timeout=5)
response.raise_for_status()
public_ip = response.text.strip()
cache.set(
"public_ip",
public_ip,
timedelta(hours=get_cache_expiration()),
)
return public_ip
except requests.RequestException as e:
kodilog(f"Error getting public IP from {url}: {e}")
return None
def export_to_kodi_paste(log_content):
try:
response = requests.post(
"https://paste.kodi.tv/documents",
data=log_content,
headers=USER_AGENT_HEADER,
)
kodilog(f"Paste.kodi.tv response: {response.status_code} - {response.text}")
if response.status_code == 200 and response.text:
paste_id = response.json().get("key", {})
if paste_id:
return f"https://paste.kodi.tv/{paste_id}"
return None
except Exception as e:
kodilog(f"Error exporting logs to paste.kodi.tv: {e}")
return None
def show_log_export_dialog(params):
log_file = params.get("log_file")
kodi_log_path = os.path.join(translatePath("special://logpath"), log_file)
if os.path.exists(kodi_log_path):
try:
with open(kodi_log_path, encoding="utf-8", errors="ignore") as f:
lines = deque(f, maxlen=500)
content = "".join(reversed(lines))
# Ask user if they want to export
from xbmcgui import Dialog
dialog = Dialog()
choice = dialog.select(
translation(90566),
[translation(90567), translation(90568)],
)
if choice == 1:
paste_url = export_to_kodi_paste(content)
qr_code = make_qrcode(paste_url)
copy2clip(paste_url)
progressDialog = QRProgressDialog("qr_dialog.xml", ADDON_PATH)
progressDialog.setup(
translation(90569),
qr_code,
paste_url,
is_debrid=False,
)
if paste_url:
count = 20
progressDialog.show_dialog()
while not progressDialog.iscanceled and count >= 0:
try:
count -= 1
progressDialog.update_progress(count)
except Exception:
pass
sleep(1000 * count)
else:
notification(translation(90601))
else:
dialog_text(translation(90566), content)
except Exception as e:
notification(translation(90602) % e)
else:
notification(translation(90603))
def extract_publish_date(date):
if not date:
return ""
match = re.search(r"\d{4}-\d{2}-\d{2}", date)
return match.group() if match else ""
def extract_release_group(title):
if not title:
return ""
# Remove file extension if present
title = re.sub(r"\.[a-z0-9]{3,4}$", "", title, flags=re.IGNORECASE)
# Search for -GroupName at the end (common in scene releases)
match = re.search(r"-([a-zA-Z0-9]+)$", title)
if match:
return match.group(1)
# Search for [GroupName] at the beginning (common in anime/p2p)
match = re.search(r"^\[([a-zA-Z0-9]+)\]", title)
if match:
return match.group(1)
return ""
def translate_weekday(weekday_name, lang="eng"):
sub_language = str(get_setting("subtitle_language"))
if sub_language and sub_language.lower() != "none":
lang = get_language_code(sub_language)
return WEEKDAY_TRANSLATIONS.get(lang, WEEKDAY_TRANSLATIONS["eng"]).get(
weekday_name, weekday_name
)
WEEKDAY_TRANSLATIONS = {
"eng": {
"Monday": "Monday",
"Tuesday": "Tuesday",
"Wednesday": "Wednesday",
"Thursday": "Thursday",
"Friday": "Friday",
"Saturday": "Saturday",
"Sunday": "Sunday",
},
"spa": {
"Monday": "Lunes",
"Tuesday": "Martes",
"Wednesday": "Miércoles",
"Thursday": "Jueves",
"Friday": "Viernes",
"Saturday": "Sábado",
"Sunday": "Domingo",
},
"por": {
"Monday": "Segunda-feira",
"Tuesday": "Terça-feira",
"Wednesday": "Quarta-feira",
"Thursday": "Quinta-feira",
"Friday": "Sexta-feira",
"Saturday": "Sábado",
"Sunday": "Domingo",
},
}
def parse_time(item):
ts = item[1].get("timestamp")
if ts:
try:
normalized = (
unicodedata.normalize("NFKC", ts)
.replace("a. m.", "AM")
.replace("p. m.", "PM")
.replace("a.m.", "AM")
.replace("p.m.", "PM")
.replace("a. m", "AM")
.replace("p. m", "PM")
)
normalized = re.sub(r"\s+", " ", normalized).strip()
day_replacements = {
"lun": "Mon",
"mar": "Tue",
"mie": "Wed",
"mié": "Wed",
"jue": "Thu",
"vie": "Fri",
"sab": "Sat",
"sáb": "Sat",
"dom": "Sun",
}
month_replacements = {
"ene": "Jan",
"feb": "Feb",
"mar": "Mar",
"abr": "Apr",
"may": "May",
"jun": "Jun",
"jul": "Jul",
"ago": "Aug",
"sep": "Sep",
"oct": "Oct",
"nov": "Nov",
"dic": "Dec",
}
if "," in normalized:
weekday_part, remainder = normalized.split(",", 1)
weekday_key = weekday_part.strip().lower()
normalized = (
f"{day_replacements.get(weekday_key, weekday_part.strip())},{remainder}"
)
normalized = re.sub(
r"(?<=,\s\d{2}\s)([A-Za-zÁÉÍÓÚáéíóú]{3})(?=\s\d{4})",
lambda match: month_replacements.get(match.group(1).lower(), match.group(1)),
normalized,
)
try:
return datetime.strptime(normalized, "%a, %d %b %Y %I:%M %p")
except Exception:
parsed = parsedate_to_datetime(normalized)
if parsed:
return parsed.replace(tzinfo=None) if parsed.tzinfo else parsed
except Exception as e:
kodilog(e)
return datetime.min
return datetime.min
def format_season_episode(season, episode):
def normalize(value):
if value in (None, ""):
return None
if isinstance(value, str):
value = value.strip()
if not value:
return None
try:
numeric_value = float(value)
except (TypeError, ValueError):
return str(value)
if numeric_value.is_integer():
return int(numeric_value)
return str(value)
season_value = normalize(season)
episode_value = normalize(episode)
if isinstance(season_value, int) and isinstance(episode_value, int):
return f"S{season_value:02d}E{episode_value:02d}"
if season_value is not None and episode_value is not None:
return f"S{season_value}E{episode_value}"
if season_value is not None:
if isinstance(season_value, int):
return f"S{season_value:02d}"
return f"S{season_value}"
if episode_value is not None:
if isinstance(episode_value, int):
return f"E{episode_value:02d}"
return f"E{episode_value}"
return ""
def safe_json_loads(value, default=None):
"""
Safely load JSON string. Returns default if parsing fails or value is None/empty.
"""
if default is None:
default = {}
if not value:
return default
try:
return json.loads(value)
except (json.JSONDecodeError, TypeError, ValueError):
return default
def normalize_tv_data(tv_data):
if not isinstance(tv_data, dict):
return {}
normalized = dict(tv_data)
# URL-decode the name to handle & in episode titles (issue #198)
name = normalized.get("name", "")
if name and isinstance(name, str):
normalized["name"] = unquote(name)
for key in ("season", "episode"):
value = _coerce_int(normalized.get(key))
if value is None:
normalized.pop(key, None)
else:
normalized[key] = value
return normalized