bulk_apk_ripper.py

#!/usr/bin/env python3
"""Rip APKs en masse over ADB.

Requires an authorized ADB connection before running (USB, or `adb connect`
for Wi-Fi / Bluetooth debugging on a watch). Needs a platform-tools release
that forwards stdin to `adb shell` (anything from 2016 onwards) and a device
shell with `grep` (Android 6+ toybox).

All per-package metadata (paths, versionCode, versionName, ABI) is gathered in
ONE `adb shell` round trip instead of one-per-package-per-split, which is where
nearly all of the old script's time went on a slow link.

Output filenames keep the original scheme so existing rips are recognised and
skipped on re-runs:

    <package>_<versionName>-<versionCode>_minAPI<minSdk>(<abi>)[_<split>].apk
"""
from __future__ import annotations

import argparse
import re
import shutil
import subprocess
import sys
import time
from dataclasses import dataclass, field
from pathlib import Path

# Runs on the device inside a single `adb shell` call. Package names arrive on
# stdin, one per line; metadata comes back as @@PKG-delimited records.
COLLECT_SCRIPT = r"""
while IFS= read -r p; do
  [ -z "$p" ] && continue
  echo "@@PKG $p"
  pm path "$p" 2>/dev/null
  dumpsys package "$p" 2>/dev/null | grep -E '^ *(versionCode=|versionName=|primaryCpuAbi=)'
done
exit 0
"""

# Characters that are illegal in filenames on at least one of Windows/macOS/Linux,
# plus whitespace (the old script already turned spaces into underscores).
UNSAFE_CHARS = re.compile(r'[\\/:*?"<>|\s]+')


# --------------------------------------------------------------------------- #
# Data
# --------------------------------------------------------------------------- #
@dataclass
class Device:
    serial: str
    state: str
    transport_id: str | None
    model: str | None

    def describe(self) -> str:
        bits = [self.serial, self.state]
        if self.model:
            bits.append(f"model:{self.model}")
        if self.transport_id:
            bits.append(f"transport_id:{self.transport_id}")
        return "  ".join(bits)


@dataclass
class Package:
    name: str
    paths: list[str] = field(default_factory=list)
    version_code: str | None = None
    version_name: str | None = None
    min_sdk: str | None = None
    abi: str | None = None

    def label(self) -> str:
        abi = "noarch" if self.abi in (None, "null") else self.abi
        return f"{self.version_name or 'null'}-{self.version_code or '0'} minAPI{self.min_sdk or 'unknown'} {abi}"


@dataclass
class Summary:
    pulled: int = 0
    skipped: int = 0
    failed: list[tuple[str, str]] = field(default_factory=list)


# --------------------------------------------------------------------------- #
# ADB plumbing
# --------------------------------------------------------------------------- #
def adb_run(cmd: list[str], *, check: bool = True, input: bytes | None = None) -> tuple[int, str, str]:
    """Run an adb command, returning (returncode, stdout, stderr) as text.

    Everything is handled as bytes and decoded here so Windows CRLF, stray
    non-UTF-8 output and stdin line endings can't trip anything up.
    """
    try:
        proc = subprocess.run(cmd, input=input, capture_output=True)
    except FileNotFoundError:
        sys.exit("adb was not found on PATH.")
    out = proc.stdout.decode("utf-8", "replace")
    err = proc.stderr.decode("utf-8", "replace")
    if check and proc.returncode != 0:
        shown = " ".join(cmd[:5]) + (" ..." if len(cmd) > 5 else "")
        sys.exit(f"command failed (exit {proc.returncode}): {shown}\n{err.strip()}")
    return proc.returncode, out, err


def list_devices() -> list[Device]:
    _, out, _ = adb_run(["adb", "devices", "-l"])
    devices: list[Device] = []
    for line in out.splitlines()[1:]:  # first line is "List of devices attached"
        tokens = line.split()
        if len(tokens) < 2:
            continue
        attrs = dict(t.split(":", 1) for t in tokens[2:] if ":" in t)
        devices.append(Device(
            serial=tokens[0],
            state=tokens[1],
            transport_id=attrs.get("transport_id"),
            model=attrs.get("model"),
        ))
    return devices


