Merge pull request #33 from nikhilbadyal/mypy

👷 Added mypy

.env

@@ -0,0 +1,5 @@
+PATCH_APPS=youtube,twitter,reddit,youtube_music
+BUILD_EXTENDED=True
+EXCLUDE_PATCH_YOUTUBE=custom-branding,enable-debugging
+EXCLUDE_PATCH_YOUTUBE_EXTENDED=custom-branding-red,custom-branding-blue,materialyou
+EXCLUDE_PATCH_YOUTUBE_MUSIC_EXTENDED=custom-branding-music

.pre-commit-config.yaml

@@ -47,6 +47,14 @@ repos:
       - id: flake8
         args: [ "--config=setup.cfg" ]
 
+  - repo: https://github.com/pre-commit/mirrors-mypy
+    rev: v0.971
+    hooks:
+      - id: mypy
+        args:
+          - '--strict'
+        additional_dependencies: [ types-requests ]
+
 
 
 # sets up .pre-commit-ci.yaml to ensure pre-commit dependencies stay up to date

main.py

@@ -20,7 +20,7 @@ def main() -> None:
         try:
             logger.info("Trying to build %s" % app)
             app_all_patches, version, is_experimental = patcher.get_app_configs(app)
-            version = downloader.download_apk_to_patch(version, app, patcher)
+            version = downloader.download_apk_to_patch(version, app)
             patcher.include_and_exclude_patches(app, parser, app_all_patches)
             logger.info(f"Downloaded {app}, version {version}")
             parser.patch_app(app=app, version=version, is_experimental=is_experimental)

src/downloader.py

@@ -4,9 +4,10 @@ from concurrent.futures import ThreadPoolExecutor
 from pathlib import Path
 from queue import PriorityQueue
 from time import perf_counter
-from typing import Tuple
+from typing import Any, Tuple
 
 import requests
+from environs import Env
 from loguru import logger
 from requests import Session
 from selectolax.lexbor import LexborHTMLParser
@@ -14,9 +15,9 @@ from tqdm import tqdm
 
 
 class Downloader(object):
-    def __init__(self, env):
+    def __init__(self, env: Env):
         self._CHUNK_SIZE = 2**21 * 5
-        self._QUEUE: PriorityQueue[Tuple] = PriorityQueue()
+        self._QUEUE: PriorityQueue[Tuple[float, str]] = PriorityQueue()
         self._QUEUE_LENGTH = 0
         self.temp_folder = Path("apks")
 
@@ -82,7 +83,7 @@ class Downloader(object):
         self._QUEUE.put((perf_counter() - start, file_name))
         logger.debug(f"Downloaded {file_name}")
 
-    def extract_download_link(self, page: str, app: str):
+    def extract_download_link(self, page: str, app: str) -> None:
         logger.debug(f"Extracting download link from\n{page}")
         parser = LexborHTMLParser(self.session.get(page).text)
 
@@ -97,7 +98,7 @@ class Downloader(object):
         self._download(self.apk_mirror + href, f"{app}.apk")
         logger.debug("Finished Extracting link and downloading")
 
-    def get_download_page(self, parser, main_page):
+    def get_download_page(self, parser: LexborHTMLParser, main_page: str) -> str:
         apm = parser.css(".apkm-badge")
         sub_url = ""
         for is_apm in apm:
@@ -113,7 +114,7 @@ class Downloader(object):
         download_url = self.apk_mirror + sub_url
         return download_url
 
-    def __upto_down_downloader(self, app: str) -> str:
+    def __upto_down_downloader(self, app: str) -> Any:
         page = "https://spotify.en.uptodown.com/android/download"
         parser = LexborHTMLParser(self.session.get(page).text)
         main_page = parser.css_first("#detail-download-button")
@@ -123,7 +124,7 @@ class Downloader(object):
         logger.debug(f"Downloaded {app} apk from apkmirror_specific_version in rt")
         return app_version
 
-    def apkmirror_specific_version(self, app: str, version: str, patcher) -> str:
+    def apkmirror_specific_version(self, app: str, version: str) -> str:
         logger.debug(f"Trying to download {app},specific version {version}")
         version = version.replace(".", "-")
         main_page = f"{self.apk_mirror_version_urls.get(app)}-{version}-release/"
@@ -133,7 +134,7 @@ class Downloader(object):
         logger.debug(f"Downloaded {app} apk from apkmirror_specific_version")
         return version
 
-    def apkmirror_latest_version(self, app: str) -> str:
+    def apkmirror_latest_version(self, app: str) -> Any:
         logger.debug(f"Trying to download {app}'s latest version from apkmirror")
         page = self.apk_mirror_urls.get(app)
         if not page:
@@ -143,7 +144,11 @@ class Downloader(object):
         main_page = parser.css_first(".appRowVariantTag>.accent_color").attributes[
             "href"
         ]
-        int_version = re.search(r"\d", main_page).start()
+        match = re.search(r"\d", main_page)
+        if not match:
+            logger.error("Cannot find app main page")
+            sys.exit(-1)
+        int_version = match.start()
         extra_release = main_page.rfind("release") - 1
         version = main_page[int_version:extra_release]
         version = version.replace("-", ".")
@@ -167,33 +172,33 @@ class Downloader(object):
         self._download(download_url, file_name=file_name)
 
     def download_revanced(self) -> None:
-        assets = (
+        assets = [
-            ("revanced", "revanced-cli", self.normal_cli_jar),
+            ["revanced", "revanced-cli", self.normal_cli_jar],
-            ("revanced", "revanced-integrations", self.normal_integrations_apk),
+            ["revanced", "revanced-integrations", self.normal_integrations_apk],
-            ("revanced", "revanced-patches", self.normal_patches_jar),
+            ["revanced", "revanced-patches", self.normal_patches_jar],
-            ("inotia00", "VancedMicroG", "VancedMicroG.apk"),
+            ["inotia00", "VancedMicroG", "VancedMicroG.apk"],
-        )
+        ]
         if self.build_extended:
