From 711e2d8c3c295445f4c5873f294951c6396dade6 Mon Sep 17 00:00:00 2001 From: Nikhil Badyal Date: Sun, 2 Jul 2023 17:51:26 +0530 Subject: [PATCH] =?UTF-8?q?=E2=99=BB=EF=B8=8F=20Modularized=20downloader?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- main.py | 10 +- src/config.py | 3 +- src/downloader.py | 342 ------------------------------------ src/downloader/__init__.py | 0 src/downloader/apkmirror.py | 107 +++++++++++ src/downloader/apkpure.py | 29 +++ src/downloader/apksos.py | 42 +++++ src/downloader/download.py | 145 +++++++++++++++ src/downloader/factory.py | 33 ++++ src/downloader/uptodown.py | 47 +++++ 10 files changed, 412 insertions(+), 346 deletions(-) delete mode 100644 src/downloader.py create mode 100644 src/downloader/__init__.py create mode 100644 src/downloader/apkmirror.py create mode 100644 src/downloader/apkpure.py create mode 100644 src/downloader/apksos.py create mode 100644 src/downloader/download.py create mode 100644 src/downloader/factory.py create mode 100644 src/downloader/uptodown.py diff --git a/main.py b/main.py index 98a757f..c62e41b 100644 --- a/main.py +++ b/main.py @@ -5,7 +5,8 @@ from environs import Env from loguru import logger from src.config import RevancedConfig -from src.downloader import Downloader +from src.downloader.download import Downloader +from src.downloader.factory import DownloaderFactory from src.parser import Parser from src.patches import Patches from src.utils import AppNotFound @@ -17,15 +18,18 @@ def main() -> None: config = RevancedConfig(env) patcher = Patches(config) - downloader = Downloader(patcher, config) parser = Parser(patcher, config) + Downloader(patcher, config).download_revanced() logger.info(f"Will Patch only {patcher.config.apps}") for app in patcher.config.apps: try: logger.info("Trying to build %s" % app) app_all_patches, version, is_experimental = patcher.get_app_configs(app) - version = downloader.download_apk_to_patch(version, app) + downloader = DownloaderFactory.create_downloader( + app=app, patcher=patcher, config=config + ) + downloader.download(version, app) config.app_versions[app] = version patcher.include_exclude_patch(app, parser, app_all_patches) logger.info(f"Downloaded {app}, version {version}") diff --git a/src/config.py b/src/config.py index 67a0f21..b581b65 100644 --- a/src/config.py +++ b/src/config.py @@ -28,9 +28,10 @@ class RevancedConfig(object): "irplus", "meme-generator-free", "yuka", + "facebook", ] self.apk_pure = ["hex-editor", "androidtwelvewidgets"] - self.apk_sos = ["expensemanager"] + self.apk_sos = ["expensemanager", "candyvpn"] self.keystore_name = env.str("KEYSTORE_FILE_NAME", "revanced.keystore") self.ci_test = env.bool("CI_TEST", False) self.apps = env.list("PATCH_APPS", default_build) diff --git a/src/downloader.py b/src/downloader.py deleted file mode 100644 index e211955..0000000 --- a/src/downloader.py +++ /dev/null @@ -1,342 +0,0 @@ -"""Downloader Class.""" -import os -import re -import sys -from concurrent.futures import ThreadPoolExecutor -from queue import PriorityQueue -from time import perf_counter -from typing import Tuple - -import requests -from bs4 import BeautifulSoup -from loguru import logger -from selectolax.lexbor import LexborHTMLParser -from tqdm import tqdm - -from src.config import RevancedConfig -from src.patches import Patches -from src.utils import AppNotFound, 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.download_revanced() - 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.debug(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: - """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: - """ - 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}" - ) - sys.exit(-1) - download_url = self.config.apk_mirror + sub_url - return download_url - - def __upto_down_specific_version(self, app: str, version: str) -> str: - """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.__upto_down_downloader(download_url, app) - logger.debug(f"Downloaded {app} apk from upto_down_downloader in rt") - return version - - def __upto_down_latest_downloader(self, app: str) -> str: - page = f"https://{app}.en.uptodown.com/android/download" - return self.__upto_down_downloader(page, app) - - def __upto_down_downloader(self, page: str, app: str) -> str: - parser = LexborHTMLParser(self.config.session.get(page).text) - main_page = parser.css_first("#detail-download-button") - download_url = main_page.attributes["data-url"] - app_version: str = parser.css_first(".version").text() - self._download(download_url, f"{app}.apk") - logger.debug(f"Downloaded {app} apk from upto_down_downloader in rt") - return app_version - - def __apk_pure_downloader(self, app: str) -> str: - 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") - return "latest" - - def __apk_sos_downloader(self, app: str) -> str: - 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}" - parser = LexborHTMLParser(self.config.session.get(download_url).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") - return "latest" - - def apkmirror_specific_version(self, app: str, version: str) -> str: - """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).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") - return version - - def apkmirror_latest_version(self, app: str) -> str: - """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") - sys.exit(1) - 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() - int_version = match.start() - extra_release = main_page.rfind("release") - 1 - version: str = main_page[int_version:extra_release] - version = version.replace("-", ".") - 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") - return version - - 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.") - - def upto_down_downloader(self, version: str, app: str) -> str: - """Function to download from UptoDown. - - :param version: version to download - :param app: Name of the application - :return: Version of downloaded APK - """ - if version and version != "latest": - return self.__upto_down_specific_version(app, version) - else: - return self.__upto_down_latest_downloader(app) - - def apk_pure_downloader(self, app: str) -> str: - """Function to download from Apk Pure. - - :param app: Name of the application - :return: Version of downloaded APK - """ - return self.__apk_pure_downloader(app) - - def download_from_apkmirror(self, version: str, app: str) -> str: - """Function to download from apkmirror. - - :param version: version to download - :param app: App to download - :return: Version of downloaded APK - """ - if version and version != "latest": - return self.apkmirror_specific_version(app, version) - else: - return self.apkmirror_latest_version(app) - - def apk_sos_downloader(self, app: str) -> str: - """Function to download from Apk Pure. - - :param app: Name of the application - :return: Version of downloaded APK - """ - return self.__apk_sos_downloader(app) - - def download_apk_to_patch(self, version: str, app: str) -> str: - """Public function to download apk to patch. - - :param version: version to download - :param app: App to download - :return: Version of apk. - """ - if app in self.config.existing_downloaded_apks: - logger.debug("Will not download apk from the internet as it already exist.") - # Returning Latest as I don't know, which version user provided. - return "latest" - if app in self.config.upto_down: - return self.upto_down_downloader(version, app) - elif app in self.config.apk_pure: - return self.apk_pure_downloader(app) - elif app in self.config.apk_sos: - return self.apk_sos_downloader(app) - else: - return self.download_from_apkmirror(version, app) diff --git a/src/downloader/__init__.py b/src/downloader/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/downloader/apkmirror.py b/src/downloader/apkmirror.py new file mode 100644 index 0000000..45378c8 --- /dev/null +++ b/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") diff --git a/src/downloader/apkpure.py b/src/downloader/apkpure.py new file mode 100644 index 0000000..25e4f2c --- /dev/null +++ b/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") diff --git a/src/downloader/apksos.py b/src/downloader/apksos.py new file mode 100644 index 0000000..3ce14c6 --- /dev/null +++ b/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) diff --git a/src/downloader/download.py b/src/downloader/download.py new file mode 100644 index 0000000..aa4dfcc --- /dev/null +++ b/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.") diff --git a/src/downloader/factory.py b/src/downloader/factory.py new file mode 100644 index 0000000..1b312ef --- /dev/null +++ b/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) diff --git a/src/downloader/uptodown.py b/src/downloader/uptodown.py new file mode 100644 index 0000000..b672146 --- /dev/null +++ b/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)