11

core/aiosqlite.py

@@ -1,105 +0,0 @@
-import aiosqlite
-import logging
-import re
-
-logging.basicConfig(level=logging.DEBUG)
-logger = logging.getLogger("AsyncSQLite")
-
-
-class AsyncSQLite:
-    def __init__(self, db_path: str):
-        self.db_path = db_path
-        self.conn = None
-
-    async def init(self):
-        if self.conn is None:
-            self.conn = await aiosqlite.connect(self.db_path)
-            self.conn.row_factory = aiosqlite.Row
-
-    async def close(self):
-        if self.conn:
-            await self.conn.close()
-            self.conn = None
-
-    def _format_sql(self, sql: str, params):
-        """生成完整 SQL（安全打印）"""
-
-        if not params:
-            return sql
-
-        def escape(value):
-            if value is None:
-                return "NULL"
-            if isinstance(value, (int, float)):
-                return str(value)
-            # 转义单引号
-            value = str(value).replace("'", "''")
-            return f"'{value}'"
-
-        final_sql = sql
-        for v in params:
-            final_sql = final_sql.replace("?", escape(v), 1)
-
-        return final_sql
-
-    async def _log_and_execute(self, sql: str, params=None, fetch: str = None):
-        await self.init()
-
-        # 打印原始 SQL 和参数
-        logger.debug(f"[SQLite] SQL Raw: {sql}")
-        logger.debug(f"[SQLite] Params: {params}")
-
-        # 打印最终执行 SQL
-        final_sql = self._format_sql(sql, params or ())
-        logger.debug(f"[SQLite] SQL Final: {final_sql}")
-
-        async with self.conn.execute(sql, params or ()) as cursor:
-            if fetch == "one":
-                row = await cursor.fetchone()
-                return dict(row) if row else None
-
-            if fetch == "all":
-                rows = await cursor.fetchall()
-                return [dict(r) for r in rows]
-
-            await self.conn.commit()
-            return True
-
-    async def fetch_one(self, sql: str, params=None):
-        return await self._log_and_execute(sql, params, fetch="one")
-
-    async def fetch_all(self, sql: str, params=None):
-        return await self._log_and_execute(sql, params, fetch="all")
-
-    async def execute(self, sql: str, params=None):
-        return await self._log_and_execute(sql, params)
-
-    async def executemany(self, sql: str, params_list):
-        await self.init()
-        logger.debug(f"[SQLite] SQL (executemany): {sql}")
-        logger.debug(f"[SQLite] Params List: {params_list}")
-        await self.conn.executemany(sql, params_list)
-        await self.conn.commit()
-        return True
-
-    async def insert_record(self, table: str, data: dict):
-        keys = ", ".join(f"`{k}`" for k in data.keys())
-        placeholders = ", ".join(['?'] * len(data))
-        sql = f"INSERT INTO `{table}` ({keys}) VALUES ({placeholders})"
-        return await self.execute(sql, tuple(data.values()))
-
-    async def update_record(self, table: str, data: dict, where: dict):
-        set_clause = ", ".join(f"`{k}`=?" for k in data.keys())
-        where_clause = " AND ".join(f"`{k}`=?" for k in where.keys())
-        sql = f"UPDATE `{table}` SET {set_clause} WHERE {where_clause}"
-        params = tuple(data.values()) + tuple(where.values())
-        return await self.execute(sql, params)
-
-    async def delete_record(self, table: str, where: dict):
-        where_clause = " AND ".join(f"`{k}`=?" for k in where.keys())
-        sql = f"DELETE FROM `{table}` WHERE {where_clause}"
-        return await self.execute(sql, tuple(where.values()))
-
-    async def clear_table(self, table: str):
-        sql = f"DELETE FROM `{table}`"
-        return await self.execute(sql)

core/jx3_service.py

@@ -1,3 +1,4 @@
+
 from datetime import datetime
 from typing import Dict, Any, Optional, List, Union
 
@@ -114,7 +115,6 @@ class JX3Service:
         data: Optional[Dict[str, Any]] = await self._base_request(
             "jx3_richang", "GET", params=params
         )
-        logger.info(f"richang 接口返回数据: {data}")
         if not data:
             return_data["msg"] = "获取接口信息失败"
             return return_data
@@ -148,6 +148,31 @@ class JX3Service:
 
         return return_data
     
+    async def richangyuche(self,server: str) -> Dict[str, Any]:
+        """日常预测"""
+        return_data = self._init_return_data()
+
+        # 1. 构造请求参数
+        params = { "num": 30}
+
+        # 2. 调用基础请求
+        data: Optional[Dict[str, Any]] = await self._base_request(
+            "jx3_richangyuche", "GET", params=params
+        )
+        logger.info(f"richang 接口返回数据: {data}")
+        if not data:
+            return_data["msg"] = "获取接口信息失败"
+            return return_data
+    
+        # 3. 处理返回数据
+        try:
+            # 格式化字符串，利用字典的 get 方法提供默认值
+            return_data["code"] = 0
+        except Exception as e:
+            logger.error(f"richang 数据处理时出错: {e}")
+            return_data["msg"] = "处理接口返回信息时出错"
+
+        return return_data
 
     async def shapan(self, server: str ) -> Dict[str, Any]:
         """区服沙盘"""

data/api_config.json

@@ -203,5 +203,13 @@
         "params":{
             "server": ""
         }
+    },
+    "jx3_richangyuche":{
+        "url":"https://www.jx3api.com/data/active/list/calendar",
+        "method":"GET",
+        "description":"该 API 用于预测每天的日常任务，您可以通过指定时间范围来获取相关任务数据。",
+        "params":{
+            "num": "30"
+        }
     }
 }

main.py

@@ -149,6 +149,19 @@ class Jx3ApiPlugin(Star):
             logger.error(f"功能函数执行错误: {e}")
             yield event.plain_result("猪脑过载，请稍后再试")
     
+    @jx3.command("月历")
+    async def jx3_richangyuche(self, event: AstrMessageEvent,server: str = "" ,num: int = 0):
+        """剑三 月历 服务器 """
+        try:
+            data= await self.jx3fun.richangyuche(await self.serverdefault(server))
+            if data["code"] == 200:
+                yield event.plain_result(data["data"])
+            else:
+                yield event.plain_result(data["msg"])
+            return
+        except Exception as e:
+            logger.error(f"功能函数执行错误: {e}")
+            yield event.plain_result("猪脑过载，请稍后再试")
 
     @jx3.command("开服")
     async def jx3_kaifu(self, event: AstrMessageEvent,server: str = ""):