def select_device(serial: str | None, tid: str | None) -> tuple[Device, list[str]]:
    """Return the chosen device and the adb argv prefix that targets it."""
    devices = list_devices()
    listing = "\n".join(f"  {d.describe()}" for d in devices) or "  (none)"

    if serial:
        match = [d for d in devices if d.serial == serial]
        prefix = ["adb", "-s", serial]
        what = f"serial {serial}"
    elif tid:
        match = [d for d in devices if d.transport_id == tid]  # whole-token match: 1 != 10
        prefix = ["adb", "-t", tid]
        what = f"transport id {tid}"
    else:
        online = [d for d in devices if d.state == "device"]
        if len(online) == 1:
            match = online
            prefix = ["adb", "-s", online[0].serial]
        elif not online:
            sys.exit(f"No authorized devices online.\nDevices seen:\n{listing}")
        else:
            sys.exit(f"More than one device online; pick one with -s or -t.\nDevices seen:\n{listing}")
        what = "the only connected device"

    if not match:
        sys.exit(f"No device with {what}.\nDevices seen:\n{listing}")
    dev = match[0]
    if dev.state != "device":
        hint = ""
        if dev.state == "unauthorized":
            hint = " (accept the USB/wireless debugging prompt on the device)"
        elif dev.state == "offline":
            hint = " (try `adb disconnect` / `adb connect` again, or replug)"
        sys.exit(f"{dev.serial} is '{dev.state}', not ready{hint}.")
    return dev, prefix


# --------------------------------------------------------------------------- #
# Package discovery
# --------------------------------------------------------------------------- #
def list_packages(adb: list[str], third_party_only: bool) -> list[str]:
    cmd = adb + ["shell", "pm", "list", "packages"] + (["-3"] if third_party_only else [])
    _, out, _ = adb_run(cmd)
    names = {line.split(":", 1)[1].strip() for line in out.splitlines() if line.startswith("package:")}
    return sorted(names)


def collect_metadata(adb: list[str], names: list[str]) -> list[Package]:
    """One round trip: feed package names on stdin, get paths + dumpsys fields back."""
    payload = ("\n".join(names) + "\n").encode("utf-8")
    rc, out, err = adb_run(adb + ["shell", COLLECT_SCRIPT], check=False, input=payload)
    packages = parse_records(out)
    if names and not packages:
        sys.exit(
            "The on-device collector returned nothing (exit {rc}). This usually means adb is too\n"
            "old to forward stdin to `adb shell`, or the device shell lacks grep.\n"
            f"stderr:\n{err.strip()}"
        )
    return packages


def parse_records(text: str) -> list[Package]:
    packages: list[Package] = []
    cur: Package | None = None
    for raw in text.splitlines():
        line = raw.strip()
        if not line:
            continue
        if line.startswith("@@PKG "):
            cur = Package(name=line[6:].strip())
            packages.append(cur)
            continue
        if cur is None:
            continue
        # Only the first occurrence of each dumpsys field is taken: that is the
        # active package; "Hidden system packages:" sections come later.
        if line.startswith("package:"):
            cur.paths.append(line[8:].strip())
        elif line.startswith("versionCode=") and cur.version_code is None:
            fields = dict(tok.split("=", 1) for tok in line.split() if "=" in tok)
            cur.version_code = fields.get("versionCode")
            cur.min_sdk = fields.get("minSdk")  # absent on old Android; don't guess targetSdk
        elif line.startswith("versionName=") and cur.version_name is None:
            cur.version_name = line.split("=", 1)[1].strip()
        elif line.startswith("primaryCpuAbi=") and cur.abi is None:
            cur.abi = line.split("=", 1)[1].strip()
    return packages


# --------------------------------------------------------------------------- #
# Naming
# --------------------------------------------------------------------------- #
def sanitize(text: str) -> str:
    cleaned = UNSAFE_CHARS.sub("_", text).strip("._")
    return cleaned or "unknown"


def apk_filename(pkg: Package, remote_path: str, is_base: bool) -> str:
    version_name = sanitize(pkg.version_name or "null")
    version_code = sanitize(pkg.version_code or "0")
    min_sdk = sanitize(pkg.min_sdk or "unknown")
    abi = "noarch" if pkg.abi in (None, "null") else sanitize(pkg.abi)
    split = ""
    if not is_base:
        split = "_" + sanitize(remote_path.rsplit("/", 1)[-1].removesuffix(".apk"))
    return f"{pkg.name}_{version_name}-{version_code}_minAPI{min_sdk}({abi}){split}.apk"


# --------------------------------------------------------------------------- #
# Pulling
# --------------------------------------------------------------------------- #
class Puller:
    def __init__(self, adb: list[str], compress: bool) -> None:
        self.adb = adb
        self.compress = compress

    def pull(self, remote: str, dest: Path) -> tuple[bool, str]:
        """Pull to a .part file and rename on success so a dead transfer never
        leaves a truncated file that later runs would mistake for a finished rip."""
        tmp = dest.with_name(dest.name + ".part")
        attempts = [True, False] if self.compress else [False]
        last_error = ""
        for use_compression in attempts:
            cmd = self.adb + ["pull", "-a"]
            if use_compression:
                cmd += ["-z", "any"]
            cmd += [remote, str(tmp)]
            rc, out, err = adb_run(cmd, check=False)
            if rc == 0 and tmp.exists():
                if self.compress and not use_compression:
                    print("          note: compressed pull failed, uncompressed worked; disabling -z for the rest of the run")
                    self.compress = False
                tmp.replace(dest)
                return True, ""
            last_error = (err.strip() or out.strip() or f"exit {rc}").splitlines()[-1]
        tmp.unlink(missing_ok=True)
        return False, last_error


