feat: improve simadmin real-device status output
This commit is contained in:
@@ -11,7 +11,7 @@
|
||||
#[priority: 999]
|
||||
#[platform: qq,qb,wx,wb,tg,tb,fs,we,web,wxmp]
|
||||
#[open_source: true]
|
||||
#[version: 1.2.4]
|
||||
#[version: 1.2.5]
|
||||
#[public: false]
|
||||
#[price: 0]
|
||||
#[description: 管理员专用：兼容多个 SimAdmin 实例，支持多级菜单配置管理、状态查询和短信发送。配置桶 otto/simadmin_instances。]
|
||||
@@ -273,7 +273,7 @@ def pick(data, *keys, **kwargs):
|
||||
|
||||
|
||||
def fmt_percent(value):
|
||||
if value in (None, ""):
|
||||
if value in (None, "", "-"):
|
||||
return "-"
|
||||
try:
|
||||
n = float(value)
|
||||
@@ -284,35 +284,169 @@ def fmt_percent(value):
|
||||
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】" % instance["name"]]
|
||||
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"))
|
||||
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")
|
||||
|
||||
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)
|
||||
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"),
|
||||
|
||||
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="未连接"),
|
||||
))
|
||||
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="-"),
|
||||
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)
|
||||
|
||||
+154
-20
@@ -11,7 +11,7 @@
|
||||
#[priority: 999]
|
||||
#[platform: qq,qb,wx,wb,tg,tb,fs,we,web,wxmp]
|
||||
#[open_source: true]
|
||||
#[version: 1.2.4]
|
||||
#[version: 1.2.5]
|
||||
#[public: false]
|
||||
#[price: 0]
|
||||
#[description: 管理员专用：兼容多个 SimAdmin 实例，支持多级菜单配置管理、状态查询和短信发送。配置桶 otto/simadmin_instances。]
|
||||
@@ -273,7 +273,7 @@ def pick(data, *keys, **kwargs):
|
||||
|
||||
|
||||
def fmt_percent(value):
|
||||
if value in (None, ""):
|
||||
if value in (None, "", "-"):
|
||||
return "-"
|
||||
try:
|
||||
n = float(value)
|
||||
@@ -284,35 +284,169 @@ def fmt_percent(value):
|
||||
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】" % instance["name"]]
|
||||
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"))
|
||||
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")
|
||||
|
||||
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)
|
||||
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"),
|
||||
|
||||
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="未连接"),
|
||||
))
|
||||
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="-"),
|
||||
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)
|
||||
|
||||
Reference in New Issue
Block a user