From 4398ad32bc3a6183a4c4fbb22b3281cf20a2633b Mon Sep 17 00:00:00 2001 From: qsc Date: Mon, 29 Sep 2025 11:52:41 +0800 Subject: [PATCH] 11 --- api_config.json | 24 ++++++ core/api_data.py | 138 +++++++++++++++++------------ core/jx3_Function.py | 121 ++++++++++++++++++++++++++ main.py | 200 +++++++++++-------------------------------- 4 files changed, 278 insertions(+), 205 deletions(-) create mode 100644 core/jx3_Function.py diff --git a/api_config.json b/api_config.json index 763d244..ce6e6ca 100644 --- a/api_config.json +++ b/api_config.json @@ -7,5 +7,29 @@ "server": "眉间雪", "num": 0 } + }, + "jx3_shaohua":{ + "url":"https://www.jx3api.com/data/saohua/random", + "method":"GET", + "description":"获取剑网三骚话", + "params":{ + "name": "万花" + } + }, + "jx3_jigai":{ + "url":"https://www.jx3api.com/data/skills/records", + "method":"GET", + "description":"获取剑网三技能记录", + "params":{ + } + }, + "aijx3_shapan":{ + "url":"https://www.jianxiachaguan.cn/api2/aijx3-jxcg/game/get-sand-table-img", + "method":"POST", + "description":"获取剑网三沙盘状态", + "params":{ + "serverName": "眉间雪" + } } + } \ No newline at end of file diff --git a/core/api_data.py b/core/api_data.py index 5655a96..320a351 100644 --- a/core/api_data.py +++ b/core/api_data.py @@ -20,14 +20,14 @@ class APIClient: self.base_timeout = base_timeout self.ssl_verify = ssl_verify - async def _make_request(self, method, url, **kwargs): + async def _make_request(self, method, url, params_data=None): """ - 内部请求方法，处理通用请求逻辑 + 内部请求方法，统一处理请求逻辑 Args: method: 请求方法 ('GET', 'POST') url: 请求URL - **kwargs: 其他请求参数 + params_data: 统一的参数字典，如 {"name": "万花"} Returns: 成功时返回解析后的数据，失败返回None @@ -35,24 +35,77 @@ class APIClient: timeout = ClientTimeout(total=self.base_timeout) try: + # 添加请求日志 + logger.debug(f"发起 {method} 请求: {url}") + logger.debug(f"参数数据: {params_data}") + async with aiohttp.ClientSession(timeout=timeout) as session: - async with session.request(method, url, ssl=self.ssl_verify, **kwargs) as response: - response.raise_for_status() - data = await response.json() - - # 检查API响应状态 - return self._check_response_data(data) + if method.upper() == 'GET': + # GET请求：参数作为查询字符串 + async with session.get( + url, + params=params_data, + ssl=self.ssl_verify + ) as response: + return await self._handle_response(response) + + elif method.upper() == 'POST': + # POST请求：参数作为JSON数据 + headers = {'Content-Type': 'application/json'} + async with session.post( + url, + json=params_data, # 使用json参数而不是data + headers=headers, + ssl=self.ssl_verify + ) as response: + return await self._handle_response(response) + + else: + logger.error(f"不支持的HTTP方法: {method}") + return None except aiohttp.ClientError as e: logger.error(f"请求出错: {e}") return None - except json.JSONDecodeError as e: - logger.error(f"JSON解析错误: {e}") - return None except Exception as e: logger.error(f"未知错误: {e}") return None + async def _handle_response(self, response): + """ + 处理HTTP响应 + + Args: + response: aiohttp响应对象 + + Returns: + 解析后的数据或None + """ + try: + # 添加响应日志 + logger.debug(f"响应状态: {response.status}") + + response.raise_for_status() + data = await response.json() + + logger.debug(f"响应数据: {data}") + + # 检查API响应状态 + return self._check_response_data(data) + + except aiohttp.ClientError as e: + logger.error(f"HTTP错误: {e}") + return None + except json.JSONDecodeError as e: + logger.error(f"JSON解析错误: {e}") + # 尝试读取原始文本内容 + try: + text_content = await response.text() + logger.error(f"原始响应内容: {text_content}") + except: + pass + return None + def _check_response_data(self, data): """ 检查API响应数据的通用逻辑 @@ -63,8 +116,16 @@ class APIClient: Returns: 检查通过返回数据，否则返回None """ + # 如果数据是字符串，尝试解析为JSON + if isinstance(data, str): + try: + data = json.loads(data) + except json.JSONDecodeError: + logger.error("响应数据是无效的JSON字符串") + return None + # 检查是否有code字段 - if 'code' in data: + if data and 'code' in data: # 有code字段时，检查是否成功 if data.get('code') not in [200, "0", 0]: logger.error(f"API返回错误:{data.get('code', '未知状态')} {data.get('msg', '未知错误')}") @@ -77,19 +138,19 @@ class APIClient: return data - async def post(self, api_url, json_data=None, outdata=None): + async def post(self, api_url, params_data=None, outdata=None): """ POST请求方法 Args: api_url: API地址 - json_data: POST请求的JSON数据 + params_data: 参数字典，如 {"name": "万花"} outdata: 返回数据中要提取的字段 Returns: 成功时返回outdata字段的数据，失败返回None """ - data = await self._make_request('POST', api_url, json=json_data) + data = await self._make_request('POST', api_url, params_data) if data is None: return None @@ -98,42 +159,19 @@ class APIClient: return data return data.get(outdata, {}) - async def get(self, api_url, params=None, outdata=None): + async def get(self, api_url, params_data=None, outdata=None): """ GET请求方法 Args: api_url: API地址 - params: 其他查询参数 + params_data: 参数字典，如 {"name": "万花"} outdata: 返回数据中要提取的字段 Returns: 成功时返回outdata字段的数据，失败返回None """ - data = await self._make_request('GET', api_url, params=params) - - if data is None: - return None - - if outdata is None or outdata == "": - return data - return data.get(outdata, {}) - - async def request(self, api_config, outdata=None): - """ - 通用请求方法，根据api_config中的method选择GET或POST - - Args: - api_config: 包含url, method, params/json等信息的字典 - outdata: 返回数据中要提取的字段 - - Returns: - 成功时返回outdata字段的数据，失败返回None - """ - if 'url' not in api_config or 'method' not in api_config: - logger.error("api_config必须包含url和method字段") - return None - data = await self._make_request(api_config['method'], api_config['url'], params=api_config['params']) + data = await self._make_request('GET', api_url, params_data) if data is None: return None @@ -143,19 +181,13 @@ class APIClient: return data.get(outdata, {}) - - - - - - -# 保持原有函数接口的兼容性（可选） -async def api_data_post(api_url, json_data=None, outdata=None): +# 保持原有函数接口的兼容性 +async def api_data_post(api_url, params_data=None, outdata=None): """兼容原有函数的POST请求""" client = APIClient() - return await client.post(api_url, json_data, outdata) + return await client.post(api_url, params_data, outdata) -async def api_data_get(api_url, params=None, outdata=None): +async def api_data_get(api_url, params_data=None, outdata=None): """兼容原有函数的GET请求""" client = APIClient() - return await client.get(api_url, params, outdata) \ No newline at end of file + return await client.get(api_url, params_data, outdata) \ No newline at end of file diff --git a/core/jx3_Function.py b/core/jx3_Function.py new file mode 100644 index 0000000..83cba30 --- /dev/null +++ b/core/jx3_Function.py @@ -0,0 +1,121 @@ +from astrbot.api import logger +from astrbot.api.event import filter, AstrMessageEvent, MessageEventResult + +from .api_data import APIClient + + +class JX3Function: + def __init__(self, api_config): + self.__api = APIClient() + self.__api_config = api_config + + async def richang(self,server: str = "眉间雪",num: int = 0): + return_data = { + "code": 0, + "msg": "功能函数未执行", + "data": {} + } + #在配置文件中获取接口配置 + api_config = self.__api_config["jx3_richang"] + #更新参数 + api_config["params"]["server"] = server + api_config["params"]["num"] = num + # 获取数据 + data = await self.__api.get(api_config["url"],api_config["params"],"data") + if not data: + return_data["msg"] = "获取接口信息失败" + return return_data + # 处理返回数据 + try: + result_msg = f"{server}-{data.get('date')}-星期{data.get('week')}\n" + result_msg += f"大战：{data.get('war')}\n" + result_msg += f"战场：{data.get('battle')}\n" + result_msg += f"阵营：{data.get('orecar')}\n" + result_msg += f"宗门：{data.get('school')}\n" + result_msg += f"驰援：{data.get('rescue')}\n" + result_msg += f"画像：{data.get('draw')}\n" + result_msg += f"宠物福缘：\n{data.get('luck')}\n" + result_msg += f"家园声望：\n{data.get('card')}\n" + result_msg += f"武林通鉴：\n{data.get('team')}\n" + return_data["data"] = result_msg + return_data["code"] = 200 + except Exception as e: + logger.error(f"处理数据时出错: {e}") + return_data["msg"] = "处理接口返回信息时出错" + return return_data + + + async def shapan(self,server: str = "眉间雪"): + return_data = { + "code": 0, + "msg": "功能函数未执行", + "data": {} + } + #在配置文件中获取接口配置 + api_config = self.__api_config["aijx3_shapan"] + #更新参数 + api_config["params"]["serverName"] = server + # 获取数据 + data = await self.__api.post(api_config["url"],api_config["params"],"data") + if not data: + return_data["msg"] = "获取接口信息失败" + return return_data + # 处理返回数据 + try: + return_data["data"] = data.get("picUrl") + return_data["code"] = 200 + except Exception as e: + logger.error(f"处理数据时出错: {e}") + return_data["msg"] = "处理接口返回信息时出错" + return return_data + + + async def shaohua(self): + return_data = { + "code": 0, + "msg": "功能函数未执行", + "data": {} + } + #在配置文件中获取接口配置 + api_config = self.__api_config["jx3_shaohua"] + # 获取数据 + data = await self.__api.get(api_config["url"],api_config["params"],"data") + if not data: + return_data["msg"] = "获取接口信息失败" + return return_data + # 处理返回数据 + try: + return_data["data"] = data.get("text") + return_data["code"] = 200 + except Exception as e: + logger.error(f"处理数据时出错: {e}") + return_data["msg"] = "处理接口返回信息时出错" + return return_data + + + async def jigai(self): + return_data = { + "code": 0, + "msg": "功能函数未执行", + "data": {} + } + #在配置文件中获取接口配置 + api_config = self.__api_config["jx3_jigai"] + # 获取数据 + data = await self.__api.get(api_config["url"],api_config["params"],"data") + if not data: + return_data["msg"] = "获取接口信息失败" + return return_data + # 处理返回数据 + try: + result_msg = f"剑网三最近技改\n" + for i, item in enumerate(data[:2], 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"处理数据时出错: {e}") + return_data["msg"] = "处理接口返回信息时出错" + return return_data \ No newline at end of file diff --git a/main.py b/main.py index 0a127c7..f5158fb 100644 --- a/main.py +++ b/main.py @@ -11,6 +11,7 @@ from .core.load_template import load_template from .core.jx3_data import jx3_data_jiaoyihang,jx3_data_wujia from .core.api_data import APIClient,api_data_get, api_data_post from .core.sql_data import sql_data_searchdata,sql_data_select +from .core.jx3_Function import JX3Function @register("astrbot_plugin_jx3api", @@ -36,32 +37,43 @@ class Jx3ApiPlugin(Star): async def initialize(self): """可选择实现异步的插件初始化方法，当实例化该插件类之后会自动调用该方法。""" #创建类实例 - self.api_data = APIClient() + self.jx3fun = JX3Function(self.api_config) logger.info("jx3api插件创建实例完成") - async def message_str(self, event: AstrMessageEvent): - """可选择实现异步的消息预处理方法，在消息分发到具体命令前会调用该方法。""" - # 预处理消息内容 - message_str = event.message_str.strip() - return message_str.split() + + + @filter.command_group("剑三") + def jx3(self): + pass - @filter.command("剑三日常") - async def jx3_richang(self, event: AstrMessageEvent): - """剑三日常""" - # 接口URL - custom_url = "https://www.jx3api.com/data/active/calendar" - # 接口参数 - params = { - "server": "眉间雪", # 服务器 - "num": 0 # 当天 - } - # 获取消息内容 - parts = await self.message_str(event) - # 解析消息内容 - if len(parts) > 1: - params["server"] = parts[1] + @jx3.command("日常") + async def jx3_richang(self, event: AstrMessageEvent,server: str = "眉间雪",num: int = 0): + """剑三 日常 服务器 天数""" + try: + data= await self.jx3fun.richang(server,num) + if data["code"] == 200: + yield event.plain_result(data["data"]) + else: + yield event.plain_result("msg") + return + except Exception as e: + logger.error(f"功能函数执行错误: {e}") + yield event.plain_result("猪脑过载，请稍后再试") + + @jx3.command("沙盘") + async def jx3_shapan(self, event: AstrMessageEvent,server: str = "眉间雪"): + """剑三 沙盘 服务器""" + try: + data= await self.jx3fun.shapan(server) + if data["code"] == 200: + yield event.image_result(data["data"]) + else: + yield event.plain_result(data["msg"]) + return + except Exception as e: + logger.error(f"功能函数执行错误: {e}") + yield event.plain_result("猪脑过载，请稍后再试") -<<<<<<< HEAD @jx3.command("骚话") async def jx3_shaohua(self, event: AstrMessageEvent,): @@ -72,110 +84,25 @@ class Jx3ApiPlugin(Star): yield event.plain_result(data["data"]) else: yield event.plain_result(data["msg"]) -======= - if len(parts) > 2: - try: - params["num"] = int(parts[2]) - except ValueError: - params["num"] = 0 - - # 获取数据 - data = await self.api_data.request(self.api_config["jx3_richang"],"data") - - if not data: - yield event.plain_result("获取获取接口信息失败，请稍后再试") ->>>>>>> ecb573761ec04986a07da4a19511ef52e9f36ec6 return - - # 处理返回数据 - try: - # 构建回复消息 - result_msg = f"{params['server']}-{data.get('date')}-星期{data.get('week')}\n" - result_msg += f"大战：{data.get('war')}\n" - result_msg += f"战场：{data.get('battle')}\n" - result_msg += f"阵营：{data.get('orecar')}\n" - result_msg += f"宗门：{data.get('school')}\n" - result_msg += f"驰援：{data.get('rescue')}\n" - result_msg += f"画像：{data.get('draw')}\n" - result_msg += f"宠物福缘：\n{data.get('luck')}\n" - result_msg += f"家园声望：\n{data.get('card')}\n" - result_msg += f"武林通鉴：\n{data.get('team')}\n" - - yield event.plain_result(result_msg) - except Exception as e: - logger.error(f"处理数据时出错: {e}") - yield event.plain_result("处理接口返回信息时出错") + logger.error(f"功能函数执行错误: {e}") + yield event.plain_result("猪脑过载，请稍后再试") - @filter.command("剑三骚话") - async def jx3_shaohua(self, event: AstrMessageEvent): - """剑三骚话""" - # 接口URL - custom_url = "https://www.jx3api.com/data/saohua/random" - # 接口参数 - params = { - "name": "万花", # 默认值 - } - - # 获取消息内容 - - # 解析消息内容 - - # 获取数据 - data = api_data_get(custom_url, params,"data") - - if not data: - yield event.plain_result("获取获取接口信息失败，请稍后再试") + @jx3.command("技改") + async def jx3_jigai(self, event: AstrMessageEvent,): + """剑三 技改""" + try: + data= await self.jx3fun.jigai() + if data["code"] == 200: + yield event.plain_result(data["data"]) + else: + yield event.plain_result(data["msg"]) return - - # 处理返回数据 - try: - # 构建回复消息 - result_msg = f"{data.get('text')}\n" - - yield event.plain_result(result_msg) - except Exception as e: - logger.error(f"处理数据时出错: {e}") - yield event.plain_result("处理接口返回信息时出错") - - - @filter.command("剑三技改") - async def jx3_jigai(self, event: AstrMessageEvent): - """剑三技改""" - # 接口URL - custom_url = "https://www.jx3api.com/data/skills/records" - # 接口参数 - params = { - - } - - # 获取消息内容 - - # 解析消息内容 - - # 获取数据 - data = api_data_get(custom_url,params,"data") - - if not data: - yield event.plain_result("获取获取接口信息失败，请稍后再试") - return - - # 处理返回数据 - try: - # 构建回复消息 - result_msg = f"剑网三最近技改\n" - for i, item in enumerate(data[:2], 1): - result_msg += f"{i}. {item.get('title', '无标题')}\n" - result_msg += f"时间：{item.get('time', '未知时间')}\n" - result_msg += f"链接：{item.get('url', '无链接')}\n\n" - - yield event.plain_result(result_msg) - - except Exception as e: - logger.error(f"处理数据时出错: {e}") - yield event.plain_result("处理接口返回信息时出错") + logger.error(f"功能函数执行错误: {e}") + yield event.plain_result("猪脑过载，请稍后再试") @filter.command("剑三交易行") @@ -234,38 +161,7 @@ class Jx3ApiPlugin(Star): yield event.plain_result("查询交易行数据时出错，请稍后再试") - @filter.command("剑三沙盘") - async def jx3_shapan(self, event: AstrMessageEvent): - """剑三沙盘 服务器""" - # 接口URL - custom_url = "https://www.jianxiachaguan.cn/api2/aijx3-jxcg/game/get-sand-table-img" - # 接口参数 - params = { - "serverName": "眉间雪" # 默认服务器 - } - - # 获取消息内容 - message_str = event.message_str.strip() - parts = message_str.split() - # 解析消息内容 - if len(parts) > 1: - params["serverName"] = parts[1] # 第二个参数为服务器 - - # 获取数据 - data = api_data_post(custom_url,params,"data") - - if not data: - yield event.plain_result("获取获取接口信息失败，请稍后再试") - return - - # 处理返回数据 - try: - # 构建回复消息 - yield event.image_result(data.get("picUrl")) - - except Exception as e: - logger.error(f"处理数据时出错: {e}") - yield event.plain_result("处理接口返回信息时出错") + @filter.command("剑三金价")