张家训
287 lines
11 KiB
Python
287 lines
11 KiB
Python
"""安装 Airtest 辅助工具（minicap/yosemite）到设备"""
|
||
import argparse
|
||
import json
|
||
import os
|
||
import subprocess
|
||
import sys
|
||
import time
|
||
|
||
|
||
def emit(event_type, level="INFO", message="", **extra):
|
||
payload = {"type": event_type, "level": level, "message": message}
|
||
payload.update(extra)
|
||
print(json.dumps(payload, ensure_ascii=False), flush=True)
|
||
|
||
|
||
def find_adb():
|
||
"""查找 adb 可执行文件"""
|
||
venv_dir = os.path.join(os.path.dirname(__file__), "..", ".venv")
|
||
candidates = [
|
||
os.path.join(venv_dir, "Lib", "site-packages", "airtest", "core", "android", "static", "adb", "windows", "adb.exe"),
|
||
os.path.join(venv_dir, "lib", "python3.11", "site-packages", "airtest", "core", "android", "static", "adb", "linux", "adb"),
|
||
os.path.join(venv_dir, "lib", "python3.10", "site-packages", "airtest", "core", "android", "static", "adb", "linux", "adb"),
|
||
]
|
||
for c in candidates:
|
||
if os.path.exists(c):
|
||
return c
|
||
return "adb"
|
||
|
||
|
||
def install_minicap(device_uri):
|
||
"""通过 Airtest 安装 minicap 并验证"""
|
||
try:
|
||
from airtest.core.api import auto_setup
|
||
import logging
|
||
logging.getLogger("airtest").setLevel(logging.INFO)
|
||
|
||
# 提取 serial
|
||
serial = device_uri.replace("Android:///", "").strip()
|
||
serial = serial.split("?")[0] if "?" in serial else serial
|
||
|
||
emit("log", "INFO", "正在连接设备并安装 minicap...")
|
||
# 不使用 cap_method=adbcap，让 Airtest 尝试安装 minicap
|
||
auto_setup(__file__, devices=[device_uri])
|
||
|
||
# 等待安装完成
|
||
time.sleep(3)
|
||
|
||
# 通过 ADB 检查 minicap 二进制文件是否存在于设备上
|
||
adb = find_adb()
|
||
check_cmd = [adb]
|
||
if serial:
|
||
check_cmd.extend(["-s", serial])
|
||
check_cmd.extend(["shell", "ls", "/data/local/tmp/minicap"])
|
||
|
||
emit("log", "INFO", "检查 minicap 是否已安装到设备...")
|
||
result = subprocess.run(check_cmd, capture_output=True, text=True, timeout=10)
|
||
combined = (result.stdout + result.stderr).strip()
|
||
|
||
if "No such file" in combined or "not found" in combined.lower():
|
||
emit("log", "WARN", "minicap 二进制文件未找到，安装失败")
|
||
return False
|
||
|
||
# minicap 文件存在，进一步验证是否能正常运行
|
||
# 尝试启动 minicap 并立即停止
|
||
test_cmd = [adb]
|
||
if serial:
|
||
test_cmd.extend(["-s", serial])
|
||
test_cmd.extend(["shell", "LD_LIBRARY_PATH=/data/local/tmp", "/data/local/tmp/minicap", "-i"])
|
||
|
||
emit("log", "INFO", "验证 minicap 是否可正常运行...")
|
||
try:
|
||
test_result = subprocess.run(test_cmd, capture_output=True, text=True, timeout=5)
|
||
test_output = (test_result.stdout + test_result.stderr).strip()
|
||
if "pid=" in test_output or "web socket" in test_output.lower() or result.returncode == 0:
|
||
emit("log", "INFO", "minicap 安装成功且可正常运行")
|
||
return True
|
||
else:
|
||
emit("log", "WARN", f"minicap 运行测试输出: {test_output}")
|
||
# 即使测试输出不明确，文件存在也算安装成功
|
||
return True
|
||
except subprocess.TimeoutExpired:
|
||
# minicap 作为服务运行会超时，这是正常的
|
||
emit("log", "INFO", "minicap 安装成功（服务进程已启动）")
|
||
return True
|
||
|
||
except Exception as e:
|
||
emit("log", "ERROR", f"安装 minicap 失败: {e}")
|
||
return False
|
||
|
||
|
||
def install_yosemite(device_uri):
|
||
"""安装 Yosemite APK（用于 javacap 截图）"""
|
||
try:
|
||
from airtest.core.api import auto_setup, connect_device
|
||
import logging
|
||
logging.getLogger("airtest").setLevel(logging.INFO)
|
||
|
||
serial = device_uri.replace("Android:///", "").strip()
|
||
serial = serial.split("?")[0] if "?" in serial else serial
|
||
|
||
emit("log", "INFO", "正在连接设备...")
|
||
auto_setup(__file__, devices=[device_uri])
|
||
time.sleep(1)
|
||
|
||
# 通过 Airtest 的 Android 设备对象直接安装 Yosemite
|
||
from airtest.core.helper import G
|
||
device = G.DEVICE
|
||
|
||
emit("log", "INFO", "正在安装 Yosemite APK...")
|
||
# 使用设备对象的 yosemite 安装方法
|
||
try:
|
||
from airtest.core.android.yosemite import Yosemite
|
||
yosemite = Yosemite(device.adb)
|
||
yosemite.install_or_upgrade()
|
||
emit("log", "INFO", "Yosemite 安装成功")
|
||
return True
|
||
except TypeError:
|
||
# 旧版本 API 兼容
|
||
from airtest.core.android.yosemite import Yosemite
|
||
yosemite = Yosemite(device)
|
||
yosemite.install_or_upgrade()
|
||
emit("log", "INFO", "Yosemite 安装成功")
|
||
return True
|
||
except Exception as e:
|
||
emit("log", "WARN", f"Yosemite 安装失败: {e}")
|
||
return False
|
||
|
||
|
||
def main():
|
||
parser = argparse.ArgumentParser(description="安装 Airtest 辅助工具")
|
||
parser.add_argument("--device", default="Android:///")
|
||
parser.add_argument("--tool", default="minicap", choices=["minicap", "yosemite", "all"])
|
||
parser.add_argument("--marker-dir", required=True, help="标记文件存放目录")
|
||
args = parser.parse_args()
|
||
|
||
os.makedirs(args.marker_dir, exist_ok=True)
|
||
# 从设备 URI 中提取 serial 用于标记文件名
|
||
serial = args.device.replace("Android:///", "").strip()
|
||
serial = serial.split("?")[0] if "?" in serial else serial
|
||
serial = serial if serial else "default"
|
||
safe_serial = serial.replace(":", "_").replace("/", "_")
|
||
|
||
results = {"minicap": False, "yosemite": False}
|
||
|
||
if args.tool in ("minicap", "all"):
|
||
results["minicap"] = install_minicap(args.device)
|
||
|
||
if args.tool in ("yosemite", "all"):
|
||
results["yosemite"] = install_yosemite(args.device)
|
||
|
||
# 写入标记文件
|
||
for tool_name, success in results.items():
|
||
marker_file = os.path.join(args.marker_dir, f".{tool_name}-{safe_serial}")
|
||
if success:
|
||
with open(marker_file, "w") as f:
|
||
f.write(json.dumps({"installed": True, "time": time.strftime("%Y-%m-%d %H:%M:%S")}))
|
||
emit("log", "INFO", f"{tool_name} 安装成功，已写入标记文件")
|
||
else:
|
||
# 如果失败，删除已有标记文件
|
||
if os.path.exists(marker_file):
|
||
os.remove(marker_file)
|
||
emit("log", "WARN", f"{tool_name} 安装失败，已清除标记")
|
||
|
||
# 输出最终结果
|
||
any_success = any(results.values())
|
||
emit("result", "INFO" if any_success else "ERROR",
|
||
json.dumps(results, ensure_ascii=False),
|
||
success=any_success)
|
||
return 0 if any_success else 1
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|
||
"""安装 Airtest 辅助工具（minicap/yosemite）到设备"""
|
||
import argparse
|
||
import json
|
||
import os
|
||
import sys
|
||
import time
|
||
|
||
|
||
def emit(event_type, level="INFO", message="", **extra):
|
||
payload = {"type": event_type, "level": level, "message": message}
|
||
payload.update(extra)
|
||
print(json.dumps(payload, ensure_ascii=False), flush=True)
|
||
|
||
|
||
def install_minicap(device_uri):
|
||
"""通过 Airtest 安装 minicap 并验证"""
|
||
try:
|
||
from airtest.core.api import auto_setup, snapshot
|
||
from airtest.core.android import Android as AndroidDevice
|
||
import logging
|
||
logging.getLogger("airtest").setLevel(logging.INFO)
|
||
|
||
emit("log", "INFO", "正在连接设备...")
|
||
# 不使用 cap_method=adbcap，让 Airtest 尝试安装 minicap
|
||
auto_setup(__file__, devices=[device_uri])
|
||
|
||
emit("log", "INFO", "设备已连接，正在初始化截图组件...")
|
||
# Airtest 在 auto_setup 时会自动安装 minicap，
|
||
# 尝试截图来验证是否安装成功
|
||
time.sleep(2)
|
||
test_path = os.path.join(os.path.dirname(__file__), "..", "test_minicap.png")
|
||
try:
|
||
snapshot(filename=test_path)
|
||
if os.path.exists(test_path) and os.path.getsize(test_path) > 100:
|
||
os.remove(test_path)
|
||
emit("log", "INFO", "minicap 安装成功，截图验证通过")
|
||
return True
|
||
else:
|
||
emit("log", "WARN", "截图文件过小，minicap 可能未正确安装")
|
||
if os.path.exists(test_path):
|
||
os.remove(test_path)
|
||
return False
|
||
except Exception as e:
|
||
emit("log", "WARN", f"minicap 截图测试失败: {e}")
|
||
return False
|
||
except Exception as e:
|
||
emit("log", "ERROR", f"安装 minicap 失败: {e}")
|
||
return False
|
||
|
||
|
||
def install_yosemite(device_uri):
|
||
"""安装 Yosemite APK（用于 javacap 截图）"""
|
||
try:
|
||
from airtest.core.api import auto_setup
|
||
from airtest.core.android.yosemite import Yosemite
|
||
import logging
|
||
logging.getLogger("airtest").setLevel(logging.INFO)
|
||
|
||
emit("log", "INFO", "正在安装 Yosemite APK...")
|
||
auto_setup(__file__, devices=[device_uri])
|
||
yosemite = Yosemite()
|
||
yosemite.install_or_upgrade()
|
||
emit("log", "INFO", "Yosemite 安装成功")
|
||
return True
|
||
except Exception as e:
|
||
emit("log", "WARN", f"Yosemite 安装失败: {e}")
|
||
return False
|
||
|
||
|
||
def main():
|
||
parser = argparse.ArgumentParser(description="安装 Airtest 辅助工具")
|
||
parser.add_argument("--device", default="Android:///")
|
||
parser.add_argument("--tool", default="minicap", choices=["minicap", "yosemite", "all"])
|
||
parser.add_argument("--marker-dir", required=True, help="标记文件存放目录")
|
||
args = parser.parse_args()
|
||
|
||
os.makedirs(args.marker_dir, exist_ok=True)
|
||
# 从设备 URI 中提取 serial 用于标记文件名
|
||
serial = args.device.replace("Android:///", "").strip()
|
||
serial = serial.split("?")[0] if "?" in serial else serial
|
||
serial = serial if serial else "default"
|
||
safe_serial = serial.replace(":", "_").replace("/", "_")
|
||
|
||
results = {"minicap": False, "yosemite": False}
|
||
|
||
if args.tool in ("minicap", "all"):
|
||
results["minicap"] = install_minicap(args.device)
|
||
|
||
if args.tool in ("yosemite", "all"):
|
||
results["yosemite"] = install_yosemite(args.device)
|
||
|
||
# 写入标记文件
|
||
for tool_name, success in results.items():
|
||
marker_file = os.path.join(args.marker_dir, f".{tool_name}-{safe_serial}")
|
||
if success:
|
||
with open(marker_file, "w") as f:
|
||
f.write(json.dumps({"installed": True, "time": time.strftime("%Y-%m-%d %H:%M:%S")}))
|
||
emit("log", "INFO", f"{tool_name} 安装成功，已写入标记文件")
|
||
else:
|
||
# 如果失败，删除已有标记文件
|
||
if os.path.exists(marker_file):
|
||
os.remove(marker_file)
|
||
|
||
# 输出最终结果
|
||
any_success = any(results.values())
|
||
emit("result", "INFO" if any_success else "ERROR",
|
||
json.dumps(results, ensure_ascii=False),
|
||
success=any_success)
|
||
return 0 if any_success else 1
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|