AntoineVe/adb-sideload-free-apps
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add APK download progress bar

Browse Source
This commit is contained in:
Antoine Van Elstraete
2026-08-08 17:08:15 +02:00
parent 8d7d06bdbf
commit 968da658aa
2 changed files with 148 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

110
src/sideload.py
View File

@@ -12,6 +12,7 @@ import sys
import urllib.error
import urllib.parse
import urllib.request
from collections.abc import Callable
from dataclasses import dataclass
from html.parser import HTMLParser
from pathlib import Path
@@ -140,6 +141,8 @@ def _atomic_download(
conditional_headers: dict[str, str] | None = None,
same_origin: bool = False,
return_metadata: bool = False,
expected_size: int | None = None,
progress: Callable[[int, int | None], None] | None = None,
) -> Path | tuple[Path | None, str, dict[str, str], int]:
parsed = urllib.parse.urlparse(url)
if parsed.scheme != "https":
@@ -165,8 +168,25 @@ def _atomic_download(
if return_metadata:
return None, final_url, response_headers, status
raise ValueError("received HTTP 304 without a reusable cache entry")
content_length = next(
(value for key, value in response_headers.items() if key.lower() == "content-length"),
None,
)
try:
total_size = int(content_length) if content_length is not None else expected_size
except (TypeError, ValueError):
total_size = expected_size
if total_size is not None and total_size <= 0:
total_size = None
downloaded_size = 0
if progress:
progress(downloaded_size, total_size)
with tmp.open("wb") as fh:
shutil.copyfileobj(resp, fh)
while chunk := resp.read(65536):
fh.write(chunk)
downloaded_size += len(chunk)
if progress:
progress(downloaded_size, total_size)
tmp.rename(dest)
if return_metadata:
return dest, final_url, response_headers, status
@@ -448,7 +468,13 @@ class GenericSource:
expected_sha256=self.app_entry.config.get("sha256"),
)
def download_apk(self, release: ReleaseInfo, cache: DownloadCache) -> Path:
def download_apk(
self,
release: ReleaseInfo,
cache: DownloadCache,
*,
progress: Callable[[int, int | None], None] | None = None,
) -> Path:
if release.expected_sha256:
cached = cache.find_cached(release.app_entry.app_id, release.expected_sha256)
if cached:
@@ -481,6 +507,8 @@ class GenericSource:
conditional_headers=conditional,
same_origin=True,
return_metadata=True,
expected_size=release.file_size,
progress=progress,
)
if status == 304 and cached:
return cached
@@ -598,7 +626,13 @@ class FdroidSource:
file_size=file_info.get("size"),
)
def download_apk(self, release: ReleaseInfo, cache: DownloadCache) -> Path:
def download_apk(
self,
release: ReleaseInfo,
cache: DownloadCache,
*,
progress: Callable[[int, int | None], None] | None = None,
) -> Path:
if release.expected_sha256:
cached = cache.find_cached(
release.app_entry.app_id, release.expected_sha256
@@ -608,7 +642,9 @@ class FdroidSource:
dest = cache.cached_path(
release.app_entry.app_id, release.expected_sha256 or "downloading"
)
dest = _atomic_download(release.download_url, dest)
dest = _atomic_download(
release.download_url, dest, expected_size=release.file_size, progress=progress
)
actual = _sha256_file(dest)
if release.expected_sha256 and actual != release.expected_sha256:
dest.unlink()
@@ -673,11 +709,19 @@ class GitHubSource:
file_size=apk_assets[0].get("size"),
)
def download_apk(self, release: ReleaseInfo, cache: DownloadCache) -> Path:
def download_apk(
self,
release: ReleaseInfo,
cache: DownloadCache,
*,
progress: Callable[[int, int | None], None] | None = None,
) -> Path:
stem = _make_apk_stem(release)
dest = cache.root / release.app_entry.app_id / f"{stem}.apk"
if not dest.is_file():
dest = _atomic_download(release.download_url, dest)
dest = _atomic_download(
release.download_url, dest, expected_size=release.file_size, progress=progress
)
sha = _sha256_file(dest)
if release.expected_sha256 and sha.lower() != release.expected_sha256.lower():
dest.unlink(missing_ok=True)
@@ -733,11 +777,19 @@ class GitLabSource:
file_size=None,
)
def download_apk(self, release: ReleaseInfo, cache: DownloadCache) -> Path:
def download_apk(
self,
release: ReleaseInfo,
cache: DownloadCache,
*,
progress: Callable[[int, int | None], None] | None = None,
) -> Path:
stem = _make_apk_stem(release)
dest = cache.root / release.app_entry.app_id / f"{stem}.apk"
if not dest.is_file():
dest = _atomic_download(release.download_url, dest)
dest = _atomic_download(
release.download_url, dest, expected_size=release.file_size, progress=progress
)
sha = _sha256_file(dest)
if release.expected_sha256 and sha.lower() != release.expected_sha256.lower():
dest.unlink(missing_ok=True)
@@ -799,11 +851,19 @@ class CodebergSource:
file_size=asset.get("size"),
)
def download_apk(self, release: ReleaseInfo, cache: DownloadCache) -> Path:
def download_apk(
self,
release: ReleaseInfo,
cache: DownloadCache,
*,
progress: Callable[[int, int | None], None] | None = None,
) -> Path:
stem = _make_apk_stem(release)
dest = cache.root / release.app_entry.app_id / f"{stem}.apk"
if not dest.is_file():
dest = _atomic_download(release.download_url, dest)
dest = _atomic_download(
release.download_url, dest, expected_size=release.file_size, progress=progress
)
sha = _sha256_file(dest)
if release.expected_sha256 and sha.lower() != release.expected_sha256.lower():
dest.unlink(missing_ok=True)
@@ -961,6 +1021,20 @@ def _make_source(app: AppEntry) -> FdroidSource | GitHubSource | GitLabSource |
# ── orchestration ─────────────────────────────────────────────────────────────
def _download_progress(downloaded: int, total: int | None) -> None:
"""Render one compact download update on stderr."""
if total is not None:
percent = min(100, downloaded * 100 // total)
width = 20
filled = min(width, downloaded * width // total)
bar = "#" * filled + "-" * (width - filled)
message = f"\rDownloading APK: [{bar}] {percent:3d}%"
else:
message = f"\rDownloading APK: {downloaded:,} bytes"
sys.stderr.write(message)
sys.stderr.flush()
def sideload(
app: AppEntry,
config: Config,
@@ -987,11 +1061,25 @@ def sideload(
# 2 — download
print("Downloading APK…")
progress_seen = False
def report_progress(downloaded: int, total: int | None) -> None:
nonlocal progress_seen
progress_seen = True
_download_progress(downloaded, total)
try:
apk_path = source.download_apk(release, cache)
apk_path = source.download_apk(release, cache, progress=report_progress)
except Exception as exc:
if progress_seen:
sys.stderr.write("\n")
sys.stderr.flush()
print(f"Error downloading: {exc}", file=sys.stderr)
return 1
if progress_seen:
# Leave the terminal cursor at the next line after the last update.
sys.stderr.write("\n")
sys.stderr.flush()
print(f" -> cached at {apk_path}")
sha = _sha256_file(apk_path)
print(f" -> sha256: {sha}")

50
tests/test_sideload.py
View File

@@ -393,8 +393,10 @@ class TestFdroidSource:
download_url="https://example.com/x.apk",
expected_sha256=sha,
)
result = source.download_apk(release, cache)
progress = mock.Mock()
result = source.download_apk(release, cache, progress=progress)
assert result == cache.cached_path(app_entry.app_id, sha)
progress.assert_not_called()
def test_download_apk_sha256_mismatch(self, app_entry: sideload.AppEntry, cache: sideload.DownloadCache):
apk_data = b"real-content"
@@ -1293,6 +1295,52 @@ class TestUtilities:
assert dest.read_bytes() == data
assert not dest.with_suffix(".apk.tmp").exists()
def test_atomic_download_reports_content_length_progress(self, tmp_path: Path):
data = b"downloaded-content"
response = _mock_urlopen_response(data)
response.headers = {"Content-Length": str(len(data))}
updates = []
with _mock_download_opener(response):
result = sideload._atomic_download(
"https://example.com/x.apk",
tmp_path / "test.apk",
progress=lambda downloaded, total: updates.append((downloaded, total)),
)
assert result.read_bytes() == data
assert updates[0] == (0, len(data))
assert updates[-1] == (len(data), len(data))
def test_atomic_download_reports_indeterminate_progress(self, tmp_path: Path):
data = b"downloaded-content"
response = _mock_urlopen_response(data)
response.headers = {}
updates = []
with _mock_download_opener(response):
result = sideload._atomic_download(
"https://example.com/x.apk",
tmp_path / "test.apk",
expected_size=None,
progress=lambda downloaded, total: updates.append((downloaded, total)),
)
assert result.read_bytes() == data
assert updates[0] == (0, None)
assert updates[-1] == (len(data), None)
def test_atomic_download_uses_expected_size_without_content_length(self, tmp_path: Path):
data = b"downloaded-content"
response = _mock_urlopen_response(data)
response.headers = {}
updates = []
with _mock_download_opener(response):
sideload._atomic_download(
"https://example.com/x.apk",
tmp_path / "test.apk",
expected_size=len(data),
progress=lambda downloaded, total: updates.append((downloaded, total)),
)
assert updates[0] == (0, len(data))
assert updates[-1] == (len(data), len(data))
def test_atomic_download_cleans_up_on_error(self, tmp_path: Path):
dest = tmp_path / "file.apk"
with _mock_download_opener(error=RuntimeError("fail")):