From 77377cdd48514fb227f587676aba4817b9eac5a0 Mon Sep 17 00:00:00 2001 From: Nikhil Badyal <59223300+nikhilbadyal@users.noreply.github.com> Date: Fri, 25 Aug 2023 15:36:50 +0530 Subject: [PATCH] =?UTF-8?q?=F0=9F=9A=A8=20Updated=20lints=20(#308)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .DS_Store | Bin 0 -> 6148 bytes .pre-commit-config.yaml | 28 ++++++--- Dockerfile | 3 +- README.md | 2 +- main.py | 19 +++--- pyproject.toml | 84 ++++++++++++++++++++++++++ scripts/__init__.py | 1 + scripts/status_check.py | 52 ++++++++-------- setup.cfg | 29 --------- src/__init__.py | 1 + src/app.py | 117 ++++++++++++------------------------ src/config.py | 26 +++----- src/downloader/__init__.py | 1 + src/downloader/apkmirror.py | 65 ++++++++------------ src/downloader/apkpure.py | 7 +-- src/downloader/apksos.py | 18 +++--- src/downloader/download.py | 44 +++++--------- src/downloader/factory.py | 29 ++++----- src/downloader/github.py | 29 +++------ src/downloader/sources.py | 1 + src/downloader/uptodown.py | 39 +++++++----- src/downloader/utils.py | 2 +- src/exceptions.py | 58 +++++++++--------- src/parser.py | 57 +++++++----------- src/patches.py | 77 +++++++++--------------- src/utils.py | 102 +++++++++++-------------------- 26 files changed, 404 insertions(+), 487 deletions(-) create mode 100644 .DS_Store create mode 100644 pyproject.toml delete mode 100644 setup.cfg diff --git a/.DS_Store b/.DS_Store new file mode 100644 index 0000000000000000000000000000000000000000..203664b176af3063d21ad83c2d27f44a1dc7613a GIT binary patch literal 6148 zcmeH~F$w}f3`G;&V!>uh%V|7-Hy9Q@ffrB^EEE+*>p8kVnIO1Yi^vZof0CK7>??LQ zBBK3mI~VChq=lQx%)-DFc_SM+%U-Vc)#Z3N9e1c%eH3SH4ew;IAKMfXAOR8}0TLjA zA0lA)HmsdgWh4O-Ab}?V`#vPNX$~z_{nLTqBLK8P*$r!-C7{U)&>UK-q5{*H9yD6j z#}KP~J2b_)99pW@cF`C;cVzub;QgOGwy`I(AQMGk}L;X0y%TE9jyNai9H|!@{KyzrRiVBQB0*--!1inh( E0eyZF!T= 17 - 2. Install Python + 2. Install Python >= 3.11 3. Create virtual environment ``` python3 -m venv venv diff --git a/main.py b/main.py index d6832fa..36700d4 100644 --- a/main.py +++ b/main.py @@ -5,7 +5,7 @@ from environs import Env from loguru import logger from src.config import RevancedConfig -from src.exceptions import AppNotFound, PatchesJsonLoadFailed, PatchingFailed +from src.exceptions import AppNotFoundError, PatchesJsonLoadError, PatchingFailedError, UnknownError from src.parser import Parser from src.patches import Patches from src.utils import check_java, extra_downloads @@ -19,13 +19,14 @@ def main() -> None: env.read_env() config = RevancedConfig(env) extra_downloads(config) - check_java(config.dry_run) + if not config.dry_run: + check_java() logger.info(f"Will Patch only {config.apps}") - for app in config.apps: - logger.info(f"Trying to build {app}") + for possible_app in config.apps: + logger.info(f"Trying to build {possible_app}") try: - app = APP(app_name=app, config=config) + app = APP(app_name=possible_app, config=config) patcher = Patches(config, app) parser = Parser(patcher, config) app_all_patches = patcher.get_app_configs(app) @@ -33,13 +34,13 @@ def main() -> None: patcher.include_exclude_patch(app, parser, app_all_patches) logger.info(app) parser.patch_app(app) - except AppNotFound as e: + except AppNotFoundError as e: logger.info(e) - except PatchesJsonLoadFailed: + except PatchesJsonLoadError: logger.exception("Patches.json not found") - except PatchingFailed as e: + except PatchingFailedError as e: logger.exception(e) - except Exception as e: + except UnknownError as e: logger.exception(f"Failed to build {app} because of {e}") diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..b331503 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,84 @@ +[tool.ruff] +line-length = 120 +select = [ + "F", # pyflakes + "E", # pycodestyle Error + "W", # pycodestyle Warning + "C90", #mccabe + "I", #isort + "N", #isort + "D", #isort + "UP", # pyupgrade + "ANN", # flake8-annotations + "ASYNC", # flake8-async + "S", # flake8-bandit + "BLE", # flake8-blind-except + "FBT", #flake8-boolean-trap + "B", #flake8-bugbear + "A" , #flake8-builtins + "C4" , #flake8-comprehensions + "DTZ" , #flake8-datetimez + "T10" , #flake8-debugger + "EM" , #flake8-errmsg + "EXE" , #flake8-executable + "ISC" , #flake8-implicit-str-concat + "ICN" , #flake8-import-conventions + "G" , #flake8-logging-format + "INP" , #flake8-no-pep420 + "PIE" , #flake8-pie + "PYI" , #flake8-pyi + "RSE" , #flake8-raise + "RET" , #flake8-return + "SLF" , #flake8-self + "SIM" , #flake8-simplify + "TCH" , #flake8-type-checking + "INT" , #flake8-gettext + "ARG" , #flake8-unused-arguments + "PTH" , #flake8-use-pathlib + "ERA" , #eradicate + "PGH" , #pygrep-hooks + "PL" , #Pylint + "TRY" , #tryceratops + "FLY" , #flynt + "PERF" , #flynt + "RUF" , #Ruff-specific rules +] +ignore = [ + "D401", + "ANN401", + "S603", + "S607", + "ARG002", #unused-method-argument + "PTH122", #os-path-splitext + "TRY301", #raise-within-try + "PERF203", #try-except-in-loop +] +fix = true +show-fixes = true +[tool.ruff.pydocstyle] +convention = "numpy" + +[tool.docformatter] +recursive = true +wrap-summaries = 120 +wrap-descriptions = 120 + +[tool.black] +line-length = 120 + +[pycodestyle] +max-line-length = 120 +exclude = ["venv"] + + +[tool.mypy] +ignore_missing_imports = true +check_untyped_defs = true +warn_unused_ignores = true +warn_redundant_casts = true +warn_unused_configs = true + +[tool.isort] +line_length = 120 +skip = ["venv"] +profile = "black" diff --git a/scripts/__init__.py b/scripts/__init__.py index e69de29..e7035af 100644 --- a/scripts/__init__.py +++ b/scripts/__init__.py @@ -0,0 +1 @@ +"""Common utilities.""" diff --git a/scripts/status_check.py b/scripts/status_check.py index 566186a..be5abe9 100644 --- a/scripts/status_check.py +++ b/scripts/status_check.py @@ -1,5 +1,6 @@ """Status check.""" import re +from pathlib import Path from typing import List import requests @@ -7,30 +8,26 @@ from bs4 import BeautifulSoup from google_play_scraper import app as gplay_app from google_play_scraper.exceptions import GooglePlayScraperException -from src.exceptions import APKMirrorIconScrapFailure +from src.exceptions import APKComboIconScrapError, APKMirrorIconScrapError, UnknownError from src.patches import Patches -from src.utils import ( - apk_mirror_base_url, - apkmirror_status_check, - bs4_parser, - handle_request_response, - request_header, -) +from src.utils import apk_mirror_base_url, apkmirror_status_check, bs4_parser, handle_request_response, request_header not_found_icon = "https://img.icons8.com/bubbles/500/android-os.png" +no_of_col = 6 def apkcombo_scrapper(package_name: str) -> str: """Apkcombo scrapper.""" try: apkcombo_url = f"https://apkcombo.com/genericApp/{package_name}" - r = requests.get( - apkcombo_url, headers=request_header, allow_redirects=True, timeout=10 - ) + r = requests.get(apkcombo_url, headers=request_header, allow_redirects=True, timeout=10) soup = BeautifulSoup(r.text, bs4_parser) - url = soup.select_one("div.avatar > img")["data-src"] - return re.sub(r"=.*$", "", url) - except Exception: + icon_element = soup.select_one("div.bubble-wrap > img") + if not icon_element: + raise APKComboIconScrapError(url=apkcombo_url) + url = icon_element["data-src"] + return re.sub(r"=.*$", "", url) # type: ignore[arg-type] + except UnknownError: return not_found_icon @@ -40,13 +37,16 @@ def apkmirror_scrapper(package_name: str) -> str: search_url = f"{apk_mirror_base_url}/?s={package_name}" if response["data"][0]["exists"]: return _extracted_from_apkmirror_scrapper(search_url) - raise APKMirrorIconScrapFailure(url=search_url) + raise APKMirrorIconScrapError(url=search_url) def _extracted_from_apkmirror_scrapper(search_url: str) -> str: r = requests.get(search_url, headers=request_header, timeout=60) soup = BeautifulSoup(r.text, bs4_parser) - sub_url = soup.select_one("div.bubble-wrap > img")["src"] + icon_element = soup.select_one("div.bubble-wrap > img") + if not icon_element: + raise APKMirrorIconScrapError(url=search_url) + sub_url = str(icon_element["src"]) new_width = 500 new_height = 500 new_quality = 100 @@ -54,9 +54,7 @@ def _extracted_from_apkmirror_scrapper(search_url: str) -> str: # regular expression pattern to match w=xx&h=xx&q=xx pattern = r"(w=\d+&h=\d+&q=\d+)" - return apk_mirror_base_url + re.sub( - pattern, f"w={new_width}&h={new_height}&q={new_quality}", sub_url - ) + return apk_mirror_base_url + re.sub(pattern, f"w={new_width}&h={new_height}&q={new_quality}", sub_url) def gplay_icon_scrapper(package_name: str) -> str: @@ -68,13 +66,13 @@ def gplay_icon_scrapper(package_name: str) -> str: ) if result["icon"]: return str(result["icon"]) - raise GooglePlayScraperException() + raise GooglePlayScraperException except GooglePlayScraperException: try: return apkmirror_scrapper(package_name) - except APKMirrorIconScrapFailure: + except APKMirrorIconScrapError: return apkcombo_scrapper(package_name) - except Exception: + except UnknownError: return not_found_icon @@ -85,11 +83,12 @@ def generate_markdown_table(data: List[List[str]]) -> str: table = ( "| Package Name | App Icon | PlayStore link | APKMirror link|APKCombo Link| Supported?|\n" - + "|-------------|----------|----------------|---------------|------------------|----------|\n" + "|-------------|----------|----------------|---------------|------------------|----------|\n" ) for row in data: - if len(row) != 6: - raise ValueError("Each row must contain 6 columns of data.") + if len(row) != no_of_col: + msg = "Each row must contain 6 columns of data." + raise ValueError(msg) table += f"| {row[0]} | {row[1]} | {row[2]} | {row[3]} |{row[4]} |{row[5]} |\n" @@ -97,6 +96,7 @@ def generate_markdown_table(data: List[List[str]]) -> str: def main() -> None: + """Entrypoint.""" repo_url = "https://releases.revanced.app/patches" response = requests.get(repo_url, timeout=10) handle_request_response(response) @@ -124,7 +124,7 @@ def main() -> None: ] table = generate_markdown_table(data) output += table - with open("status.md", "w", encoding="utf_8") as status: + with Path("status.md").open("w", encoding="utf_8") as status: status.write(output) print(output) diff --git a/setup.cfg b/setup.cfg deleted file mode 100644 index 8d91657..0000000 --- a/setup.cfg +++ /dev/null @@ -1,29 +0,0 @@ -[flake8] -max-line-length = 120 -exclude = .tox,.git,*/settings/*,*/static/CACHE/*,docs,node_modules,venv -extend-ignore = E203 - -[pycodestyle] -max-line-length = 120 -exclude = .tox,.git,*/settings/*,*/static/CACHE/*,docs,venv - -[isort] -line_length = 88 -known_first_party = config -multi_line_output = 3 -default_section = THIRDPARTY -skip = venv/ -include_trailing_comma = true -force_grid_wrap = 0 -use_parentheses = true -profile = black - - -[mypy] -python_version = 3.10 -check_untyped_defs = True -ignore_missing_imports = True -warn_unused_ignores = True -warn_redundant_casts = True -warn_unused_configs = True -exclude = ['venv/'] diff --git a/src/__init__.py b/src/__init__.py index e69de29..16f43b1 100644 --- a/src/__init__.py +++ b/src/__init__.py @@ -0,0 +1 @@ +"""Main Source Code.""" diff --git a/src/app.py b/src/app.py index 861be54..cc4060a 100644 --- a/src/app.py +++ b/src/app.py @@ -3,49 +3,42 @@ import concurrent import hashlib import pathlib from concurrent.futures import ThreadPoolExecutor -from typing import Dict, List +from typing import Dict, List, Self from loguru import logger from src.config import RevancedConfig from src.downloader.sources import apk_sources -from src.exceptions import DownloadFailure, PatchingFailed +from src.exceptions import DownloadError, PatchingFailedError, UnknownError from src.utils import slugify -class APP(object): +class APP: """Patched APK.""" - def __init__(self, app_name: str, config: RevancedConfig): + def __init__(self: Self, app_name: str, config: RevancedConfig) -> None: + """Initialize APP. + + Args: + ---- + app_name (str): Name of the app. + config (RevancedConfig): Configuration object. + """ from src.patches import Patches self.app_name = app_name self.app_version = config.env.str(f"{app_name}_VERSION".upper(), None) self.experiment = False self.cli_dl = config.env.str(f"{app_name}_CLI_DL".upper(), config.global_cli_dl) - self.patches_dl = config.env.str( - f"{app_name}_PATCHES_DL".upper(), config.global_patches_dl - ) - self.integrations_dl = config.env.str( - f"{app_name}_INTEGRATIONS_DL".upper(), config.global_integrations_dl - ) - self.patches_json_dl = config.env.str( - f"{app_name}_PATCHES_JSON_DL".upper(), config.global_patches_json_dl - ) - self.exclude_request: List[str] = config.env.list( - f"{app_name}_EXCLUDE_PATCH".upper(), [] - ) - self.include_request: List[str] = config.env.list( - f"{app_name}_INCLUDE_PATCH".upper(), [] - ) + self.patches_dl = config.env.str(f"{app_name}_PATCHES_DL".upper(), config.global_patches_dl) + self.integrations_dl = config.env.str(f"{app_name}_INTEGRATIONS_DL".upper(), config.global_integrations_dl) + self.patches_json_dl = config.env.str(f"{app_name}_PATCHES_JSON_DL".upper(), config.global_patches_json_dl) + self.exclude_request: List[str] = config.env.list(f"{app_name}_EXCLUDE_PATCH".upper(), []) + self.include_request: List[str] = config.env.list(f"{app_name}_INCLUDE_PATCH".upper(), []) self.resource: Dict[str, str] = {} self.no_of_patches: int = 0 - self.keystore_name = config.env.str( - f"{app_name}_KEYSTORE_FILE_NAME".upper(), config.global_keystore_name - ) - self.archs_to_build = config.env.list( - f"{app_name}_ARCHS_TO_BUILD".upper(), config.global_archs_to_build - ) + self.keystore_name = config.env.str(f"{app_name}_KEYSTORE_FILE_NAME".upper(), config.global_keystore_name) + self.archs_to_build = config.env.list(f"{app_name}_ARCHS_TO_BUILD".upper(), config.global_archs_to_build) self.download_file_name = "" self.download_dl = config.env.str(f"{app_name}_DL".upper(), "") self.download_patch_resources(config) @@ -53,7 +46,7 @@ class APP(object): env_package_name = config.env.str(f"{app_name}_PACKAGE_NAME".upper(), None) self.package_name = env_package_name or Patches.get_package_name(app_name) - def download_apk_for_patching(self, config: RevancedConfig) -> None: + def download_apk_for_patching(self: Self, config: RevancedConfig) -> None: """Download apk to be patched.""" from src.downloader.download import Downloader from src.downloader.factory import DownloaderFactory @@ -61,31 +54,22 @@ class APP(object): if self.download_dl: logger.info("Downloading apk to be patched using provided dl") self.download_file_name = f"{self.app_name}.apk" - Downloader(config).direct_download( - self.download_dl, self.download_file_name - ) + Downloader(config).direct_download(self.download_dl, self.download_file_name) else: logger.info("Downloading apk to be patched by scrapping") try: if not self.download_source: - self.download_source = apk_sources[self.app_name].format( - self.package_name - ) - except KeyError: - raise DownloadFailure( - f"App {self.app_name} not supported officially yet. Please provide download " - "source in env." - ) - downloader = DownloaderFactory.create_downloader( - config=config, apk_source=self.download_source - ) - self.download_file_name, self.download_dl = downloader.download( - self.app_version, self - ) + self.download_source = apk_sources[self.app_name].format(self.package_name) + except KeyError as key: + msg = f"App {self.app_name} not supported officially yet. Please provide download source in env." + raise DownloadError( + msg, + ) from key + downloader = DownloaderFactory.create_downloader(config=config, apk_source=self.download_source) + self.download_file_name, self.download_dl = downloader.download(self.app_version, self) - def get_output_file_name(self) -> str: - """The function returns a string representing the output file name for - an APK file appended with version. + def get_output_file_name(self: Self) -> str: + """The function returns a string representing the output file name. Returns ------- @@ -93,32 +77,14 @@ class APP(object): """ return f"Re-{self.app_name}-{slugify(self.app_version)}-output.apk" - def set_recommended_version(self, version: str, exp: bool = False) -> None: - """The function sets the recommended version and experiment flag for an - app. - - Parameters - ---------- - version : str - The version parameter is a string that represents the recommended version of the app. - exp : bool, optional - The "exp" parameter is a boolean flag that indicates whether the specified version is for an - experimental or regular release. If "exp" is set to True, it means the version is for an - experimental release. If "exp" is set to False or not provided, it means the version is for - """ - self.app_version = version - self.experiment = exp - def __str__(self: "APP") -> str: + """Returns the str representation of the app.""" attrs = vars(self) return ", ".join([f"{key}: {value}" for key, value in attrs.items()]) @staticmethod - def download( - url: str, config: RevancedConfig, assets_filter: str, file_name: str = "" - ) -> str: - """The `download` function downloads a file from a given URL using a - specified configuration and filters the assets based on a given filter. + def download(url: str, config: RevancedConfig, assets_filter: str, file_name: str = "") -> str: + """The `download` function downloads a file from a given URL & filters the assets based on a given filter. Parameters ---------- @@ -156,9 +122,8 @@ class APP(object): Downloader(config).direct_download(url, file_name) return file_name - def download_patch_resources(self, config: RevancedConfig) -> None: - """The function `download_patch_resources` downloads various resources - for patching in parallel using a ThreadPoolExecutor. + def download_patch_resources(self: Self, config: RevancedConfig) -> None: + """The function `download_patch_resources` downloads various resources req. for patching. Parameters ---------- @@ -177,10 +142,7 @@ class APP(object): # Using a ThreadPoolExecutor for parallelism with ThreadPoolExecutor(4) as executor: - futures = { - resource_name: executor.submit(self.download, *args) - for resource_name, *args in download_tasks - } + futures = {resource_name: executor.submit(self.download, *args) for resource_name, *args in download_tasks} # Wait for all tasks to complete concurrent.futures.wait(futures.values()) @@ -189,13 +151,12 @@ class APP(object): for resource_name, future in futures.items(): try: self.resource[resource_name] = future.result() - except Exception as e: - raise PatchingFailed(e) from e + except UnknownError as e: + raise PatchingFailedError(e) from e @staticmethod def generate_filename(url: str) -> str: - """The function `generate_filename` takes a URL as input and returns a - hashed version of the URL as the filename. + """The function `generate_filename` takes URL as input and returns a hashed version of the URL as the filename. Parameters ---------- diff --git a/src/config.py b/src/config.py index 0ec4f67..5c00287 100644 --- a/src/config.py +++ b/src/config.py @@ -1,6 +1,6 @@ """Revanced Configurations.""" from pathlib import Path -from typing import List +from typing import List, Self from environs import Env from requests import Session @@ -8,15 +8,13 @@ from requests import Session default_cli = "https://github.com/revanced/revanced-cli/releases/latest" default_patches = "https://github.com/revanced/revanced-patches/releases/latest" default_patches_json = default_patches -default_integrations = ( - "https://github.com/revanced/revanced-integrations/releases/latest" -) +default_integrations = "https://github.com/revanced/revanced-integrations/releases/latest" -class RevancedConfig(object): +class RevancedConfig: """Revanced Configurations.""" - def __init__(self, env: Env) -> None: + def __init__(self: Self, env: Env) -> None: from src.utils import default_build, request_header self.env = env @@ -34,18 +32,10 @@ class RevancedConfig(object): self.dry_run = env.bool("DRY_RUN", False) self.global_cli_dl = env.str("GLOBAL_CLI_DL", default_cli) self.global_patches_dl = env.str("GLOBAL_PATCHES_DL", default_patches) - self.global_patches_json_dl = env.str( - "GLOBAL_PATCHES_JSON_DL", default_patches_json - ) - self.global_integrations_dl = env.str( - "GLOBAL_INTEGRATIONS_DL", default_integrations - ) - self.global_keystore_name = env.str( - "GLOBAL_KEYSTORE_FILE_NAME", "revanced.keystore" - ) + self.global_patches_json_dl = env.str("GLOBAL_PATCHES_JSON_DL", default_patches_json) + self.global_integrations_dl = env.str("GLOBAL_INTEGRATIONS_DL", default_integrations) + self.global_keystore_name = env.str("GLOBAL_KEYSTORE_FILE_NAME", "revanced.keystore") self.global_archs_to_build = env.list("GLOBAL_ARCHS_TO_BUILD", []) self.extra_download_files: List[str] = env.list("EXTRA_FILES", []) self.apk_editor = "apkeditor-output.jar" - self.extra_download_files.append( - "https://github.com/REAndroid/APKEditor@apkeditor.jar" - ) + self.extra_download_files.append("https://github.com/REAndroid/APKEditor@apkeditor.jar") diff --git a/src/downloader/__init__.py b/src/downloader/__init__.py index e69de29..f48af95 100644 --- a/src/downloader/__init__.py +++ b/src/downloader/__init__.py @@ -0,0 +1 @@ +"""Downloader files.""" diff --git a/src/downloader/apkmirror.py b/src/downloader/apkmirror.py index 2e20f1c..1307f73 100644 --- a/src/downloader/apkmirror.py +++ b/src/downloader/apkmirror.py @@ -1,5 +1,5 @@ """Downloader Class.""" -from typing import Any, Dict, Tuple +from typing import Any, Dict, Self, Tuple import requests from bs4 import BeautifulSoup, Tag @@ -8,31 +8,29 @@ from loguru import logger from src.app import APP from src.downloader.download import Downloader from src.downloader.sources import APK_MIRROR_BASE_URL -from src.exceptions import APKMirrorAPKDownloadFailure +from src.downloader.utils import status_code_200 +from src.exceptions import APKMirrorAPKDownloadError from src.utils import bs4_parser, contains_any_word, request_header class ApkMirror(Downloader): """Files downloader.""" - def _extract_force_download_link(self, link: str, app: str) -> Tuple[str, str]: + def _extract_force_download_link(self: Self, link: str, app: str) -> Tuple[str, str]: """Extract force download link.""" notes_divs = self._extracted_search_div(link, "tab-pane") apk_type = self._extracted_search_div(link, "apkm-badge").get_text() extension = "zip" if apk_type == "BUNDLE" else "apk" possible_links = notes_divs.find_all("a") for possible_link in possible_links: - if possible_link.get("href") and "download.php?id=" in possible_link.get( - "href" - ): + if possible_link.get("href") and "download.php?id=" in possible_link.get("href"): file_name = f"{app}.{extension}" self._download(APK_MIRROR_BASE_URL + possible_link["href"], file_name) return file_name, APK_MIRROR_BASE_URL + possible_link["href"] - raise APKMirrorAPKDownloadFailure( - f"Unable to extract force download for {app}", url=link - ) + msg = f"Unable to extract force download for {app}" + raise APKMirrorAPKDownloadError(msg, url=link) - def extract_download_link(self, page: str, app: str) -> Tuple[str, str]: + def extract_download_link(self: Self, page: str, app: str) -> Tuple[str, str]: """Function to extract the download link from apkmirror html page. :param page: Url of the page @@ -45,19 +43,15 @@ class ApkMirror(Downloader): ( download_link["href"] for download_link in download_links - if download_link.get("href") - and "download/?key=" in download_link.get("href") + if download_link.get("href") and "download/?key=" in download_link.get("href") ), None, ): - return self._extract_force_download_link( - APK_MIRROR_BASE_URL + final_download_link, app - ) - raise APKMirrorAPKDownloadFailure( - f"Unable to extract link from {app} version list", url=page - ) + return self._extract_force_download_link(APK_MIRROR_BASE_URL + final_download_link, app) + msg = f"Unable to extract link from {app} version list" + raise APKMirrorAPKDownloadError(msg, url=page) - def get_download_page(self, main_page: str) -> str: + def get_download_page(self: Self, main_page: str) -> str: """Function to get the download page in apk_mirror. :param main_page: Main Download Page in APK mirror(Index) @@ -77,26 +71,23 @@ class ApkMirror(Downloader): links[apk_type] = f"{APK_MIRROR_BASE_URL}{sub_url}" if preferred_link := links.get("APK", links.get("BUNDLE")): return preferred_link - raise APKMirrorAPKDownloadFailure( - "Unable to extract download page", url=main_page - ) + msg = "Unable to extract download page" + raise APKMirrorAPKDownloadError(msg, url=main_page) @staticmethod def _extracted_search_div(url: str, search_class: str) -> Tag: """Extract search div.""" r = requests.get(url, headers=request_header, timeout=60) - if r.status_code != 200: - raise APKMirrorAPKDownloadFailure( - f"Unable to connect with {url} on ApkMirror. Are you blocked by APKMirror or abused apkmirror " - f"?.Reason - {r.text}", + if r.status_code != status_code_200: + msg = f"Unable to connect with {url}. Are you blocked by APKMirror or abused apkmirror ?.Reason - {r.text}" + raise APKMirrorAPKDownloadError( + msg, url=url, ) soup = BeautifulSoup(r.text, bs4_parser) - return soup.find(class_=search_class) + return soup.find(class_=search_class) # type: ignore[return-value] - def specific_version( - self, app: APP, version: str, main_page: str = "" - ) -> Tuple[str, str]: + def specific_version(self: Self, app: APP, version: str, main_page: str = "") -> Tuple[str, str]: """Function to download the specified version of app from apkmirror. :param app: Name of the application @@ -112,18 +103,14 @@ class ApkMirror(Downloader): download_page = self.get_download_page(main_page) return self.extract_download_link(download_page, app.app_name) - def latest_version(self, app: APP, **kwargs: Any) -> Tuple[str, str]: - """Function to download whatever the latest version of app from - apkmirror. + def latest_version(self: Self, app: APP, **kwargs: Any) -> Tuple[str, str]: + """Function to download whatever the latest version of app from apkmirror. :param app: Name of the application :return: Version of downloaded apk """ - app_main_page = app.download_source - versions_div = self._extracted_search_div( - app_main_page, "listWidget p-relative" - ) + versions_div = self._extracted_search_div(app_main_page, "listWidget p-relative") app_rows = versions_div.find_all(class_="appRow") version_urls = [ app_row.find(class_="downloadLink")["href"] @@ -131,6 +118,4 @@ class ApkMirror(Downloader): if "beta" not in app_row.find(class_="appRowTitle").get_text().lower() and "alpha" not in app_row.find(class_="appRowTitle").get_text().lower() ] - return self.specific_version( - app, "latest", APK_MIRROR_BASE_URL + max(version_urls) - ) + return self.specific_version(app, "latest", APK_MIRROR_BASE_URL + max(version_urls)) diff --git a/src/downloader/apkpure.py b/src/downloader/apkpure.py index ee61df3..fd150e7 100644 --- a/src/downloader/apkpure.py +++ b/src/downloader/apkpure.py @@ -1,5 +1,5 @@ """APK Pure Downloader Class.""" -from typing import Any, Tuple +from typing import Any, Self, Tuple from src.app import APP from src.downloader.download import Downloader @@ -8,9 +8,8 @@ from src.downloader.download import Downloader class ApkPure(Downloader): """Files downloader.""" - def latest_version(self, app: APP, **kwargs: Any) -> Tuple[str, str]: - """Function to download whatever the latest version of app from - apkmirror. + def latest_version(self: Self, app: APP, **kwargs: Any) -> Tuple[str, str]: + """Function to download whatever the latest version of app from apkmirror. :param app: Name of the application :return: Version of downloaded apk diff --git a/src/downloader/apksos.py b/src/downloader/apksos.py index 0194611..5f0bb9b 100644 --- a/src/downloader/apksos.py +++ b/src/downloader/apksos.py @@ -1,38 +1,38 @@ """APK SOS Downloader Class.""" -from typing import Any, Tuple +from typing import Any, Self, Tuple import requests from bs4 import BeautifulSoup from src.app import APP from src.downloader.download import Downloader -from src.exceptions import APKSosAPKDownloadFailure +from src.exceptions import APKSosAPKDownloadError from src.utils import bs4_parser, request_header class ApkSos(Downloader): """Files downloader.""" - def extract_download_link(self, page: str, app: str) -> Tuple[str, str]: + def extract_download_link(self: Self, page: str, app: str) -> Tuple[str, str]: """Function to extract the download link from apkmirror html page. :param page: Url of the page :param app: Name of the app """ - r = requests.get(page, headers=request_header, allow_redirects=True) + r = requests.get(page, headers=request_header, allow_redirects=True, timeout=60) soup = BeautifulSoup(r.text, bs4_parser) download_button = soup.find(class_="col-sm-12 col-md-8 text-center") - possible_links = download_button.find_all("a") + possible_links = download_button.find_all("a") # type: ignore[union-attr] for possible_link in possible_links: if possible_link.get("href"): file_name = f"{app}.apk" self._download(possible_link["href"], file_name) return file_name, possible_link["href"] - raise APKSosAPKDownloadFailure(f"Unable to download {app}", url=page) + msg = f"Unable to download {app}" + raise APKSosAPKDownloadError(msg, url=page) - def latest_version(self, app: APP, **kwargs: Any) -> Tuple[str, str]: - """Function to download whatever the latest version of app from - apkmirror. + def latest_version(self: Self, app: APP, **kwargs: Any) -> Tuple[str, str]: + """Function to download whatever the latest version of app from apkmirror. :param app: Name of the application :return: Version of downloaded apk diff --git a/src/downloader/download.py b/src/downloader/download.py index 2bfc82a..c147574 100644 --- a/src/downloader/download.py +++ b/src/downloader/download.py @@ -4,7 +4,7 @@ import subprocess from pathlib import Path from queue import PriorityQueue from time import perf_counter -from typing import Any, Tuple +from typing import Any, Self, Tuple from loguru import logger from tqdm import tqdm @@ -12,35 +12,25 @@ from tqdm import tqdm from src.app import APP from src.config import RevancedConfig from src.downloader.utils import implement_method -from src.exceptions import DownloadFailure +from src.exceptions import DownloadError from src.utils import handle_request_response -class Downloader(object): +class Downloader: """Files downloader.""" - def __init__(self, config: RevancedConfig): + def __init__(self: Self, config: RevancedConfig) -> None: self._CHUNK_SIZE = 10485760 self._QUEUE: PriorityQueue[Tuple[float, str]] = PriorityQueue() self._QUEUE_LENGTH = 0 self.config = config - @staticmethod - def file_status_check(file_name: Path, dry_run: bool, url: str) -> bool: - """Check if file already exists.""" - if os.path.exists(file_name) or dry_run: - logger.debug( - f"Skipping download of {file_name} from {url}. File already exists or dry running." - ) - return True - return False - - def _download(self, url: str, file_name: str) -> None: + def _download(self: Self, url: str, file_name: str) -> None: if not url: - raise DownloadFailure("No url provided to download") - if self.file_status_check( - self.config.temp_folder.joinpath(file_name), self.config.dry_run, url - ): + msg = "No url provided to download" + raise DownloadError(msg) + if self.config.dry_run or Path(file_name).exists(): + logger.debug(f"Skipping download of {file_name} from {url}. File already exists or dry running.") return logger.info(f"Trying to download {file_name} from {url}") self._QUEUE_LENGTH += 1 @@ -71,11 +61,11 @@ 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) -> Tuple[str, str]: + def extract_download_link(self: Self, page: str, app: str) -> Tuple[str, str]: """Extract download link from web page.""" raise NotImplementedError(implement_method) - def specific_version(self, app: APP, version: str) -> Tuple[str, str]: + def specific_version(self: Self, app: APP, version: str) -> Tuple[str, str]: """Function to download the specified version of app from apkmirror. :param app: Name of the application @@ -84,7 +74,7 @@ class Downloader(object): """ raise NotImplementedError(implement_method) - def latest_version(self, app: APP, **kwargs: Any) -> Tuple[str, str]: + def latest_version(self: Self, app: APP, **kwargs: Any) -> Tuple[str, str]: """Function to download the latest version of app. :param app: Name of the application @@ -92,7 +82,7 @@ class Downloader(object): """ raise NotImplementedError(implement_method) - def convert_to_apk(self, file_name: str) -> str: + def convert_to_apk(self: Self, file_name: str) -> str: """Convert apks to apk.""" if file_name.endswith(".apk"): return file_name @@ -122,7 +112,7 @@ class Downloader(object): base_name, _ = os.path.splitext(filename) return base_name + new_extension - def download(self, version: str, app: APP, **kwargs: Any) -> Tuple[str, str]: + def download(self: Self, version: str, app: APP, **kwargs: Any) -> Tuple[str, str]: """Public function to download apk to patch. :param version: version to download @@ -131,9 +121,7 @@ class Downloader(object): if self.config.dry_run: return "", "" if app in self.config.existing_downloaded_apks: - logger.debug( - f"Will not download {app.app_name} -v{version} from the internet." - ) + logger.debug(f"Will not download {app.app_name} -v{version} from the internet.") return app.app_name, f"local://{app.app_name}" if version and version != "latest": file_name, app_dl = self.specific_version(app, version) @@ -141,6 +129,6 @@ class Downloader(object): file_name, app_dl = self.latest_version(app, **kwargs) return self.convert_to_apk(file_name), app_dl - def direct_download(self, dl: str, file_name: str) -> None: + def direct_download(self: Self, dl: str, file_name: str) -> None: """Download from DL.""" self._download(dl, file_name) diff --git a/src/downloader/factory.py b/src/downloader/factory.py index b159c4f..cafc48e 100644 --- a/src/downloader/factory.py +++ b/src/downloader/factory.py @@ -5,37 +5,32 @@ from src.downloader.apkpure import ApkPure from src.downloader.apksos import ApkSos from src.downloader.download import Downloader from src.downloader.github import Github -from src.downloader.sources import ( - APK_MIRROR_BASE_URL, - APK_PURE_BASE_URL, - APKS_SOS_BASE_URL, - GITHUB_BASE_URL, -) +from src.downloader.sources import APK_MIRROR_BASE_URL, APK_PURE_BASE_URL, APKS_SOS_BASE_URL, GITHUB_BASE_URL from src.downloader.uptodown import UptoDown -from src.exceptions import DownloadFailure +from src.exceptions import DownloadError -class DownloaderFactory(object): +class DownloaderFactory: """Downloader Factory.""" @staticmethod def create_downloader(config: RevancedConfig, apk_source: str) -> Downloader: """Returns appropriate downloader. - Parameters - ---------- - app : App Name - config : Config - apk_source : Source URL for APK + Args: + ---- + config : Config + apk_source : Source URL for APK """ if apk_source.startswith(GITHUB_BASE_URL): return Github(config) if apk_source.startswith(APK_PURE_BASE_URL): return ApkPure(config) - elif apk_source.startswith(APKS_SOS_BASE_URL): + if apk_source.startswith(APKS_SOS_BASE_URL): return ApkSos(config) - elif apk_source.endswith("en.uptodown.com/android"): + if apk_source.endswith("en.uptodown.com/android"): return UptoDown(config) - elif apk_source.startswith(APK_MIRROR_BASE_URL): + if apk_source.startswith(APK_MIRROR_BASE_URL): return ApkMirror(config) - raise DownloadFailure("No download factory found.") + msg = "No download factory found." + raise DownloadError(msg) diff --git a/src/downloader/github.py b/src/downloader/github.py index c95305c..5041c14 100644 --- a/src/downloader/github.py +++ b/src/downloader/github.py @@ -1,6 +1,6 @@ """Github Downloader.""" import re -from typing import Dict, Tuple +from typing import Dict, Self, Tuple from urllib.parse import urlparse import requests @@ -9,23 +9,21 @@ from loguru import logger from src.app import APP from src.config import RevancedConfig from src.downloader.download import Downloader -from src.exceptions import DownloadFailure +from src.exceptions import DownloadError from src.utils import handle_request_response, update_changelog class Github(Downloader): """Files downloader.""" - def latest_version(self, app: APP, **kwargs: Dict[str, str]) -> Tuple[str, str]: + def latest_version(self: Self, app: APP, **kwargs: Dict[str, str]) -> Tuple[str, str]: """Function to download files from GitHub repositories. :param app: App to download """ logger.debug(f"Trying to download {app.app_name} from github") if self.config.dry_run: - logger.debug( - f"Skipping download of {app.app_name}. File already exists or dry running." - ) + logger.debug(f"Skipping download of {app.app_name}. File already exists or dry running.") return app.app_name, f"local://{app.app_name}" owner = str(kwargs["owner"]) repo_name = str(kwargs["name"]) @@ -56,11 +54,7 @@ class Github(Downloader): github_repo_name = path_segments[1] release_tag = next( - ( - f"tags/{path_segments[i + 1]}" - for i, segment in enumerate(path_segments) - if segment == "tag" - ), + (f"tags/{path_segments[i + 1]}" for i, segment in enumerate(path_segments) if segment == "tag"), "latest", ) return github_repo_owner, github_repo_name, release_tag @@ -86,9 +80,8 @@ class Github(Downloader): try: filter_pattern = re.compile(asset_filter) except re.error as e: - raise DownloadFailure( - f"Invalid regex {asset_filter} pattern provided." - ) from e + msg = f"Invalid regex {asset_filter} pattern provided." + raise DownloadError(msg) from e for asset in assets: assets_url = asset["browser_download_url"] assets_name = asset["name"] @@ -98,11 +91,7 @@ class Github(Downloader): return "" @staticmethod - def patch_resource( - repo_url: str, assets_filter: str, config: RevancedConfig - ) -> str: + def patch_resource(repo_url: str, assets_filter: str, config: RevancedConfig) -> str: """Fetch patch resource from repo url.""" repo_owner, repo_name, tag = Github._extract_repo_owner_and_tag(repo_url) - return Github._get_release_assets( - repo_owner, repo_name, tag, assets_filter, config - ) + return Github._get_release_assets(repo_owner, repo_name, tag, assets_filter, config) diff --git a/src/downloader/sources.py b/src/downloader/sources.py index eca824e..4a40f5e 100644 --- a/src/downloader/sources.py +++ b/src/downloader/sources.py @@ -1,3 +1,4 @@ +"""APK Sources used.""" APK_MIRROR_BASE_URL = "https://www.apkmirror.com" APK_MIRROR_BASE_APK_URL = f"{APK_MIRROR_BASE_URL}/apk" UPTODOWN_BASE_URL = "https://{}.en.uptodown.com/android" diff --git a/src/downloader/uptodown.py b/src/downloader/uptodown.py index 8f65bac..fda670e 100644 --- a/src/downloader/uptodown.py +++ b/src/downloader/uptodown.py @@ -1,5 +1,5 @@ """Upto Down Downloader.""" -from typing import Any, Tuple +from typing import Any, Self, Tuple import requests from bs4 import BeautifulSoup @@ -7,27 +7,33 @@ from loguru import logger from src.app import APP from src.downloader.download import Downloader -from src.exceptions import UptoDownAPKDownloadFailure +from src.exceptions import UptoDownAPKDownloadError from src.utils import bs4_parser, request_header class UptoDown(Downloader): """Files downloader.""" - def extract_download_link(self, page: str, app: str) -> Tuple[str, str]: + def extract_download_link(self: Self, page: str, app: str) -> Tuple[str, str]: + """Extract download link from uptodown url.""" r = requests.get(page, headers=request_header, allow_redirects=True, timeout=60) soup = BeautifulSoup(r.text, bs4_parser) - soup = soup.find(id="detail-download-button") - download_url = soup.get("data-url") + download_button = soup.find(id="detail-download-button") + if not download_button: + msg = f"Unable to download {app} from uptodown." + raise UptoDownAPKDownloadError(msg, url=page) + download_url = download_button.get("data-url") # type: ignore[union-attr] if not download_url: - raise UptoDownAPKDownloadFailure( - f"Unable to download {app} from uptodown.", url=page - ) + msg = f"Unable to download {app} from uptodown." + raise UptoDownAPKDownloadError(msg, url=page) file_name = f"{app}.apk" - self._download(download_url, file_name) - return file_name, download_url + if isinstance(download_url, str): + self._download(download_url, file_name) + return file_name, download_url + msg = f"Unable to download {app} from uptodown." + raise UptoDownAPKDownloadError(msg, url=page) - def specific_version(self, app: APP, version: str) -> Tuple[str, str]: + def specific_version(self: Self, app: APP, version: str) -> Tuple[str, str]: """Function to download the specified version of app from apkmirror. :param app: Name of the application @@ -40,17 +46,18 @@ class UptoDown(Downloader): soup = BeautifulSoup(html, bs4_parser) versions_list = soup.find("section", {"id": "versions"}) download_url = None - for version_item in versions_list.find_all("div", {"data-url": True}): + for version_item in versions_list.find_all("div", {"data-url": True}): # type: ignore[union-attr] extracted_version = version_item.find("span", {"class": "version"}).text if extracted_version == version: download_url = version_item["data-url"] break if download_url is None: - raise UptoDownAPKDownloadFailure( - f"Unable to download {app.app_name} from uptodown.", url=url - ) + msg = f"Unable to download {app.app_name} from uptodown." + raise UptoDownAPKDownloadError(msg, url=url) return self.extract_download_link(download_url, app.app_name) - def latest_version(self, app: APP, **kwargs: Any) -> Tuple[str, str]: + def latest_version(self: Self, app: APP, **kwargs: Any) -> Tuple[str, str]: + """Function to download the latest version of app from uptodown.""" + logger.debug("downloading latest version of app from uptodown.") page = f"{app.download_source}/download" return self.extract_download_link(page, app.app_name) diff --git a/src/downloader/utils.py b/src/downloader/utils.py index 667199b..4292655 100644 --- a/src/downloader/utils.py +++ b/src/downloader/utils.py @@ -1,4 +1,4 @@ """Utility class.""" - implement_method = "Please implement the method" +status_code_200 = 200 diff --git a/src/exceptions.py b/src/exceptions.py index 475ba1e..4d486ef 100644 --- a/src/exceptions.py +++ b/src/exceptions.py @@ -1,13 +1,15 @@ -from typing import Any +"""Possible Exceptions.""" +from typing import Any, Self -class APKMirrorIconScrapFailure(Exception): +class APKMirrorIconScrapError(Exception): """Exception raised when the icon cannot be scraped from apkmirror.""" - def __init__(self, *args: Any, **kwargs: Any) -> None: + def __init__(self: Self, *args: Any, **kwargs: Any) -> None: """Initialize the APKMirrorIconScrapFailure exception. Args: + ---- *args: Variable length argument list. **kwargs: Arbitrary keyword arguments. url (str, optional): The URL of the failed icon scraping. Defaults to None. @@ -16,13 +18,18 @@ class APKMirrorIconScrapFailure(Exception): self.url = kwargs.get("url", None) -class DownloadFailure(Exception): +class APKComboIconScrapError(APKMirrorIconScrapError): + """Exception raised when the icon cannot be scraped from apkcombo.""" + + +class DownloadError(Exception): """Generic Download failure.""" - def __init__(self, *args: Any, **kwargs: Any) -> None: + def __init__(self: Self, *args: Any, **kwargs: Any) -> None: """Initialize the APKMirrorAPKDownloadFailure exception. Args: + ---- *args: Variable length argument list. **kwargs: Arbitrary keyword arguments. url (str, optional): The URL of the failed icon scraping. Defaults to None. @@ -31,64 +38,53 @@ class DownloadFailure(Exception): self.url = kwargs.get("url", None) -class APKDownloadFailure(DownloadFailure): +class APKDownloadError(DownloadError): """Exception raised when the apk cannot be scraped.""" - pass - -class APKMirrorAPKDownloadFailure(APKDownloadFailure): +class APKMirrorAPKDownloadError(APKDownloadError): """Exception raised when downloading an APK from apkmirror failed.""" - pass - -class APKMirrorAPKNotFound(APKDownloadFailure): +class APKMirrorAPKNotFoundError(APKDownloadError): """Exception raised when apk doesn't exist on APKMirror.""" - pass - -class UptoDownAPKDownloadFailure(APKDownloadFailure): +class UptoDownAPKDownloadError(APKDownloadError): """Exception raised when downloading an APK from uptodown failed.""" - pass - -class APKPureAPKDownloadFailure(APKDownloadFailure): +class APKPureAPKDownloadError(APKDownloadError): """Exception raised when downloading an APK from apkpure failed.""" - pass - -class APKSosAPKDownloadFailure(APKDownloadFailure): +class APKSosAPKDownloadError(APKDownloadError): """Exception raised when downloading an APK from apksos failed.""" - pass - -class PatchingFailed(Exception): +class PatchingFailedError(Exception): """Patching Failed.""" - pass - -class AppNotFound(ValueError): +class AppNotFoundError(ValueError): """Not a valid Revanced App.""" - pass - -class PatchesJsonLoadFailed(ValueError): +class PatchesJsonLoadError(ValueError): """Failed to load patches json.""" - def __init__(self, *args: Any, **kwargs: Any) -> None: + def __init__(self: Self, *args: Any, **kwargs: Any) -> None: """Initialize the PatchesJsonLoadFailed exception. Args: + ---- *args: Variable length argument list. **kwargs: Arbitrary keyword arguments. file_name (str, optional): The name of json file. Defaults to None. """ super().__init__(*args) self.file_name = kwargs.get("file_name", None) + + +class UnknownError(Exception): + """Some unknown error.""" diff --git a/src/parser.py b/src/parser.py index f600b8f..3e62d6e 100644 --- a/src/parser.py +++ b/src/parser.py @@ -1,18 +1,18 @@ """Revanced Parser.""" from subprocess import PIPE, Popen from time import perf_counter -from typing import List +from typing import List, Self from loguru import logger from src.app import APP from src.config import RevancedConfig -from src.exceptions import PatchingFailed +from src.exceptions import PatchingFailedError from src.patches import Patches from src.utils import possible_archs -class Parser(object): +class Parser: """Revanced Parser.""" CLI_JAR = "-jar" @@ -23,13 +23,13 @@ class Parser(object): KEYSTORE_ARG = "--keystore" OPTIONS_ARG = "--options" - def __init__(self, patcher: Patches, config: RevancedConfig) -> None: + def __init__(self: Self, patcher: Patches, config: RevancedConfig) -> None: self._PATCHES: List[str] = [] self._EXCLUDED: List[str] = [] self.patcher = patcher self.config = config - def include(self, name: str) -> None: + def include(self: Self, name: str) -> None: """The function `include` adds a given patch to a list of patches. Parameters @@ -39,9 +39,8 @@ class Parser(object): """ self._PATCHES.extend(["-i", name]) - def exclude(self, name: str) -> None: - """The `exclude` function adds a given patch to the list of excluded - patches. + def exclude(self: Self, name: str) -> None: + """The `exclude` function adds a given patch to the list of excluded patches. Parameters ---------- @@ -51,9 +50,8 @@ class Parser(object): self._PATCHES.extend(["-e", name]) self._EXCLUDED.append(name) - def get_excluded_patches(self) -> List[str]: - """The function `get_excluded_patches` is a getter method that returns - a list of excluded patches. + def get_excluded_patches(self: Self) -> List[str]: + """The function `get_excluded_patches` is a getter method that returns a list of excluded patches. Returns ------- @@ -61,9 +59,8 @@ class Parser(object): """ return self._EXCLUDED - def get_all_patches(self) -> List[str]: - """The function "get_all_patches" is a getter method that returns a - list of all patches. + def get_all_patches(self: Self) -> List[str]: + """The function "get_all_patches" is a getter method that returns a ist of all patches. Returns ------- @@ -71,9 +68,8 @@ class Parser(object): """ return self._PATCHES - def invert_patch(self, name: str) -> bool: - """The function `invert_patch` takes a name as input, it toggles the - status of the patch and returns True, otherwise it returns False. + def invert_patch(self: Self, name: str) -> bool: + """The function `invert_patch` takes a name as input, it toggles the status of the patch. Parameters ---------- @@ -94,27 +90,23 @@ class Parser(object): self._PATCHES[patch_index - 1] = "-i" else: self._PATCHES[patch_index - 1] = "-e" - return True except ValueError: return False + else: + return True - def exclude_all_patches(self) -> None: - """The function `exclude_all_patches` replaces all occurrences of "-i" - with "-e" in the list `self._PATCHES`. - - Hence exclude all patches - """ + def exclude_all_patches(self: Self) -> None: + """The function `exclude_all_patches` exclude all the patches.""" for idx, item in enumerate(self._PATCHES): if item == "-i": self._PATCHES[idx] = "-e" # noinspection IncorrectFormatting def patch_app( - self, + self: Self, app: APP, ) -> None: - """The function `patch_app` is used to patch an app using the Revanced - CLI tool. + """The function `patch_app` is used to patch an app using the Revanced CLI tool. Parameters ---------- @@ -151,16 +143,13 @@ class Parser(object): for arch in excluded: args.extend(("--rip-lib", arch)) start = perf_counter() - logger.debug( - f"Sending request to revanced cli for building with args java {args}" - ) + logger.debug(f"Sending request to revanced cli for building with args java {args}") process = Popen(["java", *args], stdout=PIPE) output = process.stdout if not output: - raise PatchingFailed("Failed to send request for patching.") + msg = "Failed to send request for patching." + raise PatchingFailedError(msg) for line in output: logger.debug(line.decode(), flush=True, end="") process.wait() - logger.info( - f"Patching completed for app {app} in {perf_counter() - start:.2f} seconds." - ) + logger.info(f"Patching completed for app {app} in {perf_counter() - start:.2f} seconds.") diff --git a/src/patches.py b/src/patches.py index 03143cc..2ddade9 100644 --- a/src/patches.py +++ b/src/patches.py @@ -2,19 +2,20 @@ import contextlib import json -from typing import Any, Dict, List, Tuple +from pathlib import Path +from typing import Any, ClassVar, Dict, List, Self, Tuple from loguru import logger from src.app import APP from src.config import RevancedConfig -from src.exceptions import AppNotFound, PatchesJsonLoadFailed +from src.exceptions import AppNotFoundError, PatchesJsonLoadError -class Patches(object): +class Patches: """Revanced Patches.""" - revanced_package_names = { + revanced_package_names: ClassVar[Dict[str, str]] = { "com.reddit.frontpage": "reddit", "com.ss.android.ugc.trill": "tiktok", "com.twitter.android": "twitter", @@ -59,9 +60,7 @@ class Patches(object): @staticmethod def get_package_name(app: str) -> str: - """The function `get_package_name` takes an app name as input and - returns the corresponding package name, or raises an exception if the - app is not supported. + """The function `get_package_name` takes an app name as input and returns the corresponding package name. Parameters ---------- @@ -75,14 +74,12 @@ class Patches(object): for package, app_name in Patches.revanced_package_names.items(): if app_name == app: return package - raise AppNotFound( - f"App {app} not supported officially yet. Please provide package name in env to proceed." - ) + msg = f"App {app} not supported officially yet. Please provide package name in env to proceed." + raise AppNotFoundError(msg) @staticmethod def support_app() -> Dict[str, str]: - """The function `support_app()` returns a dictionary of supported app - IDs. + """The function returns a dictionary of supported app IDs. Returns ------- @@ -90,9 +87,8 @@ class Patches(object): """ return Patches.revanced_package_names - def fetch_patches(self, config: RevancedConfig, app: APP) -> None: - """The function fetches patches from a JSON file and organizes them - based on compatibility with different applications. + def fetch_patches(self: Self, config: RevancedConfig, app: APP) -> None: + """The function fetches patches from a JSON file. Parameters ---------- @@ -104,9 +100,7 @@ class Patches(object): """ self.patches_dict[app.app_name] = [] patch_loader = PatchLoader() - patches = patch_loader.load_patches( - f'{config.temp_folder}/{app.resource["patches_json"]}' - ) + patches = patch_loader.load_patches(f'{config.temp_folder}/{app.resource["patches_json"]}') for patch in patches: if not patch["compatiblePackages"]: @@ -115,9 +109,7 @@ class Patches(object): p["version"] = "all" self.patches_dict["universal_patch"].append(p) else: - for compatible_package, version in [ - (x["name"], x["versions"]) for x in patch["compatiblePackages"] - ]: + for compatible_package, version in [(x["name"], x["versions"]) for x in patch["compatiblePackages"]]: if app.package_name == compatible_package: p = {x: patch[x] for x in ["name", "description"]} p["app"] = compatible_package @@ -126,13 +118,12 @@ class Patches(object): app.no_of_patches = len(self.patches_dict[app.app_name]) - def __init__(self, config: RevancedConfig, app: APP) -> None: + def __init__(self: Self, config: RevancedConfig, app: APP) -> None: self.patches_dict: Dict[str, Any] = {"universal_patch": []} self.fetch_patches(config, app) - def get(self, app: str) -> Tuple[List[Dict[str, str]], str]: - """The function `get` retrieves all patches for a given application and - returns them along with the latest version. + def get(self: Self, app: str) -> Tuple[List[Dict[str, str]], str]: + """The function `get` returns all patches and version for a given application. Parameters ---------- @@ -146,18 +137,14 @@ class Patches(object): patches for the given app. The second element is a string representing the version of the patches. """ - patches = self.patches_dict[app] version = "latest" with contextlib.suppress(StopIteration): version = next(i["version"] for i in patches if i["version"] != "all") return patches, version - def include_exclude_patch( - self, app: APP, parser: Any, patches: List[Dict[str, str]] - ) -> None: - """The function `include_exclude_patch` includes and excludes patches - for a given app based on certain conditions. + def include_exclude_patch(self: Self, app: APP, parser: Any, patches: List[Dict[str, str]]) -> None: + """The function `include_exclude_patch` includes and excludes patches for a given app. Parameters ---------- @@ -173,20 +160,14 @@ class Patches(object): """ for patch in patches: normalized_patch = patch["name"].lower().replace(" ", "-") - parser.include( - normalized_patch - ) if normalized_patch not in app.exclude_request else parser.exclude( + parser.include(normalized_patch) if normalized_patch not in app.exclude_request else parser.exclude( normalized_patch ) for normalized_patch in app.include_request: - parser.include( - normalized_patch - ) if normalized_patch not in self.patches_dict["universal_patch"] else () + parser.include(normalized_patch) if normalized_patch not in self.patches_dict["universal_patch"] else () - def get_app_configs(self, app: "APP") -> List[Dict[str, str]]: - """The function `get_app_configs` retrieves configurations for a given - app, including patches, version information, and whether it is - experimental. + def get_app_configs(self: Self, app: "APP") -> List[Dict[str, str]]: + """The function `get_app_configs` returns configurations for a given app. Parameters ---------- @@ -211,7 +192,8 @@ class Patches(object): ): experiment = True recommended_version = app.app_version - app.set_recommended_version(recommended_version, experiment) + app.app_version = recommended_version + app.experiment = experiment return total_patches @@ -220,8 +202,7 @@ class PatchLoader: @staticmethod def load_patches(file_name: str) -> Any: - """The function `load_patches` loads patches from a file and returns - them. + """The function `load_patches` loads patches from a file and returns them. Parameters ---------- @@ -234,8 +215,8 @@ class PatchLoader: the patches loaded from the file. """ try: - with open(file_name) as f: - patches = json.load(f) - return patches + with Path(file_name).open() as f: + return json.load(f) except FileNotFoundError as e: - raise PatchesJsonLoadFailed("File not found", file_name=file_name) from e + msg = "File not found" + raise PatchesJsonLoadError(msg, file_name=file_name) from e diff --git a/src/utils.py b/src/utils.py index c3dbbb4..c343fab 100644 --- a/src/utils.py +++ b/src/utils.py @@ -2,6 +2,8 @@ import os import re import subprocess +import sys +from pathlib import Path from typing import Any, Dict, List import requests @@ -9,7 +11,8 @@ from loguru import logger from requests import Response from src.config import RevancedConfig -from src.exceptions import DownloadFailure +from src.downloader.utils import status_code_200 +from src.exceptions import DownloadError default_build = [ "youtube", @@ -28,8 +31,7 @@ bs4_parser = "html.parser" def update_changelog(name: str, response: Dict[str, str]) -> None: - """The function `update_changelog` updates the changelog file with the - provided name and response. + """The function `update_changelog` updates the changelog file. Parameters ---------- @@ -46,8 +48,7 @@ def update_changelog(name: str, response: Dict[str, str]) -> None: def format_changelog(name: str, response: Dict[str, str], parent_repo: str) -> str: - """The `format_changelog` function takes in a name, a response dictionary, - and a parent repository, and returns a formatted changelog string. + """The `format_changelog` returns formatted changelog string. Parameters ---------- @@ -67,43 +68,28 @@ def format_changelog(name: str, response: Dict[str, str], parent_repo: str) -> s a formatted changelog as a string. """ collapse_start = f"\n
👀 {name} \n\n" - release_version = ( - f"**Release Version** - [{response['tag_name']}]({response['html_url']})
" - ) + release_version = f"**Release Version** - [{response['tag_name']}]({response['html_url']})
" change_log = f"**Changelog** -
{response['body']}" publish_time = f"**Published at** -
{response['published_at']}" - footer = ( - f"
Change logs generated by [Docker Py Revanced]({parent_repo})\n" - ) + footer = f"
Change logs generated by [Docker Py Revanced]({parent_repo})\n" collapse_end = "
" - return "".join( - [ - collapse_start, - release_version, - change_log, - publish_time, - footer, - collapse_end, - ] - ) + return f"{collapse_start}{release_version}{change_log}{publish_time}{footer}{collapse_end}" def write_to_file(change_log: str) -> None: - """The function `write_to_file` writes a given changelog string to a file - named "changelog.md". + """The function `write_to_file` writes a given changelog string to a file. Parameters ---------- change_log : str A string representing the changelog that you want to write to the file. """ - with open("changelog.md", "w", encoding="utf_8") as file1: + with Path("changelog.md").open("w", encoding="utf_8") as file1: file1.write(change_log) def get_parent_repo() -> str: - """The function `get_parent_repo()` returns the URL of the parent - repository. + """The `get_parent_repo()` function returns the URL of the parent repository. Returns ------- @@ -113,8 +99,7 @@ def get_parent_repo() -> str: def handle_request_response(response: Response) -> None: - """The function handles the response of a GET request and raises an - exception if the response code is not 200. + """The function handles the response of a GET request and raises an exception if the response code is not 200. Parameters ---------- @@ -124,14 +109,13 @@ def handle_request_response(response: Response) -> None: server, such as the status code, headers, and response body. """ response_code = response.status_code - if response_code != 200: - raise DownloadFailure(f"Unable to downloaded assets. Reason - {response.text}") + if response_code != status_code_200: + msg = f"Unable to downloaded assets. Reason - {response.text}" + raise DownloadError(msg) def slugify(string: str) -> str: - """The `slugify` function converts a string to a slug format by converting - it to lowercase, removing special characters, replacing spaces with dashes, - removing consecutive dashes, and removing leading and trailing dashes. + """The `slugify` function converts a string to a slug format. Parameters ---------- @@ -155,47 +139,36 @@ def slugify(string: str) -> str: modified_string = re.sub(r"-+", "-", modified_string) # Remove leading and trailing dashes - modified_string = modified_string.strip("-") - - return modified_string + return modified_string.strip("-") -def check_java(dry_run: bool) -> None: - """The function `check_java` checks if Java version 17 or higher is - installed and logs an error message if it is not. +def _check_version(output: str) -> None: + """Check version.""" + if "Runtime Environment" not in output: + raise subprocess.CalledProcessError(-1, "java -version") + if "17" not in output and "20" not in output: + raise subprocess.CalledProcessError(-1, "java -version") - Parameters - ---------- - dry_run : bool - The `dry_run` parameter is a boolean flag that determines whether the function should actually - check if Java is installed or just simulate the check. If `dry_run` is `True`, the function will - return without performing the check. If `dry_run` is `False`, the function will execute the + +def check_java() -> None: + """The function `check_java` checks if Java version 17 or higher is installed. Returns ------- - The function `check_java` does not return any value. It has a return type annotation of `None`, - indicating that it does not return anything. + The function `check_java` does not return any value. """ try: - if dry_run: - return - jd = subprocess.check_output( - ["java", "-version"], stderr=subprocess.STDOUT - ).decode("utf-8") + jd = subprocess.check_output(["java", "-version"], stderr=subprocess.STDOUT).decode("utf-8") jd = jd[1:-1] - if "Runtime Environment" not in jd: - raise subprocess.CalledProcessError(-1, "java -version") - if "17" not in jd and "20" not in jd: - raise subprocess.CalledProcessError(-1, "java -version") + _check_version(jd) logger.debug("Cool!! Java is available") except subprocess.CalledProcessError: logger.error("Java>= 17 must be installed") - exit(-1) + sys.exit(-1) def extra_downloads(config: RevancedConfig) -> None: - """The function `extra_downloads` downloads extra files specified in the - `config` object using the `APP.download` method. + """The function `extra_downloads` downloads extra files specified. Parameters ---------- @@ -217,15 +190,11 @@ def extra_downloads(config: RevancedConfig) -> None: file_name=new_file_name, ) except (ValueError, IndexError): - logger.info( - "Unable to download extra file. Provide input in url@name.apk format." - ) + logger.info("Unable to download extra file. Provide input in url@name.apk format.") def apkmirror_status_check(package_name: str) -> Any: - """The `apkmirror_status_check` function checks if an app exists on - APKMirror by making a POST request to the APKMirror API with the package - name as a parameter. + """The `apkmirror_status_check` function checks if an app exists on APKMirror. Parameters ---------- @@ -239,9 +208,10 @@ def apkmirror_status_check(package_name: str) -> Any: """ api_url = f"{apk_mirror_base_url}/wp-json/apkm/v1/app_exists/" body = {"pnames": [package_name]} - response = requests.post(api_url, json=body, headers=request_header) + response = requests.post(api_url, json=body, headers=request_header, timeout=60) return response.json() def contains_any_word(string: str, words: List[str]) -> bool: + """Checks if a string contains any word.""" return any(word in string for word in words)