"""监控域 API：状态/运行控制/设备操作/远程看屏。""" import time import threading import re import subprocess import shlex from concurrent.futures import ThreadPoolExecutor, as_completed from flask import (Blueprint, jsonify, request, Response, render_template) from flask_login import login_required from core import device_pool from core.adb_helper import (screenshot, list_installed_apps, adb_connect, adb_connect_light, _ADB_LOCK, _adb) from core.device_worker import (get_all_worker_status, clear_worker_error, clear_all_worker_errors) from web import context from web.auth import perm_required, PERM_DEVICES from web.common import _merged_device_list, _device_screen_state from config import ADB_PATH from core.logger import get_logger _log = get_logger("web") bp = Blueprint("monitor", __name__) # 远程看屏参数 _SCREEN_JPEG_QUALITY = 65 _SCREEN_FRAME_GAP = 0.05 @bp.route("/api/health") def api_health(): """轻量健康检查：返回进程/设备/任务摘要，不暴露敏感信息，供运维探活。""" try: devices, err = context.mgr.get_status() devs = devices or [] return jsonify({ "ok": True, "status": "up", "time": time.time(), "device_total": len(devs), "device_online": sum(1 for d in devs if d.get("present")), "device_running": sum(1 for d in devs if d.get("worker_status") in ("running", "connecting")), "device_error": sum(1 for d in devs if d.get("worker_status") in ("error", "failed")), "jobs": len(context.mgr.jobs), }) except Exception as e: return jsonify({"ok": False, "status": "down", "error": str(e)}), 500 # ================== API：失败任务汇总 ================== @bp.route("/api/summary") @login_required def api_summary(): """失败/异常任务汇总：各状态计数 + 异常设备列表（last_error / 重试次数）。 供监控页"异常汇总"面板使用，便于长期运行观察设备健康度。 """ try: sts = get_all_worker_status() # 已删除/不在当前 STF 设备池的设备，其陈旧失败记录不应展示。 # 复用 get_status 缓存（顺带触发上面的状态清理），STF 异常时不隐藏任何记录。 try: devices, derr = context.mgr.get_status() if not derr: present = {d.get("serial") for d in devices} sts = [s for s in sts if s.get("serial") in present] except Exception: pass counts = {"total": 0, "running": 0, "done": 0, "error": 0, "failed": 0, "idle": 0} errors = [] for s in sts: st = s.get("status", "idle") counts["total"] += 1 if st in counts: counts[st] += 1 if st in ("error", "failed") and s.get("last_error"): errors.append({ "serial": s.get("serial"), "model": s.get("model", ""), "status": st, "last_error": s.get("last_error", ""), "task": s.get("task_job", ""), "attempt": s.get("attempt", 0), "updated": s.get("last_heartbeat", 0), }) errors.sort(key=lambda x: x.get("updated", 0), reverse=True) return jsonify({"ok": True, "counts": counts, "errors": errors[:50]}) except Exception as e: return jsonify({"ok": False, "error": str(e)}), 500 # ================== API：状态（监控大屏用）================== @bp.route("/api/status") @login_required def api_status(): devices, err = context.mgr.get_status() if err: return jsonify({"ok": False, "error": err}), 500 return jsonify({ "ok": True, "devices": devices, "server_time": time.time(), "fg_scanning": context.mgr._fg_scanner.is_scanning, "fg_last_scan": context.mgr._fg_scanner.last_scan_time, }) @bp.route("/api/scan_foreground", methods=["POST"]) @perm_required(PERM_DEVICES) def api_scan_foreground(): """手动触发前台 App 扫描（不打扰设备）。""" started = context.mgr._fg_scanner.scan_once() if started: return jsonify({"ok": True, "msg": "扫描已启动"}) return jsonify({"ok": False, "error": "已有扫描在进行中"}) @bp.route("/api/stop_device", methods=["POST"]) @perm_required(PERM_DEVICES) def api_stop_device(): serial = (request.json or {}).get("serial", "") if context.mgr.stop_device(serial): return jsonify({"ok": True, "msg": f"已发送停止信号给 {serial}"}) return jsonify({"ok": False, "error": f"{serial} 没有运行中的任务"}), 400 @bp.route("/locate") def locate_page(): """设备端定位页（免登录）：点亮后浏览器打开，大字显示设备 IP/序列号。 只展示 serial 文本，无任何平台数据，免登录可接受。 """ serial = request.args.get("serial", "") return render_template_string( """ 设备定位
{{ serial }}
{{ ip }}
定位完成 · 按返回键退出
""", serial=_esc(serial), ip=_esc(serial.split(":")[0])) # 常见浏览器包名（结束定位时 force-stop 用） _BROWSER_PKGS = {"com.android.browser", "com.miui.browser", "com.android.chrome", "com.brave.browser", "com.opera.browser", "org.mozilla.firefox", "com.UCMobile", "com.tencent.mtt", "com.baidu.browser.apps"} # 定位时启动的浏览器：serial -> 包名（结束定位时无论前后台都关闭它） _locate_browsers = {} _locate_lock = threading.Lock() def _current_focus_pkg(serial): """取设备当前前台包名（dumpsys window 解析，失败返回空串）。""" try: r = subprocess.run([ADB_PATH, "-s", serial, "shell", "dumpsys", "window"], capture_output=True, timeout=20) out = (r.stdout or b"").decode("utf-8", errors="replace") m = re.search(r"mCurrentFocus=.*?([\w.]+)/", out) return m.group(1) if m else "" except Exception: return "" @bp.route("/api/device/locate/stop", methods=["POST"]) @perm_required(PERM_DEVICES) def api_device_locate_stop(): """结束定位：关闭定位启动的浏览器（无论当前是否前台）。 优先 force-stop 定位时记录下的浏览器包名（切到后台也能关掉）； 无记录时退回：前台是浏览器 → force-stop；否则按返回键（不误杀任务应用）。 """ serial = str((request.json or {}).get("serial", "")).strip() if not serial: return jsonify({"ok": False, "error": "缺少设备 serial"}), 400 try: if ":" in serial: adb_connect(serial) # 1. 优先关闭定位时启动的浏览器（无论前后台） with _locate_lock: started = _locate_browsers.pop(serial, None) if started: subprocess.run([ADB_PATH, "-s", serial, "shell", "am", "force-stop", started], capture_output=True, timeout=20) msg = f"已关闭定位浏览器 {started}" else: # 2. 退回：前台是浏览器则关闭，否则返回键轻量退出 pkg = _current_focus_pkg(serial) if pkg in _BROWSER_PKGS: subprocess.run([ADB_PATH, "-s", serial, "shell", "am", "force-stop", pkg], capture_output=True, timeout=20) msg = f"已关闭浏览器 {pkg}" else: subprocess.run([ADB_PATH, "-s", serial, "shell", "input", "keyevent", "4"], capture_output=True, timeout=15) # BACK 轻量退出 msg = "已按返回键退出" + (f"（前台 {pkg}）" if pkg else "") _log.info(f"结束定位 {serial}: {msg}") return jsonify({"ok": True, "msg": msg}) except Exception as e: return jsonify({"ok": False, "error": f"结束定位失败: {e}"}), 502 @bp.route("/api/device/screen_all", methods=["POST"]) @perm_required(PERM_DEVICES) def api_device_screen_all(): """批量亮屏/息屏（并发）。mode: on=亮屏解锁 / off=息屏。 请求可带 serials 指定设备（前端勾选）；不带则作用于全部在线设备。 息屏会中断运行中的任务（u2 无法操作），前端已有确认提示。 """ data = request.json or {} mode = data.get("mode", "on") try: online = set(device_pool.list_online()) except Exception as e: return jsonify({"ok": False, "error": f"获取设备列表失败: {e}"}), 502 serials = data.get("serials") or [] if serials: # 指定设备：过滤掉不在线的（离线设备操作无意义，直接跳过） serials = [s for s in serials if s in online] else: serials = list(online) if not serials: return jsonify({"ok": False, "error": "无在线设备"}), 400 def _do(serial): try: if ":" in serial: adb_connect(serial) if mode == "off": subprocess.run([ADB_PATH, "-s", serial, "shell", "input", "keyevent", "26"], capture_output=True, timeout=15) # KEYCODE_POWER 息屏 else: subprocess.run([ADB_PATH, "-s", serial, "shell", "input", "keyevent", "224"], capture_output=True, timeout=15) # WAKEUP subprocess.run([ADB_PATH, "-s", serial, "shell", "wm", "dismiss-keyguard"], capture_output=True, timeout=15) return True except Exception: return False with ThreadPoolExecutor(max_workers=min(10, len(serials))) as pool: results = list(pool.map(_do, serials)) ok_n = sum(1 for x in results if x) _log.info(f"批量{'息屏' if mode == 'off' else '亮屏'}: 成功 {ok_n}/{len(serials)}") return jsonify({"ok": True, "success": ok_n, "total": len(serials)}) @bp.route("/api/device/locate", methods=["POST"]) @perm_required(PERM_DEVICES) def api_device_locate(): """定位设备：点亮屏幕并解除锁屏，可选打开大字定位页（查找设备用）。 请求 {"serial": "...", "show": true}：show=true 时额外用浏览器打开 平台 /locate 定位页，全屏大字显示设备 IP（更醒目，但会切换前台， 任务运行中慎用）。点亮组合：WAKEUP → dismiss-keyguard → MENU 兜底。 绝不 disconnect，红线。 """ data = request.json or {} serial = str(data.get("serial", "")).strip() if not serial: return jsonify({"ok": False, "error": "缺少设备 serial"}), 400 try: if ":" in serial: adb_connect(serial) for cmd in (["shell", "input", "keyevent", "224"], ["shell", "wm", "dismiss-keyguard"], ["shell", "input", "keyevent", "82"]): try: subprocess.run([ADB_PATH, "-s", serial, *cmd], capture_output=True, timeout=15) except Exception: pass # 单步失败不阻塞，继续下一步 msgs = ["屏幕已点亮"] if data.get("show"): # 用设备浏览器打开定位页（设备走 tailnet 访问本机 100.100.10.2:18050） locate_url = f"http://100.100.10.2:18050/locate?serial={urllib.parse.quote(serial)}" r = subprocess.run([ADB_PATH, "-s", serial, "shell", "am", "start", "-a", "android.intent.action.VIEW", "-d", locate_url], capture_output=True, timeout=20) out = ((r.stdout or b"") + (r.stderr or b"")).decode("utf-8", errors="replace") if "error" in out.lower() or "exception" in out.lower(): msgs.append(f"打开定位页失败: {out.strip()[:80]}") else: msgs.append("已打开大字定位页（按返回键退出）") # 记录启动的浏览器包名：结束定位时无论前后台都关闭它。 # 优先从 am start 输出解析 pkg=（浏览器可能尚未到前台，前台检测会扑空） pkg = "" m2 = re.search(r"pkg=([\w.]+)", out) if m2 and m2.group(1) in _BROWSER_PKGS: pkg = m2.group(1) else: time.sleep(1.5) # 等浏览器切到前台再查 pkg = _current_focus_pkg(serial) if pkg in _BROWSER_PKGS: with _locate_lock: _locate_browsers[serial] = pkg _log.info(f"定位设备 {serial}: {'; '.join(msgs)}") return jsonify({"ok": True, "msg": f"{serial} {'；'.join(msgs)}"}) except Exception as e: return jsonify({"ok": False, "error": f"定位失败: {e}"}), 502 @bp.route("/api/stop_all", methods=["POST"]) @perm_required(PERM_DEVICES) def api_stop_all(): stopped = context.mgr.stop_all() return jsonify({"ok": True, "stopped": stopped}) @bp.route("/api/device/clear_error", methods=["POST"]) @perm_required(PERM_DEVICES) def api_device_clear_error(): """清除单台设备的异常状态（error/failed → idle），供设备列表"清除异常"按钮使用。""" serial = (request.json or {}).get("serial", "") if not serial: return jsonify({"ok": False, "error": "缺少 serial"}), 400 if serial in context.mgr.get_running(): return jsonify({"ok": False, "error": "设备正在运行或等待重试，无法清除"}), 400 if not clear_worker_error(serial): return jsonify({"ok": False, "error": "设备正在运行任务，无法清除"}), 400 _log.info(f"清除设备 {serial} 的异常状态") return jsonify({"ok": True, "msg": f"已清除 {serial} 的异常状态"}) @bp.route("/api/device/clear_all_errors", methods=["POST"]) @perm_required(PERM_DEVICES) def api_device_clear_all_errors(): """一键清除所有异常/失败设备（跳过正在运行/等待重试的）。""" running = set(context.mgr.get_running()) cleared = clear_all_worker_errors(exclude=running) _log.info(f"一键清除异常：共清除 {cleared} 台") return jsonify({"ok": True, "cleared": cleared, "msg": f"已清除 {cleared} 台设备的异常状态"}) @bp.route("/api/device/screenshot") @login_required def api_device_screenshot(): """获取设备当前画面截图（PNG）。 用 adb exec-out screencap -p，只读操作，不抢占 u2 的 atx-agent 通道， 任务运行中调用安全。直接返回 image/png，前端用 加载。 ?serial=xxx 设备 serial ?t=123 时间戳，避免浏览器缓存（前端自动加） """ serial = request.args.get("serial", "") if not serial: return jsonify({"ok": False, "error": "缺少 serial"}), 400 ok, data = screenshot(serial) if ok: return Response(data, mimetype="image/png", headers={"Cache-Control": "no-store"}) return jsonify({"ok": False, "error": data}), 500 @bp.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 # ================== 自定义动作（步骤打包） ================== @bp.route("/api/screen/stream") @perm_required(PERM_DEVICES) def api_screen_stream(): """远程看屏：MJPEG 实时画面流（multipart/x-mixed-replace）。 ?serial=xxx 设备 serial 浏览器 直接渲染；客户端断开（GeneratorExit）自动停止，不占资源。 设备离线/atx-agent 无响应时流自然结束，前端提示重新连接。 """ serial = request.args.get("serial", "") if not serial: return jsonify({"ok": False, "error": "缺少 serial"}), 400 # 参数：q=JPEG 质量（大屏弹窗用 85 更清晰）；fps=目标帧率（0=默认节流） try: quality = max(30, min(95, int(request.args.get("q", _SCREEN_JPEG_QUALITY)))) fps = float(request.args.get("fps", 0) or 0) except ValueError: quality, fps = _SCREEN_JPEG_QUALITY, 0 gap = (1.0 / fps) if fps > 0 else _SCREEN_FRAME_GAP def generate(): import io import uiautomator2 as u2 try: d = u2.connect(serial) except Exception as e: _log.warning(f"远程看屏 {serial} u2 连接失败: {e}") return while True: try: img = d.screenshot() if img is None: break if img.mode != "RGB": img = img.convert("RGB") buf = io.BytesIO() img.save(buf, "JPEG", quality=quality) yield (b"--frame\r\nContent-Type: image/jpeg\r\n\r\n" + buf.getvalue() + b"\r\n") except GeneratorExit: break except Exception as e: _log.debug(f"远程看屏 {serial} 流中断: {e}") break time.sleep(gap) return Response(generate(), mimetype="multipart/x-mixed-replace; boundary=frame") _screen_u2_cache = {} # serial -> (ts, device)：大屏高频触控复用连接（60s TTL） def _screen_get_device(serial): """远程看屏用的 u2 连接（60s 缓存复用，避免每次 tap/swipe 都重建连接）。 大屏高频操作（点击/滑动/按键）每次 u2.connect 有 1-3s 握手开销（atx-agent 探测），体验卡顿。缓存 60s 内复用；连接失败（设备离线/重启）时清缓存重建， 由调用方把异常转 503。 """ import time import uiautomator2 as u2 now = time.time() cached = _screen_u2_cache.get(serial) if cached and now - cached[0] < 60: return cached[1] d = u2.connect(serial) _screen_u2_cache[serial] = (now, d) if len(_screen_u2_cache) > 30: # 惰性清理过期项 for k in list(_screen_u2_cache): if now - _screen_u2_cache[k][0] >= 60: del _screen_u2_cache[k] return d def _screen_invalidate(serial): """u2 调用失败时清掉缓存（设备离线/atx-agent 重启），下次请求重新连接。""" _screen_u2_cache.pop(serial, None) @bp.route("/api/screen/thumb") @perm_required(PERM_DEVICES) def api_screen_thumb(): """大屏缩略图：单张 JPEG（360px 宽，质量 55），供监控大屏轮询。 一次性请求（非流），与任务并发安全（与任务截图同走 u2 minicap）。 大屏按需轮询：每设备约 2.5s 一帧，20 台 ≈ 0.2 核 CPU + 100KB/s。 响应头 X-Screen-State: on/off/unknown —— 大屏据此显示「亮屏中/熄屏中」。 """ serial = request.args.get("serial", "") if not serial: return jsonify({"ok": False, "error": "缺少 serial"}), 400 import io import uiautomator2 as u2 try: d = u2.connect(serial) img = d.screenshot() if img is None: return jsonify({"ok": False, "error": "截图失败"}), 503 if img.width > 360: ratio = 360 / img.width img = img.resize((360, int(img.height * ratio))) buf = io.BytesIO() img.convert("RGB").save(buf, "JPEG", quality=55) return Response(buf.getvalue(), mimetype="image/jpeg", headers={"Cache-Control": "no-store", "X-Screen-State": _device_screen_state(serial)}) except Exception as e: return jsonify({"ok": False, "error": str(e)}), 503 def _snap_to_clickable(d, x, y): """坐标吸附：找包含 (x,y) 的最小可点击元素，返回其 bounds 中心。 AI/触控给的坐标常偏离目标 20-50px（模型视觉定位误差）。dump 当前 UI 树后 遍历 clickable 节点：点击点落在哪个可点元素内就点它的中心——偏了也点得准； 取包含元素中面积最小者（最具体的那个）。点空白处（收键盘等）无包含元素则 原坐标返回。dump 失败也不阻塞，直接原坐标。返回 (cx, cy, snapped, label)。 """ try: import xml.etree.ElementTree as ET xml_str = d.dump_hierarchy() best = None # (area, cx, cy, label) for node in ET.fromstring(xml_str).iter("node"): if node.get("clickable") != "true": continue if node.get("enabled") == "false": continue m = re.match(r"\[(\d+),(\d+)\]\[(\d+),(\d+)\]", node.get("bounds", "")) if not m: continue x1, y1, x2, y2 = map(int, m.groups()) if x1 <= x <= x2 and y1 <= y <= y2: area = (x2 - x1) * (y2 - y1) if best is None or area < best[0]: best = (area, (x1 + x2) // 2, (y1 + y2) // 2, (node.get("text") or node.get("content-desc") or "")[:40]) if best: return best[1], best[2], True, best[3] except Exception: pass return x, y, False, "" @bp.route("/api/screen/tap", methods=["POST"]) @perm_required(PERM_DEVICES) def api_screen_tap(): """点击：{serial, x, y, snap?}（设备原生分辨率坐标）。 snap=1 时先吸附：点落在可点击元素内则改点元素中心（MCP/AI 场景用，粗略 坐标也能点准）；大屏精确触控不带 snap 保持原行为。 返回 {ok, snapped, x, y, label}。 """ data = request.json or {} serial, x, y = data.get("serial", ""), data.get("x"), data.get("y") snap = int(data.get("snap", 0) or 0) == 1 if not serial or x is None or y is None: return jsonify({"ok": False, "error": "缺少 serial/x/y"}), 400 try: d = _screen_get_device(serial) sx, sy = int(x), int(y) label = "" snapped = False if snap: sx, sy, snapped, label = _snap_to_clickable(d, sx, sy) d.click(sx, sy) return jsonify({"ok": True, "snapped": snapped, "x": sx, "y": sy, "label": label}) except Exception as e: _screen_invalidate(serial) return jsonify({"ok": False, "error": f"点击失败: {e}"}), 503 @bp.route("/api/screen/tap_text", methods=["POST"]) @perm_required(PERM_DEVICES) def api_screen_tap_text(): """按屏幕文字点击：{serial, text}——找到显示该文字的位置并点中心（子串匹配）。 两步：① UI 树 textContains/descriptionContains 命中 → 点元素中心（原生控件）； ② 未命中 → 截图 OCR 找文字中心（WebView/图片/画布渲染的文字）。 返回 {ok, found, method: ui|ocr, matched, x, y}；found=false 表示屏幕确实 没有该文字（业务结果非设备错误）；OCR 不可用且 UI 未命中时也返回 found=false。 """ data = request.json or {} serial = (data.get("serial") or "").strip() text = (data.get("text") or "").strip() if not serial or not text: return jsonify({"ok": False, "error": "缺少 serial/text"}), 400 if len(text) > 100: return jsonify({"ok": False, "error": "文字过长（≤100 字符）"}), 400 try: d = _screen_get_device(serial) # ① UI 树：text / content-desc 子串匹配（原生控件最快最准） for kw in ({"textContains": text}, {"descriptionContains": text}): try: el = d(**kw) if el.exists(timeout=1.5): # u2 版本差异：3.x 部分版本 bounds 是方法（el.bounds()） b = el.bounds() if callable(el.bounds) else el.bounds el.click() return jsonify({"ok": True, "found": True, "method": "ui", "matched": text, "x": (b[0] + b[2]) // 2, "y": (b[1] + b[3]) // 2}) except Exception: continue # ② OCR 兜底：截图找文字中心（UI 树没有的渲染文字） img = d.screenshot() if img is not None: from core.ocr import find_on_screen found, center, matched = find_on_screen(img, text) if found: d.click(*center) return jsonify({"ok": True, "found": True, "method": "ocr", "matched": matched, "x": center[0], "y": center[1]}) return jsonify({"ok": True, "found": False, "method": "", "matched": "", "x": 0, "y": 0}) except Exception as e: _screen_invalidate(serial) return jsonify({"ok": False, "error": f"文字点击失败: {e}"}), 503 @bp.route("/api/screen/swipe", methods=["POST"]) @perm_required(PERM_DEVICES) def api_screen_swipe(): """滑动：{serial, x1, y1, x2, y2, duration?}（设备原生分辨率坐标）。""" data = request.json or {} serial = data.get("serial", "") x1, y1, x2, y2 = (data.get(k) for k in ("x1", "y1", "x2", "y2")) if not serial or None in (x1, y1, x2, y2): return jsonify({"ok": False, "error": "缺少 serial/x1/y1/x2/y2"}), 400 try: _screen_get_device(serial).swipe(int(x1), int(y1), int(x2), int(y2), duration=float(data.get("duration", 0.2))) return jsonify({"ok": True}) except Exception as e: _screen_invalidate(serial) return jsonify({"ok": False, "error": f"滑动失败: {e}"}), 503 _SCREEN_KEYS = {"back", "home", "recent", "menu", "power", "volume_up", "volume_down", "enter", "delete", "search", "camera"} @bp.route("/api/screen/key", methods=["POST"]) @perm_required(PERM_DEVICES) def api_screen_key(): """按键：{serial, key}（back/home/recent/menu/power 等，见 _SCREEN_KEYS）。""" data = request.json or {} serial, key = data.get("serial", ""), (data.get("key") or "").strip().lower() if not serial or key not in _SCREEN_KEYS: return jsonify({"ok": False, "error": "缺少 serial 或不支持的按键"}), 400 try: _screen_get_device(serial).press(key) return jsonify({"ok": True}) except Exception as e: _screen_invalidate(serial) return jsonify({"ok": False, "error": f"按键失败: {e}"}), 503 @bp.route("/api/screen/text", methods=["POST"]) @perm_required(PERM_DEVICES) def api_screen_text(): """输入文字：{serial, text}（需焦点在输入框，u2 send_keys）。""" data = request.json or {} serial, text = data.get("serial", ""), (data.get("text") or "").strip() if not serial or not text: return jsonify({"ok": False, "error": "缺少 serial/text"}), 400 try: _screen_get_device(serial).send_keys(text) return jsonify({"ok": True}) except Exception as e: return jsonify({"ok": False, "error": f"输入失败: {e}"}), 503 @bp.route("/api/screen/size") @perm_required(PERM_DEVICES) def api_screen_size(): """屏幕原生分辨率（只读，供 MCP 坐标换算：截图是缩放图，操作需原生坐标）。 返回 {ok, width, height}（如 1080x2400）；离线/不可达 → 503。 """ serial = request.args.get("serial", "") if not serial: return jsonify({"ok": False, "error": "缺少 serial"}), 400 try: d = _screen_get_device(serial) w, h = d.window_size() return jsonify({"ok": True, "width": w, "height": h}) except Exception as e: _screen_invalidate(serial) return jsonify({"ok": False, "error": f"获取分辨率失败: {e}"}), 503 # ================== API：应用管理（APK 上传/安装）================== def _current_focus_pkg(serial): """取设备当前前台包名（dumpsys window 解析，失败返回空串）。""" try: r = subprocess.run([ADB_PATH, "-s", serial, "shell", "dumpsys", "window"], capture_output=True, timeout=20) out = (r.stdout or b"").decode("utf-8", errors="replace") m = re.search(r"mCurrentFocus=.*?([\w.]+)/", out) return m.group(1) if m else "" except Exception: return "" _screen_u2_cache = {} # serial -> (ts, device)：大屏高频触控复用连接（60s TTL） def _screen_get_device(serial): """远程看屏用的 u2 连接（60s 缓存复用，避免每次 tap/swipe 都重建连接）。 大屏高频操作（点击/滑动/按键）每次 u2.connect 有 1-3s 握手开销（atx-agent 探测），体验卡顿。缓存 60s 内复用；连接失败（设备离线/重启）时清缓存重建， 由调用方把异常转 503。 """ import time import uiautomator2 as u2 now = time.time() cached = _screen_u2_cache.get(serial) if cached and now - cached[0] < 60: return cached[1] d = u2.connect(serial) _screen_u2_cache[serial] = (now, d) if len(_screen_u2_cache) > 30: # 惰性清理过期项 for k in list(_screen_u2_cache): if now - _screen_u2_cache[k][0] >= 60: del _screen_u2_cache[k] return d def _screen_invalidate(serial): """u2 调用失败时清掉缓存（设备离线/atx-agent 重启），下次请求重新连接。""" _screen_u2_cache.pop(serial, None)