feat: add simadmin interactive config menu

This commit is contained in:
2026-05-15 22:01:04 +08:00
parent 54e708744e
commit b375016239
2 changed files with 354 additions and 121 deletions
+101 -13
View File
@@ -2,6 +2,7 @@
管理员专用 Python 插件,支持多个 SimAdmin 实例:
- 多级菜单交互式配置管理
- 查询全部或指定实例状态
- 通过指定实例发送短信
- 兼容 SimAdmin API
@@ -28,33 +29,115 @@ plugin/scripts/simadmin_multi.py
# [admin: true]
# [rule: ^sim帮助$]
# [rule: ^sim配置$]
# [rule: ^sim管理$]
# [rule: ^sim状态(?:\s+\S+)?$]
# [rule: ^sim短信\s+\S+\s+\S+\s+[\s\S]+$]
```
即:仅管理员可用。
即:仅管理员可用。代码内也会再次判断 `sender.isAdmin()`
## 配置
## 配置存储
Autman 数据桶写入
插件配置存入 Autman 数据桶:
- bucket`otto`
- key`simadmin_instances`
### 推荐:JSON 数组
插件会用 JSON 数组保存配置。也兼容读取旧的多行简写配置。
## 多级菜单配置管理
发送:
```text
sim配置
```
或:
```text
sim管理
```
进入菜单:
```text
SimAdmin 配置管理(多级菜单)
当前实例数:0
1. 查看实例列表
2. 添加实例
3. 编辑实例
4. 删除实例
5. 启用/禁用实例
6. 测试状态查询
7. 导出配置 JSON
q. 退出
```
### 添加实例流程
```text
sim配置
2
请输入实例名称,例如:主卡
主卡
请输入 SimAdmin 地址,例如:http://192.168.68.1:3000
http://192.168.68.1:3000
请输入超时时间秒数,直接回车默认 10
10
请输入 token,可直接回车留空
是否启用?输入 1/是 启用,0/否 禁用;直接回车默认启用
1
确认添加?
yes
```
### 编辑实例流程
菜单选择 `3. 编辑实例` 后,选择实例编号,再进入二级菜单:
```text
编辑【主卡】
1. 修改名称
2. 修改地址
3. 修改 timeout
4. 修改 token
5. 切换启用/禁用(当前:启用)
6. 保存并返回
q. 放弃返回
```
### 删除实例流程
菜单选择 `4. 删除实例`,选择编号后,需要输入 `yes` 二次确认。
### 启用/禁用实例
菜单选择 `5. 启用/禁用实例`,选择编号后立即切换。
### 测试状态查询
菜单选择 `6. 测试状态查询`,选择编号后会调用该实例的状态接口并返回摘要。
### 导出配置
菜单选择 `7. 导出配置 JSON`,会输出当前完整 JSON,方便备份或迁移。
## 兼容的手工配置格式
虽然推荐直接使用菜单管理,但仍可手工写入数据桶。
### 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": "可选,如果前面有反代/网关鉴权"
"token": "可选,如果前面有反代/网关鉴权",
"enabled": true
}
]
```
@@ -70,18 +153,21 @@ plugin/scripts/simadmin_multi.py
| `enabled` | 否 | 是否启用,默认 true |
| `headers` | 否 | 自定义请求头对象 |
### 简写:多行 alias=url
### 旧版简写
```text
主卡=http://192.168.68.1:3000
副卡=http://192.168.68.2:3000,token=xxx,timeout=10
```
## 指令
插件能读取这种格式;通过菜单保存后会自动转换为 JSON 数组。
## 日常指令
```text
sim帮助
sim配置
sim管理
sim状态
sim状态 主卡
sim短信 主卡 10086 CXLL
@@ -93,7 +179,7 @@ sim短信 all 10086 测试短信
- `sim状态`:查询全部启用实例。
- `sim状态 <实例名>`:查询指定实例;支持名称包含匹配。
- `sim短信 <实例名|all> <号码> <内容>`:发送短信。
- `all` 会向所有实例发送同一条短信,谨慎使用。
- `all` 会向所有启用实例发送同一条短信,谨慎使用。
## 返回示例
@@ -119,7 +205,9 @@ IMSI: 460...
## 安全约定
- 插件头注和代码内都做了管理员限制。
- 配置管理、短信发送都仅管理员可用。
- 不在源码中硬编码任何 SimAdmin 地址或密钥。
- `token` 只用于兼容反代/网关鉴权;当前 SimAdmin 本身 API 如无鉴权也可留空。
- 短信号码只允许数字、空格、`-`、国际区号 `+`
- 短信内容最大 1000 字,避免误发超长内容。
- 删除实例需要输入 `yes` 二次确认。
+253 -108
View File
@@ -5,40 +5,27 @@
# [admin: true]
# [rule: ^sim帮助$]
# [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]
# [version: 1.1.0]
# [public: false]
# [price: 0]
# [description: 管理员专用:兼容多个 SimAdmin 实例,支持状态查询和短信发送。配置桶 otto/simadmin_instancesJSON数组或别名=URL多行。]
# [description: 管理员专用:兼容多个 SimAdmin 实例,支持多级菜单配置管理、状态查询和短信发送。配置桶 otto/simadmin_instances。]
"""
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
Autman Python 插件:SimAdmin 多实例状态查询短信发送、多级菜单配置管理
配置桶:otto / simadmin_instances
指令:
- sim帮助
- sim配置
- sim状态 查询全部实例
- sim状态 主卡 查询指定实例
- sim配置 / sim管理 进入多级菜单配置管理
- sim状态 查询全部实例
- sim状态 主卡 查询指定实例
- sim短信 主卡 10086 CXLL
- sim短信 all 10086 测试短信 (向全部实例发送,不建议滥用)
- sim短信 all 10086 测试短信
"""
import json
@@ -46,7 +33,6 @@ import re
import sys
import traceback
import urllib.error
import urllib.parse
import urllib.request
import middleware
@@ -54,6 +40,7 @@ import middleware
BUCKET = "otto"
CONFIG_KEY = "simadmin_instances"
DEFAULT_TIMEOUT = 10
LISTEN_TIMEOUT_MS = 120000
class SimAdminError(Exception):
@@ -68,6 +55,27 @@ def reply(sender, text):
sender.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('"')
if value.lower() in ("q", "quit", "exit", "退出", "取消"):
raise SimAdminError("已取消")
if value == "" and default is not None:
return default
if value == "" and not allow_empty:
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() == ""
@@ -134,11 +142,10 @@ def parse_line_config(text):
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]}
raw = {"name": name.strip(), "base_url": parts[0]}
for part in parts[1:]:
if "=" in part:
k, v = part.split("=", 1)
@@ -147,12 +154,12 @@ def parse_line_config(text):
return instances
def load_instances():
def load_instances(include_disabled=False, allow_empty=False):
raw = middleware.bucketGet(BUCKET, CONFIG_KEY)
if is_blank(raw):
raise SimAdminError(
"未配置 SimAdmin 实例。请在数据桶 %s/%s 写入 JSON数组或多行 alias=url 配置。" % (BUCKET, CONFIG_KEY)
)
if allow_empty:
return []
raise SimAdminError("未配置 SimAdmin 实例。发送 sim管理 通过多级菜单添加。")
text = str(raw).strip()
if text.startswith("[") or text.startswith("{"):
try:
@@ -169,33 +176,49 @@ def load_instances():
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:
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"])
return instances
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):
base_url = instance["base_url"]
url = base_url + path
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 {})
# 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:
@@ -206,7 +229,6 @@ def request_json(instance, method, path, body=None):
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:
@@ -234,21 +256,6 @@ def pick(data, *keys, **kwargs):
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 "-"
@@ -262,38 +269,34 @@ def fmt_percent(value):
def format_instance_status(instance):
name = instance["name"]
lines = ["%s" % name]
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"))
health_status = pick(health, "status", default="ok")
version = pick(health, "version", default="-")
iccid = pick(sim, "iccid", "ICCID")
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")
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))
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)
@@ -312,18 +315,15 @@ def find_instances(instances, target):
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)
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 ""))
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):
@@ -337,33 +337,178 @@ def handle_sms(sender, message):
raise SimAdminError("短信内容不能为空")
if len(content) > 1000:
raise SimAdminError("短信内容过长,已拒绝发送")
instances = load_instances()
selected = find_instances(instances, target)
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,
})
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))
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):
instances = load_instances(include_disabled=True, allow_empty=True)
if not instances:
reply(sender, "当前没有 SimAdmin 实例。")
return
reply(sender, "SimAdmin 实例列表:\n" + "\n".join(instance_brief(item, idx) for idx, item in enumerate(instances, 1)))
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):
instances = load_instances(include_disabled=True, allow_empty=True)
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):
instances = load_instances(include_disabled=True, allow_empty=True)
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_toggle(sender):
instances = load_instances(include_disabled=True, allow_empty=True)
idx = choose_instance_by_menu(sender, instances, "请选择要启用/禁用的实例编号")
instances[idx]["enabled"] = not instances[idx].get("enabled")
save_instances(instances)
reply(sender, "已将【%s】切换为:%s" % (instances[idx]["name"], "启用" if instances[idx]["enabled"] else "禁用"))
def menu_test(sender):
instances = load_instances(include_disabled=True, allow_empty=True)
idx = choose_instance_by_menu(sender, instances, "请选择要测试连通性的实例编号")
item = instances[idx]
reply(sender, format_instance_status(item))
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":
menu_list(sender)
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):
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))
# BOSS 希望通过多级菜单交互管理配置,因此 sim配置 直接进入菜单。
handle_config_menu(sender)
def handle_help(sender):
@@ -372,11 +517,11 @@ def handle_help(sender):
"配置桶:otto / simadmin_instances",
"指令:",
"- sim帮助",
"- sim配置",
"- sim状态 查询全部实例",
"- sim配置 / sim管理 进入多级菜单配置管理",
"- sim状态 查询全部启用实例",
"- sim状态 <实例名> 查询指定实例",
"- sim短信 <实例名|all> <号码> <内容>",
"示例:sim短信 主卡 10086 CXLL",
"菜单支持:查看、添加、编辑、删除、启用/禁用、测试、导出 JSON。",
]))
@@ -389,7 +534,7 @@ def main():
return
if message == "sim帮助":
handle_help(sender)
elif message == "sim配置":
elif message in ("sim配置", "sim管理"):
handle_config(sender)
elif message.startswith("sim状态"):
handle_status(sender, message)
@@ -398,7 +543,7 @@ def main():
else:
handle_help(sender)
except SimAdminError as e:
reply(sender, "SimAdmin插件错误" + str(e))
reply(sender, "SimAdmin插件提示" + str(e))
except Exception as e:
traceback.print_exc(file=sys.stderr)
reply(sender, "SimAdmin插件异常:" + str(e))