forked from zaclys/searxng
145 lines
4.1 KiB
Python
145 lines
4.1 KiB
Python
|
# SPDX-License-Identifier: AGPL-3.0-or-later
|
||
|
"""Torznab WebAPI
|
||
|
|
||
|
A engine that implements the `torznab WebAPI`_.
|
||
|
|
||
|
.. _torznab WebAPI: https://torznab.github.io/spec-1.3-draft/torznab
|
||
|
|
||
|
"""
|
||
|
|
||
|
from urllib.parse import quote
|
||
|
from lxml import etree
|
||
|
from datetime import datetime
|
||
|
|
||
|
from searx.exceptions import SearxEngineAPIException
|
||
|
|
||
|
# about
|
||
|
about = {
|
||
|
"website": None,
|
||
|
"wikidata_id": None,
|
||
|
"official_api_documentation": "https://torznab.github.io/spec-1.3-draft/torznab/Specification-v1.3.html#torznab-api-specification",
|
||
|
"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 request(query, params):
|
||
|
if len(base_url) < 1:
|
||
|
raise SearxEngineAPIException('missing torznab base_url')
|
||
|
|
||
|
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(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:
|
||
|
pass
|
||
|
|
||
|
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
|