# SPDX-License-Identifier: AGPL-3.0-or-later
"""Torznab_ is an API specification that provides a standardized way to query
torrent site for content. It is used by a number of torrent applications,
including Prowlarr_ and Jackett_.
Using this engine together with Prowlarr_ or Jackett_ allows you to search
a huge number of torrent sites which are not directly supported.
Configuration
=============
The engine has the following settings:
``base_url``:
Torznab endpoint URL.
``api_key``:
The API key to use for authentication.
``torznab_categories``:
The categories to use for searching. This is a list of category IDs. See
Prowlarr-categories_ or Jackett-categories_ for more information.
``show_torrent_files``:
Whether to show the torrent file in the search results. Be careful as using
this with Prowlarr_ or Jackett_ leaks the API key. This should be used only
if you are querying a Torznab endpoint without authentication or if the
instance is private. Be aware that private trackers may ban you if you share
the torrent file. Defaults to ``false``.
``show_magnet_links``:
Whether to show the magnet link in the search results. Be aware that private
trackers may ban you if you share the magnet link. Defaults to ``true``.
.. _Torznab:
https://torznab.github.io/spec-1.3-draft/index.html
.. _Prowlarr:
https://github.com/Prowlarr/Prowlarr
.. _Jackett:
https://github.com/Jackett/Jackett
.. _Prowlarr-categories:
https://wiki.servarr.com/en/prowlarr/cardigann-yml-definition#categories
.. _Jackett-categories:
https://github.com/Jackett/Jackett/wiki/Jackett-Categories
Implementations
===============
"""
from __future__ import annotations
from typing import TYPE_CHECKING
from typing import List, Dict, Any
from datetime import datetime
from urllib.parse import quote
from lxml import etree # type: ignore
from searx.exceptions import SearxEngineAPIException
from searx.utils import humanize_bytes
if TYPE_CHECKING:
import httpx
import logging
logger: logging.Logger
# engine settings
about: Dict[str, Any] = {
"website": None,
"wikidata_id": None,
"official_api_documentation": "https://torznab.github.io/spec-1.3-draft",
"use_official_api": True,
"require_api_key": False,
"results": 'XML',
}
categories: List[str] = ['files']
paging: bool = False
time_range_support: bool = False
# defined in settings.yml
# example (Jackett): "http://localhost:9117/api/v2.0/indexers/all/results/torznab"
base_url: str = ''
api_key: str = ''
# https://newznab.readthedocs.io/en/latest/misc/api/#predefined-categories
torznab_categories: List[str] = []
show_torrent_files: bool = False
show_magnet_links: bool = True
[docs]
def init(engine_settings=None): # pylint: disable=unused-argument
"""Initialize the engine."""
if len(base_url) < 1:
raise ValueError('missing torznab base_url')
[docs]
def request(query: str, params: Dict[str, Any]) -> Dict[str, Any]:
"""Build the request params."""
search_url: str = base_url + '?t=search&q={search_query}'
if len(api_key) > 0:
search_url += '&apikey={api_key}'
if len(torznab_categories) > 0:
search_url += '&cat={torznab_categories}'
params['url'] = search_url.format(
search_query=quote(query), api_key=api_key, torznab_categories=",".join([str(x) for x in torznab_categories])
)
return params
[docs]
def response(resp: httpx.Response) -> List[Dict[str, Any]]:
"""Parse the XML response and return a list of results."""
results = []
search_results = etree.XML(resp.content)
# handle errors: https://newznab.readthedocs.io/en/latest/misc/api/#newznab-error-codes
if search_results.tag == "error":
raise SearxEngineAPIException(search_results.get("description"))
channel: etree.Element = search_results[0]
item: etree.Element
for item in channel.iterfind('item'):
result: Dict[str, Any] = build_result(item)
results.append(result)
return results
[docs]
def build_result(item: etree.Element) -> Dict[str, Any]:
"""Build a result from a XML item."""
# extract attributes from XML
# see https://torznab.github.io/spec-1.3-draft/torznab/Specification-v1.3.html#predefined-attributes
enclosure: etree.Element | None = item.find('enclosure')
enclosure_url: str | None = None
if enclosure is not None:
enclosure_url = enclosure.get('url')
filesize = get_attribute(item, 'size')
if not filesize and enclosure:
filesize = enclosure.get('length')
guid = get_attribute(item, 'guid')
comments = get_attribute(item, 'comments')
pubDate = get_attribute(item, 'pubDate')
seeders = get_torznab_attribute(item, 'seeders')
leechers = get_torznab_attribute(item, 'leechers')
peers = get_torznab_attribute(item, 'peers')
# map attributes to searx result
result: Dict[str, Any] = {
'template': 'torrent.html',
'title': get_attribute(item, 'title'),
'filesize': humanize_bytes(int(filesize)) if filesize else None,
'files': get_attribute(item, 'files'),
'seed': seeders,
'leech': _map_leechers(leechers, seeders, peers),
'url': _map_result_url(guid, comments),
'publishedDate': _map_published_date(pubDate),
'torrentfile': None,
'magnetlink': None,
}
link = get_attribute(item, 'link')
if show_torrent_files:
result['torrentfile'] = _map_torrent_file(link, enclosure_url)
if show_magnet_links:
magneturl = get_torznab_attribute(item, 'magneturl')
result['magnetlink'] = _map_magnet_link(magneturl, guid, enclosure_url, link)
return result
def _map_result_url(guid: str | None, comments: str | None) -> str | None:
if guid and guid.startswith('http'):
return guid
if comments and comments.startswith('http'):
return comments
return None
def _map_leechers(leechers: str | None, seeders: str | None, peers: str | None) -> str | None:
if leechers:
return leechers
if seeders and peers:
return str(int(peers) - int(seeders))
return None
def _map_published_date(pubDate: str | None) -> datetime | None:
if pubDate is not None:
try:
return datetime.strptime(pubDate, '%a, %d %b %Y %H:%M:%S %z')
except (ValueError, TypeError) as e:
logger.debug("ignore exception (publishedDate): %s", e)
return None
def _map_torrent_file(link: str | None, enclosure_url: str | None) -> str | None:
if link and link.startswith('http'):
return link
if enclosure_url and enclosure_url.startswith('http'):
return enclosure_url
return None
def _map_magnet_link(
magneturl: str | None,
guid: str | None,
enclosure_url: str | None,
link: str | None,
) -> str | None:
if magneturl and magneturl.startswith('magnet'):
return magneturl
if guid and guid.startswith('magnet'):
return guid
if enclosure_url and enclosure_url.startswith('magnet'):
return enclosure_url
if link and link.startswith('magnet'):
return link
return None
[docs]
def get_attribute(item: etree.Element, property_name: str) -> str | None:
"""Get attribute from item."""
property_element: etree.Element | None = item.find(property_name)
if property_element is not None:
return property_element.text
return None
[docs]
def get_torznab_attribute(item: etree.Element, attribute_name: str) -> str | None:
"""Get torznab special attribute from item."""
element: etree.Element | None = item.find(
'.//torznab:attr[@name="{attribute_name}"]'.format(attribute_name=attribute_name),
{'torznab': 'http://torznab.com/schemas/2015/feed'},
)
if element is not None:
return element.get("value")
return None
