♻️ Modularized downloader

src/downloader/apkmirror.py

@@ -0,0 +1,107 @@
+"""Downloader Class."""
+import re
+
+from loguru import logger
+from selectolax.lexbor import LexborHTMLParser
+
+from src.downloader.download import Downloader
+from src.utils import AppNotFound
+
+
+class ApkMirror(Downloader):
+    """Files downloader."""
+
+    def extract_download_link(self, page: str, app: str) -> None:
+        """Function to extract the download link from apkmirror html page.
+
+        :param page: Url of the page
+        :param app: Name of the app
+        """
+        logger.debug(f"Extracting download link from\n{page}")
+        parser = LexborHTMLParser(self.config.session.get(page).text)
+
+        resp = self.config.session.get(
+            self.config.apk_mirror + parser.css_first("a.accent_bg").attributes["href"]
+        )
+        parser = LexborHTMLParser(resp.text)
+
+        href = parser.css_first(
+            "p.notes:nth-child(3) > span:nth-child(1) > a:nth-child(1)"
+        ).attributes["href"]
+        self._download(self.config.apk_mirror + href, f"{app}.apk")
+        logger.debug("Finished Extracting link and downloading")
+
+    def get_download_page(self, parser: LexborHTMLParser, main_page: str) -> str:
+        """Function to get the download page in apk_mirror.
+
+        :param parser: Parser
+        :param main_page: Main Download Page in APK mirror(Index)
+        :return:
+        """
+        logger.debug(f"Getting download page from {main_page}")
+        apm = parser.css(".apkm-badge")
+        sub_url = ""
+        for is_apm in apm:
+            parent_text = is_apm.parent.parent.text()
+            if "APK" in is_apm.text() and (
+                "arm64-v8a" in parent_text
+                or "universal" in parent_text
+                or "noarch" in parent_text
+            ):
+                parser = is_apm.parent
+                sub_url = parser.css_first(".accent_color").attributes["href"]
+                break
+        if sub_url == "":
+            logger.exception(
+                f"Unable to find any apk on apkmirror_specific_version on {main_page}"
+            )
+            raise AppNotFound("Unable to find apk on apkmirror site.")
+        download_url = self.config.apk_mirror + sub_url
+        return download_url
+
+    def specific_version(self, app: str, version: str) -> None:
+        """Function to download the specified version of app from  apkmirror.
+
+        :param app: Name of the application
+        :param version: Version of the application to download
+        :return: Version of downloaded apk
+        """
+        logger.debug(f"Trying to download {app},specific version {version}")
+        version = version.replace(".", "-")
+        main_page = f"{self.config.apk_mirror_version_urls.get(app)}-{version}-release/"
+        parser = LexborHTMLParser(
+            self.config.session.get(main_page, allow_redirects=True).text
+        )
+        download_page = self.get_download_page(parser, main_page)
+        self.extract_download_link(download_page, app)
+        logger.debug(f"Downloaded {app} apk from apkmirror_specific_version")
+
+    def latest_version(self, app: str) -> None:
+        """Function to download whatever the latest version of app from
+        apkmirror.
+
+        :param app: Name of the application
+        :return: Version of downloaded apk
+        """
+        logger.debug(f"Trying to download {app}'s latest version from apkmirror")
+        page = self.config.apk_mirror_urls.get(app)
+        if not page:
+            logger.debug("Invalid app")
+            raise AppNotFound("Invalid app")
+        parser = LexborHTMLParser(self.config.session.get(page).text)
+        try:
+            main_page = parser.css_first(".appRowVariantTag>.accent_color").attributes[
+                "href"
+            ]
+        except AttributeError:
+            # Handles a case when variants are not available
+            main_page = parser.css_first(".downloadLink").attributes["href"]
+        match = re.search(r"\d", main_page)
+        if not match:
+            logger.error("Cannot find app main page")
+            raise AppNotFound()
+        main_page = f"{self.config.apk_mirror}{main_page}"
+        parser = LexborHTMLParser(self.config.session.get(main_page).text)
+        download_page = self.get_download_page(parser, main_page)
+        self.extract_download_link(download_page, app)
+        logger.debug(f"Downloaded {app} apk from apkmirror_specific_version in rt")

