#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ APKPure 下载器 通过 APKPure 的直连 URL 下载应用安装包，作为 US Download Worker 的最终回退源。 当 Google Play 和 Aurora Store 都失败时使用此模块。 参考 scripts/download.py 中的下载逻辑，适配 us_download_worker 的返回格式。 """ from __future__ import annotations import logging import os import time from datetime import datetime from typing import Any, Dict, List, Optional, Tuple import requests from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry logger = logging.getLogger("apkpure_downloader") # APKPure 直连下载 URL 模板 _APKPURE_DOWNLOAD_URL = "https://d.apkpure.net/b/XAPK/{package_name}?version=latest" # 请求头 _HEADERS = { "User-Agent": ( "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " "AppleWebKit/537.36 (KHTML, like Gecko) " "Chrome/108.0.0.0 Safari/537.36" ), } # 下载超时（秒）- 对 stream 下载的每个 chunk 读取生效 _DOWNLOAD_CONNECT_TIMEOUT = 15 _DOWNLOAD_READ_TIMEOUT = 30 _DOWNLOAD_PROGRESS_TIMEOUT = 200 # 单次下载最大重试次数 _MAX_RETRIES = 2 # 每次下载间的延迟（秒），避免被限流 DOWNLOAD_DELAY_SECONDS = 3 def _create_session(retries: int = 3, backoff_factor: float = 0.3, pool_connections: int = 20, pool_maxsize: int = 20) -> requests.Session: """创建带自动重试和连接池的 requests Session。""" session = requests.Session() retry = Retry( total=retries, read=retries, connect=retries, backoff_factor=backoff_factor, status_forcelist=(500, 502, 504), ) adapter = HTTPAdapter(max_retries=retry, pool_connections=pool_connections, pool_maxsize=pool_maxsize) session.mount("http://", adapter) session.mount("https://", adapter) return session # 模块级共享 session，避免每次下载重复创建连接池 _shared_session: Optional[requests.Session] = None _session_lock = __import__('threading').Lock() def _get_shared_session() -> requests.Session: global _shared_session if _shared_session is None: with _session_lock: if _shared_session is None: _shared_session = _create_session() return _shared_session def download_from_apkpure( package_name: str, export_dir: str, ) -> Tuple[Optional[Dict[str, Any]], Optional[str]]: """从 APKPure 下载应用安装包。 Args: package_name: 应用包名 export_dir: 导出目录（APK 文件保存到 {export_dir}/{package_name}/） Returns: (result_dict, failure_reason) - 成功时 result_dict 包含 package_name, download_date, version_name, files, local_dir - 失败时 result_dict 为 None，failure_reason 为失败原因 """ download_url = _APKPURE_DOWNLOAD_URL.format(package_name=package_name) pkg_dir = os.path.join(export_dir, package_name) final_filename = os.path.join(pkg_dir, f"{package_name}.xapk") if os.path.exists(final_filename) and os.path.getsize(final_filename) > 0: logger.info("[APKPure] 文件已存在，跳过下载: %s", package_name) return _build_result(package_name, final_filename, export_dir), None for old_file in os.listdir(pkg_dir) if os.path.isdir(pkg_dir) else []: if old_file.endswith(('.apk', '.xapk', '.apkm')): try: os.remove(os.path.join(pkg_dir, old_file)) except OSError: pass os.makedirs(pkg_dir, exist_ok=True) partial_filename = final_filename + ".part" headers = dict(_HEADERS) headers["Referer"] = f"https://apkpure.net/1/{package_name}" # 支持断点续传 resumable_size = 0 if os.path.exists(partial_filename): resumable_size = os.path.getsize(partial_filename) headers["Range"] = f"bytes={resumable_size}-" try: logger.info("[APKPure] 开始下载: %s (URL: %s)", package_name, download_url) start_time = time.time() with _get_shared_session().get( download_url, headers=headers, stream=True, timeout=(_DOWNLOAD_CONNECT_TIMEOUT, _DOWNLOAD_READ_TIMEOUT), verify=False, ) as response: # 检查 HTTP 状态 if response.status_code == 404: return None, f"APKPure 404: 应用不存在或不可用" if response.status_code == 403: return None, f"APKPure 403: 访问被拒绝" response.raise_for_status() # 获取文件总大小 content_length = response.headers.get("Content-Length") total_size = int(content_length) + resumable_size if content_length else 0 mode = "ab" if resumable_size > 0 else "wb" with open(partial_filename, mode) as f: downloaded = resumable_size for chunk in response.iter_content(chunk_size=1024 * 1024): if not chunk: break f.write(chunk) downloaded += len(chunk) if time.time() - start_time > _DOWNLOAD_PROGRESS_TIMEOUT: logger.warning("[APKPure] 下载总时长超限: %s (%.0fs)", package_name, time.time() - start_time) return None, f"下载总时长超限: {time.time() - start_time:.0f}s" if total_size and downloaded % (10 * 1024 * 1024) == 0: logger.debug("[APKPure] 下载进度: %s %.0f%%", package_name, downloaded / total_size * 100) # 验证文件完整性 actual_size = os.path.getsize(partial_filename) if total_size > 0 and actual_size != total_size: return None, f"文件下载不完整: 期望 {total_size} 字节, 实际 {actual_size} 字节" if actual_size == 0: # 空文件，清理并报告失败 try: os.remove(partial_filename) except OSError: pass return None, "下载文件为空" # 重命名为最终文件 os.rename(partial_filename, final_filename) logger.info("[APKPure] 下载完成: %s (%.1f MB)", package_name, actual_size / (1024 * 1024)) return _build_result(package_name, final_filename, export_dir), None except requests.exceptions.ConnectionError as e: logger.warning("[APKPure] 连接错误: %s - %s", package_name, e) return None, f"连接错误: {e}" except requests.exceptions.ReadTimeout as e: logger.warning("[APKPure] 读取超时(卡住): %s - %s", package_name, e) return None, f"下载卡住超时: {e}" except requests.exceptions.Timeout as e: logger.warning("[APKPure] 超时: %s - %s", package_name, e) return None, f"下载超时: {e}" except requests.exceptions.HTTPError as e: logger.warning("[APKPure] HTTP错误: %s - %s", package_name, e) return None, f"HTTP错误: {e}" except Exception as e: logger.warning("[APKPure] 下载异常: %s - %s", package_name, e) return None, f"下载异常: {e}" def _build_result( package_name: str, file_path: str, export_dir: str, ) -> Dict[str, Any]: """构建与 download_and_export 兼容的返回结果。""" file_size = os.path.getsize(file_path) if os.path.isfile(file_path) else 0 download_date = datetime.now().strftime("%Y-%m-%d %H:%M:%S") return { "package_name": package_name, "download_date": download_date, "version_name": "", # APKPure 直连无法获取版本号 "files": [ { "filename": os.path.basename(file_path), "size": file_size, } ], "local_dir": export_dir, }