158 lines
5.0 KiB
Python
158 lines
5.0 KiB
Python
# -*- coding: utf-8 -*-
|
||
import subprocess
|
||
from loguru import logger
|
||
from typing import Dict, Any, List, Optional, Tuple
|
||
|
||
from base.plugin_common.message_plugin_interface import MessagePluginInterface
|
||
from base.plugin_common.plugin_interface import PluginStatus
|
||
from utils.decorator.plugin_decorators import plugin_stats_decorator
|
||
from utils.decorator.points_decorator import plugin_points_cost
|
||
from utils.robot_cmd.robot_command import Feature, PermissionStatus, GroupBotManager
|
||
from wechat_ipad import WechatAPIClient
|
||
|
||
|
||
|
||
class SystemUpdaterPlugin(MessagePluginInterface):
|
||
"""系统更新插件"""
|
||
|
||
# 功能权限常量
|
||
FEATURE_KEY = "SYSTEM_UPDATER"
|
||
FEATURE_DESCRIPTION = "🔄 系统更新功能 [更新系统, 系统更新]"
|
||
|
||
@property
|
||
def name(self) -> str:
|
||
return "系统更新"
|
||
|
||
@property
|
||
def version(self) -> str:
|
||
return "1.0.0"
|
||
|
||
@property
|
||
def description(self) -> str:
|
||
return "提供系统更新和自动登录功能"
|
||
|
||
@property
|
||
def author(self) -> str:
|
||
return "AI Assistant"
|
||
|
||
@property
|
||
def command_prefix(self) -> Optional[str]:
|
||
return ""
|
||
|
||
@property
|
||
def commands(self) -> List[str]:
|
||
return self._commands
|
||
|
||
@property
|
||
def feature_key(self) -> Optional[str]:
|
||
return self.FEATURE_KEY
|
||
|
||
@property
|
||
def feature_description(self) -> Optional[str]:
|
||
return self.FEATURE_DESCRIPTION
|
||
|
||
def __init__(self):
|
||
super().__init__()
|
||
self.admin_wxids = []
|
||
self.wait_time = 5 # 默认等待15秒
|
||
self.bot: WechatAPIClient = None
|
||
# 注册功能权限
|
||
self.feature = self.register_feature()
|
||
|
||
def initialize(self, context: Dict[str, Any]) -> bool:
|
||
"""初始化插件"""
|
||
self.LOG = logger
|
||
self.LOG.info(f"正在初始化 {self.name} 插件...")
|
||
|
||
# 保存上下文对象
|
||
self.config = self._config
|
||
|
||
# 从配置中获取命令和其他设置
|
||
plugin_config = self._config.get("SystemUpdater", {})
|
||
self._commands = plugin_config.get("commands", ["更新系统", "系统更新"])
|
||
self._shell_path = plugin_config.get("shell_path", "")
|
||
self.wait_time = plugin_config.get("wait_time", 5)
|
||
self.admin_wxids = plugin_config.get("admin_wxids", [])
|
||
self.enable = plugin_config.get("enable", True)
|
||
|
||
self.LOG.info(f"[{self.name}] 插件初始化完成，指令：{self._commands}")
|
||
return True
|
||
|
||
def start(self) -> bool:
|
||
self.status = PluginStatus.RUNNING
|
||
return True
|
||
|
||
def stop(self) -> bool:
|
||
self.status = PluginStatus.STOPPED
|
||
return True
|
||
|
||
def can_process(self, message: Dict[str, Any]) -> bool:
|
||
"""检查是否可以处理该消息"""
|
||
if not self.enable:
|
||
return False
|
||
|
||
content = str(message.get("content", "")).strip()
|
||
|
||
# 检查是否是命令
|
||
if content in self._commands:
|
||
return True
|
||
|
||
# 检查是否是带参数的更新命令
|
||
for cmd in self._commands:
|
||
if content.startswith(f"{cmd} "):
|
||
return True
|
||
|
||
return False
|
||
|
||
@plugin_stats_decorator(plugin_name="系统更新")
|
||
@plugin_points_cost(1, "系统更新消耗积分", FEATURE_KEY)
|
||
async def process_message(self, message: Dict[str, Any]) -> Tuple[bool, Optional[str]]:
|
||
"""处理消息"""
|
||
content = str(message.get("content", "")).strip()
|
||
self.LOG.debug(f"插件执行： {self.name}：{content}")
|
||
sender = message.get("sender")
|
||
roomid = message.get("roomid", "")
|
||
gbm: GroupBotManager = message.get("gbm")
|
||
bot: WechatAPIClient = message.get("bot")
|
||
|
||
# 检查权限
|
||
if roomid and gbm.get_group_permission(roomid, self.feature) == PermissionStatus.DISABLED:
|
||
return False, "没有权限"
|
||
|
||
# 检查是否是命令
|
||
if content in self._commands:
|
||
return True, "命令已接收"
|
||
|
||
# 检查是否是带参数的更新命令
|
||
for cmd in self._commands:
|
||
if content.startswith(f"{cmd} "):
|
||
return True, "带参数的命令已接收"
|
||
|
||
return False, "无法处理的消息"
|
||
|
||
def _execute_system_update(self):
|
||
"""执行系统更新操作"""
|
||
try:
|
||
# 使用相对路径调用 restart.sh 脚本
|
||
result = subprocess.run(
|
||
[self._shell_path], # 使用绝对路径
|
||
check=True,
|
||
stdout=subprocess.PIPE,
|
||
stderr=subprocess.PIPE
|
||
)
|
||
|
||
# 打印执行结果
|
||
logger.info(f"更新命令执行成功: {result.stdout.decode()}")
|
||
|
||
except subprocess.CalledProcessError as e:
|
||
# 如果 subprocess 执行出错，输出 stderr 信息
|
||
logger.error(f"执行更新命令时出错: {e.stderr.decode()}")
|
||
except Exception as e:
|
||
# 捕获其他异常
|
||
logger.error(f"系统更新失败: {str(e)}")
|
||
|
||
|
||
# 插件入口点
|
||
def get_plugin():
|
||
return SystemUpdaterPlugin()
|