Use Dict for storing patches (#304)

This commit is contained in:
Nikhil Badyal
2023-08-23 23:23:13 +05:30
committed by GitHub
parent ecd80dffda
commit 2ace3aa05a
10 changed files with 85 additions and 73 deletions
+11 -4
View File
@@ -8,8 +8,8 @@ from typing import Dict, List
from loguru import logger
from src.config import RevancedConfig
from src.downloader.download import Downloader
from src.exceptions import PatchingFailed
from src.downloader.sources import apk_sources
from src.exceptions import DownloadFailure, PatchingFailed
from src.utils import slugify
@@ -47,9 +47,11 @@ class APP(object):
self.download_file_name = ""
self.download_dl = config.env.str(f"{app_name}_DL".upper(), "")
self.download_patch_resources(config)
self.download_source = config.env.str(f"{app_name}_DL_SOURCE".upper(), "")
def download_apk_for_patching(self, config: RevancedConfig) -> None:
"""Download apk to be patched."""
from src.downloader.download import Downloader
from src.downloader.factory import DownloaderFactory
if self.download_dl:
@@ -60,11 +62,16 @@ class APP(object):
)
else:
logger.info("Downloading apk to be patched by scrapping")
try:
if not self.download_source:
self.download_source = apk_sources[self.app_name]
except KeyError:
raise DownloadFailure(f"No download source found for {self.app_name}")
downloader = DownloaderFactory.create_downloader(
app=self.app_name, config=config
config=config, apk_source=self.download_source
)
self.download_file_name, self.download_dl = downloader.download(
self.app_version, self.app_name
self.app_version, self
)
def get_output_file_name(self) -> str:
+7 -6
View File
@@ -5,8 +5,9 @@ import requests
from bs4 import BeautifulSoup, Tag
from loguru import logger
from src.app import APP
from src.downloader.download import Downloader
from src.downloader.sources import APK_MIRROR_BASE_URL, apk_sources
from src.downloader.sources import APK_MIRROR_BASE_URL
from src.exceptions import APKMirrorAPKDownloadFailure
from src.utils import bs4_parser, contains_any_word, request_header
@@ -94,7 +95,7 @@ class ApkMirror(Downloader):
return soup.find(class_=search_class)
def specific_version(
self, app: str, version: str, main_page: str = ""
self, app: APP, version: str, main_page: str = ""
) -> Tuple[str, str]:
"""Function to download the specified version of app from apkmirror.
@@ -105,13 +106,13 @@ class ApkMirror(Downloader):
"""
if not main_page:
version = version.replace(".", "-")
apk_main_page = apk_sources[app]
apk_main_page = app.download_source
version_page = apk_main_page + apk_main_page.split("/")[-2]
main_page = f"{version_page}-{version}-release/"
download_page = self.get_download_page(main_page)
return self.extract_download_link(download_page, app)
return self.extract_download_link(download_page, app.app_name)
def latest_version(self, app: str, **kwargs: Any) -> Tuple[str, str]:
def latest_version(self, app: APP, **kwargs: Any) -> Tuple[str, str]:
"""Function to download whatever the latest version of app from
apkmirror.
@@ -119,7 +120,7 @@ class ApkMirror(Downloader):
:return: Version of downloaded apk
"""
app_main_page = apk_sources[app]
app_main_page = app.download_source
versions_div = self._extracted_search_div(
app_main_page, "listWidget p-relative"
)
+5 -5
View File
@@ -1,23 +1,23 @@
"""APK Pure Downloader Class."""
from typing import Any, Tuple
from src.app import APP
from src.downloader.download import Downloader
from src.downloader.sources import apk_sources
from src.patches import Patches
class ApkPure(Downloader):
"""Files downloader."""
def latest_version(self, app: str, **kwargs: Any) -> Tuple[str, str]:
def latest_version(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
"""
package_name = Patches.get_package_name(app)
download_url = apk_sources[app].format(package_name)
file_name = f"{app}.apk"
package_name = Patches.get_package_name(app.app_name)
download_url = app.download_source.format(package_name)
file_name = f"{app.app_name}.apk"
self._download(download_url, file_name)
return file_name, download_url
+5 -5
View File
@@ -4,8 +4,8 @@ from typing import Any, Tuple
import requests
from bs4 import BeautifulSoup
from src.app import APP
from src.downloader.download import Downloader
from src.downloader.sources import apk_sources
from src.exceptions import APKSosAPKDownloadFailure
from src.patches import Patches
from src.utils import bs4_parser, request_header
@@ -31,13 +31,13 @@ class ApkSos(Downloader):
return file_name, possible_link["href"]
raise APKSosAPKDownloadFailure(f"Unable to download {app}", url=page)
def latest_version(self, app: str, **kwargs: Any) -> Tuple[str, str]:
def latest_version(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
"""
package_name = Patches.get_package_name(app)
download_url = apk_sources[app].format(package_name)
return self.extract_download_link(download_url, app)
package_name = Patches.get_package_name(app.app_name)
download_url = app.download_source.format(package_name)
return self.extract_download_link(download_url, app.app_name)
+8 -5
View File
@@ -9,6 +9,7 @@ from typing import Any, Tuple
from loguru import logger
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
@@ -74,7 +75,7 @@ class Downloader(object):
"""Extract download link from web page."""
raise NotImplementedError(implement_method)
def specific_version(self, app: str, version: str) -> Tuple[str, str]:
def specific_version(self, app: APP, version: str) -> Tuple[str, str]:
"""Function to download the specified version of app from apkmirror.
:param app: Name of the application
@@ -83,7 +84,7 @@ class Downloader(object):
"""
raise NotImplementedError(implement_method)
def latest_version(self, app: str, **kwargs: Any) -> Tuple[str, str]:
def latest_version(self, app: APP, **kwargs: Any) -> Tuple[str, str]:
"""Function to download the latest version of app.
:param app: Name of the application
@@ -121,7 +122,7 @@ class Downloader(object):
base_name, _ = os.path.splitext(filename)
return base_name + new_extension
def download(self, version: str, app: str, **kwargs: Any) -> Tuple[str, str]:
def download(self, version: str, app: APP, **kwargs: Any) -> Tuple[str, str]:
"""Public function to download apk to patch.
:param version: version to download
@@ -130,8 +131,10 @@ class Downloader(object):
if self.config.dry_run:
return "", ""
if app in self.config.existing_downloaded_apks:
logger.debug(f"Will not download {app} -v{version} from the internet.")
return app, f"local://{app}"
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)
else:
+10 -10
View File
@@ -7,10 +7,9 @@ from src.downloader.download import Downloader
from src.downloader.github import Github
from src.downloader.sources import (
APK_MIRROR_BASE_URL,
APK_PURE_URL,
APK_SOS_URL,
APK_PURE_BASE_URL,
APKS_SOS_BASE_URL,
GITHUB_BASE_URL,
apk_sources,
)
from src.downloader.uptodown import UptoDown
from src.exceptions import DownloadFailure
@@ -20,22 +19,23 @@ class DownloaderFactory(object):
"""Downloader Factory."""
@staticmethod
def create_downloader(app: str, config: RevancedConfig) -> Downloader:
def create_downloader(config: RevancedConfig, apk_source: str) -> Downloader:
"""Returns appropriate downloader.
Parameters
----------
app : App Name
config : Config
apk_source : Source URL for APK
"""
if apk_sources[app].startswith(GITHUB_BASE_URL):
if apk_source.startswith(GITHUB_BASE_URL):
return Github(config)
if apk_sources[app].startswith(APK_PURE_URL):
if apk_source.startswith(APK_PURE_BASE_URL):
return ApkPure(config)
elif apk_sources[app].startswith(APK_SOS_URL):
elif apk_source.startswith(APKS_SOS_BASE_URL):
return ApkSos(config)
elif apk_sources[app].endswith("en.uptodown.com/android"):
elif apk_source.endswith("en.uptodown.com/android"):
return UptoDown(config)
elif apk_sources[app].startswith(APK_MIRROR_BASE_URL):
elif apk_source.startswith(APK_MIRROR_BASE_URL):
return ApkMirror(config)
raise DownloadFailure(f"No download factory found for {app}")
raise DownloadFailure("No download factory found.")
+8 -7
View File
@@ -6,6 +6,7 @@ from urllib.parse import urlparse
import requests
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
@@ -15,17 +16,17 @@ from src.utils import handle_request_response, update_changelog
class Github(Downloader):
"""Files downloader."""
def latest_version(self, app: str, **kwargs: Dict[str, str]) -> Tuple[str, str]:
def latest_version(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} from github")
if self.config.dry_run or app == "microg":
logger.debug(f"Trying to download {app.app_name} from github")
if self.config.dry_run:
logger.debug(
f"Skipping download of {app}. File already exists or dry running."
f"Skipping download of {app.app_name}. File already exists or dry running."
)
return app, f"local://{app}"
return app.app_name, f"local://{app.app_name}"
owner = str(kwargs["owner"])
repo_name = str(kwargs["name"])
repo_url = f"https://api.github.com/repos/{owner}/{repo_name}/releases/latest"
@@ -42,8 +43,8 @@ class Github(Downloader):
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)
return app, download_url
self._download(download_url, file_name=app.app_name)
return app.app_name, download_url
@staticmethod
def _extract_repo_owner_and_tag(url: str) -> Tuple[str, str, str]:
+4 -2
View File
@@ -1,8 +1,10 @@
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"
APK_PURE_URL = "https://d.apkpure.com/b/APK/{}?version=latest"
APK_SOS_URL = "https://apksos.com/download-app/{}"
APK_PURE_BASE_URL = "https://d.apkpure.com/b/APK"
APK_PURE_URL = APK_PURE_BASE_URL + "/{}?version=latest"
APKS_SOS_BASE_URL = "https://apksos.com/download-app"
APK_SOS_URL = APKS_SOS_BASE_URL + "/{}"
GITHUB_BASE_URL = "https://github.com"
apk_sources = {
"backdrops": f"{APK_MIRROR_BASE_APK_URL}/backdrops/backdrops-wallpapers/",
+8 -8
View File
@@ -5,8 +5,8 @@ import requests
from bs4 import BeautifulSoup
from loguru import logger
from src.app import APP
from src.downloader.download import Downloader
from src.downloader.sources import apk_sources
from src.exceptions import UptoDownAPKDownloadFailure
from src.utils import bs4_parser, request_header
@@ -27,7 +27,7 @@ class UptoDown(Downloader):
self._download(download_url, file_name)
return file_name, download_url
def specific_version(self, app: str, version: str) -> Tuple[str, str]:
def specific_version(self, app: APP, version: str) -> Tuple[str, str]:
"""Function to download the specified version of app from apkmirror.
:param app: Name of the application
@@ -35,7 +35,7 @@ class UptoDown(Downloader):
:return: Version of downloaded apk
"""
logger.debug("downloading specified version of app from uptodown.")
url = f"{apk_sources[app]}/versions"
url = f"{app.download_source}/versions"
html = self.config.session.get(url).text
soup = BeautifulSoup(html, bs4_parser)
versions_list = soup.find("section", {"id": "versions"})
@@ -47,10 +47,10 @@ class UptoDown(Downloader):
break
if download_url is None:
raise UptoDownAPKDownloadFailure(
f"Unable to download {app} from uptodown.", url=url
f"Unable to download {app.app_name} from uptodown.", url=url
)
return self.extract_download_link(download_url, app)
return self.extract_download_link(download_url, app.app_name)
def latest_version(self, app: str, **kwargs: Any) -> Tuple[str, str]:
page = f"{apk_sources[app]}/download"
return self.extract_download_link(page, app)
def latest_version(self, app: APP, **kwargs: Any) -> Tuple[str, str]:
page = f"{app.download_source}/download"
return self.extract_download_link(page, app.app_name)
+19 -21
View File
@@ -14,7 +14,7 @@ from src.exceptions import AppNotFound, PatchesJsonLoadFailed
class Patches(object):
"""Revanced Patches."""
_revanced_app_ids = {
revanced_package_names = {
"com.reddit.frontpage": "reddit",
"com.ss.android.ugc.trill": "tiktok",
"com.twitter.android": "twitter",
@@ -56,9 +56,6 @@ class Patches(object):
"com.mgoogle.android.gms": "microg",
"jp.pxv.android": "pixiv",
}
revanced_app_ids = {
key: (value, "_" + value) for key, value in _revanced_app_ids.items()
}
@staticmethod
def get_package_name(app: str) -> str:
@@ -75,8 +72,8 @@ class Patches(object):
-------
a string, which is the package name corresponding to the given app name.
"""
for package, app_tuple in Patches.revanced_app_ids.items():
if app_tuple[0] == app:
for package, app_name in Patches.revanced_package_names.items():
if app_name == app:
return package
raise AppNotFound(f"App {app} not supported yet.")
@@ -89,7 +86,7 @@ class Patches(object):
-------
a dictionary of supported apps.
"""
return Patches._revanced_app_ids
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
@@ -107,29 +104,30 @@ class Patches(object):
patches = patch_loader.load_patches(
f'{config.temp_folder}/{app.resource["patches_json"]}'
)
for app_name in (self.revanced_app_ids[x][1] for x in self.revanced_app_ids):
setattr(self, app_name, [])
setattr(self, "universal_patch", [])
for patch in patches:
if not patch["compatiblePackages"]:
p = {x: patch[x] for x in ["name", "description"]}
p["app"] = "universal"
p["version"] = "all"
getattr(self, "universal_patch").append(p)
self.patches_dict["universal_patch"].append(p)
for compatible_package, version in [
(x["name"], x["versions"]) for x in patch["compatiblePackages"]
]:
if compatible_package in self.revanced_app_ids:
app_name = self.revanced_app_ids[compatible_package][1]
if compatible_package in self.revanced_package_names.keys():
app_name = self.revanced_package_names[compatible_package]
if not self.patches_dict.get(app_name, None):
self.patches_dict[app_name] = []
p = {x: patch[x] for x in ["name", "description"]}
p["app"] = compatible_package
p["version"] = version[-1] if version else "all"
getattr(self, app_name).append(p)
n_patches = len(getattr(self, f"_{app.app_name}"))
self.patches_dict[app_name].append(p)
n_patches = len(self.patches_dict[app.app_name])
app.no_of_patches = n_patches
def __init__(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]:
@@ -148,12 +146,12 @@ class Patches(object):
patches for the given app. The second element is a string representing the version of the
patches.
"""
app_names = {value[0]: value[1] for value in self.revanced_app_ids.values()}
app_names = self.revanced_package_names
if not (app_name := app_names.get(app)):
if app not in app_names.values():
raise AppNotFound(f"App {app} not supported yet.")
patches = getattr(self, app_name)
patches = self.patches_dict[app]
version = "latest"
with contextlib.suppress(StopIteration):
version = next(i["version"] for i in patches if i["version"] != "all")
@@ -185,9 +183,9 @@ class Patches(object):
normalized_patch
)
for normalized_patch in app.include_request:
parser.include(normalized_patch) if normalized_patch not in getattr(
self, "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