963 lines
37 KiB
Python
963 lines
37 KiB
Python
#[title: SimAdmin多实例管理]
|
||
#[language: python]
|
||
#[service: BOSS]
|
||
#[disable: false]
|
||
#[admin: false]
|
||
#[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.2]
|
||
#[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.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 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 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_brief(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)
|
||
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",
|
||
"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_brief(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()
|