535 lines
23 KiB
Python
535 lines
23 KiB
Python
"""监控域 API：状态/运行控制/设备操作/远程看屏。"""
|
||
import time
|
||
import threading
|
||
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
|
||
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(
|
||
"""<!DOCTYPE html><html><head><meta charset="utf-8">
|
||
<meta name="viewport" content="width=device-width,initial-scale=1">
|
||
<title>设备定位</title><style>
|
||
body{margin:0;background:#000;color:#fff;display:flex;flex-direction:column;
|
||
align-items:center;justify-content:center;height:100vh;font-family:monospace}
|
||
.serial{font-size:min(9vw,72px);font-weight:700;color:#ffd700;word-break:break-all;padding:0 20px;text-align:center}
|
||
.ip{font-size:min(5vw,36px);color:#7dd3fc;margin-top:24px;word-break:break-all;padding:0 20px;text-align:center}
|
||
.hint{font-size:min(3vw,16px);color:#64748b;margin-top:40px}
|
||
</style></head><body>
|
||
<div class="serial">{{ serial }}</div>
|
||
<div class="ip">{{ ip }}</div>
|
||
<div class="hint">定位完成 · 按返回键退出</div>
|
||
</body></html>""",
|
||
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，前端用 <img> 加载。
|
||
?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/<serial>/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
|
||
浏览器 <img> 直接渲染；客户端断开（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")
|
||
|
||
|
||
def _screen_get_device(serial):
|
||
"""远程看屏用的 u2 连接（连接失败抛异常由调用方转 503）。"""
|
||
import uiautomator2 as u2
|
||
return u2.connect(serial)
|
||
|
||
@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
|
||
|
||
|
||
@bp.route("/api/screen/tap", methods=["POST"])
|
||
@perm_required(PERM_DEVICES)
|
||
def api_screen_tap():
|
||
"""点击：{serial, x, y}（设备原生分辨率坐标）。"""
|
||
data = request.json or {}
|
||
serial, x, y = data.get("serial", ""), data.get("x"), data.get("y")
|
||
if not serial or x is None or y is None:
|
||
return jsonify({"ok": False, "error": "缺少 serial/x/y"}), 400
|
||
try:
|
||
_screen_get_device(serial).click(int(x), int(y))
|
||
return jsonify({"ok": True})
|
||
except Exception as e:
|
||
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:
|
||
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:
|
||
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
|
||
|
||
|
||
# ================== 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 ""
|
||
|
||
def _screen_get_device(serial):
|
||
"""远程看屏用的 u2 连接（连接失败抛异常由调用方转 503）。"""
|
||
import uiautomator2 as u2
|
||
return u2.connect(serial)
|
||
|