✨ Github Downloader

main.py

@@ -5,11 +5,11 @@ from environs import Env
 from loguru import logger
 
 from src.config import RevancedConfig
-from src.downloader.download import Downloader
 from src.downloader.factory import DownloaderFactory
+from src.downloader.utils import download_revanced
 from src.parser import Parser
 from src.patches import Patches
-from src.utils import AppNotFound
+from src.utils import AppNotFound, PatcherDownloadFailed
 
 
 def main() -> None:
@@ -18,7 +18,11 @@ def main() -> None:
     config = RevancedConfig(env)
 
     patcher = Patches(config)
-    Downloader(patcher, config).download_revanced()
+    try:
+        download_revanced(config, patcher)
+    except PatcherDownloadFailed as e:
+        logger.error(f"Failed to download {e}")
+        sys.exit(1)
 
     logger.info(f"Will Patch only {patcher.config.apps}")
     for app in patcher.config.apps:

src/downloader/apkmirror.py

@@ -1,5 +1,6 @@
 """Downloader Class."""
 import re
+from typing import Any
 
 from loguru import logger
 from selectolax.lexbor import LexborHTMLParser
@@ -76,7 +77,7 @@ class ApkMirror(Downloader):
         self.extract_download_link(download_page, app)
         logger.debug(f"Downloaded {app} apk from apkmirror_specific_version")
 
-    def latest_version(self, app: str) -> None:
+    def latest_version(self, app: str, **kwargs: Any) -> None:
         """Function to download whatever the latest version of app from
         apkmirror.
 

src/downloader/apkpure.py

@@ -1,5 +1,5 @@
 """APK Pure Downloader Class."""
-
+from typing import Any
 
 from loguru import logger
 
@@ -10,7 +10,7 @@ from src.utils import AppNotFound
 class ApkPure(Downloader):
     """Files downloader."""
 
-    def latest_version(self, app: str) -> None:
+    def latest_version(self, app: str, **kwargs: Any) -> None:
         """Function to download whatever the latest version of app from
         apkmirror.
 

src/downloader/apksos.py

@@ -1,4 +1,5 @@
 """APK SOS Downloader Class."""
+from typing import Any
 
 from loguru import logger
 from selectolax.lexbor import LexborHTMLParser
@@ -24,7 +25,7 @@ class ApkSos(Downloader):
         self._download(download_url, f"{app}.apk")
         logger.debug(f"Downloaded {app} apk from apk_combo_downloader in rt")
 
-    def latest_version(self, app: str) -> None:
+    def latest_version(self, app: str, **kwargs: Any) -> None:
         """Function to download whatever the latest version of app from
         apkmirror.
 

src/downloader/download.py

@@ -1,17 +1,16 @@
 """Downloader Class."""
 import os
-from concurrent.futures import ThreadPoolExecutor
 from queue import PriorityQueue
 from time import perf_counter
-from typing import Tuple
+from typing import Any, Tuple
 
-import requests
 from loguru import logger
 from tqdm import tqdm
 
 from src.config import RevancedConfig
+from src.downloader.utils import implement_method
 from src.patches import Patches
-from src.utils import handle_response, update_changelog
+from src.utils import handle_response
 
 
 class Downloader(object):
@@ -66,7 +65,7 @@ class Downloader(object):
 
     def extract_download_link(self, page: str, app: str) -> None:
         """Extract download link from web page."""
-        raise NotImplementedError("Please implement the method")
+        raise NotImplementedError(implement_method)
 
     def specific_version(self, app: str, version: str) -> None:
         """Function to download the specified version of app from  apkmirror.
