sjva/docker-py-revanced
1
0
Fork 0
mirror of https://github.com/sotam0316/docker-py-revanced.git synced 2026-04-25 03:48:37 +09:00
Code Issues Packages Projects Releases Wiki Activity

🧹 More Cleanup

Browse Source
This commit is contained in:
Nikhil Badyal
2022-10-06 22:11:08 +05:30
parent 2e41d8ff9f
commit 0d9688470a
7 changed files with 112 additions and 122 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/downloader.py
+26 -67
View File
@@ -1,72 +1,31 @@
import re
import sys
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
from queue import PriorityQueue
from time import perf_counter
from typing import Tuple
import requests
from environs import Env
from loguru import logger
from requests import Session
from selectolax.lexbor import LexborHTMLParser
from tqdm import tqdm
from src.config import RevancedConfig
class Downloader(object):
def __init__(self, env: Env):
def __init__(self, config: RevancedConfig):
self._CHUNK_SIZE = 2**21 * 5
self._QUEUE: PriorityQueue[Tuple[float, str]] = PriorityQueue()
self._QUEUE_LENGTH = 0
self.temp_folder = Path("apks")
self.session = Session()
self.session.headers["User-Agent"] = "anything"
self.build_extended = env.bool("BUILD_EXTENDED", False)
self.apk_mirror = "https://www.apkmirror.com"
self.normal_cli_jar = "revanced-cli.jar"
self.normal_patches_jar = "revanced-patches.jar"
self.normal_integrations_apk = "revanced-integrations.apk"
self.cli_jar = (
f"inotia00-{self.normal_cli_jar}"
if self.build_extended
else self.normal_cli_jar
)
self.patches_jar = (
f"inotia00-{self.normal_patches_jar}"
if self.build_extended
else self.normal_patches_jar
)
self.integrations_apk = (
f"inotia00-{self.normal_integrations_apk}"
if self.build_extended
else self.normal_integrations_apk
)
self.apk_mirror_urls = {
"reddit": f"{self.apk_mirror}/apk/redditinc/reddit/",
"twitter": f"{self.apk_mirror}/apk/twitter-inc/twitter/",
"tiktok": f"{self.apk_mirror}/apk/tiktok-pte-ltd/tik-tok-including-musical-ly/",
"warnwetter": f"{self.apk_mirror}/apk/deutscher-wetterdienst/warnwetter/",
"youtube": f"{self.apk_mirror}/apk/google-inc/youtube/",
"youtube_music": f"{self.apk_mirror}/apk/google-inc/youtube-music/",
}
self.apk_mirror_version_urls = {
"reddit": f"{self.apk_mirror_urls.get('reddit')}reddit",
"twitter": f"{self.apk_mirror_urls.get('twitter')}twitter",
"tiktok": f"{self.apk_mirror_urls.get('tiktok')}tik-tok-including-musical-ly",
"warnwetter": f"{self.apk_mirror_urls.get('warnwetter')}warnwetter",
"youtube": f"{self.apk_mirror_urls.get('youtube')}youtube",
"youtube_music": f"{self.apk_mirror_urls.get('youtube_music')}youtube-music",
}
self.upto_down = ["spotify"]
self.download_revanced()
self.config = config
def _download(self, url: str, file_name: str) -> None:
logger.debug(f"Trying to download {file_name} from {url}")
self._QUEUE_LENGTH += 1
start = perf_counter()
resp = self.session.get(url, stream=True)
resp = self.config.session.get(url, stream=True)
total = int(resp.headers.get("content-length", 0))
bar = tqdm(
desc=file_name,
@@ -76,7 +35,7 @@ class Downloader(object):
unit_divisor=1024,
colour="green",
)
with self.temp_folder.joinpath(file_name).open("wb") as dl_file, bar:
with self.config.temp_folder.joinpath(file_name).open("wb") as dl_file, bar:
for chunk in resp.iter_content(self._CHUNK_SIZE):
size = dl_file.write(chunk)
bar.update(size)
@@ -85,17 +44,17 @@ class Downloader(object):
def extract_download_link(self, page: str, app: str) -> None:
logger.debug(f"Extracting download link from\n{page}")
parser = LexborHTMLParser(self.session.get(page).text)
parser = LexborHTMLParser(self.config.session.get(page).text)
resp = self.session.get(
self.apk_mirror + parser.css_first("a.accent_bg").attributes["href"]
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.apk_mirror + href, f"{app}.apk")
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:
@@ -111,12 +70,12 @@ class Downloader(object):
f"Unable to find any apk on apkmirror_specific_version on {main_page}"
)
sys.exit(-1)
download_url = self.apk_mirror + sub_url
download_url = self.config.apk_mirror + sub_url
return download_url
def __upto_down_downloader(self, app: str) -> str:
page = "https://spotify.en.uptodown.com/android/download"
parser = LexborHTMLParser(self.session.get(page).text)
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()
@@ -127,8 +86,8 @@ class Downloader(object):
def apkmirror_specific_version(self, app: str, version: str) -> str:
logger.debug(f"Trying to download {app},specific version {version}")
version = version.replace(".", "-")
main_page = f"{self.apk_mirror_version_urls.get(app)}-{version}-release/"
parser = LexborHTMLParser(self.session.get(main_page).text)
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")
@@ -136,11 +95,11 @@ class Downloader(object):
def apkmirror_latest_version(self, app: str) -> str:
logger.debug(f"Trying to download {app}'s latest version from apkmirror")
page = self.apk_mirror_urls.get(app)
page = self.config.apk_mirror_urls.get(app)
if not page:
logger.debug("Invalid app")
sys.exit(1)
parser = LexborHTMLParser(self.session.get(page).text)
parser = LexborHTMLParser(self.config.session.get(page).text)
main_page = parser.css_first(".appRowVariantTag>.accent_color").attributes[
"href"
]
@@ -152,8 +111,8 @@ class Downloader(object):
extra_release = main_page.rfind("release") - 1
version: str = main_page[int_version:extra_release]
version = version.replace("-", ".")
main_page = f"{self.apk_mirror}{main_page}"
parser = LexborHTMLParser(self.session.get(main_page).text)
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")
@@ -173,16 +132,16 @@ class Downloader(object):
def download_revanced(self) -> None:
assets = [
["revanced", "revanced-cli", self.normal_cli_jar],
["revanced", "revanced-integrations", self.normal_integrations_apk],
["revanced", "revanced-patches", self.normal_patches_jar],
["revanced", "revanced-cli", self.config.normal_cli_jar],
["revanced", "revanced-integrations", self.config.normal_integrations_apk],
["revanced", "revanced-patches", self.config.normal_patches_jar],
["inotia00", "VancedMicroG", "VancedMicroG.apk"],
]
if self.build_extended:
if self.config.build_extended:
assets += [
["inotia00", "revanced-cli", self.cli_jar],
["inotia00", "revanced-integrations", self.integrations_apk],
["inotia00", "revanced-patches", self.patches_jar],
["inotia00", "revanced-cli", self.config.cli_jar],
["inotia00", "revanced-integrations", self.config.integrations_apk],
["inotia00", "revanced-patches", self.config.patches_jar],
]
with ThreadPoolExecutor(7) as executor:
executor.map(lambda repo: self.repository(*repo), assets)
@@ -198,7 +157,7 @@ class Downloader(object):
return self.apkmirror_latest_version(app)
def download_apk_to_patch(self, version: str, app: str) -> str:
if app in self.upto_down:
if app in self.config.upto_down:
return self.upto_down_downloader(app)
else:
return self.download_from_apkmirror(version, app)