def human_size(num_bytes: int) -> str:
    if num_bytes < 1_000_000:
        return f"{num_bytes / 1_000:.0f} kB"
    return f"{num_bytes / 1_000_000:.1f} MB"


def pull_packages(adb: list[str], packages: list[Package], out_dir: Path,
                  dry_run: bool, compress: bool) -> Summary:
    out_dir.mkdir(parents=True, exist_ok=True)
    existing = {p.name for p in out_dir.iterdir()}
    puller = Puller(adb, compress)
    summary = Summary()
    width = len(str(len(packages)))

    for index, pkg in enumerate(packages, 1):
        print(f"[{index:>{width}}/{len(packages)}] {pkg.name}  {pkg.label()}")
        if not pkg.paths:
            print("          no APK path reported (not installed for this user?), skipping")
            summary.skipped += 1
            continue

        for position, remote in enumerate(pkg.paths):
            if "/overlay/" in remote:
                summary.skipped += 1
                continue
            # `pm path` always lists the base APK first.
            name = apk_filename(pkg, remote, is_base=(position == 0))
            dest = out_dir / name
            if name in existing:
                summary.skipped += 1
                continue
            if dry_run:
                print(f"          would pull {remote}\n                  -> {name}")
                summary.pulled += 1
                continue

            print(f"          {remote.rsplit('/', 1)[-1]} ... ", end="", flush=True)
            started = time.monotonic()
            ok, error = puller.pull(remote, dest)
            elapsed = time.monotonic() - started
            if ok:
                existing.add(name)
                summary.pulled += 1
                print(f"ok ({human_size(dest.stat().st_size)}, {elapsed:.1f}s)")
            else:
                summary.failed.append((f"{pkg.name} {remote}", error))
                print(f"FAILED: {error}")
    return summary


# --------------------------------------------------------------------------- #
# CLI
# --------------------------------------------------------------------------- #
def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Rip APKs en masse using ADB.")
    target = parser.add_mutually_exclusive_group()
    target.add_argument("-s", "--serial", help="device serial (adb devices); stable across adb restarts")
    target.add_argument("-t", "--transport-id", help="transport id (adb devices -l)")
    parser.add_argument("-o", "--out", type=Path, default=Path.cwd(), help="output directory (default: cwd)")
    parser.add_argument("-3", "--third-party", action="store_true", help="only third-party (non-system) packages")
    parser.add_argument("--include", metavar="REGEX", help="only packages whose name matches")
    parser.add_argument("--exclude", metavar="REGEX", help="skip packages whose name matches")
    parser.add_argument("-n", "--dry-run", action="store_true", help="show what would be pulled, pull nothing")
    parser.add_argument("--no-compress", action="store_true", help="never pass -z to adb pull")
    args = parser.parse_args(argv)
    for opt in ("include", "exclude"):
        pattern = getattr(args, opt)
        if pattern:
            try:
                setattr(args, opt, re.compile(pattern))
            except re.error as exc:
                parser.error(f"--{opt}: bad regex: {exc}")
    return args


def main(argv: list[str] | None = None) -> int:
    args = parse_args(argv)
    if shutil.which("adb") is None:
        sys.exit("adb was not found on PATH.")

    device, adb = select_device(args.serial, args.transport_id)
    print(f"CONNECTED TO {device.describe()}")

    print("Listing packages... ", end="", flush=True)
    names = list_packages(adb, args.third_party)
    total = len(names)
    if args.include:
        names = [n for n in names if args.include.search(n)]
    if args.exclude:
        names = [n for n in names if not args.exclude.search(n)]
    print(f"{len(names)} selected" + (f" of {total}" if len(names) != total else ""))
    if not names:
        print("Nothing to do.")
        return 0

    print("Collecting paths and versions (one round trip)... ", end="", flush=True)
    started = time.monotonic()
    packages = collect_metadata(adb, names)
    print(f"done in {time.monotonic() - started:.1f}s")

    print(f"{'Would pull' if args.dry_run else 'Pulling'} into {args.out.resolve()}")
    summary = pull_packages(adb, packages, args.out, args.dry_run, compress=not args.no_compress)

    verb = "would pull" if args.dry_run else "pulled"
    print(f"\nCOMPLETE: {verb} {summary.pulled}, skipped {summary.skipped}, failed {len(summary.failed)}")
    for what, why in summary.failed:
        print(f"  FAILED {what}\n         {why}")
    return 1 if summary.failed else 0


if __name__ == "__main__":
    sys.exit(main())