Hermes/autman-plugins
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add simadmin restart controls

Browse Source
This commit is contained in:
Hermes Agent
2026-05-16 14:41:44 +08:00
parent 614c043068
commit e64187c75e
2 changed files with 258 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files
plugins/simadmin/simadmin_multi.py
+129 -5
View File
@@ -8,10 +8,11 @@
#[rule: ^sim管理$]
#[rule: ^sim状态.*$]
#[rule: ^sim短信.*$]
#[rule: ^sim重启.*$]
#[priority: 999]
#[platform: qq,qb,wx,wb,tg,tb,fs,we,web,wxmp]
#[open_source: true]
#[version: 1.2.5]
#[version: 1.3.0]
#[public: false]
#[price: 0]
#[description: 管理员专用：兼容多个 SimAdmin 实例，支持多级菜单配置管理、状态查询和短信发送。配置桶 otto/simadmin_instances。]
@@ -27,6 +28,7 @@ Autman Python 插件：SimAdmin 多实例状态查询、短信发送、多级菜
- sim状态 主卡 查询指定实例
- sim短信 主卡 10086 CXLL
- sim短信 all 10086 测试短信
- sim重启 <实例名|all> <基带|服务|系统>
"""
import json
@@ -469,6 +471,105 @@ def valid_phone(phone):
return bool(re.match(r"^\+?[0-9][0-9\- ]{1,30}$", phone or ""))
RESTART_ACTIONS = {
"基带": {"path": "/api/baseband/restart", "body": {}, "label": "重启基带"},
"baseband": {"path": "/api/baseband/restart", "body": {}, "label": "重启基带"},
"服务": {"path": "/api/service/restart", "body": {}, "label": "重启服务"},
"service": {"path": "/api/service/restart", "body": {}, "label": "重启服务"},
"系统": {"path": "/api/system/reboot", "body": {"delay_seconds": 1}, "label": "重启系统"},
"设备": {"path": "/api/system/reboot", "body": {"delay_seconds": 1}, "label": "重启系统"},
"system": {"path": "/api/system/reboot", "body": {"delay_seconds": 1}, "label": "重启系统"},
"reboot": {"path": "/api/system/reboot", "body": {"delay_seconds": 1}, "label": "重启系统"},
}
def normalize_restart_action(action):
key = str(action or "").strip().lower()
for name, spec in RESTART_ACTIONS.items():
if key == name.lower():
return spec
raise SimAdminError("重启类型不支持：%s。可用：基带 / 服务 / 系统" % action)
def perform_restart(instance, action_spec):
payload = request_json(instance, "POST", action_spec["path"], action_spec.get("body", {}))
msg = payload.get("message") if isinstance(payload, dict) else ""
status = payload.get("status") if isinstance(payload, dict) else ""
if not msg:
msg = status or "请求已提交"
return msg
def restart_instances(sender, instances, action_spec):
results = []
for item in instances:
try:
msg = perform_restart(item, action_spec)
results.append("✅ %s：%s" % (item["name"], msg))
except Exception as e:
results.append("❌ %s：%s" % (item["name"], str(e)))
reply(sender, "%s 结果：\n%s" % (action_spec["label"], "\n".join(results)))
def handle_restart(sender, message):
m = re.match(r"^sim重启\s+(\S+)\s+(\S+)\s*$", message)
if not m:
raise SimAdminError("格式错误：sim重启 <实例名|all> <基带|服务|系统>")
target, action = m.group(1), m.group(2)
action_spec = normalize_restart_action(action)
selected = find_instances(load_instances(), target)
names = "、".join(x["name"] for x in selected)
warning = "确认对【%s】执行【%s】？\n" % (names, action_spec["label"])
if action_spec["path"].endswith("/system/reboot"):
warning += "系统重启会导致设备离线，请确认后再继续。输入 yes 确认"
elif target.lower() in ("all", "*", "全部"):
warning += "这是批量操作，会影响全部启用设备。输入 yes 确认"
else:
warning += "输入 yes 确认"
if not confirm(sender, warning):
reply(sender, "已取消重启操作")
return
restart_instances(sender, selected, action_spec)
def menu_restart(sender, idx=None, action_name=None):
instances = load_instances(include_disabled=True, allow_empty=True)
if not instances:
raise SimAdminError("当前没有 SimAdmin 实例")
if idx is None:
idx = choose_instance_by_menu(sender, instances, "请选择要重启的实例编号")
item = instances[idx]
if not item.get("enabled"):
raise SimAdminError("实例【%s】已禁用，请先启用后再操作" % item["name"])
if not action_name:
action_name = listen_text(sender, "请选择重启类型：\n1. 基带\n2. 服务\n3. 系统\n输入 1/2/3 或 基带/服务/系统")
action_name = {"1": "基带", "2": "服务", "3": "系统"}.get(action_name, action_name)
action_spec = normalize_restart_action(action_name)
if not confirm(sender, "确认对【%s】执行【%s】？输入 yes 确认" % (item["name"], action_spec["label"])):
reply(sender, "已取消重启操作")
return
restart_instances(sender, [item], action_spec)
def menu_restart_all(sender, action_name=None):
instances = load_instances()
if not instances:
raise SimAdminError("没有启用的 SimAdmin 实例")
if not action_name:
action_name = listen_text(sender, "请选择一键重启全部设备类型：\n1. 基带\n2. 服务\n3. 系统\n输入 1/2/3 或 基带/服务/系统")
action_name = {"1": "基带", "2": "服务", "3": "系统"}.get(action_name, action_name)
action_spec = normalize_restart_action(action_name)
names = "、".join(x["name"] for x in instances)
prompt = "确认对全部启用设备【%s】执行【%s】？\n这是批量操作" % (names, action_spec["label"])
if action_spec["path"].endswith("/system/reboot"):
prompt += "，系统重启会导致设备离线"
prompt += "。输入 yes 确认"
if not confirm(sender, prompt):
reply(sender, "已取消一键重启")
return
restart_instances(sender, instances, action_spec)
def handle_status(sender, message):
instances = load_instances()
parts = message.split(None, 1)
@@ -532,6 +633,8 @@ def menu_list(sender):
"e <编号>. 编辑实例，例如 e 1",
"d <编号>. 删除实例，例如 d 1",
"t <编号>. 测试状态，例如 t 1",
"rb <编号>. 重启实例，例如 rb 1",
"rba. 一键重启全部启用设备",
"on <编号> / off <编号>. 启用或禁用",
"r. 刷新列表",
"b. 返回上级菜单",
@@ -548,6 +651,9 @@ def menu_list(sender):
if lower in ("a", "add", "新增", "添加"):
menu_add(sender)
continue
if lower in ("rba", "restartall", "重启全部", "一键重启"):
menu_restart_all(sender)
continue
if choice.isdigit():
idx = int(choice) - 1
if 0 <= idx < len(instances):
@@ -557,7 +663,7 @@ def menu_list(sender):
continue
reply(sender, "编号超出范围")
continue
m = re.match(r"^(e|edit|编辑|d|del|delete|删除|t|test|测试|on|启用|off|禁用)\s+(\d+)$", lower)
m = re.match(r"^(e|edit|编辑|d|del|delete|删除|t|test|测试|rb|restart|重启|on|启用|off|禁用)\s+(\d+)$", lower)
if not m:
reply(sender, "无法识别的列表操作，请重新输入")
continue
@@ -572,6 +678,8 @@ def menu_list(sender):
menu_delete(sender, idx)
elif op in ("t", "test", "测试"):
menu_test(sender, idx)
elif op in ("rb", "restart", "重启"):
menu_restart(sender, idx)
elif op in ("on", "启用"):
menu_set_enabled(sender, idx, True)
elif op in ("off", "禁用"):
@@ -695,6 +803,9 @@ def instance_action_menu(sender, idx):
"3. 删除实例",
"4. 启用/禁用",
"5. 复制/新增一个实例",
"6. 重启基带",
"7. 重启服务",
"8. 重启系统",
"b. 返回实例列表",
"q. 退出配置管理",
]))
@@ -710,6 +821,12 @@ def instance_action_menu(sender, idx):
menu_toggle(sender, idx)
elif choice == "5":
menu_add(sender)
elif choice == "6":
menu_restart(sender, idx, "基带")
elif choice == "7":
menu_restart(sender, idx, "服务")
elif choice == "8":
menu_restart(sender, idx, "系统")
elif choice.lower() in ("b", "back", "返回"):
return "back"
else:
@@ -734,6 +851,7 @@ def handle_config_menu(sender):
"5. 启用/禁用实例",
"6. 测试状态查询",
"7. 导出配置 JSON",
"8. 一键重启全部启用设备",
"q. 退出",
]))
choice = listen_text(sender, "请输入菜单编号")
@@ -754,6 +872,8 @@ def handle_config_menu(sender):
menu_test(sender)
elif choice == "7":
menu_export(sender)
elif choice == "8":
menu_restart_all(sender)
else:
reply(sender, "已退出 SimAdmin 配置管理")
return
@@ -774,7 +894,8 @@ def handle_help(sender):
"- sim状态 查询全部启用实例",
"- sim状态 <实例名> 查询指定实例",
"- sim短信 <实例名|all> <号码> <内容>",
"菜单支持：查看、添加、编辑、删除、启用/禁用、测试、导出 JSON。",
"- sim重启 <实例名|all> <基带|服务|系统>",
"菜单支持：查看、添加、编辑、删除、启用/禁用、测试、导出 JSON、单设备重启、一键重启全部启用设备。",
]))
@@ -797,7 +918,7 @@ def main():
message = _safe_get_message()
# 参考已有实机插件：头注 admin 只负责触发，敏感权限在代码里判断。
# 这样即使框架层 admin 识别异常，也至少会给出“仅管理员可用”，不会表现为无回应。
if message in ("sim帮助", "sim配置", "sim管理") or message.startswith("sim状态") or message.startswith("sim短信"):
if message in ("sim帮助", "sim配置", "sim管理") or message.startswith("sim状态") or message.startswith("sim短信") or message.startswith("sim重启"):
if not _safe_is_admin():
reply(sender, "仅管理员可用")
return
@@ -813,7 +934,10 @@ def main():
if message.startswith("sim短信"):
handle_sms(sender, message)
sys.exit(0)
reply(sender, "指令不支持，请发送：sim帮助 / sim配置 / sim状态 / sim短信")
if message.startswith("sim重启"):
handle_restart(sender, message)
sys.exit(0)
reply(sender, "指令不支持，请发送：sim帮助 / sim配置 / sim状态 / sim短信 / sim重启")
except SimAdminError as e:
reply(sender, "SimAdmin插件提示：" + str(e))
except Exception as e: