297 lines
11 KiB
Python
297 lines
11 KiB
Python
#[title: mi10rb-aut]
|
|
#[language: python]
|
|
#[class: 工具类]
|
|
#[service: 10010] ADB设备重启管理
|
|
#[disable: false]
|
|
#[admin: false]
|
|
#[rule: ^重启Mi10$]
|
|
#[priority: 999]
|
|
#[platform: qq,qb,wx,tb,tg,web,wxmp]
|
|
#[open_source: false]
|
|
#[icon: 图标url]
|
|
#[version: 1.0.0]
|
|
#[public: false]
|
|
#[price: 0.00]
|
|
#[description: 通过ADB(TCP)连接到设备并执行重启；发送"重启Mi10"执行。<br>]
|
|
|
|
# [param: {"required":true,"key":"mi10rb.address","placeholder":"127.0.0.1:5555","name":"ADB地址：","desc":"设备的 ADB 地址，如 192.168.1.100:5555"}]
|
|
# [param: {"required":false,"key":"mi10rb.adb_path","placeholder":"adb","name":"adb路径：","desc":"adb 可执行文件路径，默认使用系统 adb"}]
|
|
# [param: {"required":false,"key":"mi10rb.timeout","placeholder":"20","name":"超时：","desc":"单次命令超时时间（秒），默认20"}]
|
|
|
|
import os
|
|
import sys
|
|
import shlex
|
|
import logging
|
|
import subprocess
|
|
import time
|
|
import colorlog
|
|
import middleware
|
|
import sender
|
|
from typing import Tuple, Optional
|
|
|
|
|
|
# 读取插件参数
|
|
CONF_ADDRESS = middleware.bucketGet('mi10rb', 'address')
|
|
CONF_ADB_PATH = middleware.bucketGet('mi10rb', 'adb_path') or 'adb'
|
|
CONF_TIMEOUT = int(middleware.bucketGet('mi10rb', 'timeout') or '20')
|
|
|
|
|
|
sender = middleware.Sender(middleware.getSenderID())
|
|
|
|
|
|
class TOOL:
|
|
def __init__(self):
|
|
self.plugin_pre = "mi10rb"
|
|
self.color_formatter = colorlog.ColoredFormatter(
|
|
'%(log_color)s%(levelname)s: %(message)s',
|
|
log_colors={
|
|
'DEBUG': 'cyan',
|
|
'INFO': 'green',
|
|
'WARNING': 'yellow',
|
|
'ERROR': 'red',
|
|
'CRITICAL': 'red,bg_white',
|
|
}
|
|
)
|
|
self.logger = self._configure_logger()
|
|
|
|
def _configure_logger(self):
|
|
logger_obj = logging.getLogger(self.plugin_pre)
|
|
logger_obj.setLevel(logging.DEBUG)
|
|
if not any(isinstance(h, logging.StreamHandler) for h in logger_obj.handlers):
|
|
console_handler = logging.StreamHandler()
|
|
console_handler.setLevel(logging.DEBUG)
|
|
console_handler.setFormatter(self.color_formatter)
|
|
logger_obj.addHandler(console_handler)
|
|
return logger_obj
|
|
|
|
def log(self, level, message):
|
|
self.logger.log(level, message)
|
|
|
|
def reply(self, content):
|
|
return sender.reply(content)
|
|
|
|
|
|
def run_command(command: str, timeout_seconds: int = 20) -> Tuple[int, str, str]:
|
|
"""
|
|
以非交互方式执行命令，返回 (exit_code, stdout, stderr)
|
|
"""
|
|
try:
|
|
process = subprocess.Popen(
|
|
shlex.split(command),
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
text=True,
|
|
)
|
|
stdout, stderr = process.communicate(timeout=timeout_seconds)
|
|
return process.returncode, (stdout or '').strip(), (stderr or '').strip()
|
|
except subprocess.TimeoutExpired:
|
|
try:
|
|
process.kill()
|
|
except Exception:
|
|
pass
|
|
return 124, "", f"Command timed out after {timeout_seconds}s: {command}"
|
|
except FileNotFoundError:
|
|
return 127, "", f"Command not found: {command.split()[0]}"
|
|
|
|
|
|
class ADBRebootClient:
|
|
def __init__(self, address: str, adb_path: str = 'adb', timeout: int = 20, tool: Optional[TOOL] = None):
|
|
self.address = address
|
|
self.adb = adb_path or 'adb'
|
|
self.timeout = timeout
|
|
self.tool = tool
|
|
|
|
def _log(self, level, message: str):
|
|
if self.tool:
|
|
self.tool.log(level, message)
|
|
|
|
def disconnect(self) -> None:
|
|
# 断开指定地址，忽略错误
|
|
run_command(f"{self.adb} disconnect {self.address}", timeout_seconds=min(self.timeout, 5))
|
|
|
|
def kill_server(self) -> None:
|
|
# 尝试重启 adb 服务端，忽略错误
|
|
run_command(f"{self.adb} kill-server", timeout_seconds=min(self.timeout, 5))
|
|
|
|
def ensure_adb(self) -> bool:
|
|
code, out, err = run_command(f"{self.adb} version", timeout_seconds=self.timeout)
|
|
if code != 0:
|
|
self._log(logging.ERROR, "未找到可用的 adb，请安装或在参数中指定 adb 路径")
|
|
if err:
|
|
self._log(logging.ERROR, err)
|
|
return False
|
|
self._log(logging.INFO, (out or 'adb 可用'))
|
|
return True
|
|
|
|
def connect(self) -> bool:
|
|
# 先尝试断开可能残留的连接
|
|
self.disconnect()
|
|
|
|
# 首次连接
|
|
code, out, err = run_command(f"{self.adb} connect {self.address}", timeout_seconds=self.timeout)
|
|
msg = ((out or '') + ("\n" + err if err else '')).strip()
|
|
if code != 0:
|
|
self._log(logging.WARNING, f"adb connect 首次失败(code={code})，尝试重启 adb 并重试...\n{msg}")
|
|
# 重启 adb 后重试一次
|
|
self.kill_server()
|
|
self.disconnect()
|
|
code, out, err = run_command(f"{self.adb} connect {self.address}", timeout_seconds=self.timeout)
|
|
msg = ((out or '') + ("\n" + err if err else '')).strip()
|
|
if code != 0:
|
|
self._log(logging.ERROR, f"adb connect 重试失败(code={code}):\n{msg}")
|
|
return False
|
|
|
|
low = msg.lower()
|
|
if ("connected to" in low) or ("already connected to" in low):
|
|
self._log(logging.INFO, "✓ ADB 连接成功")
|
|
else:
|
|
self._log(logging.WARNING, "adb connect 返回0，但输出未包含已连接提示")
|
|
if msg:
|
|
self._log(logging.INFO, msg)
|
|
|
|
# 连接校验：确认设备在线
|
|
ok, detail = self.get_state()
|
|
if not ok:
|
|
self._log(logging.ERROR, f"连接校验失败：{detail}")
|
|
return False
|
|
self._log(logging.INFO, f"设备状态：{detail}")
|
|
return True
|
|
|
|
def get_state(self) -> Tuple[bool, str]:
|
|
code, out, err = run_command(f"{self.adb} -s {self.address} get-state", timeout_seconds=min(self.timeout, 10))
|
|
text = (out or err or '').strip()
|
|
if code == 0 and (out or '').strip() == 'device':
|
|
return True, 'device'
|
|
# 某些 ROM 会返回 unknown/unauthorized，这里给出更友好的提示
|
|
return False, (text or f"get-state 失败(code={code})")
|
|
|
|
def reboot(self) -> bool:
|
|
def looks_like_reboot_sent(message: str) -> bool:
|
|
low = (message or '').lower()
|
|
# 设备重启会导致连接被关闭，adb 可能返回如下错误，但指令已下发
|
|
hints = [
|
|
'closed',
|
|
'device offline',
|
|
'no devices/emulators found',
|
|
'cannot connect to daemon',
|
|
'daemon not running',
|
|
'failed to connect to',
|
|
'connection refused',
|
|
'connection reset',
|
|
'transport endpoint is not connected',
|
|
'broken pipe',
|
|
'read failed',
|
|
'i/o error',
|
|
'more than one device/emulator',
|
|
]
|
|
return any(h in low for h in hints)
|
|
|
|
def observe_post_reboot_disconnect(max_attempts: int = 8, interval_seconds: float = 0.5) -> bool:
|
|
# 在短时间内观察设备从在线到离线/不可达的过渡，以此判断重启已生效
|
|
for _ in range(max_attempts):
|
|
code_s, out_s, err_s = run_command(
|
|
f"{self.adb} -s {self.address} get-state",
|
|
timeout_seconds=min(self.timeout, 5)
|
|
)
|
|
text_s = ((out_s or '') + ('\n' + err_s if err_s else '')).strip().lower()
|
|
if code_s != 0:
|
|
# 命令失败，多半是设备掉线，视为重启进行中
|
|
return True
|
|
if (out_s or '').strip() != 'device':
|
|
# 非 device 状态（例如 offline/unknown），也视为重启进行中
|
|
return True
|
|
# 仍为在线，继续等待
|
|
time.sleep(interval_seconds)
|
|
return False
|
|
|
|
code, out, err = run_command(f"{self.adb} -s {self.address} reboot", timeout_seconds=self.timeout)
|
|
combined = ((out or '') + ("\n" + err if err else '')).strip()
|
|
|
|
if code == 0:
|
|
self._log(logging.INFO, "✓ 已发送重启指令")
|
|
return True
|
|
|
|
# 非零返回，但疑似已下发
|
|
if looks_like_reboot_sent(combined):
|
|
self._log(logging.WARNING, f"adb reboot 返回非零(code={code})，但检测到重启迹象，视为成功：\n{combined}")
|
|
return True
|
|
|
|
# 回退方案：走 shell reboot
|
|
self._log(logging.WARNING, f"adb reboot 失败(code={code})，尝试使用 shell reboot 回退...\n{combined}")
|
|
code2, out2, err2 = run_command(f"{self.adb} -s {self.address} shell reboot", timeout_seconds=self.timeout)
|
|
combined2 = ((out2 or '') + ("\n" + err2 if err2 else '')).strip()
|
|
if code2 == 0 or looks_like_reboot_sent(combined2):
|
|
self._log(logging.INFO, "✓ 已通过 shell reboot 发送重启指令")
|
|
return True
|
|
|
|
# 最终失败前，观察设备是否已开始断连（重启进行中）
|
|
if observe_post_reboot_disconnect():
|
|
self._log(logging.WARNING, "检测到重启后的断连迹象，尽管命令返回失败，仍视为已下发成功")
|
|
return True
|
|
|
|
self._log(logging.ERROR, f"shell reboot 仍失败(code={code2}):\n{combined2}")
|
|
return False
|
|
|
|
|
|
def run_reboot_flow(tool: TOOL):
|
|
"""执行 ADB 设备重启流程"""
|
|
tool.reply("开始执行 Mi10 ADB 重启流程...")
|
|
|
|
# 校验参数
|
|
if not CONF_ADDRESS:
|
|
tool.reply("❌ 配置参数不完整，请设置 mi10rb.address (如 192.168.1.100:5555)")
|
|
return
|
|
|
|
tool.log(logging.INFO, "配置信息:")
|
|
tool.log(logging.INFO, f" ADB地址: {CONF_ADDRESS}")
|
|
tool.log(logging.INFO, f" adb路径: {CONF_ADB_PATH}")
|
|
tool.log(logging.INFO, f" 超时: {CONF_TIMEOUT}秒")
|
|
|
|
client = ADBRebootClient(address=CONF_ADDRESS, adb_path=CONF_ADB_PATH, timeout=CONF_TIMEOUT, tool=tool)
|
|
|
|
try:
|
|
if not client.ensure_adb():
|
|
tool.reply("❌ 未检测到 adb，请安装或指定有效的 adb 路径")
|
|
return
|
|
|
|
tool.log(logging.INFO, "🚀 正在连接设备...")
|
|
if not client.connect():
|
|
tool.reply("❌ ADB 连接失败，请检查地址与网络")
|
|
return
|
|
|
|
tool.log(logging.INFO, "🔁 正在发送重启指令...")
|
|
if client.reboot():
|
|
tool.reply("✅ 重启指令已发送！设备即将重启")
|
|
else:
|
|
tool.reply("❌ 重启指令发送失败")
|
|
except Exception as exc:
|
|
tool.log(logging.ERROR, f"程序执行出错: {exc}")
|
|
tool.reply("❌ 程序执行出错，请检查配置与网络")
|
|
|
|
|
|
def exit_with_msg(msg: str):
|
|
if msg:
|
|
sender.reply(msg)
|
|
sender.reply("已退出！")
|
|
sys.exit(0)
|
|
|
|
|
|
def _safe_get_message():
|
|
try:
|
|
return sender.getMessage().strip('"')
|
|
except Exception:
|
|
return ""
|
|
|
|
|
|
if __name__ == "__main__":
|
|
tool = TOOL()
|
|
msg = _safe_get_message()
|
|
|
|
if "重启Mi10" in msg:
|
|
run_reboot_flow(tool)
|
|
sys.exit(0)
|
|
|
|
sender.reply("指令不支持，请发送：重启Mi10")
|
|
|
|
|