-            assets += (
+            assets += [
-                ("inotia00", "revanced-cli", self.cli_jar),
+                ["inotia00", "revanced-cli", self.cli_jar],
-                ("inotia00", "revanced-integrations", self.integrations_apk),
+                ["inotia00", "revanced-integrations", self.integrations_apk],
-                ("inotia00", "revanced-patches", self.patches_jar),
+                ["inotia00", "revanced-patches", self.patches_jar],
-            )
+            ]
-        with ThreadPoolExecutor() as executor:
+        with ThreadPoolExecutor(7) as executor:
             executor.map(lambda repo: self.repository(*repo), assets)
         logger.info("Downloaded revanced microG ,cli, integrations and patches.")
 
-    def upto_down_downloader(self, app: str) -> str:
+    def upto_down_downloader(self, app: str) -> Any:
         return self.__upto_down_downloader(app)
 
-    def download_from_apkmirror(self, version: str, app: str, patches) -> str:
+    def download_from_apkmirror(self, version: str, app: str) -> Any:
         if version and version != "latest":
-            return self.apkmirror_specific_version(app, version, patches)
+            return self.apkmirror_specific_version(app, version)
         else:
             return self.apkmirror_latest_version(app)
 
-    def download_apk_to_patch(self, version: str, app: str, patches) -> str:
+    def download_apk_to_patch(self, version: str, app: str) -> Any:
         if app in self.upto_down:
             return self.upto_down_downloader(app)
         else:
-            return self.download_from_apkmirror(version, app, patches)
+            return self.download_from_apkmirror(version, app)

src/parser.py

@@ -1,14 +1,19 @@
+import sys
+from pathlib import Path
 from subprocess import PIPE, Popen
 from time import perf_counter
 from typing import Any, List
 
+from environs import Env
 from loguru import logger
 
+from src.patches import Patches
+
 
 class Parser(object):
-    def __init__(self, patcher, env, temp_folder):
+    def __init__(self, patcher: Patches, env: Env, temp_folder: Path) -> None:
-        self._PATCHES = []
+        self._PATCHES: List[str] = []
-        self._EXCLUDED = []
+        self._EXCLUDED: List[str] = []
         self.patcher = patcher
         self.keystore_name = env.str("KEYSTORE_FILE_NAME", "revanced.keystore")
         self.build_extended = env.bool("BUILD_EXTENDED", False)
@@ -77,7 +82,11 @@ class Parser(object):
 
         start = perf_counter()
         process = Popen(["java", *args], stdout=PIPE)
-        for line in process.stdout:
+        output = process.stdout
+        if not output:
+            logger.error("Failed to send request for patching.")
+            sys.exit(-1)
+        for line in output:
             logger.debug(line.decode(), flush=True, end="")
         process.wait()
         logger.info(

src/patches.py

@@ -2,6 +2,7 @@ import subprocess
 import sys
 from typing import Any, Dict, List, Tuple
 
+from environs import Env
 from loguru import logger
 from requests import Session
 
@@ -11,8 +12,10 @@ from src.utils import supported_apps
 class Patches(object):
     def check_java(self) -> None:
         logger.debug("Checking if java is available")
-        jd = subprocess.check_output(["java", "-version"], stderr=subprocess.STDOUT)
+        jd = subprocess.check_output(
-        jd = str(jd)[1:-1]
+            ["java", "-version"], stderr=subprocess.STDOUT
+        ).decode("utf-8")
+        jd = jd[1:-1]
         if "Runtime Environment" not in jd:
             logger.debug("Java Must be installed")
             exit(-1)
@@ -21,7 +24,7 @@ class Patches(object):
             exit(-1)
         logger.debug("Cool!! Java is available")
 
-    def fetch_patches(self):
+    def fetch_patches(self) -> None:
         session = Session()
 
         logger.debug("fetching all patches")
@@ -85,13 +88,13 @@ class Patches(object):
             n_patches = len(getattr(self, app_id))
             logger.debug(f"Total patches in {app_name} are {n_patches}")
 
-    def __init__(self, env) -> None:
+    def __init__(self, env: Env) -> None:
         self.env = env
         self.apps = env.list("PATCH_APPS", supported_apps)
         self.build_extended = env.bool("BUILD_EXTENDED", False)
         self.check_java()
         self.fetch_patches()
-        self.extended_apps = ["youtube", "youtube_music"]
+        self.extended_apps: List[str] = ["youtube", "youtube_music"]
 
     def get(self, app: str) -> Tuple[List[Dict[str, str]], str]:
         logger.debug("Getting patches for %s" % app)
@@ -116,7 +119,9 @@ class Patches(object):
             logger.debug("No recommended version.")
         return patches, version
 
-    def include_and_exclude_patches(self, app, arg_parser, app_patches) -> None:
+    def include_and_exclude_patches(
+        self, app: str, arg_parser: Any, app_patches: List[Any]
+    ) -> None:
         logger.debug(f"Excluding patches for app {app}")
         if self.build_extended and app in self.extended_apps:
             excluded_patches = self.env.list(
@@ -134,7 +139,7 @@ class Patches(object):
         else:
             logger.debug(f"No excluded patches for {app}")
 
-    def get_app_configs(self, app) -> Any:
+    def get_app_configs(self, app: str) -> Any:
         experiment = False
         total_patches, recommended_version = self.get(app=app)
         env_version = self.env.str(f"{app}_VERSION".upper(), None)