mirror of
https://github.com/sotam0316/docker-py-revanced.git
synced 2026-04-25 03:48:37 +09:00
🎨 Handle split apks
This commit is contained in:
@@ -1,5 +1,6 @@
|
|||||||
"""Apkeep Downloader Class."""
|
"""Apkeep Downloader Class."""
|
||||||
|
|
||||||
|
import zipfile
|
||||||
from subprocess import PIPE, Popen
|
from subprocess import PIPE, Popen
|
||||||
from time import perf_counter
|
from time import perf_counter
|
||||||
from typing import Any, Self
|
from typing import Any, Self
|
||||||
@@ -25,11 +26,18 @@ class Apkeep(Downloader):
|
|||||||
|
|
||||||
file_name = f"{package_name}.apk"
|
file_name = f"{package_name}.apk"
|
||||||
file_path = self.config.temp_folder / file_name
|
file_path = self.config.temp_folder / file_name
|
||||||
|
folder_path = self.config.temp_folder / package_name
|
||||||
|
zip_path = self.config.temp_folder / f"{package_name}.zip"
|
||||||
|
|
||||||
|
# If already downloaded, return it
|
||||||
if file_path.exists():
|
if file_path.exists():
|
||||||
logger.debug(f"{file_name} already downloaded.")
|
logger.debug(f"{file_name} already downloaded.")
|
||||||
return file_name
|
return file_name
|
||||||
|
if zip_path.exists():
|
||||||
|
logger.debug(f"{zip_path.name} already zipped and exists.")
|
||||||
|
return zip_path.name
|
||||||
|
|
||||||
|
# Build apkeep command
|
||||||
cmd = [
|
cmd = [
|
||||||
"apkeep",
|
"apkeep",
|
||||||
"-a",
|
"-a",
|
||||||
@@ -40,6 +48,8 @@ class Apkeep(Downloader):
|
|||||||
email,
|
email,
|
||||||
"-t",
|
"-t",
|
||||||
token,
|
token,
|
||||||
|
"-o",
|
||||||
|
"split_apk=true",
|
||||||
self.config.temp_folder_name,
|
self.config.temp_folder_name,
|
||||||
]
|
]
|
||||||
logger.debug(f"Running command: {cmd}")
|
logger.debug(f"Running command: {cmd}")
|
||||||
@@ -57,7 +67,19 @@ class Apkeep(Downloader):
|
|||||||
msg = f"Command failed with exit code {process.returncode} for app {package_name}"
|
msg = f"Command failed with exit code {process.returncode} for app {package_name}"
|
||||||
raise DownloadError(msg)
|
raise DownloadError(msg)
|
||||||
logger.info(f"Downloading completed for app {package_name} in {perf_counter() - start:.2f} seconds.")
|
logger.info(f"Downloading completed for app {package_name} in {perf_counter() - start:.2f} seconds.")
|
||||||
|
|
||||||
|
if file_path.exists():
|
||||||
return file_name
|
return file_name
|
||||||
|
if folder_path.exists() and folder_path.is_dir():
|
||||||
|
# Zip the folder
|
||||||
|
with zipfile.ZipFile(zip_path, "w", zipfile.ZIP_DEFLATED) as zipf:
|
||||||
|
for file in folder_path.rglob("*"):
|
||||||
|
arcname = file.relative_to(self.config.temp_folder)
|
||||||
|
zipf.write(file, arcname)
|
||||||
|
logger.debug(f"Zipped {folder_path} to {zip_path}")
|
||||||
|
return zip_path.name
|
||||||
|
msg = "APK file or folder not found after apkeep execution."
|
||||||
|
raise DownloadError(msg)
|
||||||
|
|
||||||
def latest_version(self: Self, app: APP, **kwargs: Any) -> tuple[str, str]:
|
def latest_version(self: Self, app: APP, **kwargs: Any) -> tuple[str, str]:
|
||||||
"""Download latest version from Google Play via Apkeep."""
|
"""Download latest version from Google Play via Apkeep."""
|
||||||
|
|||||||
Reference in New Issue
Block a user