diff --git a/core/async_task.py b/core/async_task.py index 7d716de..9db0557 100644 --- a/core/async_task.py +++ b/core/async_task.py @@ -3,6 +3,8 @@ from pathlib import Path import json from typing import Callable, Awaitable, Optional +from apscheduler.schedulers.asyncio import AsyncIOScheduler +from apscheduler.triggers.interval import IntervalTrigger from astrbot.api.event import filter, AstrMessageEvent, MessageEventResult, MessageChain from astrbot.api.star import Context, Star, register, StarTools @@ -11,35 +13,31 @@ from astrbot.api import AstrBotConfig from .jx3_service import JX3Service + class AsyncTask: """ - 异步后台任务类（asyncio版） - - 支持后台异步循环执行 - - 支持 start/stop - - 支持设定间隔 + 基于 APScheduler 的后台异步监控任务管理类 """ def __init__(self, context: Context, config: AstrBotConfig, jx3fun: JX3Service): - """ - Args: - coro: 异步任务函数 - interval: 每次执行间隔（秒） - auto_start: 是否自动启动 - """ self.context = context self.conf = config self.jx3fun = jx3fun + self.file_path = StarTools.get_data_dir("astrbot_plugin_jx3") / "local_async.json" self._file_lock = asyncio.Lock() + + self.scheduler = AsyncIOScheduler() + self.tasks = {} # 存储 task_id 对应的状态信息 + logger.info(f"获取后台数据缓存文件路径成功：{self.file_path}") + """===================== 本地读写 =====================""" async def set_local_data(self, key: str, value): - """异步安全写入本地 JSON""" async with self._file_lock: try: self.file_path.parent.mkdir(parents=True, exist_ok=True) - if not self.file_path.exists(): local_data = {} else: @@ -47,147 +45,120 @@ class AsyncTask: local_data = json.load(f) local_data[key] = value - with open(self.file_path, 'w', encoding='utf-8') as f: json.dump(local_data, f, ensure_ascii=False, indent=4) - logger.debug(f"后台数据写入完成: {key}--{value}") - except Exception as e: logger.error(f"数据写入文件失败：{e}") - async def get_local_data(self, key: str, default=None): - """异步安全读取本地 JSON""" async with self._file_lock: try: if not self.file_path.exists(): return default - with open(self.file_path, 'r', encoding='utf-8') as f: local_data = json.load(f) - - value = local_data.get(key, default) - logger.debug(f"读取数据成功: {key}--{value}") - return value - + return local_data.get(key, default) except Exception as e: logger.error(f"读取数据文件失败：{e}") return default + """===================== 通用后台任务 =====================""" - async def _cycle_common(self,fetch_func, conf: dict, state: dict, namefun: str, local_key: str): - """后台程序""" - - # 调用函数获取最新数据 + async def _job_common(self, fetch_func, task_key: str, namefun: str): + state = self.tasks[task_key] try: - state["state_old"] = await self.get_local_data(local_key) - - except Exception as e: - logger.error(f"获取{namefun}本地缓存数据失败: {e}") - # 判断推送功能是否启用 - if conf["enable"]: - logger.info(f"{namefun}功能开启") - else: - logger.info(f"{namefun}功能关闭") - return - # 循环启用 - while conf["enable"]: - try: - # 获取最新状态 - data = await fetch_func() - state["state_new"] = data["status"] - logger.debug(f"{namefun}功能循环中,上次询问状态：{state['state_old']},本次询问状态：{state['state_new']}") - # 判断状态是否变化 - if state["state_old"] != state["state_new"]: - logger.info(f"{namefun}功能循环中,上次询问状态：{state['state_old']},本次询问状态：{state['state_new']}") - # 构建消息 - message_chain = MessageChain().message(data.get("data")) - # 推送消息 - if conf["umos"]: - for umo in conf["umos"]: - await self.context.send_message(umo, message_chain) - # 状态储存本地 - await self.set_local_data(local_key, state["state_new"]) - # 最新状态覆盖以前状态 + data = await fetch_func() + state["state_new"] = data["status"] + + if state["state_old"] != state["state_new"]: + message_chain = MessageChain().message(data.get("data")) + + for umo in state["umos"]: + await self.context.send_message(umo, message_chain) + + await self.set_local_data(task_key, state["state_new"]) state["state_old"] = state["state_new"] - - except Exception as e: - logger.error(f"{namefun}循环异常: {e}") - await asyncio.sleep(conf["time"]) - - async def _get_conf(self, conf: dict, state: dict, namefun: str) -> str: - """获取功能配置信息""" - return_msg = f"{namefun}后台状态：{conf['enable']}\n" - return_msg += f"周期询问时间：{conf['time']}秒\n" - return_msg += f"上次询问状态：{state['state_old']}\n" - return_msg += f"推送会话列表：\n{conf['umos']}" - return return_msg - - - async def cycle_kfjk(self): - """开服监控后台程序""" - # 获取配置信息 - conf = self.conf.get("kfjk", {}) - self.kfjk_conf = { - "enable": conf.get("enable", True), - "time": conf.get("time", 60), - "umos": conf.get("umos", []), - } - # 状态记录 - self.kfjk = { - "state_old": False, - "state_new": False - } - - await self._cycle_common( - fetch_func=lambda: self.jx3fun.kaifu("梦江南"), - conf=self.kfjk_conf, - state=self.kfjk, - namefun="开服监控", - local_key="kfjk" - ) - - - async def cycle_xwzx(self): - """最新新闻资讯后台程序""" - # 获取配置信息 - conf = self.conf.get("xwzx", {}) - self.xwzx_conf = { - "enable": conf.get("enable", True), - "time": conf.get("time", 280), - "umos": conf.get("umos", []), - } - # 状态记录 - self.xwzx = { - "state_old": 0, - "state_new": 0 - } - # 后台进程开启 - await self._cycle_common( - fetch_func=lambda: self.jx3fun.xinwei(), - conf=self.xwzx_conf, - state=self.xwzx, - namefun="新闻资讯", - local_key="xwzx" - ) - - - async def get_kfjk_conf(self) -> str: - """获取开服监控配置信息""" - try: - return_msg = await self._get_conf(self.kfjk_conf, self.kfjk, "开服监控") except Exception as e: - return_msg = f"获取后台配置状态失败：{e}" - return return_msg - + logger.error(f"{namefun}后台任务执行异常: {e}") - async def get_xwzx_conf(self) -> str: - """获取开服监控配置信息""" + """===================== 初始化任务 =====================""" + + async def init_tasks(self): + settings = [ + ("kfjk", "开服监控", lambda: self.jx3fun.kaifu("梦江南")), + ("xwzx", "新闻资讯", lambda: self.jx3fun.xinwei()), + ] + + for key, name, fetch in settings: + conf = self.conf.get(key, {}) + + state_old = await self.get_local_data(key, default=False) + self.tasks[key] = { + "enable": conf.get("enable", True), + "interval": conf.get("time", 60), + "umos": conf.get("umos", []), + "state_old": state_old, + "state_new": state_old + } + + if self.tasks[key]["enable"]: + self._add_scheduler(key, name, fetch) + + if not self.scheduler.running: + self.scheduler.start() + logger.info("后台监控调度器已启动") + + """===================== 调度操作 =====================""" + + def _add_scheduler(self, key, namefun, fetch_func): + if self.scheduler.get_job(key): + self.scheduler.remove_job(key) + + interval = self.tasks[key]["interval"] + self.scheduler.add_job( + func=self._job_common, + trigger=IntervalTrigger(seconds=interval), + id=key, + args=[fetch_func, key, namefun] + ) + + logger.info(f"{namefun}后台任务启动成功，周期：{interval}s") + + def stop_all_tasks(self): + """ + 停止并移除所有任务 + """ try: - return_msg = await self._get_conf(self.xwzx_conf, self.xwzx, "新闻资讯") + self.scheduler.remove_all_jobs() + for key in self.tasks: + self.tasks[key]["enable"] = False + logger.info("已停止全部后台任务") except Exception as e: - return_msg = f"获取后台配置状态失败：{e}" - return return_msg - \ No newline at end of file + logger.error(f"停止全部后台任务失败：{e}") + + async def destroy(self): + """ + 销毁整个调度器，适合插件卸载/重启时调用 + """ + try: + self.stop_all_tasks() + if self.scheduler.running: + self.scheduler.shutdown(wait=False) + logger.info("后台调度器已销毁") + except Exception as e: + logger.error(f"销毁调度器失败：{e}") + + async def get_task_info(self, key: str) -> str: + try: + t = self.tasks[key] + return ( + f"功能：{key}\n" + f"启用：{t['enable']}\n" + f"周期：{t['interval']} 秒\n" + f"旧状态：{t['state_old']}\n" + f"推送对象：{t['umos']}" + ) + except Exception as e: + return f"读取后台配置失败：{e}" diff --git a/main.py b/main.py index a4471a7..ac46a8b 100644 --- a/main.py +++ b/main.py @@ -65,12 +65,13 @@ class Jx3ApiPlugin(Star): try: self.jx3fun = JX3Service(self.api_config,self.conf) self.at = AsyncTask(self.context, self.conf, self.jx3fun) + await self.at.init_tasks() except Exception as e: + await self.at.destroy() logger.error(f"功能示例初始化失败: {e}") return # 周期函数调用 - self.kf_task = asyncio.create_task(self.at.cycle_kfjk()) - self.xw_task = asyncio.create_task(self.at.cycle_xwzx()) + # 初始化函数 logger.info("jx3api异步插件初始化完成") @@ -280,14 +281,14 @@ class Jx3ApiPlugin(Star): @jx3.command("开服监控") async def jx3_kaifhujiank(self, event: AstrMessageEvent): """剑三 开服监控""" - return_msg = await self.at.get_kfjk_conf() + return_msg = await self.at.get_task_info("kfjk") yield event.plain_result(return_msg) @jx3.command("新闻推送") async def jx3_xinwenzhixun(self, event: AstrMessageEvent): """剑三 新闻推送""" - return_msg = await self.at.get_xwzx_conf() + return_msg = await self.at.get_task_info("xwzx") yield event.plain_result(return_msg) @@ -460,8 +461,7 @@ class Jx3ApiPlugin(Star): async def terminate(self): """可选择实现异步的插件销毁方法，当插件被卸载/停用时会调用。""" # 关闭数据库连接 - await self.db.close() + # 后台z周期进程销毁 - self.kf_task.cancel() - self.xw_task.cancel() + await self.at.destroy() logger.info("jx3api插件已卸载/停用") \ No newline at end of file