import random import subprocess import time from datetime import datetime, timedelta import pytz from app.core.config import settings from app.plugins import _PluginBase from typing import Any, List, Dict, Tuple, Optional from app.log import logger from apscheduler.schedulers.background import BackgroundScheduler from apscheduler.triggers.cron import CronTrigger from app.schemas import NotificationType, MediaType class CustomCommand(_PluginBase): # 插件名称 plugin_name = "自定义命令" # 插件描述 plugin_desc = "自定义执行周期执行命令并推送结果。" # 插件图标 plugin_icon = "Ntfy_A.png" # 插件版本 plugin_version = "1.0" # 插件作者 plugin_author = "thsrite" # 作者主页 author_url = "https://github.com/thsrite" # 插件配置项ID前缀 plugin_config_prefix = "customcommand_" # 加载顺序 plugin_order = 39 # 可使用的用户级别 auth_level = 1 # 私有属性 _enabled: bool = False _onlyonce: bool = False _notify: bool = False _msgtype: bool = False _time_confs = None _scheduler: Optional[BackgroundScheduler] = None def init_plugin(self, config: dict = None): # 停止现有任务 self.stop_service() if config: self._enabled = config.get("enabled") self._onlyonce = config.get("onlyonce") self._notify = config.get("notify") self._msgtype = config.get("msgtype") self._time_confs = config.get("time_confs") if (self._enabled or self._onlyonce) and self._time_confs: # 周期运行 self._scheduler = BackgroundScheduler(timezone=settings.TZ) # 分别执行命令，输入结果 for time_conf in self._time_confs.split("\n"): if time_conf: for conf in str(time_conf).split("#"): name = conf[0] cron = conf[1] command = conf[2] if self._onlyonce: # 立即运行一次 logger.info(f"{name}服务启动，立即运行一次") self._scheduler.add_job(self.__execute_command, 'date', run_date=datetime.now( tz=pytz.timezone(settings.TZ)) + timedelta(seconds=3), name=name, args=[name, command]) # 关闭一次性开关 self._onlyonce = False # 保存配置 self.__update_config() else: random_delay = None if str(time_conf).count("#") == 3: random_delay = conf[3] try: self._scheduler.add_job(func=self.__execute_command, trigger=CronTrigger.from_crontab(str(cron)), name=name + ( f"随机延时{random_delay}秒" if random_delay else ""), args=[name, command, random_delay]) except Exception as err: logger.error(f"定时任务配置错误：{err}") # 推送实时消息 self.systemmessage.put(f"执行周期配置错误：{err}") # 启动任务 if self._scheduler.get_jobs(): self._scheduler.print_jobs() self._scheduler.start() def __execute_command(self, name, command, random_delay=None): """ 执行命令 """ if random_delay: random_delay = random.randint(int(str(random_delay).split("-")[0]), int(str(random_delay).split("-")[1])) logger.info(f"随机延时 {random_delay} 秒") time.sleep(random_delay) result = subprocess.call(command, shell=True) logger.info(f"执行命令：{command} 返回值：{result}") if self._notify and self._msgtype: # 发送通知 mtype = NotificationType.Manual if self._msgtype: mtype = NotificationType.__getitem__(str(self._msgtype)) or NotificationType.Manual self.post_message(title=name, mtype=mtype, text=result) def __update_config(self): self.update_config({ "enabled": self._enabled, "onlyonce": self._onlyonce, "notify": self._notify, "msgtype": self._msgtype, "time_confs": self._time_confs }) def get_state(self) -> bool: return self._enabled @staticmethod def get_command() -> List[Dict[str, Any]]: pass def get_api(self) -> List[Dict[str, Any]]: pass def get_form(self) -> Tuple[List[dict], Dict[str, Any]]: """ 拼装插件配置页面，需要返回两块数据：1、页面配置；2、数据结构 """ # 编历 NotificationType 枚举，生成消息类型选项 MsgTypeOptions = [] for item in NotificationType: MsgTypeOptions.append({ "title": item.value, "value": item.name }) return [ { 'component': 'VForm', 'content': [ { 'component': 'VRow', 'content': [ { 'component': 'VCol', 'props': { 'cols': 12, 'md': 3 }, 'content': [ { 'component': 'VSwitch', 'props': { 'model': 'enabled', 'label': '启用插件', } } ] }, { 'component': 'VCol', 'props': { 'cols': 12, 'md': 3 }, 'content': [ { 'component': 'VSwitch', 'props': { 'model': 'notify', 'label': '发送通知', } } ] }, { 'component': 'VCol', 'props': { 'cols': 12, 'md': 3 }, 'content': [ { 'component': 'VSwitch', 'props': { 'model': 'onlyonce', 'label': '立即运行一次', } } ] }, { 'component': 'VCol', 'props': { 'cols': 12, 'md': 3 }, 'content': [ { 'component': 'VSelect', 'props': { 'multiple': False, 'chips': True, 'model': 'msgtype', 'label': '消息类型', 'items': MsgTypeOptions } } ] }, ] }, { 'component': 'VRow', 'content': [ { 'component': 'VCol', 'props': { 'cols': 12, }, 'content': [ { 'component': 'VTextarea', 'props': { 'model': 'time_confs', 'label': '执行命令', 'rows': 2, 'placeholder': '命令名#0 9 * * *#python main.py\n' '命令名#0 9 * * *#python main.py#1-600' } } ] }, ] }, { 'component': 'VRow', 'content': [ { 'component': 'VCol', 'props': { 'cols': 12, }, 'content': [ { 'component': 'VAlert', 'props': { 'type': 'info', 'variant': 'tonal', 'text': '命令名#cron表达式#命令' } } ] } ] }, { 'component': 'VRow', 'content': [ { 'component': 'VCol', 'props': { 'cols': 12, }, 'content': [ { 'component': 'VAlert', 'props': { 'type': 'info', 'variant': 'tonal', 'text': '命令名#cron表达式#命令#随机延时（单位秒）' } } ] } ] } ] } ], { "enabled": False, "notify": False, "onlyonce": False, "time_confs": "", "msgtype": "" } def get_page(self) -> List[dict]: pass def stop_service(self): """ 退出插件 """ try: if self._scheduler: self._scheduler.remove_all_jobs() if self._scheduler.running: self._scheduler.shutdown() self._scheduler = None except Exception as e: logger.error("退出插件失败：%s" % str(e))