@@ -75,17 +74,17 @@ class Downloader(object):
         :param version: Version of the application to download
         :return: Version of downloaded apk
         """
-        raise NotImplementedError("Please implement the method")
+        raise NotImplementedError(implement_method)
 
-    def latest_version(self, app: str) -> None:
+    def latest_version(self, app: str, **kwargs: Any) -> None:
         """Function to download the latest version of app.
 
         :param app: Name of the application
         :return: Version of downloaded apk
         """
-        raise NotImplementedError("Please implement the method")
+        raise NotImplementedError(implement_method)
 
-    def download(self, version: str, app: str) -> None:
+    def download(self, version: str, app: str, **kwargs: Any) -> None:
         """Public function to download apk to patch.
 
         :param version: version to download
@@ -99,59 +98,4 @@ class Downloader(object):
         if version and version != "latest":
             self.specific_version(app, version)
         else:
-            self.latest_version(app)
+            self.latest_version(app, **kwargs)
-
-    def repository(self, owner: str, name: str, file_name: str) -> None:
-        """Function to download files from GitHub repositories.
-
-        :param owner: github user/organization
-        :param name: name of the repository
-        :param file_name: name of the file after downloading
-        """
-        logger.debug(f"Trying to download {name} from github")
-        if self.config.dry_run:
-            logger.debug(
-                f"Skipping download of {file_name}. File already exists or dry running."
-            )
-            return
-        repo_url = f"https://api.github.com/repos/{owner}/{name}/releases/latest"
-        headers = {
-            "Content-Type": "application/vnd.github.v3+json",
-        }
-        if self.config.personal_access_token:
-            logger.debug("Using personal access token")
-            headers.update(
-                {"Authorization": "token " + self.config.personal_access_token}
-            )
-        response = requests.get(repo_url, headers=headers)
-        handle_response(response)
-        if name == "revanced-patches":
-            download_url = response.json()["assets"][1]["browser_download_url"]
-        else:
-            download_url = response.json()["assets"][0]["browser_download_url"]
-        update_changelog(f"{owner}/{name}", response.json())
-        self._download(download_url, file_name=file_name)
-
-    def download_revanced(self) -> None:
-        """Download Revanced and Extended Patches, Integration and CLI."""
-        if os.path.exists("changelog.md") and not self.config.dry_run:
-            logger.debug("Deleting old changelog.md")
-            os.remove("changelog.md")
-        assets = [
-            ["revanced", "revanced-cli", self.config.normal_cli_jar],
-            ["revanced", "revanced-integrations", self.config.normal_integrations_apk],
-            ["revanced", "revanced-patches", self.config.normal_patches_jar],
-        ]
-        if self.config.build_extended:
-            assets += [
-                ["inotia00", "revanced-cli", self.config.cli_jar],
-                ["inotia00", "revanced-integrations", self.config.integrations_apk],
-                ["inotia00", "revanced-patches", self.config.patches_jar],
-            ]
-        if "youtube" in self.config.apps or "youtube_music" in self.config.apps:
-            assets += [
-                ["inotia00", "mMicroG", "mMicroG-output.apk"],
-            ]
-        with ThreadPoolExecutor(7) as executor:
-            executor.map(lambda repo: self.repository(*repo), assets)
-        logger.info("Downloaded revanced microG ,cli, integrations and patches.")

src/downloader/factory.py

@@ -4,6 +4,7 @@ from src.downloader.apkmirror import ApkMirror
 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.uptodown import UptoDown
 from src.patches import Patches
 