src/downloader/apkpure.py

@@ -0,0 +1,29 @@
+"""APK Pure Downloader Class."""
+
+
+from loguru import logger
+
+from src.downloader.download import Downloader
+from src.utils import AppNotFound
+
+
+class ApkPure(Downloader):
+    """Files downloader."""
+
+    def latest_version(self, app: str) -> None:
+        """Function to download whatever the latest version of app from
+        apkmirror.
+
+        :param app: Name of the application
+        :return: Version of downloaded apk
+        """
+        package_name = None
+        for package, app_tuple in self.patcher.revanced_app_ids.items():
+            if app_tuple[0] == app:
+                package_name = package
+        if not package_name:
+            logger.info("Unable to download from apkpure")
+            raise AppNotFound()
+        download_url = f"https://d.apkpure.com/b/APK/{package_name}?version=latest"
+        self._download(download_url, f"{app}.apk")
+        logger.debug(f"Downloaded {app} apk from apk_pure_downloader in rt")

src/downloader/apksos.py

@@ -0,0 +1,42 @@
+"""APK SOS Downloader Class."""
+
+from loguru import logger
+from selectolax.lexbor import LexborHTMLParser
+
+from src.downloader.download import Downloader
+from src.utils import AppNotFound
+
+
+class ApkSos(Downloader):
+    """Files downloader."""
+
+    def extract_download_link(self, page: str, app: str) -> None:
+        """Function to extract the download link from apkmirror html page.
+
+        :param page: Url of the page
+        :param app: Name of the app
+        """
+        parser = LexborHTMLParser(self.config.session.get(page).text)
+        download_url = parser.css_first(
+            r"body > div > div > div > div > div.col-sm-12.col-md-8 > div.card.fluid.\.idma > "
+            "div.section.row > div.col-sm-12.col-md-8.text-center > p > a"
+        ).attributes["href"]
+        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:
+        """Function to download whatever the latest version of app from
+        apkmirror.
+
+        :param app: Name of the application
+        :return: Version of downloaded apk
+        """
+        package_name = None
+        for package, app_tuple in self.patcher.revanced_app_ids.items():
+            if app_tuple[0] == app:
+                package_name = package
+        if not package_name:
+            logger.info("Unable to download from apkcombo")
+            raise AppNotFound()
+        download_url = f"https://apksos.com/download-app/{package_name}"
+        self.extract_download_link(download_url, app)

src/downloader/download.py

@@ -0,0 +1,145 @@
+"""Downloader Class."""
+import os
+from concurrent.futures import ThreadPoolExecutor
+from queue import PriorityQueue
+from time import perf_counter
+from typing import Tuple
+
+import requests
+from loguru import logger
+from tqdm import tqdm
+
+from src.config import RevancedConfig
+from src.patches import Patches
+from src.utils import handle_response, update_changelog
+
+
+class Downloader(object):
+    """Files downloader."""
+
+    def __init__(self, patcher: Patches, config: RevancedConfig):
+        self._CHUNK_SIZE = 10485760
+        self._QUEUE: PriorityQueue[Tuple[float, str]] = PriorityQueue()
+        self._QUEUE_LENGTH = 0
+        self.config = config
+        self.patcher = patcher
+
+    def _download(self, url: str, file_name: str) -> None:
+        if os.path.exists(self.config.temp_folder.joinpath(file_name)):
+            logger.debug(f"Skipping download of {file_name}. File already exists.")
+            return
+        logger.info(f"Trying to download {file_name} from {url}")
+        self._QUEUE_LENGTH += 1
+        start = perf_counter()
+        headers = {}
+        if self.config.personal_access_token and "github" in url:
+            logger.debug("Using personal access token")
+            headers.update(
+                {"Authorization": "token " + self.config.personal_access_token}
+            )
+        response = self.config.session.get(
+            url,
+            stream=True,
+            headers=headers,
+        )
+        handle_response(response)
+        total = int(response.headers.get("content-length", 0))
+        bar = tqdm(
+            desc=file_name,
+            total=total,
+            unit="iB",
+            unit_scale=True,
+            unit_divisor=1024,
+            colour="green",
+        )
+        with self.config.temp_folder.joinpath(file_name).open("wb") as dl_file, bar:
+            for chunk in response.iter_content(self._CHUNK_SIZE):
+                size = dl_file.write(chunk)
+                bar.update(size)
+        self._QUEUE.put((perf_counter() - start, file_name))
+        logger.debug(f"Downloaded {file_name}")
+
+    def extract_download_link(self, page: str, app: str) -> None:
+        """Extract download link from web page."""
+        raise NotImplementedError("Please implement the method")
+
+    def specific_version(self, app: str, version: str) -> None:
+        """Function to download the specified version of app from  apkmirror.
+
+        :param app: Name of the application
+        :param version: Version of the application to download
+        :return: Version of downloaded apk
+        """
+        raise NotImplementedError("Please implement the method")
+
+    def latest_version(self, app: str) -> 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")
+
+    def download(self, version: str, app: str) -> None:
+        """Public function to download apk to patch.
+
+        :param version: version to download
+        :param app: App to download
+        """
+        if app in self.config.existing_downloaded_apks:
+            logger.debug(f"Will not download {app} -v{version} from the internet.")
+            return
+        if version and version != "latest":
+            self.specific_version(app, version)
+        else:
+            self.latest_version(app)
+
+    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")
+        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"):
+            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

