@@ -0,0 +1,682 @@
#[title: SimAdmin多实例管理]
#[language: python]
#[service: BOSS]
#[disable: false]
#[admin: false]
#[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.2.3]
#[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 测试短信
"""
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 ( " 等待输入超时或监听失败，已退出 " )
# allow_empty=True 的场景允许直接回车/空消息走默认值。
if value == " " and default is not None :
return default
if value == " " and not allow_empty :
raise SimAdminError ( " 输入不能为空，已退出 " )
if value . lower ( ) in ( " q " , " quit " , " exit " , " 退出 " , " 取消 " ) :
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 ( ) == " "
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= %s s 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 format_instance_status ( instance ) :
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 " ) )
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 " )
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 " ) ,
) )
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 )
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 " " ) )
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 " ,
" 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 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|测试|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 ( " 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_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 , 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_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_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. 复制/新增一个实例 " ,
" 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 . 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 " ,
" 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 )
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> <号码> <内容> " ,
" 菜单支持：查看、添加、编辑、删除、启用/禁用、测试、导出 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短信 " ) :
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 )
reply ( sender , " 指令不支持，请发送：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 ( )
