import concurrent
from typing import Any, List, Optional
from urllib.parse import quote
import requests
from lib.clients.base import BaseClient
from lib.domain.torrent import TorrentStream
from lib.jacktook.utils import kodilog
from lib.utils.general.utils import USER_AGENT_HEADER
from lib.utils.kodi.logging import summarize_locator_for_log
from lib.utils.kodi.settings import get_jackett_timeout
from lib.utils.kodi.utils import get_setting, notification, translation
from lib.utils.parsers import xmltodict
class Jackett(BaseClient):
def __init__(
self,
host: str,
apikey: str,
port: str,
notification,
session: Optional[requests.Session] = None,
) -> None:
super().__init__(host, notification)
self.apikey = apikey
self.port = port
self.host = host
self.base_url = self._make_base_url("all")
self.session = session or requests.Session()
self.session.headers.update(USER_AGENT_HEADER.copy())
def _make_base_url(self, indexer_id: str) -> str:
return f"{self.host}:{self.port}/api/v2.0/indexers/{indexer_id}/results/torznab/api?apikey={self.apikey}"
def _build_url(
self,
base_url: str,
query: str,
mode: str,
season: Optional[int] = None,
episode: Optional[int] = None,
categories: Optional[List[int]] = None,
additional_params: Optional[dict] = None,
season_pack=False,
year: Optional[int] = None,
) -> str:
url = f"{base_url}&q={quote(query)}"
if mode == "tv":
url += "&t=tvsearch"
if season_pack:
url += f"&season={season}"
else:
url += f"&season={season}&ep={episode}"
elif mode == "movies":
url += "&t=movie"
else:
url += "&t=search"
if categories:
url += f"&cat={','.join(map(str, categories))}"
if year and mode == "movies":
url += f"&year={year}"
if additional_params:
for key, value in additional_params.items():
url += f"&{key}={value}"
return url
def search_indexer(
self,
base_url,
query: str,
mode: str,
season: Optional[int] = None,
episode: Optional[int] = None,
categories: Optional[List[int]] = None,
additional_params: Optional[dict] = None,
year: Optional[int] = None,
) -> Optional[List[TorrentStream]]:
try:
results = []
futures = []
with concurrent.futures.ThreadPoolExecutor() as executor:
if mode == "tv" and season:
if episode:
url_ep = self._build_url(
base_url,
query,
mode,
season,
episode,
categories,
additional_params,
year=year,
)
futures.append(
executor.submit(self.session.get, url_ep, timeout=get_jackett_timeout())
)
if get_setting("include_season_packs"):
url_season = self._build_url(
base_url,
query,
mode,
season,
None,
categories,
additional_params,
season_pack=True,
year=year,
)
futures.append(
executor.submit(
self.session.get,
url_season,
timeout=get_jackett_timeout(),
)
)
else:
url = self._build_url(
base_url,
query,
mode,
season,
episode,
categories,
additional_params,
year=year,
)
futures.append(
executor.submit(self.session.get, url, timeout=get_jackett_timeout())
)
for future in concurrent.futures.as_completed(futures):
response = future.result()
if response.status_code == 200:
res = self.parse_response(response)
if res:
results.extend(res)
else:
notification(f"{translation(30229)} ({response.status_code})")
return results if results else None
except Exception as e:
self.handle_exception(f"{translation(30229)}: {e!s}")
return None
def search(
self,
query: str,
mode: str,
season: Optional[int] = None,
episode: Optional[int] = None,
categories: Optional[List[int]] = None,
additional_params: Optional[dict] = None,
variant=None,
year: Optional[int] = None,
) -> Optional[List[TorrentStream]]:
try:
return self.search_indexer(
self.base_url,
query,
mode,
season,
episode,
categories,
additional_params,
year=year,
)
except Exception as e:
self.handle_exception(f"{translation(30229)}: {e!s}")
return None
def parse_response(self, res: Any) -> Optional[List[TorrentStream]]:
try:
res_dict = xmltodict.parse(res.content)
if res_dict:
channel = res_dict.get("rss", {}).get("channel", {})
items = channel.get("item")
if not items:
return []
results: List[TorrentStream] = []
for item in items if isinstance(items, list) else [items]:
self.extract_result(results, item)
kodilog(f"Parsed {len(results)} results from Jackett")
return results
except Exception as e:
self.handle_exception(f"Error parsing Jackett response: {e!s}")
return None
def extract_result(self, results: List[TorrentStream], item: dict) -> None:
attrs = item.get("torznab:attr", [])
if isinstance(attrs, dict):
attrs = [attrs]
attributes = {attr.get("@name"): attr.get("@value") for attr in attrs}
enclosure = item.get("enclosure", {})
if not isinstance(enclosure, dict):
enclosure = {}
link = item.get("link", "")
guid = item.get("guid", "")
info_hash = str(attributes.get("infohash", ""))
enclosure_url = enclosure.get("@url", "")
provider = item.get("jackettindexer", {}).get("#text", "")
kodilog(
"Jackett parsed result: title={!r}, provider={!r}, guid={!r}, link={!r}, enclosure={!r}, infohash={!r}, has_magnet={}, has_http_url={}, has_infohash={}".format(
item.get("title", ""),
provider,
summarize_locator_for_log(guid),
summarize_locator_for_log(link),
summarize_locator_for_log(enclosure_url),
info_hash[:12].lower(),
bool(str(link).startswith("magnet:?") or str(guid).startswith("magnet:?")),
bool(
str(link).startswith(("http://", "https://"))
or str(guid).startswith(("http://", "https://"))
or str(enclosure_url).startswith(("http://", "https://"))
),
bool(info_hash),
)
)
results.append(
TorrentStream(
title=item.get("title", ""),
type="Torrent",
indexer="Jackett",
publishDate=item.get("pubDate", ""),
provider=provider,
guid=guid,
url=link,
size=item.get("size", ""),
seeders=int(attributes.get("seeders", 0) or 0),
peers=int(attributes.get("peers", 0) or 0),
infoHash=info_hash,
languages=[],
fullLanguages="",
)
)