src/searchengine/nova3/nova2.py

   1 #VERSION: 1.48
   2
   3 # Author:
   4 #  Fabien Devaux <fab AT gnux DOT info>
   5 # Contributors:
   6 #  Christophe Dumez <chris@qbittorrent.org> (qbittorrent integration)
   7 #  Thanks to gab #gcu @ irc.freenode.net (multipage support on PirateBay)
   8 #  Thanks to Elias <gekko04@users.sourceforge.net> (torrentreactor and isohunt search engines)
   9 #
  10 # Licence: BSD
  11
  12 # Redistribution and use in source and binary forms, with or without
  13 # modification, are permitted provided that the following conditions are met:
  14 #
  15 #    * Redistributions of source code must retain the above copyright notice,
  16 #      this list of conditions and the following disclaimer.
  17 #    * Redistributions in binary form must reproduce the above copyright
  18 #      notice, this list of conditions and the following disclaimer in the
  19 #      documentation and/or other materials provided with the distribution.
  20 #    * Neither the name of the author nor the names of its contributors may be
  21 #      used to endorse or promote products derived from this software without
  22 #      specific prior written permission.
  23 #
  24 # THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
  25 # AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
  26 # IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
  27 # ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
  28 # LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
  29 # CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
  30 # SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
  31 # INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
  32 # CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
  33 # ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
  34 # POSSIBILITY OF SUCH DAMAGE.
  35
  36 import importlib
  37 import pathlib
  38 import sys
  39 import traceback
  40 import urllib.parse
  41 import xml.etree.ElementTree as ET
  42 from collections.abc import Iterable
  43 from enum import Enum
  44 from glob import glob
  45 from multiprocessing import Pool, cpu_count
  46 from os import path
  47 from typing import Optional
  48
  49 THREADED: bool = True
  50 try:
  51     MAX_THREADS: int = cpu_count()
  52 except NotImplementedError:
  53     MAX_THREADS = 1
  54
  55 Category = Enum('Category', ['all', 'anime', 'books', 'games', 'movies', 'music', 'pictures', 'software', 'tv'])
  56
  57
  58 ################################################################################
  59 # Every engine should have a "search" method taking
  60 # a space-free string as parameter (ex. "family+guy")
  61 # it should call prettyPrinter() with a dict as parameter.
  62 # The keys in the dict must be: link,name,size,seeds,leech,engine_url
  63 # As a convention, try to list results by decreasing number of seeds or similar
  64 ################################################################################
  65
  66
  67 EngineModuleName = str  # the filename of the engine plugin
  68
  69
  70 class Engine:
  71     url: str
  72     name: str
  73     supported_categories: dict[str, str]
  74
  75     def __init__(self) -> None:
  76         pass
  77
  78     def search(self, what: str, cat: str = Category.all.name) -> None:
  79         pass
  80
  81     def download_torrent(self, info: str) -> None:
  82         pass
  83
  84
  85 # global state
  86 engine_dict: dict[EngineModuleName, Optional[type[Engine]]] = {}
  87
  88
  89 def list_engines() -> list[EngineModuleName]:
  90     """ List all engines,
  91         including broken engines that would fail on import
  92
  93         Return list of all engines' module name
  94     """
  95
  96     names = []
  97
  98     for engine_path in glob(path.join(path.dirname(__file__), 'engines', '*.py')):
  99         engine_module_name = path.basename(engine_path).split('.')[0].strip()
 100         if len(engine_module_name) == 0 or engine_module_name.startswith('_'):
 101             continue
 102         names.append(engine_module_name)
 103
 104     return sorted(names)
 105
 106
 107 def import_engine(engine_module_name: EngineModuleName) -> Optional[type[Engine]]:
 108     if engine_module_name in engine_dict:
 109         return engine_dict[engine_module_name]
 110
 111     # when import fails, return `None`
 112     engine_class = None
 113     try:
 114         # import engines.[engine_module_name]
 115         engine_module = importlib.import_module(f"engines.{engine_module_name}")
 116         engine_class = getattr(engine_module, engine_module_name)
 117     except Exception:
 118         pass
 119
 120     engine_dict[engine_module_name] = engine_class
 121     return engine_class
 122
 123
 124 def get_capabilities(engines: Iterable[EngineModuleName]) -> str:
 125     """
 126     Return capabilities in XML format
 127     <capabilities>
 128       <engine_module_name>
 129         <name>long name</name>
 130         <url>http://example.com</url>
 131         <categories>movies music games</categories>
 132       </engine_module_name>
 133     </capabilities>
 134     """
 135
 136     capabilities_element = ET.Element('capabilities')
 137
 138     for engine_module_name in engines:
 139         engine_class = import_engine(engine_module_name)
 140         if engine_class is None:
 141             continue
 142
 143         engine_module_element = ET.SubElement(capabilities_element, engine_module_name)
 144
 145         ET.SubElement(engine_module_element, 'name').text = engine_class.name
 146         ET.SubElement(engine_module_element, 'url').text = engine_class.url
 147
 148         supported_categories = ""
 149         if hasattr(engine_class, "supported_categories"):
 150             supported_categories = " ".join((key
 151                                              for key in sorted(engine_class.supported_categories.keys())
 152                                              if key != Category.all.name))
 153         ET.SubElement(engine_module_element, 'categories').text = supported_categories
 154
 155     ET.indent(capabilities_element)
 156     return ET.tostring(capabilities_element, 'unicode')
 157
 158
 159 def run_search(search_params: tuple[type[Engine], str, Category]) -> bool:
 160     """ Run search in engine
 161
 162         @param search_params Tuple with engine, query and category
 163
 164         @retval False if any exceptions occurred
 165         @retval True  otherwise
 166     """
 167
 168     engine_class, what, cat = search_params
 169     try:
 170         engine = engine_class()
 171         # avoid exceptions due to invalid category
 172         if hasattr(engine, 'supported_categories'):
 173             if cat.name in engine.supported_categories:
 174                 engine.search(what, cat.name)
 175         else:
 176             engine.search(what)
 177         return True
 178     except Exception:
 179         traceback.print_exc()
 180         return False
 181
 182
 183 if __name__ == "__main__":
 184     def main() -> int:
 185         # qbt tend to run this script in 'isolate mode' so append the current path manually
 186         current_path = str(pathlib.Path(__file__).parent.resolve())
 187         if current_path not in sys.path:
 188             sys.path.append(current_path)
 189
 190         # https://docs.python.org/3/library/sys.html#sys.exit
 191         class ExitCode(Enum):
 192             OK = 0
 193             AppError = 1
 194             ArgError = 2
 195
 196         found_engines = list_engines()
 197
 198         prog_name = sys.argv[0]
 199         prog_usage = (f"Usage: {prog_name} all|engine1[,engine2]* <category> <keywords>\n"
 200                       f"To list available engines: {prog_name} --capabilities [--names]\n"
 201                       f"Found engines: {','.join(found_engines)}")
 202
 203         if "--capabilities" in sys.argv:
 204             if "--names" in sys.argv:
 205                 print(",".join((e for e in found_engines if import_engine(e) is not None)))
 206                 return ExitCode.OK.value
 207
 208             print(get_capabilities(found_engines))
 209             return ExitCode.OK.value
 210         elif len(sys.argv) < 4:
 211             print(prog_usage, file=sys.stderr)
 212             return ExitCode.ArgError.value
 213
 214         # get unique engines
 215         engs = set(arg.strip().lower() for arg in sys.argv[1].split(','))
 216         engines = found_engines if 'all' in engs else [e for e in found_engines if e in engs]
 217
 218         cat = sys.argv[2].lower()
 219         try:
 220             category = Category[cat]
 221         except KeyError:
 222             print(f"Invalid category: {cat}", file=sys.stderr)
 223             return ExitCode.ArgError.value
 224
 225         what = urllib.parse.quote(' '.join(sys.argv[3:]))
 226         params = ((engine_class, what, category) for e in engines if (engine_class := import_engine(e)) is not None)
 227
 228         search_success = False
 229         if THREADED:
 230             processes = max(min(len(engines), MAX_THREADS), 1)
 231             with Pool(processes) as pool:
 232                 search_success = all(pool.map(run_search, params))
 233         else:
 234             search_success = all(map(run_search, params))
 235
 236         return ExitCode.OK.value if search_success else ExitCode.AppError.value
 237
 238     sys.exit(main())