@@ -0,0 +1,33 @@
+"""Downloader Factory."""
+from src.config import RevancedConfig
+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.uptodown import UptoDown
+from src.patches import Patches
+
+
+class DownloaderFactory(object):
+    """Downloader Factory."""
+
+    @staticmethod
+    def create_downloader(
+        app: str, patcher: Patches, config: RevancedConfig
+    ) -> Downloader:
+        """Returns appropriate downloader.
+
+        Parameters
+        ----------
+        app : App Name
+        patcher : Patcher
+        config : Config
+        """
+        if app in config.apk_pure:
+            return ApkPure(patcher, config)
+        elif app in config.apk_sos:
+            return ApkSos(patcher, config)
+        elif app in config.upto_down:
+            return UptoDown(patcher, config)
+        else:
+            return ApkMirror(patcher, config)

src/downloader/uptodown.py

@@ -0,0 +1,47 @@
+"""Upto Down Downloader."""
+
+from bs4 import BeautifulSoup
+from loguru import logger
+from selectolax.lexbor import LexborHTMLParser
+
+from src.downloader.download import Downloader
+from src.utils import AppNotFound
+
+
+class UptoDown(Downloader):
+    """Files downloader."""
+
+    def extract_download_link(self, page: str, app: str) -> None:
+        parser = LexborHTMLParser(self.config.session.get(page).text)
+        main_page = parser.css_first("#detail-download-button")
+        download_url = main_page.attributes["data-url"]
+        self._download(download_url, f"{app}.apk")
+        logger.debug(f"Downloaded {app} apk from upto_down_downloader in rt")
+
+    def specific_version(self, app: str, version: str) -> None:
+        """Function to download the specified version of app from  apkmirror.
+
+        :param app: Name of the application
+        :param version: Version of the application to download
+        :return: Version of downloaded apk
+        """
+        logger.debug("downloading specified version of app from uptodown.")
+        url = f"https://{app}.en.uptodown.com/android/versions"
+        html = self.config.session.get(url).text
+        soup = BeautifulSoup(html, "html.parser")
+        versions_list = soup.find("section", {"id": "versions"})
+        download_url = None
+        for version_item in versions_list.find_all("div", {"data-url": True}):
+            extracted_version = version_item.find("span", {"class": "version"}).text
+            if extracted_version == version:
+                download_url = version_item["data-url"]
+                print(f"data-url for version {version}: {download_url}")
+                break
+        if download_url is None:
+            raise AppNotFound(f"Unable to get download url for {app}")
+        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:
+        page = f"https://{app}.en.uptodown.com/android/download"
+        self.extract_download_link(page, app)