Add static typing

Author: Chocobo1

.github/workflows/ci.yaml

@@ -29,26 +29,29 @@ jobs:
           python-version: ${{ matrix.python-version }}
 
       - name: Install tools
-        run: pip install bandit isort mypy pycodestyle pyflakes pyright pytest
+        run: |
+          pip install bandit isort mypy pycodestyle pyflakes pyright pytest
+          ./nova3/aux/fetch_aux.sh
 
       - name: Run tests
         if: matrix.python-version == '3.10' # avoid hammering the sites
         continue-on-error: true
         run: |
-          ./nova3/aux/fetch_aux.sh
+          cd nova3
           pytest \
             --showlocals \
-            nova3/tests/*.py
+            tests/*.py
 
       - name: Check typings
         run: |
+          cd nova3
           mypy \
-            --follow-imports skip \
             --strict \
-            nova3/tests/*.py
+            engines/*.py \
+            tests/*.py
           pyright \
-            --skipunannotated \
-            nova3/tests/*.py
+            engines/*.py \
+            tests/*.py
 
       - name: Lint code
         run: |
@@ -72,6 +75,7 @@ jobs:
           isort \
              --check \
              --diff \
+             --line-length 1000 \
              nova3/engines/*.py \
              nova3/tests/*.py
 

nova3/engines/eztv.py

@@ -1,20 +1,22 @@
-# VERSION: 1.19
+# VERSION: 1.20
 # AUTHORS: nindogo
 # CONTRIBUTORS: Diego de las Heras ([email protected])
 
+import http.client
 import re
 import sys
 import urllib.error
 import urllib.parse
 import urllib.request
 from datetime import datetime, timedelta
 from html.parser import HTMLParser
+from typing import Any, Callable, Dict, List, Mapping, Match, Tuple, Union
 
 from helpers import retrieve_url
 from novaprinter import prettyPrinter
 
 
-class eztv(object):
+class eztv:
     name = "EZTV"
     url = 'https://eztvx.to/'
     supported_categories = {'all': 'all', 'tv': 'tv'}
@@ -23,22 +25,26 @@ class MyHtmlParser(HTMLParser):
         A, TD, TR, TABLE = ('a', 'td', 'tr', 'table')
 
         """ Sub-class for parsing results """
-        def __init__(self, url):
+        def __init__(self, url: str) -> None:
             HTMLParser.__init__(self)
             self.url = url
 
             now = datetime.now()
-            self.date_parsers = {
+            self.date_parsers: Mapping[str, Callable[[Match[str]], datetime]] = {
                 r"(\d+)h\s+(\d+)m": lambda m: now - timedelta(hours=int(m[1]), minutes=int(m[2])),
                 r"(\d+)d\s+(\d+)h": lambda m: now - timedelta(days=int(m[1]), hours=int(m[2])),
                 r"(\d+)\s+weeks?": lambda m: now - timedelta(weeks=int(m[1])),
                 r"(\d+)\s+mo": lambda m: now - timedelta(days=int(m[1]) * 30),
                 r"(\d+)\s+years?": lambda m: now - timedelta(days=int(m[1]) * 365),
             }
             self.in_table_row = False
-            self.current_item = {}
+            self.current_item: Dict[str, Any] = {}
+
+        def handle_starttag(self, tag: str, attrs: List[Tuple[str, Union[str, None]]]) -> None:
+            def getStr(d: Mapping[str, Any], key: str) -> str:
+                value = d.get(key, '')
+                return value if value is not None else ''
 
-        def handle_starttag(self, tag, attrs):
             params = dict(attrs)
 
             if (params.get('class') == 'forum_header_border'
@@ -57,10 +63,10 @@ def handle_starttag(self, tag, attrs):
 
             if (tag == self.A
                     and self.in_table_row and params.get('class') == 'epinfo'):
-                self.current_item['desc_link'] = self.url + params.get('href')
-                self.current_item['name'] = params.get('title').split(' (')[0]
+                self.current_item['desc_link'] = self.url + getStr(params, 'href')
+                self.current_item['name'] = getStr(params, 'title').split(' (')[0]
 
-        def handle_data(self, data):
+        def handle_data(self, data: str) -> None:
             data = data.replace(',', '')
             if (self.in_table_row
                     and (data.endswith(' KB') or data.endswith(' MB') or data.endswith(' GB'))):
@@ -76,12 +82,12 @@ def handle_data(self, data):
                         self.current_item["pub_date"] = int(calc(m).timestamp())
                         break
 
-        def handle_endtag(self, tag):
+        def handle_endtag(self, tag: str) -> None:
             if self.in_table_row and tag == self.TR:
-                prettyPrinter(self.current_item)
+                prettyPrinter(self.current_item)  # type: ignore[arg-type] # refactor later
                 self.in_table_row = False
 
-    def do_query(self, what):
+    def do_query(self, what: str) -> str:
         url = f"{self.url}/search/{what.replace('%20', '-')}"
         data = b"layout=def_wlinks"
         try:
@@ -92,13 +98,13 @@ def do_query(self, what):
             user_agent = 'Mozilla/5.0 (X11; Linux x86_64; rv:125.0) Gecko/20100101 Firefox/125.0'
             req = urllib.request.Request(url, data, {'User-Agent': user_agent})
             try:
-                response = urllib.request.urlopen(req)  # nosec B310
+                response: http.client.HTTPResponse = urllib.request.urlopen(req)  # nosec B310
                 return response.read().decode('utf-8')
             except urllib.error.URLError as errno:
                 print(f"Connection error: {errno.reason}", file=sys.stderr)
             return ""
 
-    def search(self, what, cat='all'):
+    def search(self, what: str, cat: str = 'all') -> None:
         eztv_html = self.do_query(what)
 
         eztv_parser = self.MyHtmlParser(self.url)

nova3/engines/jackett.py

@@ -1,4 +1,4 @@
-# VERSION: 4.4
+# VERSION: 4.5
 # AUTHORS: Diego de las Heras ([email protected])
 # CONTRIBUTORS: ukharley
 #               hannsen (github.com/hannsen)
@@ -12,6 +12,7 @@
 from http.cookiejar import CookieJar
 from multiprocessing.dummy import Pool
 from threading import Lock
+from typing import Any, Dict, List, Union
 from urllib.parse import unquote, urlencode
 
 import helpers
@@ -53,7 +54,7 @@ def enable_proxy(self, enable: bool) -> None:
 # load configuration from file
 CONFIG_FILE = 'jackett.json'
 CONFIG_PATH = os.path.join(os.path.dirname(os.path.realpath(__file__)), CONFIG_FILE)
-CONFIG_DATA = {
+CONFIG_DATA: Dict[str, Any] = {
     'api_key': 'YOUR_API_KEY_HERE',  # jackett api
     'url': 'http://127.0.0.1:9117',  # jackett url
     'tracker_first': False,          # (False/True) add tracker name to beginning of search result
@@ -62,7 +63,7 @@ def enable_proxy(self, enable: bool) -> None:
 PRINTER_THREAD_LOCK = Lock()
 
 
-def load_configuration():
+def load_configuration() -> None:
     global CONFIG_DATA
     try:
         # try to load user data from file
@@ -85,7 +86,7 @@ def load_configuration():
         save_configuration()
 
 
-def save_configuration():
+def save_configuration() -> None:
     with open(CONFIG_PATH, 'w') as f:
         f.write(json.dumps(CONFIG_DATA, indent=4, sort_keys=True))
 
@@ -94,7 +95,7 @@ def save_configuration():
 ###############################################################################
 
 
-class jackett(object):
+class jackett:
     name = 'Jackett'
     url = CONFIG_DATA['url'] if CONFIG_DATA['url'][-1] != '/' else CONFIG_DATA['url'][:-1]
     api_key = CONFIG_DATA['api_key']
@@ -110,7 +111,7 @@ class jackett(object):
         'tv': ['5000'],
     }
 
-    def download_torrent(self, download_url):
+    def download_torrent(self, download_url: str) -> None:
         # fix for some indexers with magnet link inside .torrent file
         if download_url.startswith('magnet:?'):
             print(download_url + " " + download_url)
@@ -122,7 +123,7 @@ def download_torrent(self, download_url):
         else:
             print(helpers.download_file(download_url))
 
-    def search(self, what, cat='all'):
+    def search(self, what: str, cat: str = 'all') -> None:
         what = unquote(what)
         category = self.supported_categories[cat.lower()]
 
@@ -147,52 +148,59 @@ def search(self, what, cat='all'):
         else:
             self.search_jackett_indexer(what, category, 'all')
 
-    def get_jackett_indexers(self, what):
-        params = [
+    def get_jackett_indexers(self, what: str) -> List[str]:
+        params = urlencode([
             ('apikey', self.api_key),
             ('t', 'indexers'),
             ('configured', 'true')
-        ]
-        params = urlencode(params)
+        ])
         jacket_url = self.url + "/api/v2.0/indexers/all/results/torznab/api?%s" % params
         response = self.get_response(jacket_url)
         if response is None:
             self.handle_error("connection error getting indexer list", what)
-            return
+            return []
         # process results
         response_xml = xml.etree.ElementTree.fromstring(response)
         indexers = []
         for indexer in response_xml.findall('indexer'):
             indexers.append(indexer.attrib['id'])
         return indexers
 
-    def search_jackett_indexer(self, what, category, indexer_id):
+    def search_jackett_indexer(self, what: str, category: Union[List[str], None], indexer_id: str) -> None:
+        def toStr(s: Union[str, None]) -> str:
+            return s if s is not None else ''
+
+        def getTextProp(e: Union[xml.etree.ElementTree.Element[str], None]) -> str:
+            return toStr(e.text if e is not None else '')
+
         # prepare jackett url
-        params = [
+        params_tmp = [
             ('apikey', self.api_key),
             ('q', what)
         ]
         if category is not None:
-            params.append(('cat', ','.join(category)))
-        params = urlencode(params)
+            params_tmp.append(('cat', ','.join(category)))
+        params = urlencode(params_tmp)
         jacket_url = self.url + "/api/v2.0/indexers/" + indexer_id + "/results/torznab/api?%s" % params  # noqa
         response = self.get_response(jacket_url)
         if response is None:
             self.handle_error("connection error for indexer: " + indexer_id, what)
             return
         # process search results
         response_xml = xml.etree.ElementTree.fromstring(response)
-        for result in response_xml.find('channel').findall('item'):
-            res = {}
+        channel = response_xml.find('channel')
+        if channel is None:
+            return
+        for result in channel.findall('item'):
+            res: Dict[str, Any] = {}
 
-            title = result.find('title')
-            if title is not None:
-                title = title.text
+            title_tmp = result.find('title')
+            if title_tmp is not None:
+                title = title_tmp.text
             else:
                 continue
 
-            tracker = result.find('jackettindexer')
-            tracker = '' if tracker is None else tracker.text
+            tracker = getTextProp(result.find('jackettindexer'))
             if CONFIG_DATA['tracker_first']:
                 res['name'] = '[%s] %s' % (tracker, title)
             else:
@@ -209,7 +217,7 @@ def search_jackett_indexer(self, what, category, indexer_id):
                     continue
 
             res['size'] = result.find('size')
-            res['size'] = -1 if res['size'] is None else (res['size'].text + ' B')
+            res['size'] = -1 if res['size'] is None else (toStr(res['size'].text) + ' B')
 
             res['seeds'] = result.find(self.generate_xpath('seeders'))
             res['seeds'] = -1 if res['seeds'] is None else int(res['seeds'].attrib['value'])
@@ -231,17 +239,17 @@ def search_jackett_indexer(self, what, category, indexer_id):
             res['engine_url'] = self.url
 
             try:
-                date = datetime.strptime(result.find('pubDate').text, '%a, %d %b %Y %H:%M:%S %z')
+                date = datetime.strptime(getTextProp(result.find('pubDate')), '%a, %d %b %Y %H:%M:%S %z')
                 res['pub_date'] = int(date.timestamp())
             except Exception:
                 res['pub_date'] = -1
 
             self.pretty_printer_thread_safe(res)
 
-    def generate_xpath(self, tag):
+    def generate_xpath(self, tag: str) -> str:
         return './{http://torznab.com/schemas/2015/feed}attr[@name="%s"]' % tag
 
-    def get_response(self, query):
+    def get_response(self, query: str) -> Union[str, None]:
         response = None
         try:
             # we can't use helpers.retrieve_url because of redirects
@@ -256,24 +264,26 @@ def get_response(self, query):
             pass
         return response
 
-    def handle_error(self, error_msg, what):
+    def handle_error(self, error_msg: str, what: str) -> None:
         # we need to print the search text to be displayed in qBittorrent when
         # 'Torrent names only' is enabled
         self.pretty_printer_thread_safe({
-            'seeds': -1,
+            'link': self.url,
+            'name': "Jackett: %s! Right-click this row and select 'Open description page' to open help. Configuration file: '%s' Search: '%s'" % (error_msg, CONFIG_PATH, what),  # noqa
             'size': -1,
+            'seeds': -1,
             'leech': -1,
             'engine_url': self.url,
-            'link': self.url,
             'desc_link': 'https://github.com/qbittorrent/search-plugins/wiki/How-to-configure-Jackett-plugin',  # noqa
-            'name': "Jackett: %s! Right-click this row and select 'Open description page' to open help. Configuration file: '%s' Search: '%s'" % (error_msg, CONFIG_PATH, what)  # noqa
+            'pub_date': -1
         })
 
-    def pretty_printer_thread_safe(self, dictionary):
+    def pretty_printer_thread_safe(self, dictionary: Dict[str, Any]) -> None:
+        escaped_dict = self.escape_pipe(dictionary)
         with PRINTER_THREAD_LOCK:
-            prettyPrinter(self.escape_pipe(dictionary))
+            prettyPrinter(escaped_dict)  # type: ignore[arg-type] # refactor later
 
-    def escape_pipe(self, dictionary):
+    def escape_pipe(self, dictionary: Dict[str, Any]) -> Dict[str, Any]:
         # Safety measure until it's fixed in prettyPrinter
         for key in dictionary.keys():
             if isinstance(dictionary[key], str):