From 661d5d685a05f9ed3d2c00d213e5138edf5285f3 Mon Sep 17 00:00:00 2001 From: Hermes Agent Date: Sat, 16 May 2026 13:36:49 +0800 Subject: [PATCH] fix: add simadmin plugin to autman py directory --- README.md | 4 +- plugins/simadmin/README.md | 9 +- plugins/simadmin/simadmin_multi.py | 2 +- py/simadmin_multi.py | 682 +++++++++++++++++++++++++++++ 4 files changed, 693 insertions(+), 4 deletions(-) create mode 100644 py/simadmin_multi.py diff --git a/README.md b/README.md index f2ae893..bc1c8ae 100644 --- a/README.md +++ b/README.md @@ -7,7 +7,9 @@ Autman 插件集合仓库。 ## 目录规划 ```text -plugins/ # Autman 插件脚本 +py/ # 实机 Python 脚本插件（参考已有可用插件，优先复制这里的一层 .py 文件） +js/ # 实机 JS 插件 / 路由插件 +plugins/ # 插件说明、归档和按插件名组织的源码副本 examples/ # 示例配置 / 调用示例 docs/ # 设计说明、接口说明、参考资料 ``` diff --git a/plugins/simadmin/README.md b/plugins/simadmin/README.md index 4c95463..509cfe9 100644 --- a/plugins/simadmin/README.md +++ b/plugins/simadmin/README.md @@ -13,16 +13,21 @@ - `GET /api/sms/stats` - `POST /api/sms/send` -插件文件：`simadmin_multi.py` +插件文件： + +- 推荐实机文件：`py/simadmin_multi.py` +- 归档说明文件：`plugins/simadmin/simadmin_multi.py` ## 安装位置 -复制到 Autman 的 Python 脚本插件目录，例如： +参考仓库里已有可用插件（`py/rb-aut.py`、`py/mi10rb-aut.py`、`py/jd_rabbit_账密.py`），实机请优先复制 `py/simadmin_multi.py` 到 Autman 的 Python 脚本插件目录，例如： ```text plugin/scripts/simadmin_multi.py ``` +不要只复制 `plugins/simadmin/` 整个目录；部分 Autman 部署只扫描脚本目录的一层 `.py` 文件，子目录里的文件可能不会被加载。 + 插件头注已设置： ```text diff --git a/plugins/simadmin/simadmin_multi.py b/plugins/simadmin/simadmin_multi.py index 09dc202..db3c353 100644 --- a/plugins/simadmin/simadmin_multi.py +++ b/plugins/simadmin/simadmin_multi.py @@ -11,7 +11,7 @@ #[priority: 999] #[platform: qq,qb,wx,wb,tg,tb,fs,we,web,wxmp] #[open_source: true] -#[version: 1.2.2] +#[version: 1.2.3] #[public: false] #[price: 0] #[description: 管理员专用：兼容多个 SimAdmin 实例，支持多级菜单配置管理、状态查询和短信发送。配置桶 otto/simadmin_instances。] diff --git a/py/simadmin_multi.py b/py/simadmin_multi.py new file mode 100644 index 0000000..db3c353 --- /dev/null +++ b/py/simadmin_multi.py @@ -0,0 +1,682 @@ +#[title: SimAdmin多实例管理] +#[language: python] +#[service: BOSS] +#[disable: false] +#[admin: false] +#[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.2.3] +#[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 测试短信 +""" + +import json +import re +import sys +import traceback +import urllib.error +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("等待输入超时或监听失败，已退出") + # allow_empty=True 的场景允许直接回车/空消息走默认值。 + if value == "" and default is not None: + return default + if value == "" and not allow_empty: + raise SimAdminError("输入不能为空，已退出") + if value.lower() in ("q", "quit", "exit", "退出", "取消"): + raise SimAdminError("已取消") + return value + + +def confirm(sender, prompt="确认执行？输入 yes 确认，其他取消"): + value = listen_text(sender, prompt, allow_empty=True) + 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 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, item["base_url"], item["timeout"], token + ) + + +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 format_instance_status(instance): + lines = ["【%s】" % instance["name"]] + try: + health = request_json(instance, "GET", "/api/health") + sim = unwrap_data(request_json(instance, "GET", "/api/sim")) + network = unwrap_data(request_json(instance, "GET", "/api/network")) + stats = unwrap_data(request_json(instance, "GET", "/api/stats")) + sms_stats = unwrap_data(request_json(instance, "GET", "/api/sms/stats")) + lines.append("健康: %s 版本: %s" % (pick(health, "status", default="ok"), pick(health, "version", default="-"))) + lines.append("SIM: %s" % (pick(sim, "phone_number", "msisdn", "number", "phone") or "号码未知")) + lines.append("ICCID: %s" % pick(sim, "iccid", "ICCID")) + imsi = pick(sim, "imsi", "IMSI") + if imsi != "-": + lines.append("IMSI: %s" % imsi) + lines.append("网络: %s / %s / 信号:%s" % ( + pick(network, "operator", "operator_name", "network_operator", "provider"), + pick(network, "registration_state", "reg_state", "status", "state"), + pick(network, "signal_strength", "signal", "rssi", "rsrp"), + )) + lines.append("系统: CPU %s / MEM %s / 温度 %s / 运行 %s" % ( + fmt_percent(pick(stats, "cpu_usage", "cpu_percent", "cpu")), + fmt_percent(pick(stats, "memory_usage", "memory_percent", "mem")), + pick(stats, "temperature", "temp", "cpu_temperature"), + pick(stats, "uptime", "uptime_text", "system_uptime"), + )) + lines.append("短信: 收 %s / 发 %s" % ( + pick(sms_stats, "received", "received_count", "inbox", default="-"), + pick(sms_stats, "sent", "sent_count", "outbox", default="-"), + )) + 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 "")) + + +def handle_status(sender, message): + instances = load_instances() + parts = message.split(None, 1) + selected = find_instances(instances, parts[1].strip() if len(parts) > 1 else "") + 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_brief(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", + "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 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|测试|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 ("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_text(sender, "请输入超时时间秒数，直接回车默认 10", allow_empty=True, default=str(DEFAULT_TIMEOUT)) + token = listen_text(sender, "请输入 token，可直接回车留空", allow_empty=True, default="") + enabled_text = listen_text(sender, "是否启用？输入 1/是 启用，0/否 禁用；直接回车默认启用", allow_empty=True, default="1") + item = normalize_instance({ + "name": name, + "base_url": base_url, + "timeout": timeout, + "token": token, + "enabled": parse_bool(enabled_text, True), + }, len(instances)) + reply(sender, "即将添加：\n" + instance_brief(item) + "\n输入 yes 确认") + if not confirm(sender, "确认添加？"): + reply(sender, "已取消添加") + return + 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_text(sender, "请输入新 token；直接回车清空", allow_empty=True, default="") + 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_brief(item), + "1. 测试状态", + "2. 编辑实例", + "3. 删除实例", + "4. 启用/禁用", + "5. 复制/新增一个实例", + "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.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", + "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) + 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> <号码> <内容>", + "菜单支持：查看、添加、编辑、删除、启用/禁用、测试、导出 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短信"): + 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) + reply(sender, "指令不支持，请发送：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()