diff --git a/core/aiosqlite.py b/core/aiosqlite.py deleted file mode 100644 index d379be4..0000000 --- a/core/aiosqlite.py +++ /dev/null @@ -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) diff --git a/core/jx3_service.py b/core/jx3_service.py index fb9401e..65f74a5 100644 --- a/core/jx3_service.py +++ b/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 @@ -147,7 +147,32 @@ class JX3Service: return_data["msg"] = "处理接口返回信息时出错" 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]: """区服沙盘""" @@ -464,7 +489,7 @@ class JX3Service: try: for item in data: - inner_list = item.get("data", []) + inner_list = item.get("data", []) first = inner_list[0] if inner_list else {} new_item = { "name": item.get("name"), diff --git a/data/api_config.json b/data/api_config.json index 78b7c8b..3c26290 100644 --- a/data/api_config.json +++ b/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" + } } } \ No newline at end of file diff --git a/main.py b/main.py index 52087c6..91eb19a 100644 --- a/main.py +++ b/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 = ""):