diff --git a/core/adb_helper.py b/core/adb_helper.py index 09bcd61..6ff867f 100644 --- a/core/adb_helper.py +++ b/core/adb_helper.py @@ -154,6 +154,148 @@ def screenshot(serial, timeout=5): return False, str(e) +# 常见应用名映射：快速识别大/重要应用，避免拉取 APK 解析（慢） +COMMON_APPS = { + "com.ss.android.ugc.aweme": "抖音", + "com.ss.android.ugc.aweme.lite": "抖音极速版", + "com.smile.gifmaker": "快手", + "com.kuaishou.nebula": "快手极速版", + "com.tencent.mm": "微信", + "com.tencent.mobileqq": "QQ", + "com.tencent.qqlive": "腾讯视频", + "com.eg.android.AlipayGphone": "支付宝", + "com.taobao.taobao": "淘宝", + "com.tmall.wireless": "天猫", + "com.jingdong.app.mall": "京东", + "com.xunmeng.pinduoduo": "拼多多", + "com.ss.android.article.news": "今日头条", + "com.zhihu.android": "知乎", + "com.miui.home": "桌面", + "com.android.settings": "设置", + "com.android.systemui": "系统", + "com.github.uiautomator": "ATX", + "com.github.uiautomator.test": "ATX测试", + "jp.co.cyberagent.stf": "STF服务", + "moe.shizuku.privileged.api": "Shizuku", + "moe.shizuku.api": "Shizuku", + "io.tailscale.android": "Tailscale", + "com.immomo.momo": "陌陌", + "com.xingin.xhs": "小红书", +} +# 解析应用名时，超过该大小的 APK 不拉取（太慢），用映射或包名兜底 +_LABEL_PARSE_MAX_BYTES = 5 * 1024 * 1024 # 5MB +# 应用名缓存：(serial, package) -> label +_APP_LABEL_CACHE = {} + + +def list_installed_apps(serial, timeout=40): + """获取设备上已安装应用列表：应用名 + 包名 + versionCode + versionName + APK路径。 + + - `pm list packages --show-versioncode -f` 取包名/版本码/路径（快） + - `pm dump packages` 取人性化 versionName（实测 ~0.8s） + - 应用名 label：常见应用走映射；用户安装且较小(<=10MB)的未知应用流式拉 APK 用 + pyaxmlparser 解析（带缓存），其余用包名兜底。 + 返回 (ok, apps) 或 (False, 错误消息)。 + """ + if not serial: + return False, "缺少 serial" + try: + # 1. 包名 + versionCode + APK 路径 + r1 = subprocess.run( + [ADB_PATH, "-s", serial, "shell", "pm", "list", "packages", + "--show-versioncode", "-f"], + capture_output=True, timeout=timeout) + apps = {} + for line in (r1.stdout or b"").decode("utf-8", errors="replace").splitlines(): + line = line.strip() + m = re.match(r'package:(.+)=(\S+) versionCode:(\d+)', line) + if m: + apps[m.group(2)] = {"package": m.group(2), "path": m.group(1), + "version_code": int(m.group(3)), "version_name": "", + "label": COMMON_APPS.get(m.group(2), "")} + else: + m2 = re.match(r'package:(.+)=(\S+)', line) + if m2: + apps[m2.group(2)] = {"package": m2.group(2), "path": m2.group(1), + "version_code": 0, "version_name": "", + "label": COMMON_APPS.get(m2.group(2), "")} + # 2. 人性化版本号 versionName + try: + r2 = subprocess.run( + [ADB_PATH, "-s", serial, "shell", "pm", "dump", "packages"], + capture_output=True, timeout=timeout) + out2 = (r2.stdout or b"").decode("utf-8", errors="replace") + for m in re.finditer(r'Package \[([^\]]+)\]', out2): + pkg = m.group(1) + if pkg not in apps: + continue + nxt = out2.find("Package [", m.end()) + seg = out2[m.end():nxt if nxt != -1 else len(out2)] + vm = re.search(r'versionName=(\S+)', seg) + if vm: + apps[pkg]["version_name"] = vm.group(1) + except Exception: + pass # versionName 拿不到不阻塞 + # 3. 应用名 label：用户安装且未映射的小 APK，流式拉取解析真实名 + _resolve_app_labels(serial, apps, timeout) + return True, sorted(apps.values(), key=lambda a: a["package"]) + except subprocess.TimeoutExpired: + return False, "查询超时" + except Exception as e: + return False, str(e) + + +def _resolve_app_labels(serial, apps, timeout): + """并行解析用户安装小应用的显示名 label（带缓存）。失败/超时静默跳过。""" + import io as _io + from concurrent.futures import ThreadPoolExecutor + targets = [(pkg, a["path"]) for pkg, a in apps.items() + if a["path"].startswith("/data/") and not a["label"] + and (serial, pkg) not in _APP_LABEL_CACHE] + if not targets: + return + # 用 ls -l 一次拿所有候选 APK 大小，跳过过大的（流式拉大 APK 太慢） + sizes = {} + try: + rsz = subprocess.run( + [ADB_PATH, "-s", serial, "shell", "ls", "-l"] + [p for _, p in targets], + capture_output=True, timeout=min(timeout, 15)) + for line in (rsz.stdout or b"").decode("utf-8", errors="replace").splitlines(): + parts = line.split() + # ls -l(toybox): 权限 链接 属主 属组 大小 日期 时间 路径 = 8 字段 + if len(parts) >= 6: + try: + sizes[parts[-1]] = int(parts[4]) + except ValueError: + pass + except Exception: + pass + small = [(pkg, path) for pkg, path in targets if sizes.get(path, 0) <= _LABEL_PARSE_MAX_BYTES] + + def _fetch(pkg_path): + pkg, path = pkg_path + label = "" + try: + rp = subprocess.run( + [ADB_PATH, "-s", serial, "exec-out", "cat", path], + capture_output=True, timeout=min(timeout, 8)) + from pyaxmlparser import APK as _APKParser + label = (_APKParser(_io.BytesIO(rp.stdout)).get_app_name() or "").strip() + except Exception: + label = "" + _APP_LABEL_CACHE[(serial, pkg)] = label + return pkg, label + + if small: + try: + with ThreadPoolExecutor(max_workers=8) as pool: + for pkg, label in pool.map(_fetch, small): + if pkg in apps: + apps[pkg]["label"] = label + except Exception: + pass # label 解析失败不影响主流程 + + def identify_device(serial, duration=3): """让设备响铃以供识别（直连 serial，不经过 STF）。 diff --git a/web_server.py b/web_server.py index b2842f0..16d7fcd 100644 --- a/web_server.py +++ b/web_server.py @@ -29,7 +29,7 @@ from core.task_manager import TaskManager from core.logger import get_logger, _LOG_DIR, _MODULE_FILES from core.models import db, init_db, User, DeviceGroup, TaskJob, CustomAction from core.apk_manager import ApkManager -from core.adb_helper import screenshot +from core.adb_helper import screenshot, list_installed_apps from core import uiauto_helper from tasks import list_task_types, get_task_class @@ -370,6 +370,16 @@ def api_device_screenshot(): return jsonify({"ok": False, "error": data}), 500 +@app.route("/api/devices//apps") +@login_required +def api_device_apps(serial): + """获取指定设备上已安装的应用列表（包名 + versionCode + versionName + 路径）。""" + ok, data = list_installed_apps(serial) + if ok: + return jsonify({"ok": True, "apps": data}) + return jsonify({"ok": False, "error": data}), 500 + + # ================== 自定义动作（步骤打包） ================== @app.route("/api/custom_actions") @login_required