diff --git a/.env b/.env index e69de29..0e8a167 100644 --- a/.env +++ b/.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 diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 422a7bc..5fc7c2b 100644 --- a/.pre-commit-config.yaml +++ b/.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 diff --git a/main.py b/main.py index 253e73a..d8d1e54 100644 --- a/main.py +++ b/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) diff --git a/src/downloader.py b/src/downloader.py index b25359a..7e14b61 100644 --- a/src/downloader.py +++ b/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 = ( - ("revanced", "revanced-cli", self.normal_cli_jar), - ("revanced", "revanced-integrations", self.normal_integrations_apk), - ("revanced", "revanced-patches", self.normal_patches_jar), - ("inotia00", "VancedMicroG", "VancedMicroG.apk"), - ) + assets = [ + ["revanced", "revanced-cli", self.normal_cli_jar], + ["revanced", "revanced-integrations", self.normal_integrations_apk], + ["revanced", "revanced-patches", self.normal_patches_jar], + ["inotia00", "VancedMicroG", "VancedMicroG.apk"], + ] if self.build_extended: - assets += ( - ("inotia00", "revanced-cli", self.cli_jar), - ("inotia00", "revanced-integrations", self.integrations_apk), - ("inotia00", "revanced-patches", self.patches_jar), - ) - with ThreadPoolExecutor() as executor: + assets += [ + ["inotia00", "revanced-cli", self.cli_jar], + ["inotia00", "revanced-integrations", self.integrations_apk], + ["inotia00", "revanced-patches", self.patches_jar], + ] + 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) diff --git a/src/parser.py b/src/parser.py index 16c4bc0..2584221 100644 --- a/src/parser.py +++ b/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): - self._PATCHES = [] - self._EXCLUDED = [] + def __init__(self, patcher: Patches, env: Env, temp_folder: Path) -> None: + self._PATCHES: List[str] = [] + 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( diff --git a/src/patches.py b/src/patches.py index 62a7871..73d02bb 100644 --- a/src/patches.py +++ b/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 = str(jd)[1:-1] + jd = subprocess.check_output( + ["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)