#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 美国下载PC定时脚本 每 N 分钟从 MinIO 拉取下载任务列表，批量下载 → 导出APK → 上传到 MinIO， 并写入下载结果供中控读取。下载失败的包记录详细原因，汇总为报告定期上传。 多设备下载策略： - 第一次在模拟器A（默认 Google Play）上尝试下载 - 如果失败，自动回退到模拟器B（Aurora Store）再次尝试 - 如果仍然失败，最终回退到 APKPure HTTP 直连下载 用法: python us_download_worker.py python us_download_worker.py --export-dir D:\\mumu_apk_export --interval 600 python us_download_worker.py --once """ from __future__ import annotations import json import logging import os import shutil import subprocess import sys import threading import time from concurrent.futures import ThreadPoolExecutor, as_completed from datetime import datetime from typing import Any, Dict, List, Optional PROJECT_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) if PROJECT_ROOT not in sys.path: sys.path.insert(0, PROJECT_ROOT) from config import ( APK_EMULATOR_CLEANUP_ENABLED, APK_US_DEVICE_A_SERIAL, APK_US_DEVICE_A_VM_INDEX, APK_US_DEVICE_B_SERIAL, APK_US_DEVICE_B_VM_INDEX, APK_US_EXPORT_DIR, MUMU_MANAGER_PATH, ) from apk_cloud.storage import MinioStorage from apk_cloud.android_google_play_downloader import ( GooglePlayDownloader, STORE_GOOGLE_PLAY, STORE_AURORA, ) from apk_cloud.apkpure_downloader import download_from_apkpure, DOWNLOAD_DELAY_SECONDS logger = logging.getLogger("us_download_worker") _DEFAULT_POLL_SECONDS = 60 _EMULATOR_CLEANUP_INTERVAL = 3600 _FAILURE_REPORT_KEY = "download-reports/failures.json" _HISTORY_FILE = "download_history.json" _MAX_PIPELINE_WORKERS = 8 _APKPURE_MAX_CONCURRENT = 3 _device_a_lock = threading.Lock() _device_b_lock = threading.Lock() _apkpure_semaphore = threading.Semaphore(_APKPURE_MAX_CONCURRENT) _history_lock = threading.Lock() US_DEVICE_A_SERIAL = APK_US_DEVICE_A_SERIAL US_DEVICE_B_SERIAL = APK_US_DEVICE_B_SERIAL # 设备串号 → MuMu VM 编号 _US_SERIAL_VM_MAP = { US_DEVICE_A_SERIAL: APK_US_DEVICE_A_VM_INDEX, US_DEVICE_B_SERIAL: APK_US_DEVICE_B_VM_INDEX, } _US_EMULATOR_RESTART_LOCK = threading.Lock() _US_EMULATOR_RESTART_COOLDOWN: Dict[str, float] = {} def _is_adb_device_offline_error(error_text: str) -> bool: if not error_text: return False lowered = error_text.lower() return "device" in lowered and "not found" in lowered def _restart_emulator_if_needed(serial: str, reason: str) -> bool: vm_index = _US_SERIAL_VM_MAP.get(serial) if not vm_index: return False if not _is_adb_device_offline_error(reason): return False now = time.time() with _US_EMULATOR_RESTART_LOCK: last = _US_EMULATOR_RESTART_COOLDOWN.get(serial, 0) if now - last < 120: logger.info("[Emulator] %s 在冷却期内(%.0fs前刚重启)，跳过本次重启", serial, now - last) return False _US_EMULATOR_RESTART_COOLDOWN[serial] = now logger.info("[Emulator] 检测到 ADB 断链 (%s)，正在重启模拟器 VM-%d ...", serial, vm_index) manager = MUMU_MANAGER_PATH.strip() if not os.path.isfile(manager): logger.warning("[Emulator] MuMuManager.exe 不存在: %s", manager) return False try: subprocess.run( [manager, "control", "-v", str(vm_index), "shutdown"], capture_output=True, timeout=30, ) time.sleep(10) subprocess.run( [manager, "control", "-v", str(vm_index), "restart"], capture_output=True, timeout=30, ) time.sleep(30) subprocess.run(["adb", "disconnect", serial.split(":")[0]], capture_output=True, timeout=10) time.sleep(5) subprocess.run(["adb", "connect", serial], capture_output=True, timeout=15) time.sleep(10) logger.info("[Emulator] VM-%d 重启完成，ADB 已重连", vm_index) return True except subprocess.TimeoutExpired as exc: logger.warning("[Emulator] VM-%d 重启命令超时: %s", vm_index, exc) return False except Exception as exc: logger.warning("[Emulator] VM-%d 重启失败: %s", vm_index, exc) return False # ── Download History ──────────────────────────────────────────────── def _load_history(export_dir: str) -> Dict[str, Dict[str, str]]: path = os.path.join(export_dir, _HISTORY_FILE) if not os.path.isfile(path): return {} try: with open(path, "r", encoding="utf-8") as f: data = json.load(f) return data if isinstance(data, dict) else {} except (json.JSONDecodeError, IOError): return {} def _save_history(export_dir: str, history: Dict[str, Dict[str, str]]) -> None: path = os.path.join(export_dir, _HISTORY_FILE) tmp = path + ".tmp" with open(tmp, "w", encoding="utf-8") as f: json.dump(history, f, ensure_ascii=False, indent=2) os.replace(tmp, path) def _is_history_fresh(history: Dict[str, Dict[str, str]], package_name: str, last_updated: str) -> bool: entry = history.get(package_name) if not entry: return False hist_updated = str(entry.get("last_updated") or "") hist_date = str(entry.get("download_date") or "") if not hist_date: return False if hist_updated != last_updated: return False return True def _record_history(history: Dict[str, Dict[str, str]], package_name: str, last_updated: str, download_date: str = "", version_name: str = "") -> None: history[package_name] = { "last_updated": last_updated, "download_date": download_date or datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "version_name": version_name or "", } def download_and_export(package_name: str, export_dir: str, downloader: GooglePlayDownloader) -> tuple: """返回 (result_dict_or_None, failure_reason_or_None)。""" try: if downloader.is_installed(package_name): logger.info("Already installed, skip download: %s", package_name) return _export_only(package_name, export_dir, downloader), None logger.info("Downloading: %s (device=%s)", package_name, downloader._serial or "default") success, message = downloader.start(package_name) if not success: reason = str(message or "unknown download error").strip() logger.warning("Download failed: %s reason=%s", package_name, reason) return None, reason logger.info("Download succeeded: %s", package_name) result = _export_only(package_name, export_dir, downloader) if result is None: return None, "export failed after download" return result, None finally: # 每次下载任务完成后（无论成功失败）重置设备状态 # 清理商店页面、系统弹窗等残留，确保不影响下一个包的下载 try: downloader.reset_state() except Exception as exc: logger.warning("reset_state failed: %s", exc) def download_with_fallback(package_name: str, export_dir: str) -> tuple: """多设备下载 pipeline：先尝试模拟器A (Google Play)，失败则回退模拟器B (Aurora Store)， 最终回退到 APKPure HTTP 直连下载。 各阶段通过 device lock 协调并发： - 一个任务释放 Device A → 下一个任务立即抢占 Device A - Device A / Device B / APKPure 可同时处理互不干扰 返回 (result_dict_or_None, failure_reason_or_None, source_details_dict)， 其中 result_dict 包含 source_details 记录每个来源的尝试结果。 """ source_details = {} # ── 第一级：Google Play (Device A) ── logger.info("[Device A / Google Play] Trying: %s", package_name) with _device_a_lock: device_a = GooglePlayDownloader(serial=US_DEVICE_A_SERIAL, store_type=STORE_GOOGLE_PLAY) result, reason = download_and_export(package_name, export_dir, device_a) if result: source_details["us_play"] = { "status": "success", "error": None, "date": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), } source_details["us_aurora"] = { "status": "not_attempted", "error": None, "date": None, } source_details["apkpure"] = { "status": "not_attempted", "error": None, "date": None, } result["source_details"] = source_details return result, None, source_details source_details["us_play"] = { "status": "failed", "error": reason, "date": None, } logger.warning("[Device A] Failed (%s), falling back to Device B / Aurora Store: %s", reason, package_name) if _restart_emulator_if_needed(US_DEVICE_A_SERIAL, reason): logger.info("[Device A] 模拟器已重启，重试一次: %s", package_name) with _device_a_lock: device_a = GooglePlayDownloader(serial=US_DEVICE_A_SERIAL, store_type=STORE_GOOGLE_PLAY) result, reason = download_and_export(package_name, export_dir, device_a) if result: source_details["us_play"] = { "status": "success", "error": None, "date": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), } source_details["us_aurora"] = {"status": "not_attempted", "error": None, "date": None} source_details["apkpure"] = {"status": "not_attempted", "error": None, "date": None} result["source_details"] = source_details return result, None, source_details # ── 第二级：Aurora Store (Device B) ── with _device_b_lock: device_b = GooglePlayDownloader(serial=US_DEVICE_B_SERIAL, store_type=STORE_AURORA) result2, reason2 = download_and_export(package_name, export_dir, device_b) if result2: source_details["us_aurora"] = { "status": "success", "error": None, "date": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), } source_details["apkpure"] = { "status": "not_attempted", "error": None, "date": None, } result2["source_details"] = source_details return result2, None, source_details source_details["us_aurora"] = { "status": "failed", "error": reason2, "date": None, } logger.warning( "[Device A+B] Both failed (A=%s, B=%s), falling back to APKPure: %s", reason, reason2, package_name, ) if _restart_emulator_if_needed(US_DEVICE_B_SERIAL, reason2): logger.info("[Device B] 模拟器已重启，重试一次: %s", package_name) with _device_b_lock: device_b = GooglePlayDownloader(serial=US_DEVICE_B_SERIAL, store_type=STORE_AURORA) result2, reason2 = download_and_export(package_name, export_dir, device_b) if result2: source_details["us_aurora"] = { "status": "success", "error": None, "date": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), } source_details["apkpure"] = {"status": "not_attempted", "error": None, "date": None} result2["source_details"] = source_details return result2, None, source_details # ── 第三级（最终回退）：APKPure HTTP 直连下载 ── with _apkpure_semaphore: result3, reason3 = download_from_apkpure(package_name, export_dir) if result3: source_details["apkpure"] = { "status": "success", "error": None, "date": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), } result3["source_details"] = source_details return result3, None, source_details source_details["apkpure"] = { "status": "failed", "error": reason3, "date": None, } return ( None, f"all sources failed: Play={reason}, Aurora={reason2}, APKPure={reason3}", source_details, ) def _export_only(package_name: str, export_dir: str, downloader: GooglePlayDownloader, uninstall_after: bool = True) -> Optional[Dict[str, Any]]: version = downloader.get_apk_version(package_name) pkg_export_dir = os.path.join(export_dir, package_name) for old_file in os.listdir(pkg_export_dir) if os.path.isdir(pkg_export_dir) else []: if old_file.endswith(('.apk', '.xapk', '.apkm')): try: os.remove(os.path.join(pkg_export_dir, old_file)) except OSError: pass os.makedirs(pkg_export_dir, exist_ok=True) apk_files = downloader.export_apk(package_name, pkg_export_dir) if not apk_files: logger.warning("APK export empty: %s", package_name) return None if uninstall_after and APK_EMULATOR_CLEANUP_ENABLED: try: downloader._adb.run( ["shell", "am", "force-stop", package_name], check=False, timeout=10, ) except Exception: pass try: downloader._adb.run( ["shell", "pm", "clear", package_name], check=False, timeout=15, ) except Exception: pass try: downloader._adb.run( ["shell", "pm", "uninstall", "--user", "0", package_name], check=True, timeout=30, ) logger.info("Uninstalled after export: %s", package_name) except Exception as exc: logger.warning("Uninstall failed: %s: %s", package_name, exc) try: downloader._adb.run( ["shell", "pm", "uninstall", package_name], check=False, timeout=15, ) except Exception: pass download_date = datetime.now().strftime("%Y-%m-%d %H:%M:%S") file_infos = [] for fp in apk_files: size = os.path.getsize(fp) if os.path.isfile(fp) else 0 file_infos.append({ "filename": os.path.basename(fp), "size": size, }) return { "package_name": package_name, "download_date": download_date, "version_name": version, "files": file_infos, "local_dir": export_dir, } def _process_single_task(task: Dict[str, str], export_dir: str, storage: MinioStorage, history: Dict[str, Dict[str, str]]) -> Dict[str, Any]: """处理单个下载任务：pipeline 下载 → 上传 MinIO → 记录结果。 返回 {"success": bool, "failures": [...]} 供 run_cycle 聚合统计。 """ package_name = str(task.get("package_name") or "").strip() last_updated = str(task.get("last_updated") or "").strip() if not package_name: return {"success": False, "failures": []} try: result, fail_reason, source_details = download_with_fallback(package_name, export_dir) except Exception as exc: logger.exception("download_with_fallback panicked: %s", package_name) fail_info = { "package_name": package_name, "reason": f"pipeline error: {exc}", "failed_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), } fallback = { "status": "failed", "package_name": package_name, "download_date": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "reason": str(exc), "source_details": {}, } try: storage.write_download_result(package_name, fallback) except Exception: pass with _history_lock: _record_history(history, package_name, last_updated, download_date="none", version_name="") _save_history(export_dir, history) return {"success": False, "failures": [fail_info]} if result: result["status"] = "ok" pkg_dir = os.path.join(export_dir, package_name) try: version_code = result.get("version_name", "").replace(".", "_") or "v0" manifest = storage.upload_apk( package_name, export_dir, download_date=result["download_date"][:10], version_code=version_code, ) result["uploaded_files"] = manifest.get("files", []) result["download_date"] = manifest.get("download_date", result.get("download_date", "")) result["version_code"] = manifest.get("version_code", "") storage.cleanup_old_versions(package_name) with _history_lock: _record_history(history, package_name, last_updated, download_date=result.get("download_date", ""), version_name=result.get("version_name", "")) _save_history(export_dir, history) except Exception as exc: logger.error("Upload failed for %s: %s", package_name, exc) result["status"] = "download_ok_upload_failed" result["upload_error"] = str(exc) try: storage.write_download_result(package_name, result) except Exception: pass return { "success": False, "failures": [{ "package_name": package_name, "reason": f"upload failed: {exc}", "failed_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), }], } finally: if os.path.isdir(pkg_dir): shutil.rmtree(pkg_dir, ignore_errors=True) logger.info("[Upload] 已清理本地文件: %s", pkg_dir) storage.write_download_result(package_name, result) return {"success": True, "failures": []} else: reason = fail_reason or "download_or_export_failed" result = { "status": "failed", "package_name": package_name, "download_date": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "reason": reason, "source_details": source_details, } storage.write_download_result(package_name, result) with _history_lock: _record_history(history, package_name, last_updated, download_date="none", version_name="") _save_history(export_dir, history) return { "success": False, "failures": [{ "package_name": package_name, "reason": reason, "failed_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), }], } def run_cycle(storage: MinioStorage, export_dir: str, history: Dict[str, Dict[str, str]] = None) -> Dict[str, Any]: history = history if history is not None else {} tasks = storage.pull_download_queue() if not tasks: logger.info("No tasks in download queue, sleeping") return {"processed": 0, "uploaded": 0, "failures": []} original_count = len(tasks) filtered = [] skipped = 0 for task in tasks: package_name = str(task.get("package_name") or "").strip() last_updated = str(task.get("last_updated") or "").strip() if not package_name or ' ' in package_name or '&&' in package_name: logger.warning("Skipping invalid task entry: %s", task) continue if not task.get("force") and _is_history_fresh(history, package_name, last_updated): skipped += 1 continue filtered.append(task) if skipped: logger.info("History filter: %d/%d tasks skipped (already downloaded)", skipped, original_count) tasks = filtered logger.info("%d tasks to process (max_workers=%d, priority ordered)", len(tasks), _MAX_PIPELINE_WORKERS) processed = 0 uploaded = 0 failures: List[Dict[str, Any]] = [] if not tasks: return {"processed": 0, "uploaded": 0, "failures": []} max_workers = min(_MAX_PIPELINE_WORKERS, max(len(tasks), 1)) with ThreadPoolExecutor(max_workers=max_workers) as executor: future_map = {} for task in tasks: future = executor.submit(_process_single_task, task, export_dir, storage, history) future_map[future] = task for future in as_completed(future_map): task_result = future.result() if task_result.get("success"): processed += 1 uploaded += 1 failures.extend(task_result.get("failures", [])) return {"processed": processed, "uploaded": uploaded, "failures": failures} def upload_failure_report(storage: MinioStorage, failures: List[Dict[str, Any]]) -> None: if not failures: return try: existing = storage.download_json(_FAILURE_REPORT_KEY) or {} except Exception: existing = {} existing.setdefault("failures", []) existing_failures = existing.get("failures", []) if not isinstance(existing_failures, list): existing_failures = [] existing["failures"] = existing_failures existing_failures.extend(failures) existing["generated_at"] = datetime.now().strftime("%Y-%m-%d %H:%M:%S") existing["worker"] = "us-downloader" existing["total_failures"] = len(existing_failures) try: storage.upload_json(existing, _FAILURE_REPORT_KEY) logger.info("Failure report uploaded: %d total failures", len(existing_failures)) except Exception as exc: logger.warning("Failed to upload failure report: %s", exc) def cleanup_emulator_third_party_apps(downloader: GooglePlayDownloader) -> int: result = downloader._adb.run( ["shell", "pm", "list", "packages", "-3"], check=False, timeout=30, ) if result.returncode != 0: return 0 packages = [] for line in (result.stdout or "").splitlines(): line = line.strip() if line.startswith("package:"): pkg = line.split(":", 1)[1].strip() if pkg and pkg not in ("com.android.vending", "com.aurora.store"): packages.append(pkg) uninstalled = 0 for pkg in packages: try: r = downloader._adb.run( ["shell", "pm", "uninstall", pkg], check=False, timeout=15, ) if r.returncode == 0: uninstalled += 1 logger.info("Emulator cleanup uninstalled: %s", pkg) except Exception: pass return uninstalled def main_loop(export_dir: str, poll_seconds: int = _DEFAULT_POLL_SECONDS): storage = MinioStorage() downloader_for_cleanup = GooglePlayDownloader(serial=US_DEVICE_B_SERIAL, store_type=STORE_AURORA) last_emulator_cleanup = time.time() accumulated_failures: List[Dict[str, Any]] = [] history = _load_history(export_dir) history_count = len(history) logger.info("US Download Worker started, poll_interval=%ds, export_dir=%s", poll_seconds, export_dir) logger.info("Device A (Google Play): %s, Device B (Aurora): %s", US_DEVICE_A_SERIAL, US_DEVICE_B_SERIAL) logger.info("MinIO: %s / %s", storage.endpoint, storage.bucket) logger.info("History: %d previously downloaded packages", history_count) while True: start_time = time.time() try: stats = run_cycle(storage, export_dir, history) elapsed = time.time() - start_time logger.info("Cycle done: processed=%d uploaded=%d failed=%d elapsed=%.1fs", stats["processed"], stats["uploaded"], len(stats.get("failures", [])), elapsed) failures = stats.get("failures", []) if failures: accumulated_failures.extend(failures) upload_failure_report(storage, accumulated_failures) if stats["uploaded"] > 0: try: limit_result = storage.enforce_storage_limit() if limit_result.get("deleted_versions"): logger.info( "MinIO storage limit: %.1fGB -> %.1fGB (deleted %d versions)", limit_result["before_bytes"] / (1024 ** 3), limit_result["after_bytes"] / (1024 ** 3), limit_result["deleted_versions"], ) except Exception as exc: logger.warning("MinIO storage limit check failed: %s", exc) if APK_EMULATOR_CLEANUP_ENABLED and (time.time() - last_emulator_cleanup >= _EMULATOR_CLEANUP_INTERVAL): try: count = cleanup_emulator_third_party_apps(downloader_for_cleanup) if count > 0: logger.info("Emulator cleanup: uninstalled %d apps", count) except Exception as exc: logger.warning("Emulator cleanup failed: %s", exc) last_emulator_cleanup = time.time() except Exception: logger.exception("Cycle error") sleep_left = max(0, poll_seconds - (time.time() - start_time)) if sleep_left > 0: logger.info("Sleeping %.0fs until next poll...", sleep_left) time.sleep(sleep_left) def main() -> int: import argparse parser = argparse.ArgumentParser(description="US Download Worker") parser.add_argument("--export-dir", default=APK_US_EXPORT_DIR, help=f"APK导出目录 (default: {APK_US_EXPORT_DIR})") parser.add_argument("--interval", type=int, default=_DEFAULT_POLL_SECONDS, help=f"轮询间隔秒数 (default: {_DEFAULT_POLL_SECONDS})") parser.add_argument("--once", action="store_true", help="只执行一次循环后退出") args = parser.parse_args() logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s", datefmt="%H:%M:%S", ) export_dir = str(args.export_dir or "").strip() or APK_US_EXPORT_DIR os.makedirs(export_dir, exist_ok=True) if args.once: storage = MinioStorage() history = _load_history(export_dir) stats = run_cycle(storage, export_dir, history) failures = stats.get("failures", []) if failures: upload_failure_report(storage, failures) print(f"done: processed={stats['processed']} uploaded={stats['uploaded']} failed={len(failures)}") return 0 main_loop(export_dir, poll_seconds=args.interval) return 0 if __name__ == "__main__": raise SystemExit(main())