torznab.py 4.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145
  1. # SPDX-License-Identifier: AGPL-3.0-or-later
  2. # lint: pylint
  3. """Torznab WebAPI
  4. A engine that implements the `torznab WebAPI`_.
  5. .. _torznab WebAPI: https://torznab.github.io/spec-1.3-draft/torznab
  6. """
  7. from datetime import datetime
  8. from urllib.parse import quote
  9. from lxml import etree
  10. from searx.exceptions import SearxEngineAPIException
  11. # about
  12. about = {
  13. "website": None,
  14. "wikidata_id": None,
  15. "official_api_documentation": "https://torznab.github.io/spec-1.3-draft",
  16. "use_official_api": True,
  17. "require_api_key": False,
  18. "results": 'XML',
  19. }
  20. categories = ['files']
  21. paging = False
  22. time_range_support = False
  23. # defined in settings.yml
  24. # example (Jackett): "http://localhost:9117/api/v2.0/indexers/all/results/torznab"
  25. base_url = ''
  26. api_key = ''
  27. # https://newznab.readthedocs.io/en/latest/misc/api/#predefined-categories
  28. torznab_categories = []
  29. def init(engine_settings=None): # pylint: disable=unused-argument
  30. if len(base_url) < 1:
  31. raise ValueError('missing torznab base_url')
  32. def request(query, params):
  33. search_url = base_url + '?t=search&q={search_query}'
  34. if len(api_key) > 0:
  35. search_url += '&apikey={api_key}'
  36. if len(torznab_categories) > 0:
  37. search_url += '&cat={torznab_categories}'
  38. params['url'] = search_url.format(
  39. search_query = quote(query),
  40. api_key = api_key,
  41. torznab_categories = ",".join([str(x) for x in torznab_categories])
  42. )
  43. return params
  44. def response(resp):
  45. results = []
  46. search_results = etree.XML(resp.content)
  47. # handle errors
  48. # https://newznab.readthedocs.io/en/latest/misc/api/#newznab-error-codes
  49. if search_results.tag == "error":
  50. raise SearxEngineAPIException(search_results.get("description"))
  51. for item in search_results[0].iterfind('item'):
  52. result = {'template': 'torrent.html'}
  53. enclosure = item.find('enclosure')
  54. result["filesize"] = int(enclosure.get('length'))
  55. link = get_property(item, 'link')
  56. guid = get_property(item, 'guid')
  57. comments = get_property(item, 'comments')
  58. # define url
  59. result["url"] = enclosure.get('url')
  60. if comments is not None and comments.startswith('http'):
  61. result["url"] = comments
  62. elif guid is not None and guid.startswith('http'):
  63. result["url"] = guid
  64. # define torrent file url
  65. result["torrentfile"] = None
  66. if enclosure.get('url').startswith("http"):
  67. result["torrentfile"] = enclosure.get('url')
  68. elif link is not None and link.startswith('http'):
  69. result["torrentfile"] = link
  70. # define magnet link
  71. result["magnetlink"] = get_torznab_attr(item, 'magneturl')
  72. if result["magnetlink"] is None:
  73. if enclosure.get('url').startswith("magnet"):
  74. result["magnetlink"] = enclosure.get('url')
  75. elif link is not None and link.startswith('magnet'):
  76. result["magnetlink"] = link
  77. result["title"] = get_property(item, 'title')
  78. result["files"] = get_property(item, 'files')
  79. result["publishedDate"] = None
  80. try:
  81. result["publishedDate"] = datetime.strptime(
  82. get_property(item, 'pubDate'), '%a, %d %b %Y %H:%M:%S %z')
  83. except (ValueError, TypeError) as e:
  84. logger.debug("ignore exception (publishedDate): %s", e)
  85. result["seed"] = get_torznab_attr(item, 'seeders')
  86. # define leech
  87. result["leech"] = get_torznab_attr(item, 'leechers')
  88. if result["leech"] is None and result["seed"] is not None:
  89. peers = get_torznab_attr(item, 'peers')
  90. if peers is not None:
  91. result["leech"] = int(peers) - int(result["seed"])
  92. results.append(result)
  93. return results
  94. def get_property(item, property_name):
  95. property_element = item.find(property_name)
  96. if property_element is not None:
  97. return property_element.text
  98. return None
  99. def get_torznab_attr(item, attr_name):
  100. element = item.find(
  101. './/torznab:attr[@name="{attr_name}"]'.format(attr_name=attr_name),
  102. {
  103. 'torznab': 'http://torznab.com/schemas/2015/feed'
  104. }
  105. )
  106. if element is not None:
  107. return element.get("value")
  108. return None