11

_conf_schema.json

@@ -1,8 +1,36 @@
 {
-  "api_token": {
+  "jx3_kfjk_en": {
-    "description": "JX3API的token",
+    "description": "开服监控功能",
+    "type": "bool",
+    "default": true,
+    "hint": "是否启用剑网三开服监控功能。"
+  },
+  "jx3_kfjk_time": {
+    "description": "开服监控循环时间",
+    "type": "int",
+    "default": 10,
+    "hint": "请求服务器状态的循环时间，单位秒。"
+  },
+  "jx3_kfjk_server": {
+    "description": "开服监控服务器",
     "type": "string",
-    "default": "",
+    "default": "梦江南",
-    "hint": "填写购买的JX3API的token，用于访问的付费接口服务"
+    "hint": "请求服务器状态的名称，要官方名称。"
+  },
+  "jx3_kfjk_list": {
+    "description": "开服监控推送列表",
+    "type": "list",
+    "hint": "可以填写多个会话唯一ID。",
+    "items": {
+        "type": "string",
+        "description": "会话唯一ID"
+    },
+    "default": ["default:GroupMessage:768280021"]
+  },
+  "jx3_kfjk_test": {
+    "description": "开服监控功能测试",
+    "type": "bool",
+    "default": false,
+    "hint": "可以手动修改服务器的状态，进行推送测试。"
   }
 }

core/cless_jx3.py

@@ -77,6 +77,9 @@ class JX3Function:
         return return_data
     
     async def kaifu(self,server: str = "梦江南"):
+        """
+        开服状态查询
+        """
         return_data = {
             "code": 0,
             "msg": "功能函数未执行",
@@ -97,9 +100,11 @@ class JX3Function:
             status_time = datetime.fromtimestamp(data.get("time")).strftime("%Y-%m-%d %H:%M:%S")
             if status == 1: 
                 status_str = f"{server}服务器已开服，快冲，快冲！\n开服时间：{status_time}" 
+                status_bool = True 
             else:
                 status_str = f"{server}服务器当前维护中，等会再来吧！\n上次开服时间：{status_time}"
-                
+                status_bool = False 
+            return_data["status"] = status_bool  
             return_data["data"] = status_str
             return_data["code"] = 200       
         except Exception as e:

main.py

@@ -1,8 +1,9 @@
 import json
+import asyncio
 from pathlib import Path
 from datetime import datetime
 
-from astrbot.api.event import filter, AstrMessageEvent, MessageEventResult
+from astrbot.api.event import filter, AstrMessageEvent, MessageEventResult, MessageChain
 from astrbot.api.star import Context, Star, register, StarTools
 from astrbot.api import logger
 from astrbot.api import AstrBotConfig
@@ -11,6 +12,7 @@ from .core.cless_mysql import AsyncMySQL
 from .core.cless_jx3 import JX3Function
 
 
+
 @register("astrbot_plugin_jx3api", 
           "fxdyz", 
           "通过接口调用剑网三API接口获取游戏数据", 
@@ -29,7 +31,9 @@ class Jx3ApiPlugin(Star):
         # 读取文件内容
         with open(self.api_file_path, 'r', encoding='utf-8') as f:
             self.api_config = json.load(f)  
-
+        # 初始化数据
+        self.inidata()
+        self.kf_task = asyncio.create_task(self.cycle_kaifjiankong()) 
         logger.info("jx3api插件初始化完成")
 
 
@@ -48,9 +52,51 @@ class Jx3ApiPlugin(Star):
         #创建类实例
         self.db = AsyncMySQL(db_config)
         self.jx3fun = JX3Function(self.api_config,self.db)
+        # 周期函数调用
+    
+
         logger.info("jx3api插件创建实例完成")
 
+    def inidata(self):
+        """数据初始化"""
+        self.test_server = False
+        logger.info("数据初始化完成")
 
+    async def cycle_kaifjiankong(self):
+        """开服监控后台程序"""
+        # 获取配置信息
+        time_int = self.conf.get("jx3_kfjk_time", 10)
+        self.kfjk_en = self.conf.get("jx3_kfjk_en", True)
+        self.kfjk_umos = self.conf.get("jx3_kfjk_list", [])
+        self.kfjk_servername = self.conf.get("jx3_kfjk_server", "梦江南")
+
+        kfjk_test = self.conf.get("jx3_kfjk_test", False)
+        server_state = False    # 上一次查询的状态
+        server_state_new = False  # 最新查询的状态
+
+        if self.kfjk_en:
+            logger.info(f"开服监控功能开启")
+        while self.kfjk_en:
+            data = await self.jx3fun.kaifu(self.kfjk_servername)
+            if kfjk_test:
+                server_state_new = self.test_server
+            else:
+                server_state_new = data["status"]
+            logger.info(f"开服监控功能循环中,上次询问服务器状态{server_state},本次询问的服务器状态{server_state_new}") 
+            if server_state != server_state_new:
+                if server_state and not server_state_new:
+                    message_chain = MessageChain().message(f"{self.kfjk_servername}服务器关闭")
+                if server_state_new and not server_state:
+                    message_chain = MessageChain().message(f"{self.kfjk_servername}服务器开启")
+                if self.kfjk_umos:
+                    for umo in self.kfjk_umos:
+                        await self.context.send_message(umo, message_chain)
+            server_state = server_state_new
+            await asyncio.sleep(time_int)
+        if not self.kfjk_en: 
+            # 销毁进程
+            self.kf_task.cancel()
+            logger.info(f"开服监控功能关闭")
     
     @filter.command_group("剑三")
     def jx3(self):
@@ -193,6 +239,25 @@ class Jx3ApiPlugin(Star):
             logger.error(f"功能函数执行错误: {e}")
             yield event.plain_result("猪脑过载，请稍后再试") 
 
+    @jx3.command("开服监控")
+    async def jx3_kaifhujiank(self, event: AstrMessageEvent,en: str = "状态"):
+        """剑三 开服监控"""     
+        if en == "开启":
+            umo_set = set(self.kfjk_umos)
+            umo_set.add(event.unified_msg_origin)
+            self.kfjk_umos = list(umo_set)
+            yield event.plain_result(f"开服监控已开启") 
+        elif en =="关闭":
+            try:
+                self.kfjk_umos.remove(event.unified_msg_origin)
+                yield event.plain_result(f"开服监控已关闭")
+            except ValueError as e:
+                yield event.plain_result(f"开服监控已关闭") 
+        elif en == "状态":
+            yield event.plain_result(f"开服监控后台状态：{self.kfjk_en}\n监控服务器：{self.kfjk_servername}\n推送会话列表：\n{self.kfjk_umos}") 
+        else:
+            yield event.plain_result("开服监控指令错误，请输入 开启/关闭") 
+
 
     @filter.permission_type(filter.PermissionType.ADMIN)
     @jx3.command("外观数据同步")
@@ -209,9 +274,19 @@ class Jx3ApiPlugin(Star):
             logger.error(f"功能函数执行错误: {e}")
             yield event.plain_result("猪脑过载，请稍后再试") 
 
+    @filter.command("测试值")
+    async def jx3_testvar(self, event: AstrMessageEvent,var: str = "0"):
+        """测试程序"""   
+        if var == "1":
+            self.test_server = True
+        if var == "0":
+            self.test_server = False
         
+        yield event.plain_result(f"测试值:{self.test_server}") 
 
     async def terminate(self):
         """可选择实现异步的插件销毁方法，当插件被卸载/停用时会调用。"""
         await self.db.close_pool()
+        # 后台进程销毁
+        self.kf_task.cancel()
         logger.info("jx3api插件已卸载/停用")