diff --git a/core/async_mysql.py b/core/async_mysql.py new file mode 100644 index 0000000..75a89f1 --- /dev/null +++ b/core/async_mysql.py @@ -0,0 +1,90 @@ +import aiomysql + + +class AsyncMySQL: + def __init__(self, db_config: dict): + self.db_config = db_config + self.pool = None + + async def init_pool(self): + """初始化连接池""" + if self.pool is None: + self.pool = await aiomysql.create_pool(**self.db_config) + + async def close_pool(self): + """关闭连接池""" + if self.pool: + self.pool.close() + await self.pool.wait_closed() + self.pool = None + + async def fetch_one(self, sql: str, params=None): + """查询单条数据""" + await self.init_pool() + async with self.pool.acquire() as conn: # type: ignore + async with conn.cursor(aiomysql.DictCursor) as cursor: + await cursor.execute(sql, params or ()) + return await cursor.fetchone() + + async def fetch_all(self, sql: str, params=None): + """查询多条数据""" + await self.init_pool() + async with self.pool.acquire() as conn: # type: ignore + async with conn.cursor(aiomysql.DictCursor) as cursor: + await cursor.execute(sql, params or ()) + return await cursor.fetchall() + + async def execute(self, sql: str, params=None): + """执行 SQL(insert/update/delete)""" + await self.init_pool() + async with self.pool.acquire() as conn: # type: ignore + async with conn.cursor() as cursor: + await cursor.execute(sql, params or ()) + await conn.commit() + return cursor.rowcount + + async def executemany(self, sql: str, params_list): + """批量执行 SQL""" + await self.init_pool() + async with self.pool.acquire() as conn: # type: ignore + async with conn.cursor() as cursor: + await cursor.executemany(sql, params_list) + await conn.commit() + return cursor.rowcount + + async def truncate_table(self, table_name: str): + """清空指定表""" + await self.init_pool() + async with self.pool.acquire() as conn: # type: ignore + async with conn.cursor() as cursor: + sql = f"TRUNCATE TABLE `{table_name}`" + await cursor.execute(sql) + await conn.commit() + return True + + # ---------------------------------------------------------------------- + # 新增:自动生成 SQL 的增删改功能 + # ---------------------------------------------------------------------- + + async def insert_record(self, table: str, data: dict): + """插入记录:data 是 dict""" + keys = ", ".join(f"`{k}`" for k in data.keys()) + placeholders = ", ".join(["%s"] * 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): + """更新记录:data、where 都是 dict""" + set_clause = ", ".join(f"`{k}`=%s" for k in data.keys()) + where_clause = " AND ".join(f"`{k}`=%s" 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 是 dict""" + where_clause = " AND ".join(f"`{k}`=%s" for k in where.keys()) + sql = f"DELETE FROM `{table}` WHERE {where_clause}" + return await self.execute(sql, tuple(where.values())) \ No newline at end of file diff --git a/core/jx3_service.py b/core/jx3_service.py new file mode 100644 index 0000000..b45743b --- /dev/null +++ b/core/jx3_service.py @@ -0,0 +1,593 @@ +from datetime import datetime +from typing import Dict, Any, Optional, List, Union + +from astrbot.api import logger + +from .request import APIClient +from .AsyncMySQL import AsyncMySQL +from .function_basic import load_template,flatten_field,extract_fields,gold_to_string + +class JX3Service: + def __init__(self, api_config,db: AsyncMySQL ): + self._api = APIClient() + self._db = db + self._api_config = api_config + + + def _init_return_data(self) -> Dict[str, Any]: + """初始化标准的返回数据结构""" + return { + "code": 0, + "msg": "功能函数未执行", + "data": {} + } + + + async def _base_request( + self, + config_key: str, + method: str, + params: Optional[Dict[str, Any]] = None, + out_key: Optional[str] = "data" + ) -> Optional[Any]: + """ + 基础请求封装,处理配置获取和API调用。 + + :param config_key: 配置字典中对应 API 的键名。 + :param method: HTTP方法 ('GET' 或 'POST')。 + :param params: 请求参数或 Body 数据。 + :param out_key: 响应数据中需要提取的字段。 + :return: 成功时返回提取后的数据,失败时返回 None。 + """ + try: + api_config = self._api_config.get(config_key) + if not api_config: + logger.error(f"配置文件中未找到 key: {config_key}") + return None + + # 复制 params,避免修改原始配置模板 + request_params = api_config.get("params", {}).copy() + if params: + request_params.update(params) + + url = api_config.get("url", "") + if not url: + logger.error(f"API配置缺少 URL: {config_key}") + return None + + if method.upper() == 'POST': + data = await self._api.post(url, data=request_params, out_key=out_key) + else: # 默认为 GET + data = await self._api.get(url, params=request_params, out_key=out_key) + + if not data: + logger.warning(f"获取接口信息失败或返回空数据: {config_key}") + + return data + + except Exception as e: + logger.error(f"基础请求调用出错 ({config_key}): {e}") + return None + + # --- 业务功能函数 --- + + async def richang(self,server: str, num: int = 0) -> Dict[str, Any]: + """日常活动""" + return_data = self._init_return_data() + + # 1. 构造请求参数 + params = {"server": server, "num": num} + + # 2. 调用基础请求 + data: Optional[Dict[str, Any]] = await self._base_request( + "jx3_richang", "GET", params=params + ) + + if not data: + return_data["msg"] = "获取接口信息失败" + return return_data + + # 3. 处理返回数据 + try: + # 格式化字符串,利用字典的 get 方法提供默认值 + result_msg = ( + f"{server}\n{data.get('date', '未知日期')}-星期{data.get('week', '未知')}\n" + f"大战:{data.get('war', '无')}\n" + f"战场:{data.get('battle', '无')}\n" + f"阵营:{data.get('orecar', '无')}\n" + f"宗门:{data.get('school', '无')}\n" + f"驰援:{data.get('rescue', '无')}\n" + f"画像:{data.get('draw', '无')}\n" + ) + + # 安全地处理列表索引 + luck = data.get('luck', [None, None, None]) + luck_msg = f"[宠物福缘]:\n{luck[0] or '无'},{luck[1] or '无'},{luck[2] or '无'}\n" + card = data.get('card', [None, None, None]) + card_msg = f"[家园声望·加倍道具]:\n{card[0] or '无'},{card[1] or '无'},{card[2] or '无'}\n" + team = data.get('team', [None, None, None]) + team_msg = f"[武林通鉴·公共任务]:\n{team[0] or '无'}\n[武林通鉴·团队秘境]:\n{team[2] or '无'}\n" + + return_data["data"] = result_msg + luck_msg + card_msg + team_msg + return_data["code"] = 200 + except Exception as e: + logger.error(f"richang 数据处理时出错: {e}") + return_data["msg"] = "处理接口返回信息时出错" + + return return_data + + + async def shapan(self, server: str ) -> Dict[str, Any]: + """区服沙盘""" + return_data = self._init_return_data() + + # 1. 构造请求参数 + params = {"serverName": server} + + # 2. 调用基础请求 + data: Optional[Dict[str, Any]] = await self._base_request( + "aijx3_shapan", "POST", params=params + ) + + if not data: + return_data["msg"] = "获取接口信息失败" + return return_data + + # 3. 处理返回数据 (直接提取图片 URL) + pic_url = data.get("picUrl") + if pic_url: + return_data["data"] = pic_url + return_data["code"] = 200 + else: + return_data["msg"] = "接口未返回图片URL" + + return return_data + + async def kaifu(self, server: str) -> Dict[str, Any]: + """开服状态查询""" + return_data = self._init_return_data() + + # 1. 构造请求参数 + params = {"server": server} + + # 2. 调用基础请求 + data: Optional[Dict[str, Union[int, str]]] = await self._base_request( + "jx3_kaifu", "GET", params=params + ) + + if not data: + return_data["msg"] = "获取接口信息失败" + return return_data + + # 3. 处理返回数据 + try: + status = data.get("status", 0) + timestamp = data.get("time", 0) + + status_time = datetime.fromtimestamp(float(timestamp)).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: + logger.error(f"kaifu 数据处理时出错: {e}") + return_data["msg"] = "处理接口返回信息时出错" + + return return_data + + async def shaohua(self) -> Dict[str, Any]: + """骚话""" + return_data = self._init_return_data() + + # 因为没有参数,所以 params=None + data: Optional[Dict[str, Any]] = await self._base_request("jx3_shaohua", "GET") + + if not data: + return_data["msg"] = "获取接口信息失败" + return return_data + + text = data.get("text") + if text: + return_data["data"] = text + return_data["code"] = 200 + else: + return_data["msg"] = "接口未返回文本" + + return return_data + + + async def jigai(self) -> Dict[str, Any]: + """技改记录""" + return_data = self._init_return_data() + + # 提取字段可能返回列表 + data: Optional[List[Dict[str, Any]]] = await self._base_request("jx3_jigai", "GET") + + if not data or not isinstance(data, list): + return_data["msg"] = "获取接口信息失败或数据格式错误" + return return_data + + try: + result_msg = "剑网三最近技改\n" + # 仅展示前1条,避免消息过长 + for i, item in enumerate(data[:1], 1): + result_msg += f"{i}. {item.get('title', '无标题')}\n" + result_msg += f"时间:{item.get('time', '未知时间')}\n" + result_msg += f"链接:{item.get('url', '无链接')}\n\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) -> Dict[str, Any]: + """区服金价""" + return_data = self._init_return_data() + + params = {"serverName": server} + data_list: Optional[List[Dict[str, Any]]] = await self._base_request("aijx3_jinjia", "POST", params=params) + + if not data_list or not isinstance(data_list, list): + return_data["msg"] = "获取接口信息失败或数据格式错误" + return return_data + + # 安全处理列表切片 + display_data = data_list[:15] + + # 加载模板 + try: + return_data["temp"] = load_template("jinjia.html") + except FileNotFoundError as e: + logger.error(f"加载模板失败: {e}") + return_data["msg"] = "系统错误:模板文件不存在" + return return_data + + # 准备模板渲染数据 + try: + return_data["data"] = { + "items": display_data, + "server": server, + "update_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), + } + return_data["code"] = 200 + except Exception as e: + logger.error(f"jinjia 模板数据准备失败: {e}") + return_data["msg"] = "系统错误:模板渲染数据准备失败" + + return return_data + + + async def qiyu(self, adventureName: str = "阴阳两界", serverName: str = "眉间雪") -> Dict[str, Any]: + """区服奇遇""" + return_data = self._init_return_data() + + params = {"adventureName": adventureName, "serverName": serverName} + data_list: Optional[List[Dict[str, Any]]] = await self._base_request("aijx3_qiyu", "POST", params=params) + + if not data_list or not isinstance(data_list, list): + return_data["msg"] = "获取接口信息失败或数据格式错误" + return return_data + + # 格式化时间 + for item in data_list: + timestamp = item.get("time") + if timestamp and isinstance(timestamp, (int, float)): + # 修复时间戳:原代码显示这里是毫秒级,除以 1000 + item["time"] = datetime.fromtimestamp(timestamp / 1000).strftime("%Y-%m-%d %H:%M:%S") + else: + item["time"] = "未知时间" # 确保即使 time 字段缺失也不会报错 + + # 加载模板 + try: + return_data["temp"] = load_template("qiyuliebiao.html") + except FileNotFoundError as e: + logger.error(f"加载模板失败: {e}") + return_data["msg"] = "系统错误:模板文件不存在" + return return_data + + # 准备模板渲染数据 + try: + return_data["data"] = { + "items": data_list, + "server": serverName, + "update_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), + "qiyuname": adventureName + } + return_data["code"] = 200 + except Exception as e: + logger.error(f"qiyu 模板数据准备失败: {e}") + return_data["msg"] = "系统错误:模板渲染数据准备失败" + + return return_data + + + async def SearchData(self) -> Dict[str, Any]: + """外观数据插入数据库""" + return_data = self._init_return_data() + + # 获取数据 + data: Optional[List[Dict[str, Any]]] = await self._base_request("aijx3_SearchData", "POST") + + if not data: + return_data["msg"] = "获取接口信息失败" + return return_data + + # 提取数据 + try: + extracted_data = flatten_field(data, "dataModels") + except Exception as e: + logger.error(f"提取数据失败: {e}") + return_data["msg"] = "提取指定数据字段失败" + return return_data + + # 准备SQL和批量数据 + sql = """ + INSERT INTO searchdata + (typeName, name, showName, picUrl, searchId, searchDescType) + VALUES (%s, %s, %s, %s, %s, %s) + """ + # 增强数据提取的安全性 + values_list = [ + ( + item.get('typeName'), + item.get('name'), + item.get('showName'), + item.get('picUrl'), + item.get('searchId'), + item.get('searchDescType') + ) + for item in extracted_data + ] + + # 插入数据 + try: + await self._db.truncate_table("searchdata") + await self._db.executemany(sql, values_list) + except Exception as e: + logger.error(f"数据插入失败: {e}") + return_data["msg"] = "数据插入数据库失败" + return return_data + + return_data["msg"] = f"成功批量插入 {len(values_list)} 条数据!" + return_data["code"] = 200 + return return_data + + + async def wujia(self, Name: str) -> Dict[str, Any]: + """物价查询""" + return_data = self._init_return_data() + + # 1. 查询魔盒获取外观名称 (jx3box_exterior) + params_search = {"keyword": Name} + search_data: Optional[Dict[str, Any]] = await self._base_request("jx3box_exterior", "GET", params=params_search) + + showName = "" + searchId = None + + # 2. 确定外观名称和 ID + if search_data and search_data.get("total", 0) > 0 and search_data.get("list"): + # 从 API 获取名称 + showName = search_data["list"][0].get("name", "") + + # 进一步从 DB 查找 ID + try: + sql = "SELECT searchId FROM searchdata WHERE showName=%s" + sqldata = await self._db.fetch_one(sql, (showName,)) + searchId = sqldata["searchId"] if sqldata else None + except Exception as e: + logger.warning(f"从 DB 获取外观ID错误: {e}") # 警告而非错误,因为可能数据库同步失败 + + elif search_data and search_data.get("total", 0) == 0: + # 从 DB 查找名称和 ID + try: + sql = "SELECT showName, searchId FROM searchdata WHERE name = %s OR showName = %s" + sqldata = await self._db.fetch_one(sql, (Name, Name)) + if sqldata: + showName = sqldata["showName"] + searchId = sqldata["searchId"] + else: + raise ValueError("DB did not find data.") + except Exception as e: + logger.error(f"获取外观信息错误: {e}") + return_data["msg"] = "未找到该外观信息" + return return_data + + if not showName or not searchId: + return_data["msg"] = "无法确定外观名称或ID" + return return_data + + return_data["data"]["showName"] = showName + return_data["data"]["searchId"] = searchId + + # 3. 获取外观详细数据 (aijx3_GoodsDetail) + params_detail = {"goodsName": showName} + detail_data: Optional[Dict[str, Any]] = await self._base_request("aijx3_GoodsDetail", "POST", params=params_detail) + + if not detail_data: + return_data["msg"] = "获取外观详细数据失败" + return return_data + + # 提取外观详细数据 + try: + imgs = detail_data.get("imgs", []) + return_data["data"].update({ + "goodsDesc": detail_data.get("goodsDesc", "无描述"), + "publishTime": detail_data.get("publishTime", "无价格"), + "priceNum": detail_data.get("priceNum", 0), + "goodsId": detail_data.get("goodsId", "无数据"), + "imgs": imgs[0] if imgs and isinstance(imgs, list) else "", + "goodsAlias": detail_data.get("goodsAlias", "无别名"), + }) + except Exception as e: + logger.error(f"提取外观详细数据失败: {e}") + return_data["msg"] = "提取外观详细数据失败" + return return_data + + # 4. 查询万宝楼数据(公示和在售) + wbl_data = await self._get_wbl_data(showName) + if wbl_data: + return_data["data"]["wblgs"] = wbl_data["wblgs"] + return_data["data"]["wblzs"] = wbl_data["wblzs"] + return_data["msg"] = "获取万宝楼数据完成" + else: + logger.warning("万宝楼数据获取失败") + + # 5. 加载模板 + try: + return_data["temp"] = load_template("wujia.html") + return_data["code"] = 200 + except FileNotFoundError as e: + logger.error(f"加载模板失败: {e}") + return_data["msg"] = "系统错误:模板文件不存在" + return return_data + + return return_data + + + async def _get_wbl_data(self,showName): + """获取万宝楼数据(公示和在售)""" + try: + #在配置文件中获取接口配置 + #api_config = self._api_config["aijx3_wblwg"] + api_config = self._api_config["wbl_waiguan"] + #更新参数 + api_config["params"]["filter[role_appearance]"] = showName + # 获取公示数据 + api_config["params"]["filter[state]"] = "1" + datawblgs = await self._api.get(api_config["url"], api_config["params"], "data") + # 获取在售数据 + api_config["params"]["filter[state]"] = "2" + datawblzs = await self._api.get(api_config["url"], api_config["params"], "data") + + return { + "wblgs": await self._process_wbl_records(datawblgs.get("list", [])), + "wblzs": await self._process_wbl_records(datawblzs.get("list", [])) + } + except Exception as e: + logger.error(f"获取万宝楼数据出错: {e}") + return None + + + async def _process_wbl_records(self,records): + """处理万宝楼记录数据""" + processed = [] + for record in records: + try: + # 转换时间戳为可读格式 + timestamp = record.get("remaining_time", 0) + days = timestamp // 86400 # 每天有 86400 秒 + hours = (timestamp % 86400) // 3600 # 每小时有 3600 秒 + minutes = (timestamp % 3600) // 60 # 每分钟有 60 秒 + result = "" + if days > 0: + result = (f"{days}天") + if hours > 0: + result += (f"{hours}时") + if minutes > 0: + result += (f"{minutes}分钟") + single_unit_price = record.get("single_unit_price", 0) + processed.append({ + "priceNum": "{:.2f}".format( single_unit_price / 100), + "belongQf2": record.get("server_name", "无数据"), + "replyTime": result, + "discountRate": record.get("discountRate", 0.0), + }) + except Exception as e: + logger.error(f"处理万宝楼记录出错: {e}") + continue + + return processed + + + async def jiaoyihang(self, Name: str , server: str) -> Dict[str, Any]: + """区服交易行""" + return_data = self._init_return_data() + + # 1. 查找物品 ID 列表 (多页) + api_config_search = self._api_config["jx3box_item"] + search_params = api_config_search.get("params", {}).copy() + search_params["keyword"] = Name + + # 使用 all_pages 获取所有数据 + all_items: List[Dict[str, Any]] = await self._api.all_pages( + "GET", + api_config_search["url"], + params_data=search_params, # 对应到 APIClient.all_pages + out_key="data", + list_key="data" + ) + + if not all_items: + return_data["msg"] = "未找到该物品" + return return_data + + # 2. 提取 ID、IconID 和 Name + fields = ["IconID", "Name", "id"] + # 安全提取字段,并确保 id 存在且是数字 + result = [extract_fields([item], fields)[0] for item in all_items if item.get("id") is not None] + + lists_id = [str(item.get("id")) for item in result if item.get("id") is not None] + strlists_id = ",".join(lists_id) + + if not strlists_id: + return_data["msg"] = "未找到有效的物品ID" + return return_data + + # 3. 查询物品价格 (jx3box_itemprice) + api_config_price = self._api_config["jx3box_itemprice"] + price_params = api_config_price.get("params", {}).copy() + price_params["itemIds"] = strlists_id + price_params["server"] = server + + price_data: Optional[Dict[str, Dict[str, Any]]] = await self._api.get( + api_config_price["url"], params=price_params, out_key="data" + ) + + if not price_data: + return_data["msg"] = "未找到在售物品" + return return_data + + # 4. 合并数据和格式化 + fieldsjyh = ["ItemId", "SampleSize", "LowestPrice", "AvgPrice", "Date"] + resultjyh = [{f: v[f] for f in fieldsjyh} for v in price_data.values()] + #合并表格数据 + # 先建立一个字典映射,加快查找 + map_result = {item["id"]: {"IconID": item["IconID"], "Name": item["Name"]} for item in result} + # 遍历 resultjyh,合并字段 + for item in resultjyh: + if item["ItemId"] in map_result: + item.update(map_result[item["ItemId"]]) + #处理数据 + for item in resultjyh: + item["LowestPrice"] = gold_to_string(item["LowestPrice"]) + item["AvgPrice"] = gold_to_string(item["AvgPrice"]) + item["IconID"] = f"https://icon.jx3box.com/icon/{item['IconID']}.png" + + # 5. 模板渲染 + try: + return_data["temp"] = load_template("jiaoyihang.html") + return_data["data"] = { + "items": resultjyh, + "server": server, + "update_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S") + } + return_data["code"] = 200 + except FileNotFoundError as e: + logger.error(f"加载模板失败: {e}") + return_data["msg"] = "系统错误:模板文件不存在" + except Exception as e: + logger.error(f"模板渲染数据准备失败: {e}") + return_data["msg"] = "系统错误:数据处理失败" + + return return_data diff --git a/core/request.py b/core/request.py new file mode 100644 index 0000000..7b43000 --- /dev/null +++ b/core/request.py @@ -0,0 +1,192 @@ +# core/request.py +import json +import aiohttp +from typing import Optional, Dict, Any, Union, List + +from aiohttp import ClientTimeout, ClientSession +from astrbot.api import logger + +class APIClient: + """ + API客户端类 + + 优化说明: + 1. 复用 aiohttp.ClientSession 以提高性能。 + 2. 增加类型提示 (Type Hints)。 + 3. 支持异步上下文管理器 (Async Context Manager)。 + """ + + def __init__(self, base_timeout: int = 10, ssl_verify: bool = False): + self.base_timeout = base_timeout + self.ssl_verify = ssl_verify + self._session: Optional[ClientSession] = None + + async def get_session(self) -> ClientSession: + """获取或创建单例 Session""" + if self._session is None or self._session.closed: + timeout = ClientTimeout(total=self.base_timeout) + self._session = ClientSession(timeout=timeout) + return self._session + + async def close(self): + """关闭 Session""" + if self._session and not self._session.closed: + await self._session.close() + + async def __aenter__(self): + await self.get_session() + return self + + async def __aexit__(self, exc_type, exc_val, exc_tb): + await self.close() + + async def _request(self, method: str, url: str, params: Optional[Dict] = None, json_data: Optional[Dict] = None) -> Any: + """ + 统一的内部请求处理方法 + """ + session = await self.get_session() + method = method.upper() + + # 记录日志 + logger.debug(f"发起 {method} 请求: {url}") + if params: logger.debug(f"Query参数: {params}") + if json_data: logger.debug(f"Body数据: {json_data}") + + try: + # aiohttp 会自动处理 json=json_data 时的 Content-Type + async with session.request( + method=method, + url=url, + params=params, + json=json_data, + ssl=self.ssl_verify + ) as response: + return await self._handle_response(response) + + except aiohttp.ClientError as e: + logger.error(f"网络请求出错 ({method} {url}): {e}") + return None + except Exception as e: + logger.error(f"未知错误 ({method} {url}): {e}") + return None + + async def _handle_response(self, response: aiohttp.ClientResponse) -> Any: + """处理响应:自动识别二进制或JSON""" + try: + logger.debug(f"响应状态: {response.status}") + response.raise_for_status() # 如果是 4xx/5xx 直接抛出异常 + + content_type = response.headers.get('Content-Type', '').lower() + + # 处理二进制流 (图片、文件等) + if 'image' in content_type or 'octet-stream' in content_type: + return await response.read() + + # 处理 JSON + # 优先尝试标准 json 解析 + try: + data = await response.json() + except Exception: + # 容错:有些 API 返回 text/html 但内容是 JSON + text = await response.text() + try: + data = json.loads(text) + except json.JSONDecodeError: + logger.error(f"无法解析响应为 JSON。原始内容: {text[:100]}...") + return None + + logger.debug(f"响应数据: {data}") + return self._validate_api_payload(data) + + except aiohttp.ClientError as e: + logger.error(f"HTTP响应错误: {e}") + return None + + def _validate_api_payload(self, data: Any) -> Any: + """校验业务层面的 JSON 数据结构""" + if not data: + logger.error("API返回空数据") + return None + + # 如果返回的是 JSON 字符串而非对象,再次解析 + if isinstance(data, str): + try: + data = json.loads(data) + except json.JSONDecodeError: + return None + + if isinstance(data, dict) and 'code' in data: + # 兼容多种成功状态码:200, "0", 0, 1 + code = data.get('code') + if code not in [200, "0", 0, 1]: + msg = data.get('msg') or data.get('message', '未知错误') + logger.error(f"API业务报错: code={code}, msg={msg}") + return None + + return data + + async def get(self, url: str, params: Optional[Dict] = None, out_key: Optional[str] = None) -> Any: + """GET 请求封装""" + data = await self._request('GET', url, params=params) + return self._extract_data(data, out_key) + + async def post(self, url: str, data: Optional[Dict] = None, out_key: Optional[str] = None) -> Any: + """POST 请求封装 (默认发送 JSON)""" + data = await self._request('POST', url, json_data=data) + return self._extract_data(data, out_key) + + def _extract_data(self, data: Any, key: Optional[str]) -> Any: + """辅助方法:从结果中提取指定字段""" + if data is None: + return None + if isinstance(data, bytes): + return data + if key and isinstance(data, dict): + return data.get(key, {}) + return data + + async def all_pages( + self, + method: str, + url: str, + params_data: Optional[Dict] = None, + out_key: str = "", + list_key: str = "list", + max_pages: int = 10 + ) -> List[Any]: + """ + 分页获取所有数据 + :param method: GET 或 POST + :param list_key: 列表数据在 JSON 中的字段名,如 'data' 或 'list' + """ + all_data = [] + current_page = 1 + params = params_data.copy() if params_data else {} + + while True: + params["page"] = str(current_page) + + if method.upper() == "POST": + data = await self.post(url, data=params, out_key=out_key) + else: + data = await self.get(url, params=params, out_key=out_key) + + # 终止条件判断 + if not data or isinstance(data, bytes): + break + + # 如果 data 是列表本身(有些API直接返回列表) + page_items = data if isinstance(data, list) else data.get(list_key) + + if not page_items: + break + + all_data.extend(page_items) + + if current_page >= max_pages: + break + + current_page += 1 + logger.info(f"已获取第 {current_page} 页数据") + + return all_data \ No newline at end of file diff --git a/main.py b/main.py index 067a97a..358f21f 100644 --- a/main.py +++ b/main.py @@ -8,8 +8,8 @@ from astrbot.api.star import Context, Star, register, StarTools from astrbot.api import logger from astrbot.api import AstrBotConfig -from .core.AsyncMySQL import AsyncMySQL -from .core.JX3Function import JX3Function +from .core.async_mysql import AsyncMySQL +from .core.jx3_service import JX3Service from .core.WZRYFunction import WZRYFunction @@ -52,7 +52,7 @@ class Jx3ApiPlugin(Star): } #创建类实例 self.db = AsyncMySQL(db_config) - self.jx3fun = JX3Function(self.api_config,self.db) + self.jx3fun = JX3Service(self.api_config,self.db) self.wzry = WZRYFunction(self.api_config,self.db) # 周期函数调用 diff --git a/test.py b/test.py new file mode 100644 index 0000000..ab2efbd --- /dev/null +++ b/test.py @@ -0,0 +1,14 @@ +import http.client + +conn = http.client.HTTPSConnection("api.t1qq.com") +payload = '' +headers = { + 'User-Agent': 'Apifox/1.0.0 (https://apifox.com)', + 'Accept': '*/*', + 'Host': 'api.t1qq.com', + 'Connection': 'keep-alive' +} +conn.request("GET", "/api/tool/wzrr/ydtp?key=vBpEzoiC9z5A9c9Nn83IhLn6M9&id=489048724", payload, headers) +res = conn.getresponse() +data = res.read() +print(data) \ No newline at end of file