1173 lines
44 KiB
Python
1173 lines
44 KiB
Python
#[title: rb-aut]
|
||
#[language: python]
|
||
#[class: 工具类]
|
||
#[service: 10010] 设备重启管理（设备管理模式）
|
||
#[disable: false]
|
||
#[admin: false]
|
||
#[rule: ^(rb-now|rb管理)$]
|
||
#[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: 设备重启管理，支持多设备管理。<br>指令：rb-now（设备选单重启）、rb管理（设备管理），adb方式重启需安装依赖adbtool，并且首次运行需在设备授权]
|
||
|
||
import sys
|
||
import shlex
|
||
import time
|
||
import json
|
||
import logging
|
||
import subprocess
|
||
from typing import Tuple, Optional
|
||
|
||
import colorlog
|
||
import middleware
|
||
import sender
|
||
|
||
|
||
# 存储桶前缀
|
||
BUCKET = 'rb-aut'
|
||
|
||
|
||
sender = middleware.Sender(middleware.getSenderID())
|
||
|
||
|
||
class TOOL:
|
||
def __init__(self):
|
||
self.plugin_pre = BUCKET
|
||
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 bucket_get(key: str, default: Optional[str] = None) -> Optional[str]:
|
||
val = middleware.bucketGet(BUCKET, key)
|
||
return val if val is not None and val != '' else default
|
||
|
||
|
||
def bucket_set(key: str, value: str) -> None:
|
||
middleware.bucketSet(BUCKET, key, value)
|
||
|
||
|
||
def get_all_devices() -> dict:
|
||
"""获取所有设备配置"""
|
||
devices = {}
|
||
all_keys = middleware.bucketAllKeys(BUCKET)
|
||
for key in all_keys:
|
||
if key.startswith('device.'):
|
||
device_id = key[7:] # 去掉 'device.' 前缀
|
||
device_data = bucket_get(key)
|
||
if device_data:
|
||
try:
|
||
devices[device_id] = json.loads(device_data)
|
||
except Exception:
|
||
continue
|
||
return devices
|
||
|
||
|
||
def save_device(device_id: str, device_config: dict) -> None:
|
||
"""保存设备配置"""
|
||
bucket_set(f'device.{device_id}', json.dumps(device_config))
|
||
|
||
|
||
def delete_device(device_id: str) -> None:
|
||
"""删除设备配置"""
|
||
middleware.bucketDel(BUCKET, f'device.{device_id}')
|
||
|
||
|
||
def has_device_permission(device_config: dict, user_id: str, is_admin: bool) -> bool:
|
||
"""检查用户是否有设备权限
|
||
Args:
|
||
device_config: 设备配置字典
|
||
user_id: 用户ID
|
||
is_admin: 是否是管理员
|
||
Returns:
|
||
bool: 是否有权限
|
||
"""
|
||
# 管理员有所有权限
|
||
if is_admin:
|
||
return True
|
||
|
||
# 检查设备权限配置
|
||
admin_only = device_config.get('admin_only', False)
|
||
if admin_only:
|
||
return False
|
||
|
||
# 检查允许的用户列表
|
||
allowed_users = device_config.get('allowed_users', [])
|
||
if isinstance(allowed_users, str):
|
||
# 兼容旧格式（逗号分隔的字符串）
|
||
allowed_users = [u.strip() for u in allowed_users.split(',') if u.strip()]
|
||
if not isinstance(allowed_users, list):
|
||
allowed_users = []
|
||
|
||
# 如果允许的用户列表为空，则默认所有用户都可以访问
|
||
if not allowed_users:
|
||
return True
|
||
|
||
return user_id in allowed_users
|
||
|
||
|
||
def get_device_list(user_id: Optional[str] = None, is_admin: bool = False) -> list:
|
||
"""获取设备列表，返回格式化的设备信息
|
||
Args:
|
||
user_id: 用户ID，如果提供则只返回该用户有权限的设备
|
||
is_admin: 是否是管理员，管理员可以看到所有设备
|
||
"""
|
||
devices = get_all_devices()
|
||
device_list = []
|
||
for device_id, config in devices.items():
|
||
device_type = config.get('type', 'unknown')
|
||
name = config.get('name', device_id)
|
||
|
||
# 如果指定了用户ID且不是管理员，检查权限
|
||
if user_id and not is_admin:
|
||
if not has_device_permission(config, user_id, False):
|
||
continue
|
||
|
||
if device_type == 'adb':
|
||
device_list.append({
|
||
'id': device_id,
|
||
'name': name,
|
||
'type': 'ADB',
|
||
'info': name
|
||
})
|
||
elif device_type == 'ssh':
|
||
device_list.append({
|
||
'id': device_id,
|
||
'name': name,
|
||
'type': 'SSH',
|
||
'info': name
|
||
})
|
||
return device_list
|
||
|
||
|
||
def run_command(command: str, timeout_seconds: int = 20) -> Tuple[int, str, str]:
|
||
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:
|
||
run_command(f"{self.adb} kill-server", timeout_seconds=min(self.timeout, 5))
|
||
|
||
def cleanup(self) -> None:
|
||
"""清理ADB连接和服务"""
|
||
try:
|
||
# 先断开指定设备连接
|
||
self.disconnect()
|
||
# 等待一下确保断开完成
|
||
time.sleep(1)
|
||
# 然后kill整个adb服务
|
||
self.kill_server()
|
||
if self.tool:
|
||
self.tool.log(logging.INFO, "✓ ADB服务已清理完成")
|
||
except Exception as e:
|
||
if self.tool:
|
||
self.tool.log(logging.WARNING, f"ADB清理过程中出现异常: {e}")
|
||
|
||
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}")
|
||
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'
|
||
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()
|
||
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)
|
||
)
|
||
if code_s != 0:
|
||
return True
|
||
if (out_s or '').strip() != 'device':
|
||
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
|
||
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
|
||
|
||
|
||
class SSHRebootClient:
|
||
def __init__(self, hostname: str, username: str, password: str,
|
||
port: int = 5722, timeout: int = 30, tool: Optional[TOOL] = None):
|
||
self.hostname = hostname
|
||
self.username = username
|
||
self.password = password
|
||
self.port = port
|
||
self.timeout = timeout
|
||
self.tool = tool
|
||
self.client = None
|
||
|
||
def connect(self) -> bool:
|
||
try:
|
||
if self.tool:
|
||
self.tool.log(logging.INFO, f"正在连接到 {self.hostname}:{self.port}...")
|
||
import paramiko
|
||
self.client = paramiko.SSHClient()
|
||
self.client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
|
||
self.client.connect(
|
||
hostname=self.hostname,
|
||
username=self.username,
|
||
password=self.password,
|
||
port=self.port,
|
||
timeout=self.timeout,
|
||
allow_agent=False,
|
||
look_for_keys=False,
|
||
)
|
||
if self.tool:
|
||
self.tool.log(logging.INFO, "✓ SSH连接建立成功!")
|
||
return True
|
||
except Exception as e:
|
||
if self.tool:
|
||
self.tool.log(logging.ERROR, f"✗ SSH连接失败: {e}")
|
||
return False
|
||
|
||
def _execute_reboot_command(self, command: str) -> bool:
|
||
reboot_commands = [
|
||
command,
|
||
f"sudo {command}",
|
||
f"nohup {command} > /dev/null 2>&1 &",
|
||
f"sudo nohup {command} > /dev/null 2>&1 &",
|
||
]
|
||
for i, cmd in enumerate(reboot_commands):
|
||
try:
|
||
if self.tool:
|
||
self.tool.log(logging.INFO, f"尝试执行重启命令 ({i+1}/{len(reboot_commands)}): {cmd}")
|
||
stdin, stdout, stderr = self.client.exec_command(cmd, timeout=3)
|
||
try:
|
||
_ = stdout.read()
|
||
_ = stderr.read()
|
||
except Exception:
|
||
pass
|
||
time.sleep(1)
|
||
try:
|
||
t_in, t_out, t_err = self.client.exec_command("echo test", timeout=1)
|
||
_ = t_out.read()
|
||
except Exception:
|
||
if self.tool:
|
||
self.tool.log(logging.INFO, "✓ 检测到SSH连接断开，重启命令生效!")
|
||
self.tool.log(logging.INFO, "✓ 设备正在重启中...")
|
||
return True
|
||
if i < len(reboot_commands) - 1:
|
||
if self.tool:
|
||
self.tool.log(logging.INFO, "连接仍在，尝试下一种命令格式...")
|
||
time.sleep(1)
|
||
except Exception as e:
|
||
if self.tool:
|
||
self.tool.log(logging.WARNING, f"命令执行异常: {e}")
|
||
try:
|
||
_ = self.client.get_transport().is_active()
|
||
except Exception:
|
||
if self.tool:
|
||
self.tool.log(logging.INFO, "✓ 检测到SSH连接断开，重启命令生效!")
|
||
return True
|
||
if i < len(reboot_commands) - 1:
|
||
if self.tool:
|
||
self.tool.log(logging.INFO, "尝试下一种命令格式...")
|
||
time.sleep(1)
|
||
if self.tool:
|
||
self.tool.log(logging.WARNING, "⚠️ 所有重启命令格式都尝试过了；重启可能已发送")
|
||
return True
|
||
|
||
def send_reboot_command(self, command: str = 'reboot') -> bool:
|
||
if not self.client:
|
||
if self.tool:
|
||
self.tool.log(logging.ERROR, "✗ 未建立SSH连接")
|
||
return False
|
||
try:
|
||
if command.lower() in ['reboot', 'shutdown', 'poweroff']:
|
||
return self._execute_reboot_command(command)
|
||
stdin, stdout, stderr = self.client.exec_command(command, timeout=30)
|
||
exit_status = stdout.channel.recv_exit_status()
|
||
_ = stdout.read()
|
||
_ = stderr.read()
|
||
if exit_status == 0:
|
||
if self.tool:
|
||
self.tool.log(logging.INFO, "✓ 命令执行成功!")
|
||
return True
|
||
if self.tool:
|
||
self.tool.log(logging.ERROR, f"✗ 命令执行失败，退出码: {exit_status}")
|
||
return False
|
||
except Exception as e:
|
||
if self.tool:
|
||
self.tool.log(logging.ERROR, f"✗ 执行命令时出错: {e}")
|
||
return False
|
||
|
||
def is_connected(self) -> bool:
|
||
try:
|
||
if self.client and self.client.get_transport():
|
||
return self.client.get_transport().is_active()
|
||
return False
|
||
except Exception:
|
||
return False
|
||
|
||
def close(self) -> None:
|
||
if self.client:
|
||
try:
|
||
self.client.close()
|
||
if self.tool:
|
||
self.tool.log(logging.INFO, "SSH连接已关闭")
|
||
except Exception:
|
||
if self.tool:
|
||
self.tool.log(logging.INFO, "SSH连接已断开")
|
||
|
||
|
||
def run_adb_reboot(tool: TOOL, device_config: dict) -> bool:
|
||
"""执行ADB设备重启"""
|
||
addr = device_config.get('address', '')
|
||
name = device_config.get('name', '未知设备')
|
||
|
||
tool.reply(f"开始执行 {name} ADB 重启流程...")
|
||
if not addr:
|
||
tool.reply("❌ 设备配置不完整，缺少 address")
|
||
return
|
||
|
||
tool.log(logging.INFO, "配置信息:")
|
||
tool.log(logging.INFO, f" 设备名称: {name}")
|
||
tool.log(logging.INFO, f" ADB地址: {addr}")
|
||
|
||
client = ADBRebootClient(address=addr, adb_path='adb', timeout=20, tool=tool)
|
||
try:
|
||
if not client.ensure_adb():
|
||
tool.reply("❌ 未检测到 adb，请安装 adb")
|
||
return False
|
||
tool.log(logging.INFO, "🚀 正在连接设备...")
|
||
if not client.connect():
|
||
tool.reply("❌ ADB 连接失败，请检查地址与网络")
|
||
return False
|
||
tool.log(logging.INFO, "🔁 正在发送重启指令...")
|
||
if client.reboot():
|
||
tool.reply("✅ 重启指令已发送！设备即将重启")
|
||
# 等待一下让重启指令生效
|
||
time.sleep(2)
|
||
# 清理ADB连接和服务
|
||
tool.log(logging.INFO, "🔌 正在清理ADB连接...")
|
||
client.cleanup()
|
||
return True
|
||
else:
|
||
tool.reply("❌ 重启指令发送失败")
|
||
# 即使失败也清理连接
|
||
tool.log(logging.INFO, "🔌 正在清理ADB连接...")
|
||
client.cleanup()
|
||
return False
|
||
except Exception as exc:
|
||
tool.log(logging.ERROR, f"程序执行出错: {exc}")
|
||
tool.reply("❌ 程序执行出错，请检查配置与网络")
|
||
# 异常情况下也尝试清理
|
||
try:
|
||
tool.log(logging.INFO, "🔌 异常情况下清理ADB连接...")
|
||
client.cleanup()
|
||
except Exception:
|
||
pass
|
||
return False
|
||
|
||
|
||
def run_ssh_reboot(tool: TOOL, device_config: dict) -> bool:
|
||
"""执行SSH设备重启"""
|
||
host = device_config.get('hostname', '')
|
||
username = device_config.get('username', 'root')
|
||
password = device_config.get('password', '')
|
||
port = device_config.get('port', 22)
|
||
timeout = device_config.get('timeout', 30)
|
||
command = device_config.get('command', 'reboot')
|
||
name = device_config.get('name', '未知设备')
|
||
|
||
tool.reply(f"开始执行 {name} SSH 重启流程...")
|
||
if not host or not username or not password:
|
||
tool.reply("❌ 设备配置不完整，缺少 hostname/username/password")
|
||
return False
|
||
|
||
tool.log(logging.INFO, f"配置信息:")
|
||
tool.log(logging.INFO, f" 设备名称: {name}")
|
||
tool.log(logging.INFO, f" 主机: {host}")
|
||
tool.log(logging.INFO, f" 用户: {username}")
|
||
tool.log(logging.INFO, f" 端口: {port}")
|
||
tool.log(logging.INFO, f" 命令: {command}")
|
||
tool.log(logging.INFO, f" 超时: {timeout}秒")
|
||
|
||
client = SSHRebootClient(hostname=host, username=username, password=password,
|
||
port=port, timeout=timeout, tool=tool)
|
||
try:
|
||
if not client.connect():
|
||
tool.reply("❌ SSH连接失败，请检查设备地址和认证信息")
|
||
return False
|
||
tool.log(logging.INFO, "🚀 开始执行重启操作...")
|
||
result = client.send_reboot_command(command)
|
||
if result:
|
||
tool.reply("✅ 重启操作完成!")
|
||
tool.reply("📱 设备正在重启中，请稍后检查设备状态")
|
||
time.sleep(3)
|
||
if not client.is_connected():
|
||
tool.log(logging.INFO, "🔌 SSH连接已断开，确认设备正在重启")
|
||
else:
|
||
tool.log(logging.WARNING, "⚠️ SSH连接仍在，设备可能未重启成功")
|
||
return True
|
||
else:
|
||
tool.reply("❌ 重启命令执行失败!")
|
||
return False
|
||
except Exception as e:
|
||
tool.log(logging.ERROR, f"程序执行出错: {e}")
|
||
tool.reply("❌ 程序执行出错，请检查配置和网络连接")
|
||
return False
|
||
finally:
|
||
client.close()
|
||
tool.log(logging.INFO, "🏁 脚本执行完毕")
|
||
|
||
|
||
def reboot_device(tool: TOOL, device_config: dict) -> bool:
|
||
"""根据设备类型执行重启，返回是否成功"""
|
||
device_type = device_config.get('type')
|
||
if device_type == 'adb':
|
||
return run_adb_reboot(tool, device_config)
|
||
if device_type == 'ssh':
|
||
return run_ssh_reboot(tool, device_config)
|
||
tool.reply("❌ 未知设备类型")
|
||
return False
|
||
|
||
|
||
def cmd_rb_now(tool: TOOL, user_id: str, is_admin: bool):
|
||
"""rb-now 命令：显示设备选单并执行重启"""
|
||
device_list = get_device_list(user_id, is_admin)
|
||
|
||
if not device_list:
|
||
tool.reply("❌ 暂无可用设备，请联系管理员添加设备")
|
||
return
|
||
|
||
# 显示设备选单（只显示名称），0 为全部重启
|
||
menu_text = "请选择要重启的设备：\n0. 重启所有设备\n"
|
||
for i, device in enumerate(device_list, 1):
|
||
menu_text += f"{i}. {device['name']}\n"
|
||
|
||
tool.reply(menu_text)
|
||
|
||
# 等待用户选择
|
||
try:
|
||
choice = sender.listen(60000)
|
||
if not choice or choice == 'q' or choice == 'error':
|
||
tool.reply("已取消")
|
||
return
|
||
|
||
# 选择 0 重启所有设备
|
||
if choice == '0':
|
||
devices = get_all_devices()
|
||
total = len(device_list)
|
||
success = 0
|
||
fail = 0
|
||
for item in device_list:
|
||
cfg = devices.get(item['id'])
|
||
if not cfg:
|
||
fail += 1
|
||
continue
|
||
# 再次验证权限，确保安全
|
||
if not has_device_permission(cfg, user_id, is_admin):
|
||
fail += 1
|
||
continue
|
||
ok = reboot_device(tool, cfg)
|
||
if ok:
|
||
success += 1
|
||
else:
|
||
fail += 1
|
||
tool.reply(f"执行完成：成功 {success}/{total}，失败 {fail}/{total}")
|
||
return
|
||
|
||
choice_num = int(choice)
|
||
if 1 <= choice_num <= len(device_list):
|
||
selected_device = device_list[choice_num - 1]
|
||
device_id = selected_device['id']
|
||
devices = get_all_devices()
|
||
device_config = devices.get(device_id)
|
||
if not device_config:
|
||
tool.reply("❌ 设备配置不存在")
|
||
return
|
||
# 再次验证权限，确保安全
|
||
if not has_device_permission(device_config, user_id, is_admin):
|
||
tool.reply("❌ 您没有权限重启此设备")
|
||
return
|
||
reboot_device(tool, device_config)
|
||
else:
|
||
tool.reply("❌ 无效选择")
|
||
except ValueError:
|
||
tool.reply("❌ 请输入有效数字")
|
||
except Exception as e:
|
||
tool.reply(f"❌ 选择设备时出错: {e}")
|
||
|
||
|
||
def cmd_rb_manage(tool: TOOL, user_id: str, is_admin: bool):
|
||
"""rb管理 命令：设备管理（仅管理员可用）"""
|
||
if not is_admin:
|
||
tool.reply("❌ 此命令仅管理员可使用")
|
||
return
|
||
|
||
device_list = get_device_list(user_id, is_admin)
|
||
|
||
if not device_list:
|
||
tool.reply("设备管理菜单：\n1. 添加设备\n0. 返回")
|
||
try:
|
||
choice = sender.listen(60000)
|
||
if not choice or choice == '0' or choice == 'q' or choice == 'error':
|
||
tool.reply("已返回")
|
||
return
|
||
if choice == '1':
|
||
add_device(tool)
|
||
else:
|
||
tool.reply("❌ 无效选择")
|
||
except Exception as e:
|
||
tool.reply(f"❌ 操作出错: {e}")
|
||
return
|
||
|
||
# 显示设备列表和管理选项（仅名称），提示 + 和 - 用法
|
||
menu_text = "设备管理菜单：\n"
|
||
for i, device in enumerate(device_list, 1):
|
||
menu_text += f"{i}. {device['name']}\n"
|
||
menu_text += "+ （添加设备）\n- 序号 （删除设备）\n0. 返回"
|
||
|
||
tool.reply(menu_text)
|
||
|
||
try:
|
||
choice = sender.listen(60000)
|
||
if not choice or choice == '0' or choice == 'q' or choice == 'error':
|
||
tool.reply("已返回")
|
||
return
|
||
|
||
if choice == '+':
|
||
add_device(tool)
|
||
elif choice.startswith('-'):
|
||
# 删除设备：-序号
|
||
try:
|
||
device_num = int(choice[1:])
|
||
if 1 <= device_num <= len(device_list):
|
||
delete_device_by_number(tool, device_list, device_num - 1)
|
||
else:
|
||
tool.reply("❌ 无效的设备序号")
|
||
except ValueError:
|
||
tool.reply("❌ 请输入正确的格式：-序号")
|
||
elif choice.isdigit():
|
||
# 查看/修改设备：序号
|
||
device_num = int(choice)
|
||
if 1 <= device_num <= len(device_list):
|
||
view_edit_device(tool, device_list, device_num - 1)
|
||
else:
|
||
tool.reply("❌ 无效的设备序号")
|
||
else:
|
||
tool.reply("❌ 无效选择")
|
||
except Exception as e:
|
||
tool.reply(f"❌ 操作出错: {e}")
|
||
|
||
|
||
def add_device(tool: TOOL):
|
||
"""添加设备"""
|
||
tool.reply("请选择设备类型：\n1. ADB设备\n2. SSH设备\n0. 取消")
|
||
|
||
try:
|
||
type_choice = sender.listen(60000)
|
||
if not type_choice or type_choice == '0' or type_choice == 'q' or type_choice == 'error':
|
||
tool.reply("已取消")
|
||
return
|
||
|
||
if type_choice == '1':
|
||
add_adb_device(tool)
|
||
elif type_choice == '2':
|
||
add_ssh_device(tool)
|
||
else:
|
||
tool.reply("❌ 无效选择")
|
||
except Exception as e:
|
||
tool.reply(f"❌ 添加设备出错: {e}")
|
||
|
||
|
||
def add_adb_device(tool: TOOL):
|
||
"""添加ADB设备"""
|
||
tool.reply("请输入设备名称（q:退出）")
|
||
name = sender.listen(60000)
|
||
if not name or name == 'q' or name == 'error':
|
||
tool.reply("已退出")
|
||
return
|
||
|
||
tool.reply("请输入ADB地址（示例：192.168.1.100:5555，q:退出）")
|
||
address = sender.listen(60000)
|
||
if not address or address == 'q' or address == 'error':
|
||
tool.reply("已退出")
|
||
return
|
||
|
||
# 设置设备权限
|
||
tool.reply("请设置设备权限：\n1. 仅管理员可用\n2. 所有用户可用\n3. 指定用户可用\n0. 取消")
|
||
perm_choice = sender.listen(60000)
|
||
if not perm_choice or perm_choice == '0' or perm_choice == 'q' or perm_choice == 'error':
|
||
tool.reply("已取消")
|
||
return
|
||
|
||
device_config = {
|
||
'type': 'adb',
|
||
'name': name,
|
||
'address': address,
|
||
'admin_only': False,
|
||
'allowed_users': []
|
||
}
|
||
|
||
if perm_choice == '1':
|
||
device_config['admin_only'] = True
|
||
elif perm_choice == '2':
|
||
device_config['admin_only'] = False
|
||
device_config['allowed_users'] = [] # 空列表表示所有用户可用
|
||
elif perm_choice == '3':
|
||
tool.reply("请输入允许访问的用户ID（多个用户用逗号分隔，q:退出）")
|
||
users_input = sender.listen(60000)
|
||
if not users_input or users_input == 'q' or users_input == 'error':
|
||
tool.reply("已取消")
|
||
return
|
||
allowed_users = [u.strip() for u in users_input.split(',') if u.strip()]
|
||
device_config['admin_only'] = False
|
||
device_config['allowed_users'] = allowed_users
|
||
else:
|
||
tool.reply("❌ 无效选择，已取消")
|
||
return
|
||
|
||
# 生成设备ID
|
||
device_id = f"adb_{int(time.time())}"
|
||
|
||
save_device(device_id, device_config)
|
||
tool.reply(f"✅ ADB设备 '{name}' 添加成功！")
|
||
|
||
|
||
def add_ssh_device(tool: TOOL):
|
||
"""添加SSH设备"""
|
||
tool.reply("请输入设备名称（q:退出）")
|
||
name = sender.listen(60000)
|
||
if not name or name == 'q' or name == 'error':
|
||
tool.reply("已退出")
|
||
return
|
||
|
||
tool.reply("请输入SSH主机地址（示例：192.168.1.200，q:退出）")
|
||
hostname = sender.listen(60000)
|
||
if not hostname or hostname == 'q' or hostname == 'error':
|
||
tool.reply("已退出")
|
||
return
|
||
|
||
tool.reply("请输入SSH用户名（默认root，直接回车跳过）")
|
||
username = sender.listen(60000)
|
||
if not username or username == 'q' or username == 'error':
|
||
username = 'root'
|
||
|
||
tool.reply("请输入SSH密码（q:退出）")
|
||
password = sender.listen(60000)
|
||
if not password or password == 'q' or password == 'error':
|
||
tool.reply("已退出")
|
||
return
|
||
|
||
tool.reply("请输入SSH端口（默认22，直接回车跳过）")
|
||
port_str = sender.listen(60000)
|
||
port = 22
|
||
if port_str and port_str.isdigit():
|
||
port = int(port_str)
|
||
|
||
tool.reply("请输入重启命令（默认reboot，直接回车跳过）")
|
||
command = sender.listen(60000)
|
||
if not command or command == 'q' or command == 'error':
|
||
command = 'reboot'
|
||
|
||
tool.reply("请输入超时时间（默认30秒，直接回车跳过）")
|
||
timeout_str = sender.listen(60000)
|
||
timeout = 30
|
||
if timeout_str and timeout_str.isdigit():
|
||
timeout = int(timeout_str)
|
||
|
||
# 设置设备权限
|
||
tool.reply("请设置设备权限：\n1. 仅管理员可用\n2. 所有用户可用\n3. 指定用户可用\n0. 取消")
|
||
perm_choice = sender.listen(60000)
|
||
if not perm_choice or perm_choice == '0' or perm_choice == 'q' or perm_choice == 'error':
|
||
tool.reply("已取消")
|
||
return
|
||
|
||
device_config = {
|
||
'type': 'ssh',
|
||
'name': name,
|
||
'hostname': hostname,
|
||
'username': username,
|
||
'password': password,
|
||
'port': port,
|
||
'command': command,
|
||
'timeout': timeout,
|
||
'admin_only': False,
|
||
'allowed_users': []
|
||
}
|
||
|
||
if perm_choice == '1':
|
||
device_config['admin_only'] = True
|
||
elif perm_choice == '2':
|
||
device_config['admin_only'] = False
|
||
device_config['allowed_users'] = [] # 空列表表示所有用户可用
|
||
elif perm_choice == '3':
|
||
tool.reply("请输入允许访问的用户ID（多个用户用逗号分隔，q:退出）")
|
||
users_input = sender.listen(60000)
|
||
if not users_input or users_input == 'q' or users_input == 'error':
|
||
tool.reply("已取消")
|
||
return
|
||
allowed_users = [u.strip() for u in users_input.split(',') if u.strip()]
|
||
device_config['admin_only'] = False
|
||
device_config['allowed_users'] = allowed_users
|
||
else:
|
||
tool.reply("❌ 无效选择，已取消")
|
||
return
|
||
|
||
# 生成设备ID
|
||
device_id = f"ssh_{int(time.time())}"
|
||
|
||
save_device(device_id, device_config)
|
||
tool.reply(f"✅ SSH设备 '{name}' 添加成功！")
|
||
|
||
|
||
|
||
|
||
def show_device_list(tool: TOOL):
|
||
"""显示设备列表"""
|
||
device_list = get_device_list()
|
||
|
||
if not device_list:
|
||
tool.reply("❌ 暂无设备")
|
||
return
|
||
|
||
text = "当前设备列表：\n"
|
||
for i, device in enumerate(device_list, 1):
|
||
text += f"{i}. {device['name']}\n"
|
||
|
||
tool.reply(text)
|
||
|
||
|
||
def view_edit_device(tool: TOOL, device_list: list, device_index: int):
|
||
"""查看/编辑设备配置"""
|
||
selected_device = device_list[device_index]
|
||
device_id = selected_device['id']
|
||
device_name = selected_device['name']
|
||
device_type = selected_device['type']
|
||
|
||
# 获取设备详细配置
|
||
devices = get_all_devices()
|
||
device_config = devices.get(device_id)
|
||
|
||
if not device_config:
|
||
tool.reply("❌ 设备配置不存在")
|
||
return
|
||
|
||
# 显示设备配置
|
||
config_text = f"设备配置详情：\n"
|
||
config_text += f"设备名称: {device_config.get('name', '未知')}\n"
|
||
config_text += f"设备类型: {device_config.get('type', '未知')}\n"
|
||
|
||
# 显示权限信息
|
||
admin_only = device_config.get('admin_only', False)
|
||
allowed_users = device_config.get('allowed_users', [])
|
||
if isinstance(allowed_users, str):
|
||
allowed_users = [u.strip() for u in allowed_users.split(',') if u.strip()]
|
||
if not isinstance(allowed_users, list):
|
||
allowed_users = []
|
||
|
||
if admin_only:
|
||
config_text += f"权限设置: 仅管理员可用\n"
|
||
elif not allowed_users:
|
||
config_text += f"权限设置: 所有用户可用\n"
|
||
else:
|
||
config_text += f"权限设置: 指定用户可用 ({', '.join(allowed_users)})\n"
|
||
|
||
if device_type == 'ADB':
|
||
config_text += f"ADB地址: {device_config.get('address', '未设置')}\n"
|
||
elif device_type == 'SSH':
|
||
config_text += f"SSH主机: {device_config.get('hostname', '未设置')}\n"
|
||
config_text += f"SSH用户: {device_config.get('username', 'root')}\n"
|
||
config_text += f"SSH密码: {'******' if device_config.get('password') else '未设置'}\n"
|
||
config_text += f"SSH端口: {device_config.get('port', 22)}\n"
|
||
config_text += f"重启命令: {device_config.get('command', 'reboot')}\n"
|
||
config_text += f"超时时间: {device_config.get('timeout', 30)}秒\n"
|
||
|
||
config_text += "\n操作选项：\n1. 修改配置\n2. 删除设备\n0. 返回"
|
||
|
||
tool.reply(config_text)
|
||
|
||
try:
|
||
choice = sender.listen(60000)
|
||
if not choice or choice == '0' or choice == 'q' or choice == 'error':
|
||
tool.reply("已返回")
|
||
return
|
||
|
||
if choice == '1':
|
||
edit_device_config(tool, device_id, device_config)
|
||
elif choice == '2':
|
||
# 确认删除
|
||
tool.reply(f"确认删除设备 '{device_name}' 吗？(y/n)")
|
||
confirm = sender.listen(30000)
|
||
if confirm and confirm.lower() in ['y', 'yes', '是']:
|
||
delete_device(device_id)
|
||
tool.reply(f"✅ 设备 '{device_name}' 已删除")
|
||
else:
|
||
tool.reply("已取消删除")
|
||
else:
|
||
tool.reply("❌ 无效选择")
|
||
except Exception as e:
|
||
tool.reply(f"❌ 操作出错: {e}")
|
||
|
||
|
||
def edit_device_config(tool: TOOL, device_id: str, device_config: dict):
|
||
"""编辑设备配置"""
|
||
device_type = device_config.get('type')
|
||
device_name = device_config.get('name', '未知设备')
|
||
|
||
tool.reply(f"正在编辑设备 '{device_name}' 的配置...")
|
||
|
||
if device_type == 'adb':
|
||
edit_adb_device_config(tool, device_id, device_config)
|
||
elif device_type == 'ssh':
|
||
edit_ssh_device_config(tool, device_id, device_config)
|
||
else:
|
||
tool.reply("❌ 未知设备类型")
|
||
|
||
|
||
def edit_adb_device_config(tool: TOOL, device_id: str, device_config: dict):
|
||
"""编辑ADB设备配置"""
|
||
tool.reply("请选择要修改的配置项：\n1. 设备名称\n2. ADB地址\n3. 权限设置\n0. 返回")
|
||
|
||
try:
|
||
choice = sender.listen(60000)
|
||
if not choice or choice == '0' or choice == 'q' or choice == 'error':
|
||
tool.reply("已返回")
|
||
return
|
||
|
||
if choice == '1':
|
||
tool.reply("请输入新的设备名称（q:退出）")
|
||
new_name = sender.listen(60000)
|
||
if new_name and new_name != 'q' and new_name != 'error':
|
||
device_config['name'] = new_name
|
||
save_device(device_id, device_config)
|
||
tool.reply(f"✅ 设备名称已更新为: {new_name}")
|
||
else:
|
||
tool.reply("已取消")
|
||
elif choice == '2':
|
||
tool.reply("请输入新的ADB地址（示例：192.168.1.100:5555，q:退出）")
|
||
new_address = sender.listen(60000)
|
||
if new_address and new_address != 'q' and new_address != 'error':
|
||
device_config['address'] = new_address
|
||
save_device(device_id, device_config)
|
||
tool.reply(f"✅ ADB地址已更新为: {new_address}")
|
||
else:
|
||
tool.reply("已取消")
|
||
elif choice == '3':
|
||
tool.reply("请设置设备权限：\n1. 仅管理员可用\n2. 所有用户可用\n3. 指定用户可用\n0. 取消")
|
||
perm_choice = sender.listen(60000)
|
||
if not perm_choice or perm_choice == '0' or perm_choice == 'q' or perm_choice == 'error':
|
||
tool.reply("已取消")
|
||
return
|
||
|
||
if perm_choice == '1':
|
||
device_config['admin_only'] = True
|
||
device_config['allowed_users'] = []
|
||
save_device(device_id, device_config)
|
||
tool.reply("✅ 权限已设置为：仅管理员可用")
|
||
elif perm_choice == '2':
|
||
device_config['admin_only'] = False
|
||
device_config['allowed_users'] = []
|
||
save_device(device_id, device_config)
|
||
tool.reply("✅ 权限已设置为：所有用户可用")
|
||
elif perm_choice == '3':
|
||
tool.reply("请输入允许访问的用户ID（多个用户用逗号分隔，q:退出）")
|
||
users_input = sender.listen(60000)
|
||
if not users_input or users_input == 'q' or users_input == 'error':
|
||
tool.reply("已取消")
|
||
return
|
||
allowed_users = [u.strip() for u in users_input.split(',') if u.strip()]
|
||
device_config['admin_only'] = False
|
||
device_config['allowed_users'] = allowed_users
|
||
save_device(device_id, device_config)
|
||
tool.reply(f"✅ 权限已设置为：指定用户可用 ({', '.join(allowed_users)})")
|
||
else:
|
||
tool.reply("❌ 无效选择")
|
||
else:
|
||
tool.reply("❌ 无效选择")
|
||
except Exception as e:
|
||
tool.reply(f"❌ 操作出错: {e}")
|
||
|
||
|
||
def edit_ssh_device_config(tool: TOOL, device_id: str, device_config: dict):
|
||
"""编辑SSH设备配置"""
|
||
tool.reply("请选择要修改的配置项：\n1. 设备名称\n2. SSH主机\n3. SSH用户\n4. SSH密码\n5. SSH端口\n6. 重启命令\n7. 超时时间\n8. 权限设置\n0. 返回")
|
||
|
||
try:
|
||
choice = sender.listen(60000)
|
||
if not choice or choice == '0' or choice == 'q' or choice == 'error':
|
||
tool.reply("已返回")
|
||
return
|
||
|
||
if choice == '1':
|
||
tool.reply("请输入新的设备名称（q:退出）")
|
||
new_name = sender.listen(60000)
|
||
if new_name and new_name != 'q' and new_name != 'error':
|
||
device_config['name'] = new_name
|
||
save_device(device_id, device_config)
|
||
tool.reply(f"✅ 设备名称已更新为: {new_name}")
|
||
else:
|
||
tool.reply("已取消")
|
||
elif choice == '2':
|
||
tool.reply("请输入新的SSH主机地址（q:退出）")
|
||
new_hostname = sender.listen(60000)
|
||
if new_hostname and new_hostname != 'q' and new_hostname != 'error':
|
||
device_config['hostname'] = new_hostname
|
||
save_device(device_id, device_config)
|
||
tool.reply(f"✅ SSH主机已更新为: {new_hostname}")
|
||
else:
|
||
tool.reply("已取消")
|
||
elif choice == '3':
|
||
tool.reply("请输入新的SSH用户名（q:退出）")
|
||
new_username = sender.listen(60000)
|
||
if new_username and new_username != 'q' and new_username != 'error':
|
||
device_config['username'] = new_username
|
||
save_device(device_id, device_config)
|
||
tool.reply(f"✅ SSH用户已更新为: {new_username}")
|
||
else:
|
||
tool.reply("已取消")
|
||
elif choice == '4':
|
||
tool.reply("请输入新的SSH密码（q:退出）")
|
||
new_password = sender.listen(60000)
|
||
if new_password and new_password != 'q' and new_password != 'error':
|
||
device_config['password'] = new_password
|
||
save_device(device_id, device_config)
|
||
tool.reply("✅ SSH密码已更新")
|
||
else:
|
||
tool.reply("已取消")
|
||
elif choice == '5':
|
||
tool.reply("请输入新的SSH端口（q:退出）")
|
||
new_port = sender.listen(60000)
|
||
if new_port and new_port != 'q' and new_port != 'error':
|
||
if new_port.isdigit():
|
||
device_config['port'] = int(new_port)
|
||
save_device(device_id, device_config)
|
||
tool.reply(f"✅ SSH端口已更新为: {new_port}")
|
||
else:
|
||
tool.reply("❌ 端口必须是数字")
|
||
else:
|
||
tool.reply("已取消")
|
||
elif choice == '6':
|
||
tool.reply("请输入新的重启命令（q:退出）")
|
||
new_command = sender.listen(60000)
|
||
if new_command and new_command != 'q' and new_command != 'error':
|
||
device_config['command'] = new_command
|
||
save_device(device_id, device_config)
|
||
tool.reply(f"✅ 重启命令已更新为: {new_command}")
|
||
else:
|
||
tool.reply("已取消")
|
||
elif choice == '7':
|
||
tool.reply("请输入新的超时时间（秒，q:退出）")
|
||
new_timeout = sender.listen(60000)
|
||
if new_timeout and new_timeout != 'q' and new_timeout != 'error':
|
||
if new_timeout.isdigit():
|
||
device_config['timeout'] = int(new_timeout)
|
||
save_device(device_id, device_config)
|
||
tool.reply(f"✅ 超时时间已更新为: {new_timeout}秒")
|
||
else:
|
||
tool.reply("❌ 超时时间必须是数字")
|
||
else:
|
||
tool.reply("已取消")
|
||
elif choice == '8':
|
||
tool.reply("请设置设备权限：\n1. 仅管理员可用\n2. 所有用户可用\n3. 指定用户可用\n0. 取消")
|
||
perm_choice = sender.listen(60000)
|
||
if not perm_choice or perm_choice == '0' or perm_choice == 'q' or perm_choice == 'error':
|
||
tool.reply("已取消")
|
||
return
|
||
|
||
if perm_choice == '1':
|
||
device_config['admin_only'] = True
|
||
device_config['allowed_users'] = []
|
||
save_device(device_id, device_config)
|
||
tool.reply("✅ 权限已设置为：仅管理员可用")
|
||
elif perm_choice == '2':
|
||
device_config['admin_only'] = False
|
||
device_config['allowed_users'] = []
|
||
save_device(device_id, device_config)
|
||
tool.reply("✅ 权限已设置为：所有用户可用")
|
||
elif perm_choice == '3':
|
||
tool.reply("请输入允许访问的用户ID（多个用户用逗号分隔，q:退出）")
|
||
users_input = sender.listen(60000)
|
||
if not users_input or users_input == 'q' or users_input == 'error':
|
||
tool.reply("已取消")
|
||
return
|
||
allowed_users = [u.strip() for u in users_input.split(',') if u.strip()]
|
||
device_config['admin_only'] = False
|
||
device_config['allowed_users'] = allowed_users
|
||
save_device(device_id, device_config)
|
||
tool.reply(f"✅ 权限已设置为：指定用户可用 ({', '.join(allowed_users)})")
|
||
else:
|
||
tool.reply("❌ 无效选择")
|
||
else:
|
||
tool.reply("❌ 无效选择")
|
||
except Exception as e:
|
||
tool.reply(f"❌ 操作出错: {e}")
|
||
|
||
|
||
def delete_device_by_number(tool: TOOL, device_list: list, device_index: int):
|
||
"""按序号删除设备"""
|
||
selected_device = device_list[device_index]
|
||
device_id = selected_device['id']
|
||
device_name = selected_device['name']
|
||
|
||
# 确认删除
|
||
tool.reply(f"确认删除设备 '{device_name}' 吗？(y/n)")
|
||
try:
|
||
confirm = sender.listen(30000)
|
||
if confirm and confirm.lower() in ['y', 'yes', '是']:
|
||
delete_device(device_id)
|
||
tool.reply(f"✅ 设备 '{device_name}' 已删除")
|
||
else:
|
||
tool.reply("已取消删除")
|
||
except Exception as e:
|
||
tool.reply(f"❌ 删除设备时出错: {e}")
|
||
|
||
|
||
def _safe_get_message():
|
||
try:
|
||
return sender.getMessage().strip('"')
|
||
except Exception:
|
||
return ""
|
||
|
||
|
||
if __name__ == "__main__":
|
||
tool = TOOL()
|
||
msg = _safe_get_message()
|
||
|
||
# 获取用户信息
|
||
try:
|
||
user_id = sender.getUserID()
|
||
is_admin = sender.isAdmin()
|
||
except Exception:
|
||
user_id = ""
|
||
is_admin = False
|
||
|
||
if "rb-now" in msg:
|
||
cmd_rb_now(tool, user_id, is_admin)
|
||
sys.exit(0)
|
||
|
||
if "rb管理" in msg:
|
||
cmd_rb_manage(tool, user_id, is_admin)
|
||
sys.exit(0)
|
||
|
||
sender.reply("指令不支持，请发送：rb-now 或 rb管理")
|
||
|