forked from zaclys/searxng
3d96a9839a
This patch was generated by black [1]::
make format.python
[1] https://github.com/psf/black
Signed-off-by: Markus Heiser <markus.heiser@darmarit.de>
144 lines
4.2 KiB
Python
144 lines
4.2 KiB
Python
# 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
|