searxng/searx/engines/torznab.py

146 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