刷马推送

core/async_task.py

@@ -1,3 +1,4 @@
+# pyright: reportArgumentType=false
 import asyncio
 import json
 import aiofiles
@@ -22,7 +23,8 @@ class AsyncTask:
         self.context = context
         self.conf = config
         self.jx3fun = jx3fun
-        
+
+        self.server = self.conf.get("server", "梦江南")
         self.file_path = StarTools.get_data_dir("astrbot_plugin_jx3") / "local_async.json"
         self._file_lock = asyncio.Lock()
         
@@ -31,6 +33,7 @@ class AsyncTask:
         
         logger.info(f"获取后台数据缓存文件路径成功：{self.file_path}")
 
+    
     """===================== 本地读写 ====================="""
 
     async def set_local_data(self, key: str, value):
@@ -109,8 +112,10 @@ class AsyncTask:
 
     async def init_tasks(self):
         settings = [
-            ("kfjk", "开服监控", lambda: self.jx3fun.kaifu("梦江南")),
+            ("kfjk", "开服监控", lambda: self.jx3fun.kaifu(self.server)),
             ("xwzx", "新闻资讯", lambda: self.jx3fun.xinwen()),
+            ("smxx", "刷马消息", lambda: self.jx3fun.shuamamsg(self.server,type="horse",subtype="foreshow")),
+            ("ctxx", "赤兔消息", lambda: self.jx3fun.shuamamsg(self.server,type="chitu-horse",subtype="share_msg")),
         ]
 
         for key, name, fetch in settings:

core/jx3_service.py

@@ -411,9 +411,10 @@ class JX3Service:
     async def xinwen(self) -> Dict[str, Any]:
         """新闻资讯"""
         return_data = self._init_return_data()
-        
+
         # 提取字段可能返回列表
-        data: Optional[List[Dict[str, Any]]] = await self._base_request("jx3_xinweng", "GET")
+        data: Optional[List[Dict[str, Any]]] = await self._base_request(
+            "jx3_xinweng", "GET")
         
         if not data or not isinstance(data, list):
             return_data["msg"] = "获取接口信息失败或数据格式错误"
@@ -440,6 +441,37 @@ class JX3Service:
         return return_data
 
 
+    async def shuamamsg(self,server:str,type:str,subtype:str) -> Dict[str, Any]:
+        """刷马消息"""
+        return_data = self._init_return_data()
+
+        params = {"server": server, "type": type, "subtype": subtype}
+        # 提取字段可能返回列表
+        data: Optional[List[Dict[str, Any]]] = await self._base_request(
+            "jx3box_shuamamsg", "GET", params=params)
+        
+        if not data:
+            return_data["msg"] = "获取接口信息失败或数据格式错误"
+            return return_data
+        
+        try:
+            # 
+            new_msg = data.get("list")[0]
+            return_data["status"] = new_msg.get('id')
+
+            result_msg = f"{server}\n"
+            result_msg += f"{new_msg.get('content')}\n"
+            result_msg += f"{new_msg.get('created_at')}\n"
+                
+            return_data["data"] = result_msg
+            return_data["code"] = 200
+        except Exception as e:
+            logger.error(f"jigai 数据处理时出错: {e}")
+            return_data["msg"] = "处理接口返回信息时出错"
+        
+        return return_data
+
+
     async def jinjia(self, server: str, limit:str) -> Dict[str, Any]:
         """区服金价"""
         return_data = self._init_return_data()