# [title: SimAdmin] #[language: python] #[service: BOSS] #[disable: false] #[admin: true] #[rule: ^sim帮助$] #[rule: ^sim配置$] #[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.3.5] #[public: false] #[price: 0] #[description: 管理员专用：兼容多个 SimAdmin 实例，支持多级菜单配置管理、状态查询和短信发送。配置桶 otto/simadmin_instances。] """ Autman Python 插件：SimAdmin 多实例状态查询、短信发送、多级菜单配置管理。 配置桶：otto / simadmin_instances 指令： - sim帮助 - sim配置 / sim管理 进入多级菜单配置管理 - sim状态 查询全部实例 - sim状态 主卡 查询指定实例 - sim短信 主卡 10086 CXLL - sim短信 all 10086 测试短信 - sim重启基带 / sim重启服务 / sim重启系统 选择某个实例或全部启用实例 """ import json import re import sys import traceback import urllib.error import urllib.parse import urllib.request import middleware BUCKET = "otto" CONFIG_KEY = "simadmin_instances" DEFAULT_TIMEOUT = 10 LISTEN_TIMEOUT_MS = 120000 class SimAdminError(Exception): pass sender = middleware.Sender(middleware.getSenderID()) def reply(sender_obj, text): sender_obj.reply(str(text)) def listen_text(sender, prompt, allow_empty=False, default=None): if prompt: reply(sender, prompt) value = sender.listen(LISTEN_TIMEOUT_MS) if value is None: raise SimAdminError("等待输入超时，已退出") value = str(value).strip().strip('"') # Autman listen 超时/无效输入在不少实机环境返回字符串 error，而不是 None。 if value.lower() == "error": raise SimAdminError("等待输入超时或监听失败，已退出") if value == "" and not allow_empty: raise SimAdminError("输入不能为空，已退出") if value.lower() in ("q", "quit", "exit", "退出", "取消"): raise SimAdminError("已取消") return value def listen_with_default(sender, prompt, default_value, default_label=None, allow_clear=False): """Autman/聊天端不能发送空回车，用显式指令代替默认/跳过/清空。""" label = default_label if default_label is not None else str(default_value) tips = [prompt, "输入 d/default/默认 使用默认值：%s" % label] if allow_clear: tips.append("输入 clear/清空 表示留空") value = listen_text(sender, "\n".join(tips), allow_empty=False) lower = value.lower() if lower in ("d", "default", "默认", "跳过", "skip", "s"): return default_value if allow_clear and lower in ("clear", "清空", "空", "none", "null", "无"): return "" return value def confirm(sender, prompt="确认执行？输入 yes/y/确认，输入 n/no/取消 放弃"): value = listen_text(sender, prompt, allow_empty=False) return value.lower() in ("yes", "y", "确认", "是") def is_blank(value): return value is None or str(value).strip() == "" def mask_secret(value): if not value: return "" s = str(value) if len(s) <= 8: return "***" return s[:3] + "***" + s[-3:] def normalize_base_url(url): if is_blank(url): raise SimAdminError("base_url 不能为空") url = str(url).strip().rstrip("/") if not re.match(r"^https?://", url, re.I): raise SimAdminError("base_url 必须以 http:// 或 https:// 开头: " + url) return url def parse_bool(value, default=False): if value is None: return default s = str(value).strip().lower() if s in ("1", "true", "yes", "y", "on", "启用", "是"): return True if s in ("0", "false", "no", "n", "off", "禁用", "否"): return False return default def normalize_instance(raw, index=0): if not isinstance(raw, dict): raise SimAdminError("实例配置必须是对象") name = str(raw.get("name") or raw.get("alias") or raw.get("id") or ("sim" + str(index + 1))).strip() base_url = normalize_base_url(raw.get("base_url") or raw.get("url") or raw.get("host")) timeout = raw.get("timeout", raw.get("timeout_seconds", DEFAULT_TIMEOUT)) try: timeout = int(timeout) except Exception: timeout = DEFAULT_TIMEOUT timeout = max(3, min(timeout, 60)) token = str(raw.get("token") or raw.get("access_token") or raw.get("api_key") or "").strip() headers = raw.get("headers") if isinstance(raw.get("headers"), dict) else {} enabled = parse_bool(raw.get("enabled"), True) return { "name": name, "base_url": base_url, "timeout": timeout, "token": token, "headers": headers, "enabled": enabled, } def parse_line_config(text): instances = [] for line_no, line in enumerate(str(text).splitlines(), 1): line = line.strip() if not line or line.startswith("#"): continue if "=" not in line: raise SimAdminError("第 %s 行配置缺少 =" % line_no) name, rest = line.split("=", 1) parts = [p.strip() for p in rest.split(",") if p.strip()] if not parts: raise SimAdminError("第 %s 行 URL 不能为空" % line_no) raw = {"name": name.strip(), "base_url": parts[0]} for part in parts[1:]: if "=" in part: k, v = part.split("=", 1) raw[k.strip()] = v.strip() instances.append(normalize_instance(raw, len(instances))) return instances def load_instances(include_disabled=False, allow_empty=False): raw = middleware.bucketGet(BUCKET, CONFIG_KEY) if is_blank(raw): if allow_empty: return [] raise SimAdminError("未配置 SimAdmin 实例。发送 sim管理 通过多级菜单添加。") text = str(raw).strip() if text.startswith("[") or text.startswith("{"): try: data = json.loads(text) except Exception as e: raise SimAdminError("SimAdmin 配置 JSON 解析失败: " + str(e)) if isinstance(data, dict): if isinstance(data.get("instances"), list): data = data["instances"] else: data = [data] if not isinstance(data, list): raise SimAdminError("SimAdmin 配置必须是 JSON 数组或对象") instances = [normalize_instance(item, i) for i, item in enumerate(data)] else: instances = parse_line_config(text) if not include_disabled: instances = [x for x in instances if x.get("enabled")] if not instances and not allow_empty: raise SimAdminError("没有启用的 SimAdmin 实例") validate_unique_names(instances) return instances def validate_unique_names(instances): names = set() for item in instances: if item["name"] in names: raise SimAdminError("SimAdmin 实例名称重复: " + item["name"]) names.add(item["name"]) def save_instances(instances): validate_unique_names(instances) normalized = [normalize_instance(item, i) for i, item in enumerate(instances)] middleware.bucketSet(BUCKET, CONFIG_KEY, json.dumps(normalized, ensure_ascii=False, indent=2)) def display_host(url): try: parsed = urllib.parse.urlparse(str(url or "")) return parsed.hostname or str(url or "-") except Exception: return str(url or "-") def instance_brief(item, idx=None): prefix = "" if idx is None else "%s. " % idx state = "启用" if item.get("enabled") else "禁用" token = mask_secret(item.get("token")) or "未设置" return "%s%s [%s]\n 主机:%s timeout=%ss token=%s" % ( prefix, item["name"], state, display_host(item.get("base_url")), item["timeout"], token ) def instance_short(item, idx=None): prefix = "" if idx is None else "%s. " % idx state = "启用" if item.get("enabled") else "禁用" return "%s%s [%s]" % (prefix, item["name"], state) def request_json(instance, method, path, body=None): url = instance["base_url"] + path data = None headers = {"Accept": "application/json"} if body is not None: data = json.dumps(body, ensure_ascii=False).encode("utf-8") headers["Content-Type"] = "application/json" headers.update(instance.get("headers") or {}) token = instance.get("token") or "" if token: headers.setdefault("Authorization", "Bearer " + token) headers.setdefault("X-API-Token", token) req = urllib.request.Request(url, data=data, headers=headers, method=method.upper()) try: with urllib.request.urlopen(req, timeout=instance["timeout"]) as resp: raw = resp.read().decode("utf-8", errors="replace") status_code = getattr(resp, "status", resp.getcode()) except urllib.error.HTTPError as e: raw = e.read().decode("utf-8", errors="replace") raise SimAdminError("HTTP %s: %s" % (e.code, raw[:300])) except Exception as e: raise SimAdminError(str(e)) if not raw: return {"status_code": status_code} try: payload = json.loads(raw) except Exception: return {"status_code": status_code, "raw": raw} if isinstance(payload, dict) and payload.get("status") == "error": raise SimAdminError(str(payload.get("message") or payload)) return payload def unwrap_data(payload): if isinstance(payload, dict) and isinstance(payload.get("data"), dict): return payload.get("data") return payload if isinstance(payload, dict) else {} def pick(data, *keys, **kwargs): default = kwargs.get("default", "-") if not isinstance(data, dict): return default for key in keys: if key in data and data[key] not in (None, ""): return data[key] return default def fmt_percent(value): if value in (None, "", "-"): return "-" try: n = float(value) if n <= 1: n *= 100 return "%.1f%%" % n except Exception: return str(value) def fmt_bytes(value): if value in (None, "", "-"): return "-" try: n = float(value) except Exception: return str(value) units = ["B", "KB", "MB", "GB", "TB"] idx = 0 while n >= 1024 and idx < len(units) - 1: n /= 1024.0 idx += 1 if idx == 0: return "%d%s" % (int(n), units[idx]) return "%.1f%s" % (n, units[idx]) def fmt_speed(value): if value in (None, "", "-"): return "-" return fmt_bytes(value) + "/s" def fmt_bool(value, yes="是", no="否"): if isinstance(value, bool): return yes if value else no if value in (None, "", "-"): return "-" text = str(value).strip().lower() if text in ("true", "1", "yes", "on", "online", "powered"): return yes if text in ("false", "0", "no", "off", "offline"): return no return str(value) def fmt_temp_list(value): if value in (None, "", "-"): return "-" if isinstance(value, list): if not value: return "-" parts = [] for item in value: if isinstance(item, dict): name = pick(item, "type", "zone", default="温度") temp = pick(item, "temperature", "temp", default="-") if temp != "-": parts.append("%s %.1f℃" % (name, float(temp))) elif item not in (None, ""): parts.append(str(item)) return " / ".join(parts[:3]) if parts else "-" try: return "%.1f℃" % float(value) except Exception: return str(value) def first_list_value(data, key): value = pick(data, key, default=[]) if isinstance(value, list) and value: return "、".join(str(x) for x in value if x not in (None, "")) or "-" return "-" if value in (None, "", []) else str(value) def get_nested(data, *keys): cur = data for key in keys: if not isinstance(cur, dict): return {} cur = cur.get(key) return cur if isinstance(cur, dict) else {} def format_instance_status(instance): lines = ["【%s】%s" % (instance["name"], "（已禁用）" if not instance.get("enabled", True) else "")] try: health = request_json(instance, "GET", "/api/health") device = unwrap_data(request_json(instance, "GET", "/api/device")) sim = unwrap_data(request_json(instance, "GET", "/api/sim")) network = unwrap_data(request_json(instance, "GET", "/api/network")) data_conn = unwrap_data(request_json(instance, "GET", "/api/data")) stats = unwrap_data(request_json(instance, "GET", "/api/stats")) sms_stats = unwrap_data(request_json(instance, "GET", "/api/sms/stats")) health_status = pick(health, "status", default="ok") version = pick(health, "version", default="-") platform = pick(health, "platform", default="-") lines.append("服务: %s 版本:%s 平台:%s" % (health_status, version, platform)) lines.append("设备: %s %s 在线:%s 供电:%s" % ( pick(device, "manufacturer", default="-"), pick(device, "model", default="-"), fmt_bool(pick(device, "online", default="-")), fmt_bool(pick(device, "powered", default="-")), )) imei = pick(device, "imei", "IMEI", default="-") if imei != "-": lines.append("IMEI: %s" % imei) phone = first_list_value(sim, "phone_numbers") if phone == "-": phone = pick(sim, "phone_number", "msisdn", "number", "phone", default="号码未知") sim_present = fmt_bool(pick(sim, "present", default="-"), yes="已插卡", no="未插卡") lines.append("SIM: %s 号码:%s" % (sim_present, phone)) lines.append("ICCID: %s" % pick(sim, "iccid", "ICCID", default="-")) imsi = pick(sim, "imsi", "IMSI", default="-") if imsi != "-": lines.append("IMSI: %s" % imsi) signal = pick(network, "signal_strength", "signal", default="-") signal_text = "%s%%" % signal if signal != "-" else "-" lines.append("网络: %s(%s/%s) %s 信号:%s 数据:%s" % ( pick(network, "operator", "operator_name", "network_operator", "provider", default="-"), pick(network, "mcc", default="-"), pick(network, "mnc", default="-"), pick(network, "registration_status", "registration_state", "reg_state", "status", "state", default="-"), signal_text, fmt_bool(pick(data_conn, "active", default="-"), yes="已连接", no="未连接"), )) tech = pick(network, "technology_preference", "technology", "access_tech", default="-") if tech != "-": lines.append("制式: %s" % tech) cpu_load = get_nested(stats, "cpu_load") memory = get_nested(stats, "memory") uptime = get_nested(stats, "uptime") lines.append("系统: CPU %s(%s核) 内存 %s/%s 运行 %s" % ( fmt_percent(pick(cpu_load, "load_percent", "cpu_percent", default=pick(stats, "cpu_usage", "cpu_percent", "cpu", default="-"))), pick(cpu_load, "core_count", default="-"), fmt_percent(pick(memory, "used_percent", default=pick(stats, "memory_usage", "memory_percent", "mem", default="-"))), fmt_bytes(pick(memory, "total_bytes", default="-")), pick(uptime, "uptime_formatted", default=pick(stats, "uptime", "uptime_text", "system_uptime", default="-")), )) net_speed = get_nested(stats, "network_speed") interfaces = net_speed.get("interfaces") if isinstance(net_speed, dict) else None if isinstance(interfaces, list) and interfaces: iface = interfaces[0] lines.append("流量: %s ↓%s ↑%s 累计↓%s ↑%s" % ( pick(iface, "interface", default="-"), fmt_speed(pick(iface, "rx_bytes_per_sec", default="-")), fmt_speed(pick(iface, "tx_bytes_per_sec", default="-")), fmt_bytes(pick(iface, "total_rx_bytes", default="-")), fmt_bytes(pick(iface, "total_tx_bytes", default="-")), )) disks = stats.get("disk") if isinstance(stats, dict) else None if isinstance(disks, list) and disks: main_disk = disks[0] lines.append("磁盘: %s %s/%s" % ( pick(main_disk, "mount_point", default="/"), fmt_percent(pick(main_disk, "used_percent", default="-")), fmt_bytes(pick(main_disk, "total_bytes", default="-")), )) temp_text = fmt_temp_list(stats.get("temperature") if isinstance(stats, dict) else pick(stats, "temperature", "temp", "cpu_temperature", default="-")) if temp_text != "-": lines.append("温度: %s" % temp_text) total = pick(sms_stats, "total", default="-") incoming = pick(sms_stats, "incoming", "received", "received_count", "inbox", default="-") outgoing = pick(sms_stats, "outgoing", "sent", "sent_count", "outbox", default="-") lines.append("短信: 总 %s / 收 %s / 发 %s" % (total, incoming, outgoing)) except Exception as e: lines.append("查询失败: " + str(e)) return "\n".join(lines) def find_instances(instances, target): target = (target or "").strip() if target in ("", "全部", "all", "ALL", "*"): return instances matched = [x for x in instances if x["name"] == target] if matched: return matched matched = [x for x in instances if target.lower() in x["name"].lower()] if matched: return matched raise SimAdminError("未找到 SimAdmin 实例: %s。可用: %s" % (target, ", ".join(x["name"] for x in instances))) 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 choose_restart_targets(sender, action_spec): all_instances = load_instances(include_disabled=True, allow_empty=True) enabled_instances = [x for x in all_instances if x.get("enabled")] if not enabled_instances: raise SimAdminError("没有启用的 SimAdmin 实例") lines = ["%s：请选择设备" % action_spec["label"]] for idx, item in enumerate(all_instances, 1): lines.append(instance_short(item, idx)) lines.extend([ "", "a. 全部启用设备", "q. 取消", ]) choice = listen_text(sender, "\n".join(lines), allow_empty=False).strip() lower = choice.lower() if lower in ("a", "all", "全部", "所有", "*"): return enabled_instances, True if not choice.isdigit(): raise SimAdminError("请输入实例编号或 a/全部") idx = int(choice) - 1 if idx < 0 or idx >= len(all_instances): raise SimAdminError("编号超出范围") item = all_instances[idx] if not item.get("enabled"): raise SimAdminError("实例【%s】已禁用，请先启用后再操作" % item["name"]) return [item], False def handle_restart(sender, message): text = str(message or "").strip() direct = re.match(r"^sim重启(基带|服务|系统)$", text) legacy = re.match(r"^sim重启\s+(\S+)\s+(\S+)\s*$", text) if direct: action_spec = normalize_restart_action(direct.group(1)) selected, is_all = choose_restart_targets(sender, action_spec) elif legacy: target, action = legacy.group(1), legacy.group(2) action_spec = normalize_restart_action(action) selected = find_instances(load_instances(), target) is_all = target.lower() in ("all", "*", "全部") else: raise SimAdminError("格式错误：请发送 sim重启基带 / sim重启服务 / sim重启系统") 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) 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) restart_instances(sender, instances, action_spec) def handle_status(sender, message): instances = load_instances() parts = message.split(None, 1) target = parts[1].strip() if len(parts) > 1 else "" selected = find_instances(instances, target) if target in ("", "全部", "all", "ALL", "*") and len(selected) > 1: for item in selected: reply(sender, format_instance_status(item)) return reply(sender, "\n\n".join(format_instance_status(x) for x in selected)) def handle_sms(sender, message): m = re.match(r"^sim短信\s+(\S+)\s+(\S+)\s+([\s\S]+)$", message) if not m: raise SimAdminError("格式错误：sim短信 <实例名|all> <手机号> <内容>") target, phone, content = m.group(1), m.group(2), m.group(3).strip() if not valid_phone(phone): raise SimAdminError("手机号/号码格式不合法: " + phone) if not content: raise SimAdminError("短信内容不能为空") if len(content) > 1000: raise SimAdminError("短信内容过长，已拒绝发送") selected = find_instances(load_instances(), target) results = [] for instance in selected: try: payload = request_json(instance, "POST", "/api/sms/send", {"phone_number": phone, "content": content}) path = pick(unwrap_data(payload), "path", "message_path", "id", default="") results.append("✅ %s 发送成功%s" % (instance["name"], (" path=" + str(path)) if path else "")) except Exception as e: results.append("❌ %s 发送失败: %s" % (instance["name"], str(e))) reply(sender, "短信发送结果：\n" + "\n".join(results)) def choose_instance_by_menu(sender, instances, prompt="请选择实例编号"): if not instances: raise SimAdminError("暂无实例，请先添加") lines = [prompt] for idx, item in enumerate(instances, 1): lines.append(instance_short(item, idx)) lines.append("q. 返回") value = listen_text(sender, "\n".join(lines)) if not value.isdigit(): raise SimAdminError("编号无效") idx = int(value) if idx < 1 or idx > len(instances): raise SimAdminError("编号超出范围") return idx - 1 def menu_list(sender): """实例列表页本身就是二级操作页：看完列表后可直接编辑/新增/删除/测试。""" while True: instances = load_instances(include_disabled=True, allow_empty=True) lines = ["SimAdmin 实例列表："] if instances: lines.extend(instance_brief(item, idx) for idx, item in enumerate(instances, 1)) else: lines.append("当前没有 SimAdmin 实例。") lines.extend([ "", "列表操作：", "输入实例编号：进入该实例操作菜单", "a. 新增实例", "e <编号>. 编辑实例，例如 e 1", "d <编号>. 删除实例，例如 d 1", "t <编号>. 测试状态，例如 t 1", "rb <编号>. 重启实例，例如 rb 1", "rba. 一键重启全部启用设备", "on <编号> / off <编号>. 启用或禁用", "r. 刷新列表", "b. 返回上级菜单", "q. 退出配置管理", ]) choice = listen_text(sender, "\n".join(lines), allow_empty=True, default="r").strip() lower = choice.lower() if lower in ("r", "刷新"): continue if lower in ("b", "back", "返回"): return "back" if lower in ("q", "quit", "exit", "退出"): return "quit" 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): action = instance_action_menu(sender, idx) if action == "quit": return "quit" continue reply(sender, "编号超出范围") continue m = re.match(r"^(e|edit|编辑|d|del|delete|删除|t|test|测试|rb|restart|重启|on|启用|off|禁用)\s+(\d+)$", lower) if not m: reply(sender, "无法识别的列表操作，请重新输入") continue op, idx_text = m.group(1), m.group(2) idx = int(idx_text) - 1 if idx < 0 or idx >= len(instances): reply(sender, "编号超出范围") continue if op in ("e", "edit", "编辑"): menu_edit(sender, idx) elif op in ("d", "del", "delete", "删除"): 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", "禁用"): menu_set_enabled(sender, idx, False) def menu_add(sender): instances = load_instances(include_disabled=True, allow_empty=True) name = listen_text(sender, "请输入实例名称，例如：主卡") if any(x["name"] == name for x in instances): raise SimAdminError("实例名称已存在: " + name) base_url = listen_text(sender, "请输入 SimAdmin 地址，例如：http://192.168.68.1:3000") timeout = listen_with_default(sender, "请输入超时时间秒数", str(DEFAULT_TIMEOUT), str(DEFAULT_TIMEOUT)) token = listen_with_default(sender, "请输入 token", "", "留空", allow_clear=True) enabled_text = listen_with_default(sender, "是否启用？输入 1/是 启用，0/否 禁用", "1", "启用") item = normalize_instance({ "name": name, "base_url": base_url, "timeout": timeout, "token": token, "enabled": parse_bool(enabled_text, True), }, len(instances)) instances.append(item) save_instances(instances) reply(sender, "已添加 SimAdmin 实例：" + item["name"]) def menu_delete(sender, idx=None): instances = load_instances(include_disabled=True, allow_empty=True) if idx is None: idx = choose_instance_by_menu(sender, instances, "请选择要删除的实例编号") item = instances[idx] if not confirm(sender, "确认删除【%s】？输入 yes 确认" % item["name"]): reply(sender, "已取消删除") return removed = instances.pop(idx) save_instances(instances) reply(sender, "已删除 SimAdmin 实例：" + removed["name"]) def menu_edit(sender, idx=None): instances = load_instances(include_disabled=True, allow_empty=True) if idx is None: idx = choose_instance_by_menu(sender, instances, "请选择要编辑的实例编号") item = dict(instances[idx]) while True: reply(sender, "\n".join([ "编辑【%s】" % item["name"], "1. 修改名称", "2. 修改地址", "3. 修改 timeout", "4. 修改 token", "5. 切换启用/禁用（当前：%s）" % ("启用" if item.get("enabled") else "禁用"), "6. 保存并返回", "q. 放弃返回", ])) choice = listen_text(sender, "请输入选项") if choice == "1": new_name = listen_text(sender, "请输入新名称") if new_name != item["name"] and any(x["name"] == new_name for i, x in enumerate(instances) if i != idx): reply(sender, "名称已存在，请重新选择") else: item["name"] = new_name elif choice == "2": item["base_url"] = normalize_base_url(listen_text(sender, "请输入新地址")) elif choice == "3": item["timeout"] = normalize_instance({**item, "timeout": listen_text(sender, "请输入 timeout 秒数 3-60")})["timeout"] elif choice == "4": item["token"] = listen_with_default(sender, "请输入新 token", item.get("token", ""), "保持当前（%s）" % (mask_secret(item.get("token")) or "未设置"), allow_clear=True) elif choice == "5": item["enabled"] = not item.get("enabled") reply(sender, "已切换为：" + ("启用" if item["enabled"] else "禁用")) elif choice == "6": instances[idx] = normalize_instance(item, idx) save_instances(instances) reply(sender, "已保存：" + item["name"]) return else: reply(sender, "已放弃编辑") return def menu_set_enabled(sender, idx, enabled): instances = load_instances(include_disabled=True, allow_empty=True) instances[idx]["enabled"] = bool(enabled) save_instances(instances) reply(sender, "已将【%s】设置为：%s" % (instances[idx]["name"], "启用" if instances[idx]["enabled"] else "禁用")) def menu_toggle(sender, idx=None): instances = load_instances(include_disabled=True, allow_empty=True) if idx is None: idx = choose_instance_by_menu(sender, instances, "请选择要启用/禁用的实例编号") menu_set_enabled(sender, idx, not instances[idx].get("enabled")) def menu_test(sender, idx=None): instances = load_instances(include_disabled=True, allow_empty=True) if idx is None: idx = choose_instance_by_menu(sender, instances, "请选择要测试连通性的实例编号") item = instances[idx] reply(sender, format_instance_status(item)) def instance_action_menu(sender, idx): while True: instances = load_instances(include_disabled=True, allow_empty=True) if idx < 0 or idx >= len(instances): reply(sender, "实例已不存在，返回列表") return "back" item = instances[idx] reply(sender, "\n".join([ "实例操作：【%s】" % item["name"], instance_short(item), "1. 测试状态", "2. 编辑实例", "3. 删除实例", "4. 启用/禁用", "5. 复制/新增一个实例", "6. 重启基带", "7. 重启服务", "8. 重启系统", "b. 返回实例列表", "q. 退出配置管理", ])) choice = listen_text(sender, "请输入操作编号") if choice == "1": menu_test(sender, idx) elif choice == "2": menu_edit(sender, idx) elif choice == "3": menu_delete(sender, idx) return "back" elif choice == "4": 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: return "quit" def menu_export(sender): instances = load_instances(include_disabled=True, allow_empty=True) reply(sender, "当前配置 JSON：\n" + json.dumps(instances, ensure_ascii=False, indent=2)) def handle_config_menu(sender): while True: instances = load_instances(include_disabled=True, allow_empty=True) reply(sender, "\n".join([ "SimAdmin 配置管理（多级菜单）", "当前实例数：%s" % len(instances), "1. 查看实例列表", "2. 添加实例", "3. 编辑实例", "4. 删除实例", "5. 启用/禁用实例", "6. 测试状态查询", "7. 导出配置 JSON", "8. 一键重启全部启用设备", "q. 退出", ])) choice = listen_text(sender, "请输入菜单编号") if choice == "1": action = menu_list(sender) if action == "quit": reply(sender, "已退出 SimAdmin 配置管理") return elif choice == "2": menu_add(sender) elif choice == "3": menu_edit(sender) elif choice == "4": menu_delete(sender) elif choice == "5": menu_toggle(sender) elif choice == "6": menu_test(sender) elif choice == "7": menu_export(sender) elif choice == "8": menu_restart_all(sender) else: reply(sender, "已退出 SimAdmin 配置管理") return def handle_config(sender): # BOSS 希望通过多级菜单交互管理配置，因此 sim配置 直接进入菜单。 handle_config_menu(sender) def handle_help(sender): reply(sender, "\n".join([ "SimAdmin 多实例管理（仅管理员）", "配置桶：otto / simadmin_instances", "指令：", "- sim帮助", "- sim配置 / sim管理 进入多级菜单配置管理", "- sim状态 查询全部启用实例", "- sim状态 <实例名> 查询指定实例", "- sim短信 <实例名|all> <号码> <内容>", "- sim重启基带 / sim重启服务 / sim重启系统", " 二级菜单选择某个实例或全部启用实例", "菜单支持：查看、添加、编辑、删除、启用/禁用、测试、导出 JSON、单设备重启、一键重启全部启用设备。", ])) def _safe_get_message(): try: return (sender.getMessage() or "").strip().strip('"') except Exception: return "" def _safe_is_admin(): try: return bool(sender.isAdmin()) except Exception: return False def main(): try: message = _safe_get_message() # 参考已有实机插件：头注 admin 只负责触发，敏感权限在代码里判断。 # 这样即使框架层 admin 识别异常，也至少会给出“仅管理员可用”，不会表现为无回应。 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 if message == "sim帮助": handle_help(sender) sys.exit(0) if message in ("sim配置", "sim管理"): handle_config(sender) sys.exit(0) if message.startswith("sim状态"): handle_status(sender, message) sys.exit(0) if message.startswith("sim短信"): handle_sms(sender, message) sys.exit(0) 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: traceback.print_exc(file=sys.stderr) try: reply(sender, "SimAdmin插件异常：" + str(e)) except Exception: pass if __name__ == "__main__": main()