[mod] engines - add torznab WebAPI

1 files changed, 144 insertions, 0 deletions

diff --git a/searx/engines/torznab.py b/searx/engines/torznab.py
new file mode 100644
index 000000000..aa9919c34
--- /dev/null
+++ b/searx/engines/torznab.py
@@ -0,0 +1,144 @@
+# 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