diff --git a/plugins/simadmin/README.md b/plugins/simadmin/README.md new file mode 100644 index 0000000..8788bce --- /dev/null +++ b/plugins/simadmin/README.md @@ -0,0 +1,125 @@ +# SimAdmin 多实例 Autman 插件 + +管理员专用 Python 插件，支持多个 SimAdmin 实例： + +- 查询全部或指定实例状态 +- 通过指定实例发送短信 +- 兼容 SimAdmin API： + - `GET /api/health` + - `GET /api/sim` + - `GET /api/network` + - `GET /api/stats` + - `GET /api/sms/stats` + - `POST /api/sms/send` + +插件文件：`simadmin_multi.py` + +## 安装位置 + +复制到 Autman 的 Python 脚本插件目录，例如： + +```text +plugin/scripts/simadmin_multi.py +``` + +插件头注已设置： + +```text +# [admin: true] +# [rule: ^sim帮助$] +# [rule: ^sim配置$] +# [rule: ^sim状态(?:\s+\S+)?$] +# [rule: ^sim短信\s+\S+\s+\S+\s+[\s\S]+$] +``` + +即：仅管理员可用。 + +## 配置 + +在 Autman 数据桶写入： + +- bucket：`otto` +- key：`simadmin_instances` + +### 推荐：JSON 数组 + +```json +[ + { + "name": "主卡", + "base_url": "http://192.168.68.1:3000", + "timeout": 10 + }, + { + "name": "副卡", + "base_url": "http://192.168.68.2:3000", + "timeout": 10, + "token": "可选，如果前面有反代/网关鉴权" + } +] +``` + +字段： + +| 字段 | 必填 | 说明 | +|---|---:|---| +| `name` | 是 | 实例别名，指令里使用 | +| `base_url` | 是 | SimAdmin 地址，不要以 `/` 结尾 | +| `timeout` | 否 | 请求超时秒数，默认 10，范围 3-60 | +| `token` | 否 | 可选；会同时带 `Authorization: Bearer ` 和 `X-API-Token` | +| `enabled` | 否 | 是否启用，默认 true | +| `headers` | 否 | 自定义请求头对象 | + +### 简写：多行 alias=url + +```text +主卡=http://192.168.68.1:3000 +副卡=http://192.168.68.2:3000,token=xxx,timeout=10 +``` + +## 指令 + +```text +sim帮助 +sim配置 +sim状态 +sim状态 主卡 +sim短信 主卡 10086 CXLL +sim短信 all 10086 测试短信 +``` + +说明： + +- `sim状态`：查询全部启用实例。 +- `sim状态 <实例名>`：查询指定实例；支持名称包含匹配。 +- `sim短信 <实例名|all> <号码> <内容>`：发送短信。 +- `all` 会向所有实例发送同一条短信，谨慎使用。 + +## 返回示例 + +```text +【主卡】 +健康: ok 版本: 1.0.0 +SIM: 13800138000 +ICCID: 8986... +IMSI: 460... +网络: China Mobile / registered / 信号:-70 +系统: CPU 12.1% / MEM 45.2% / 温度 52 / 运行 1d2h +短信: 收 12 / 发 3 +``` + +短信发送： + +```text +短信发送结果： +✅ 主卡 发送成功 path=/xxx +❌ 副卡 发送失败: timed out +``` + +## 安全约定 + +- 插件头注和代码内都做了管理员限制。 +- 不在源码中硬编码任何 SimAdmin 地址或密钥。 +- `token` 只用于兼容反代/网关鉴权；当前 SimAdmin 本身 API 如无鉴权也可留空。 +- 短信号码只允许数字、空格、`-`、国际区号 `+`。 +- 短信内容最大 1000 字，避免误发超长内容。 diff --git a/plugins/simadmin/simadmin_multi.py b/plugins/simadmin/simadmin_multi.py new file mode 100644 index 0000000..63eb140 --- /dev/null +++ b/plugins/simadmin/simadmin_multi.py @@ -0,0 +1,408 @@ +# [title: SimAdmin多实例管理] +# [language: python] +# [service: BOSS] +# [disable: false] +# [admin: true] +# [rule: ^sim帮助$] +# [rule: ^sim配置$] +# [rule: ^sim状态(?:\s+\S+)?$] +# [rule: ^sim短信\s+\S+\s+\S+\s+[\s\S]+$] +# [platform: qq,wx,tg,tb,fs] +# [open_source: true] +# [version: 1.0.0] +# [public: false] +# [price: 0] +# [description: 管理员专用：兼容多个 SimAdmin 实例，支持状态查询和短信发送。配置桶 otto/simadmin_instances，JSON数组或别名=URL多行。] + +""" +Autman Python 插件：SimAdmin 多实例状态查询和短信发送。 + +配置方式（Autman 数据桶）： +- bucket: otto +- key: simadmin_instances +- value 支持两种格式： + +1) JSON 数组（推荐）： +[ + {"name":"主卡", "base_url":"http://192.168.68.1:3000", "token":"可选", "timeout":10}, + {"name":"副卡", "base_url":"http://192.168.68.2:3000"} +] + +2) 多行简写： +主卡=http://192.168.68.1:3000 +副卡=http://192.168.68.2:3000,token=xxx,timeout=10 + +指令： +- sim帮助 +- sim配置 +- sim状态 查询全部实例 +- sim状态 主卡 查询指定实例 +- sim短信 主卡 10086 CXLL +- sim短信 all 10086 测试短信 （向全部实例发送，不建议滥用） +""" + +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 + + +class SimAdminError(Exception): + pass + + +def get_sender(): + return middleware.Sender(middleware.getSenderID()) + + +def reply(sender, text): + sender.reply(str(text)) + + +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) + name = name.strip() + parts = [p.strip() for p in rest.split(",") if p.strip()] + if not parts: + raise SimAdminError("第 %s 行 URL 不能为空" % line_no) + raw = {"name": name, "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(): + raw = middleware.bucketGet(BUCKET, CONFIG_KEY) + if is_blank(raw): + raise SimAdminError( + "未配置 SimAdmin 实例。请在数据桶 %s/%s 写入 JSON数组或多行 alias=url 配置。" % (BUCKET, CONFIG_KEY) + ) + 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) + instances = [x for x in instances if x.get("enabled")] + if not instances: + raise SimAdminError("没有启用的 SimAdmin 实例") + names = set() + for item in instances: + if item["name"] in names: + raise SimAdminError("SimAdmin 实例名称重复: " + item["name"]) + names.add(item["name"]) + return instances + + +def request_json(instance, method, path, body=None): + base_url = instance["base_url"] + url = 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 {}) + + # SimAdmin 当前 API 无鉴权；这里兼容未来带 token 的反代/网关。 + 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_bytes(value): + try: + n = float(value) + except Exception: + return str(value) if value not in (None, "") else "-" + 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" % (n, units[idx]) + return "%.1f%s" % (n, units[idx]) + + +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): + name = instance["name"] + lines = ["【%s】" % 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")) + + health_status = pick(health, "status", default="ok") + version = pick(health, "version", default="-") + iccid = pick(sim, "iccid", "ICCID") + imsi = pick(sim, "imsi", "IMSI") + msisdn = pick(sim, "phone_number", "msisdn", "number", "phone") + operator = pick(network, "operator", "operator_name", "network_operator", "provider") + reg = pick(network, "registration_state", "reg_state", "status", "state") + signal = pick(network, "signal_strength", "signal", "rssi", "rsrp") + uptime = pick(stats, "uptime", "uptime_text", "system_uptime") + cpu = pick(stats, "cpu_usage", "cpu_percent", "cpu") + mem = pick(stats, "memory_usage", "memory_percent", "mem") + temp = pick(stats, "temperature", "temp", "cpu_temperature") + sent = pick(sms_stats, "sent", "sent_count", "outbox", default="-") + inbox = pick(sms_stats, "received", "received_count", "inbox", default="-") + + lines.append("健康: %s 版本: %s" % (health_status, version)) + lines.append("SIM: %s" % (msisdn if msisdn != "-" else "号码未知")) + lines.append("ICCID: %s" % iccid) + if imsi != "-": + lines.append("IMSI: %s" % imsi) + lines.append("网络: %s / %s / 信号:%s" % (operator, reg, signal)) + lines.append("系统: CPU %s / MEM %s / 温度 %s / 运行 %s" % (fmt_percent(cpu), fmt_percent(mem), temp, uptime)) + lines.append("短信: 收 %s / 发 %s" % (inbox, sent)) + 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 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) + blocks = [format_instance_status(x) for x in selected] + reply(sender, "\n\n".join(blocks)) + + +def valid_phone(phone): + # 兼容手机号、服务号、国际号码；只做安全字符校验，不强制中国手机号。 + return bool(re.match(r"^\+?[0-9][0-9\- ]{1,30}$", phone or "")) + + +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("短信内容过长，已拒绝发送") + + instances = load_instances() + selected = find_instances(instances, target) + results = [] + for instance in selected: + try: + payload = request_json(instance, "POST", "/api/sms/send", { + "phone_number": phone, + "content": content, + }) + data = unwrap_data(payload) + path = pick(data, "path", "message_path", "id", default="") + suffix = "" if not path else " path=" + str(path) + results.append("✅ %s 发送成功%s" % (instance["name"], suffix)) + except Exception as e: + results.append("❌ %s 发送失败: %s" % (instance["name"], str(e))) + reply(sender, "短信发送结果：\n" + "\n".join(results)) + + +def handle_config(sender): + instances = load_instances() + lines = ["SimAdmin 实例配置："] + for item in instances: + lines.append("- %s => %s timeout=%ss token=%s" % ( + item["name"], item["base_url"], item["timeout"], mask_secret(item.get("token")) or "未设置" + )) + reply(sender, "\n".join(lines)) + + +def handle_help(sender): + reply(sender, "\n".join([ + "SimAdmin 多实例管理（仅管理员）", + "配置桶：otto / simadmin_instances", + "指令：", + "- sim帮助", + "- sim配置", + "- sim状态 查询全部实例", + "- sim状态 <实例名> 查询指定实例", + "- sim短信 <实例名|all> <号码> <内容>", + "示例：sim短信 主卡 10086 CXLL", + ])) + + +def main(): + sender = get_sender() + try: + message = (sender.getMessage() or "").strip().strip('"') + if not sender.isAdmin(): + reply(sender, "仅管理员可用") + return + if message == "sim帮助": + handle_help(sender) + elif message == "sim配置": + handle_config(sender) + elif message.startswith("sim状态"): + handle_status(sender, message) + elif message.startswith("sim短信"): + handle_sms(sender, message) + else: + handle_help(sender) + except SimAdminError as e: + reply(sender, "SimAdmin插件错误：" + str(e)) + except Exception as e: + traceback.print_exc(file=sys.stderr) + reply(sender, "SimAdmin插件异常：" + str(e)) + + +if __name__ == "__main__": + main()