111 lines
3.7 KiB
Python
111 lines
3.7 KiB
Python
"""MCP 直连操作封装：adb/u2/OCR/剪贴板——轻量通道优先。
|
||
|
||
为什么直连而不是全部走平台 HTTP：
|
||
- adb monkey 打开 App / am force-stop：一条 adb 命令，不建 u2 连接（省时省 token）
|
||
- 输入文字：u2 EditText.set_text 直接设文本（比剪贴板+粘贴少依赖）
|
||
- OCR：平台 RapidOCR（本地模型，截图即识别）
|
||
平台 REST 无法表达的操作（无独立端点）在此封装；安全（白名单/写门控）
|
||
仍在 MCP 工具层统一把关。绝不 kill-server / 绝不 disconnect（项目红线）。
|
||
"""
|
||
import re
|
||
import subprocess
|
||
|
||
from config import ADB_PATH
|
||
|
||
_log = None # mcp 层 logging 由调用方配置
|
||
|
||
|
||
def _adb(serial, *args, timeout=25):
|
||
"""对指定设备执行 adb shell 命令，返回输出文本（超时返回空）。"""
|
||
try:
|
||
r = subprocess.run([ADB_PATH, "-s", serial, "shell", *args],
|
||
capture_output=True, timeout=timeout)
|
||
return (r.stdout or b"").decode("utf-8", errors="replace")
|
||
except subprocess.TimeoutExpired:
|
||
return ""
|
||
except Exception:
|
||
return ""
|
||
|
||
|
||
def open_app(serial, package):
|
||
"""adb monkey 打开 App（无需知道 activity，最轻量）。"""
|
||
out = _adb(serial, "monkey", "-p", package,
|
||
"-c", "android.intent.category.LAUNCHER", "1")
|
||
return out or ""
|
||
|
||
|
||
def stop_app(serial, package):
|
||
"""强制停止 App（am force-stop）。"""
|
||
return _adb(serial, "am", "force-stop", package)
|
||
|
||
|
||
def foreground_app(serial):
|
||
"""当前前台 App 包名（dumpsys window；mCurrentFocus 为空时 fallback mFocusedApp）。"""
|
||
out = _adb(serial, "dumpsys", "window", timeout=15) or ""
|
||
m = re.search(r"mCurrentFocus=.*?([\w.]+)/", out)
|
||
if m:
|
||
return m.group(1)
|
||
# 部分 MIUI 焦点在 IME/过渡时 mCurrentFocus=null，用 mFocusedApp 兜底
|
||
m2 = re.search(r"mFocusedApp=.*?([\w.]+)/", out)
|
||
return m2.group(1) if m2 else ""
|
||
|
||
|
||
def list_apps(serial, keyword=""):
|
||
"""第三方已装应用包名列表（pm list packages -3，可关键词过滤）。"""
|
||
out = _adb(serial, "pm", "list", "packages", "-3")
|
||
pkgs = []
|
||
for line in (out or "").splitlines():
|
||
p = line.replace("package:", "").strip()
|
||
if p and (not keyword or keyword.lower() in p.lower()):
|
||
pkgs.append(p)
|
||
return pkgs
|
||
|
||
|
||
def type_text(serial, text):
|
||
"""向当前界面输入框输入文字（u2 定位 EditText set_text，支持中文）。
|
||
|
||
返回 (ok, msg)。
|
||
"""
|
||
import uiautomator2 as u2
|
||
d = u2.connect(serial)
|
||
# 优先聚焦输入框；找不到则第一个 EditText（与任务 input_text 兜底一致）
|
||
try:
|
||
el = d(focused=True)
|
||
if el.exists:
|
||
el.set_text(text)
|
||
return True, "已输入到聚焦输入框"
|
||
except Exception:
|
||
pass
|
||
try:
|
||
edit = d(className="android.widget.EditText")
|
||
if edit.exists:
|
||
edit.set_text(text)
|
||
return True, "已输入到输入框"
|
||
except Exception as e:
|
||
return False, f"输入失败: {type(e).__name__}: {str(e)[:100]}"
|
||
return False, "未找到输入框（请先点击输入框或提供界面信息）"
|
||
|
||
|
||
def set_clipboard(serial, text):
|
||
"""剪贴板注入（ClipInject 通道，读回验证）。"""
|
||
from core.clipboard_helper import inject_clipboard
|
||
return inject_clipboard(serial, text)
|
||
|
||
|
||
def ocr(serial):
|
||
"""截屏 + RapidOCR 识别，返回 [{text, score, box}]。"""
|
||
import uiautomator2 as u2
|
||
from core.ocr import recognize
|
||
d = u2.connect(serial)
|
||
img = d.screenshot()
|
||
if img is None:
|
||
return []
|
||
return recognize(img)
|
||
|
||
|
||
def read_clipboard(serial):
|
||
"""读设备剪贴板（u2）。"""
|
||
import uiautomator2 as u2
|
||
d = u2.connect(serial)
|
||
return d.clipboard or ""
|