@@ -23,6 +24,8 @@ class DownloaderFactory(object):
         patcher : Patcher
         config : Config
         """
+        if app == "patches":
+            return Github(patcher, config)
         if app in config.apk_pure:
             return ApkPure(patcher, config)
         elif app in config.apk_sos:

src/downloader/github.py

@@ -0,0 +1,43 @@
+"""Github Downloader."""
+from typing import Dict
+
+import requests
+from loguru import logger
+
+from src.downloader.download import Downloader
+from src.utils import handle_response, update_changelog
+
+
+class Github(Downloader):
+    """Files downloader."""
+
+    def latest_version(self, app: str, **kwargs: Dict[str, str]) -> None:
+        """Function to download files from GitHub repositories.
+
+        :param app: App to download
+        """
+        owner = str(kwargs["owner"])
+        repo_name = str(kwargs["name"])
+        logger.debug(f"Trying to download {app} from github")
+        if self.config.dry_run:
+            logger.debug(
+                f"Skipping download of {app}. File already exists or dry running."
+            )
+            return
+        repo_url = f"https://api.github.com/repos/{owner}/{repo_name}/releases/latest"
+        headers = {
+            "Content-Type": "application/vnd.github.v3+json",
+        }
+        if self.config.personal_access_token:
+            logger.debug("Using personal access token")
+            headers.update(
+                {"Authorization": "token " + self.config.personal_access_token}
+            )
+        response = requests.get(repo_url, headers=headers)
+        handle_response(response)
+        if repo_name == "revanced-patches":
+            download_url = response.json()["assets"][1]["browser_download_url"]
+        else:
+            download_url = response.json()["assets"][0]["browser_download_url"]
+        update_changelog(f"{owner}/{repo_name}", response.json())
+        self._download(download_url, file_name=app)

src/downloader/uptodown.py

@@ -1,4 +1,5 @@
 """Upto Down Downloader."""
+from typing import Any
 
 from bs4 import BeautifulSoup
 from loguru import logger
@@ -41,6 +42,6 @@ class UptoDown(Downloader):
         self.extract_download_link(download_url, app)
         logger.debug(f"Downloaded {app} apk from upto_down_downloader in rt")
 
-    def latest_version(self, app: str) -> None:
+    def latest_version(self, app: str, **kwargs: Any) -> None:
         page = f"https://{app}.en.uptodown.com/android/download"
         self.extract_download_link(page, app)

src/downloader/utils.py

@@ -0,0 +1,54 @@
+"""Utility class."""
+import os
+from concurrent.futures import ThreadPoolExecutor, as_completed
+
+from loguru import logger
+
+from src.config import RevancedConfig
+from src.patches import Patches
+from src.utils import PatcherDownloadFailed
+
+implement_method = "Please implement the method"
+
+
+def download_revanced(config: RevancedConfig, patcher: Patches) -> None:
+    """Download Revanced and Extended Patches, Integration and CLI."""
+    from src.downloader.factory import DownloaderFactory
+
+    if os.path.exists("changelog.md") and not config.dry_run:
+        logger.debug("Deleting old changelog.md")
+        os.remove("changelog.md")
+    assets = [
+        ["revanced", "revanced-cli", config.normal_cli_jar],
+        ["revanced", "revanced-integrations", config.normal_integrations_apk],
+        ["revanced", "revanced-patches", config.normal_patches_jar],
+    ]
+    if config.build_extended:
+        assets += [
+            ["inotia00", "revanced-cli", config.cli_jar],
+            ["inotia00", "revanced-integrations", config.integrations_apk],
+            ["inotia00", "revanced-patches", config.patches_jar],
+        ]
+    if "youtube" in config.apps or "youtube_music" in config.apps:
+        assets += [
+            ["inotia00", "mMicroG", "mMicroG-output.apk"],
+        ]
+    downloader = DownloaderFactory.create_downloader(
+        app="patches", patcher=patcher, config=config
+    )
+    with ThreadPoolExecutor(7) as executor:
+        futures = [
+            executor.submit(
+                downloader.download,
+                version="latest",
+                app=repo[2],
+                **{"owner": repo[0], "name": repo[1]},
+            )
+            for repo in assets
+        ]
+        for future in as_completed(futures):
+            try:
+                future.result()
+            except Exception as e:
+                raise PatcherDownloadFailed(f"An exception occurred: {e}")
+    logger.info("Downloaded revanced microG ,cli, integrations and patches.")

src/utils.py

@@ -43,6 +43,12 @@ class AppNotFound(ValueError):
     pass
 
 
+class PatcherDownloadFailed(Exception):
+    """Not a valid Revanced App."""
+
+    pass
+
+
 def handle_response(response: Response) -> None:
     """Handle Get Request Response."""
     response_code = response.status_code