Skip to content
Back to top

# Source code for searx.engines.torznab

    # 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
    ===============

    """

    import typing as t
    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 t.TYPE_CHECKING:
        from searx.extended_types import SXNG_Response

    # engine settings
    about: dict[str, t.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

    def init(engine_settings=None):  # pylint: disable=unused-argument
        """Initialize the engine."""
        if len(base_url) < 1:
            raise ValueError('missing torznab base_url')

    def request(query: str, params: dict[str, t.Any]) -> dict[str, t.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

    def response(resp: "SXNG_Response") -> list[dict[str, t.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, t.Any] = build_result(item)
            results.append(result)

        return results

    def build_result(item: etree.Element) -> dict[str, t.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 SearXNG result
        result: dict[str, t.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

    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

    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