|  | # SPDX-License-Identifier: AGPL-3.0-or-later
 | 
						
						
						
							|  | # lint: pylint
 | 
						
						
						
							|  | """Torznab WebAPI
 | 
						
						
						
							|  | 
 | 
						
						
						
							|  | A engine that implements the `torznab WebAPI`_.
 | 
						
						
						
							|  | 
 | 
						
						
						
							|  | .. _torznab WebAPI: https://torznab.github.io/spec-1.3-draft/torznab
 | 
						
						
						
							|  | 
 | 
						
						
						
							|  | """
 | 
						
						
						
							|  | 
 | 
						
						
						
							|  | from datetime import datetime
 | 
						
						
						
							|  | from urllib.parse import quote
 | 
						
						
						
							|  | from lxml import etree
 | 
						
						
						
							|  | 
 | 
						
						
						
							|  | from searx.exceptions import SearxEngineAPIException
 | 
						
						
						
							|  | 
 | 
						
						
						
							|  | # about
 | 
						
						
						
							|  | about = {
 | 
						
						
						
							|  |     "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 = ['files']
 | 
						
						
						
							|  | paging = False
 | 
						
						
						
							|  | time_range_support = False
 | 
						
						
						
							|  | 
 | 
						
						
						
							|  | # defined in settings.yml
 | 
						
						
						
							|  | # example (Jackett): "http://localhost:9117/api/v2.0/indexers/all/results/torznab"
 | 
						
						
						
							|  | base_url = ''
 | 
						
						
						
							|  | api_key = ''
 | 
						
						
						
							|  | # https://newznab.readthedocs.io/en/latest/misc/api/#predefined-categories
 | 
						
						
						
							|  | torznab_categories = []
 | 
						
						
						
							|  | 
 | 
						
						
						
							|  | 
 | 
						
						
						
							|  | def init(engine_settings=None):  # pylint: disable=unused-argument
 | 
						
						
						
							|  |     if len(base_url) < 1:
 | 
						
						
						
							|  |         raise ValueError('missing torznab base_url')
 | 
						
						
						
							|  | 
 | 
						
						
						
							|  | 
 | 
						
						
						
							|  | def request(query, params):
 | 
						
						
						
							|  | 
 | 
						
						
						
							|  |     search_url = 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):
 | 
						
						
						
							|  |     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"))
 | 
						
						
						
							|  | 
 | 
						
						
						
							|  |     for item in search_results[0].iterfind('item'):
 | 
						
						
						
							|  |         result = {'template': 'torrent.html'}
 | 
						
						
						
							|  | 
 | 
						
						
						
							|  |         enclosure = item.find('enclosure')
 | 
						
						
						
							|  | 
 | 
						
						
						
							|  |         result["filesize"] = int(enclosure.get('length'))
 | 
						
						
						
							|  | 
 | 
						
						
						
							|  |         link = get_property(item, 'link')
 | 
						
						
						
							|  |         guid = get_property(item, 'guid')
 | 
						
						
						
							|  |         comments = get_property(item, 'comments')
 | 
						
						
						
							|  | 
 | 
						
						
						
							|  |         # define url
 | 
						
						
						
							|  |         result["url"] = enclosure.get('url')
 | 
						
						
						
							|  |         if comments is not None and comments.startswith('http'):
 | 
						
						
						
							|  |             result["url"] = comments
 | 
						
						
						
							|  |         elif guid is not None and guid.startswith('http'):
 | 
						
						
						
							|  |             result["url"] = guid
 | 
						
						
						
							|  | 
 | 
						
						
						
							|  |         # define torrent file url
 | 
						
						
						
							|  |         result["torrentfile"] = None
 | 
						
						
						
							|  |         if enclosure.get('url').startswith("http"):
 | 
						
						
						
							|  |             result["torrentfile"] = enclosure.get('url')
 | 
						
						
						
							|  |         elif link is not None and link.startswith('http'):
 | 
						
						
						
							|  |             result["torrentfile"] = link
 | 
						
						
						
							|  | 
 | 
						
						
						
							|  |         # define magnet link
 | 
						
						
						
							|  |         result["magnetlink"] = get_torznab_attr(item, 'magneturl')
 | 
						
						
						
							|  |         if result["magnetlink"] is None:
 | 
						
						
						
							|  |             if enclosure.get('url').startswith("magnet"):
 | 
						
						
						
							|  |                 result["magnetlink"] = enclosure.get('url')
 | 
						
						
						
							|  |             elif link is not None and link.startswith('magnet'):
 | 
						
						
						
							|  |                 result["magnetlink"] = link
 | 
						
						
						
							|  | 
 | 
						
						
						
							|  |         result["title"] = get_property(item, 'title')
 | 
						
						
						
							|  |         result["files"] = get_property(item, 'files')
 | 
						
						
						
							|  | 
 | 
						
						
						
							|  |         result["publishedDate"] = None
 | 
						
						
						
							|  |         try:
 | 
						
						
						
							|  |             result["publishedDate"] = datetime.strptime(get_property(item, 'pubDate'), '%a, %d %b %Y %H:%M:%S %z')
 | 
						
						
						
							|  |         except (ValueError, TypeError) as e:
 | 
						
						
						
							|  |             logger.debug("ignore exception (publishedDate): %s", e)
 | 
						
						
						
							|  | 
 | 
						
						
						
							|  |         result["seed"] = get_torznab_attr(item, 'seeders')
 | 
						
						
						
							|  | 
 | 
						
						
						
							|  |         # define leech
 | 
						
						
						
							|  |         result["leech"] = get_torznab_attr(item, 'leechers')
 | 
						
						
						
							|  |         if result["leech"] is None and result["seed"] is not None:
 | 
						
						
						
							|  |             peers = get_torznab_attr(item, 'peers')
 | 
						
						
						
							|  |             if peers is not None:
 | 
						
						
						
							|  |                 result["leech"] = int(peers) - int(result["seed"])
 | 
						
						
						
							|  | 
 | 
						
						
						
							|  |         results.append(result)
 | 
						
						
						
							|  | 
 | 
						
						
						
							|  |     return results
 | 
						
						
						
							|  | 
 | 
						
						
						
							|  | 
 | 
						
						
						
							|  | def get_property(item, property_name):
 | 
						
						
						
							|  |     property_element = item.find(property_name)
 | 
						
						
						
							|  | 
 | 
						
						
						
							|  |     if property_element is not None:
 | 
						
						
						
							|  |         return property_element.text
 | 
						
						
						
							|  | 
 | 
						
						
						
							|  |     return None
 | 
						
						
						
							|  | 
 | 
						
						
						
							|  | 
 | 
						
						
						
							|  | def get_torznab_attr(item, attr_name):
 | 
						
						
						
							|  |     element = item.find(
 | 
						
						
						
							|  |         './/torznab:attr[@name="{attr_name}"]'.format(attr_name=attr_name),
 | 
						
						
						
							|  |         {'torznab': 'http://torznab.com/schemas/2015/feed'},
 | 
						
						
						
							|  |     )
 | 
						
						
						
							|  | 
 | 
						
						
						
							|  |     if element is not None:
 | 
						
						
						
							|  |         return element.get("value")
 | 
						
						
						
							|  | 
 | 
						
						
						
							|  |     return None
 |