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

feat: add simadmin multi-instance autman plugin

Browse Source
This commit is contained in:
Hermes Agent
2026-05-15 21:49:14 +08:00
parent 5024f17e9c
commit 54e708744e
2 changed files with 533 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
plugins/simadmin/README.md
+125
View File
@@ -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 <token>` 和 `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 字，避免误发超长内容。
plugins/simadmin/simadmin_multi.py
+408
View File
@@ -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()