diff --git a/api_config.json b/api_config.json new file mode 100644 index 0000000..763d244 --- /dev/null +++ b/api_config.json @@ -0,0 +1,11 @@ +{ + "jx3_richang":{ + "url":"https://www.jx3api.com/data/active/calendar", + "method":"GET", + "description":"获取剑网三日常任务", + "params":{ + "server": "眉间雪", + "num": 0 + } + } +} \ No newline at end of file diff --git a/core/api_data.py b/core/api_data.py index d1cc9cb..5af65b7 100644 --- a/core/api_data.py +++ b/core/api_data.py @@ -1,25 +1,68 @@ # core/request.py -import requests +import aiohttp import json +from aiohttp import ClientTimeout from astrbot.api import logger -def api_data_post(api_url, json_data=None, outdata=None): +class APIClient: """ - 获取数据的POST请求函数 - - Args: - api_url: API地址 - json_data: POST请求的JSON数据 - outdata: 返回数据中要提取的字段 - - Returns: - 成功时返回outdata字段的数据，失败返回None + API客户端类，封装GET和POST请求功能 """ - try: - response = requests.post(api_url, json=json_data, timeout=10, verify=False) - response.raise_for_status() - data = response.json() + + def __init__(self, base_timeout=10, ssl_verify=False): + """ + 初始化APIClient + Args: + base_timeout: 默认超时时间（秒） + ssl_verify: SSL证书验证开关 + """ + self.base_timeout = base_timeout + self.ssl_verify = ssl_verify + + async def _make_request(self, method, url, **kwargs): + """ + 内部请求方法，处理通用请求逻辑 + + Args: + method: 请求方法 ('GET', 'POST') + url: 请求URL + **kwargs: 其他请求参数 + + Returns: + 成功时返回解析后的数据，失败返回None + """ + timeout = ClientTimeout(total=self.base_timeout) + + try: + 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) + + 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 + + def _check_response_data(self, data): + """ + 检查API响应数据的通用逻辑 + + Args: + data: API返回的数据 + + Returns: + 检查通过返回数据，否则返回None + """ # 检查是否有code字段 if 'code' in data: # 有code字段时，检查是否成功 @@ -32,53 +75,88 @@ def api_data_post(api_url, json_data=None, outdata=None): logger.error("API返回空数据") return None + return data + + async def post(self, api_url, json_data=None, outdata=None): + """ + POST请求方法 + + Args: + api_url: API地址 + json_data: POST请求的JSON数据 + outdata: 返回数据中要提取的字段 + + Returns: + 成功时返回outdata字段的数据，失败返回None + """ + data = await self._make_request('POST', api_url, json=json_data) + + if data is None: + return None + if outdata is None or outdata == "": return data return data.get(outdata, {}) - - except requests.exceptions.RequestException as e: - logger.error(f"请求出错: {e}") - return None - except json.JSONDecodeError as e: - logger.error(f"JSON解析错误: {e}") - return None -def api_data_get(api_url,params=None,outdata=None): - """ - 获取数据的GET请求函数 - - Args: - api_url: API地址 + async def get(self, api_url, params=None, outdata=None): + """ + GET请求方法 - params: 其他查询参数 - - Returns: - 成功时返回outdata字段的数据，失败返回None - """ - try: - response = requests.get(api_url, params, timeout=10, verify=False) - response.raise_for_status() - data = response.json() + Args: + api_url: API地址 + params: 其他查询参数 + outdata: 返回数据中要提取的字段 - # 检查是否有code字段 - if 'code' in data: - # 有code字段时，检查是否成功 - if data.get('code') not in [200, "0", 0]: - logger.error(f"API返回错误:{data.get('code', '未知状态')} {data.get('msg', '未知错误')}") - return None - else: - # 无code字段时，检查数据是否为空 - if not data: - logger.error("API返回空数据") - return None + 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 - except requests.exceptions.RequestException as e: - logger.error(f"请求出错: {e}") - return None - except json.JSONDecodeError as e: - logger.error(f"JSON解析错误: {e}") - return None \ No newline at end of file + 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'], api_config['params']) + + if data is None: + return None + + if outdata is None or outdata == "": + return data + return data.get(outdata, {}) + + + + + + + + + +# 保持原有函数接口的兼容性（可选） +async def api_data_post(api_url, json_data=None, outdata=None): + """兼容原有函数的POST请求""" + client = APIClient() + return await client.post(api_url, json_data, outdata) + +async def api_data_get(api_url, params=None, outdata=None): + """兼容原有函数的GET请求""" + client = APIClient() + return await client.get(api_url, params, outdata) \ No newline at end of file diff --git a/main.py b/main.py index a8c3dff..f3c5749 100644 --- a/main.py +++ b/main.py @@ -1,29 +1,45 @@ -import urllib3 +from pathlib import Path from datetime import datetime + from astrbot.api.event import filter, AstrMessageEvent, MessageEventResult -from astrbot.api.star import Context, Star, register +from astrbot.api.star import Context, Star, register, StarTools from astrbot.api import logger from astrbot.api import AstrBotConfig from .core.load_template import load_template from .core.jx3_data import jx3_data_jiaoyihang,jx3_data_wujia -from .core.api_data import api_data_get, api_data_post +from .core.api_data import APIClient,api_data_get, api_data_post from .core.sql_data import sql_data_searchdata,sql_data_select -# 禁用 SSL 警告 -urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) - -@register("jx3api", "fxdyz", "通过接口调用剑网三API接口获取游戏数据", "1.0.0") +@register("astrbot_plugin_jx3api", + "fxdyz", + "通过接口调用剑网三API接口获取游戏数据", + "1.0.0", + "https://github.com/qsc20001102/astrbot_plugin_jx3api.git" +) class Jx3ApiPlugin(Star): def __init__(self, context: Context, config: AstrBotConfig): super().__init__(context) + #获取配置 self.conf = config - print(self.conf) + # 本地数据存储路径 + self.local_data_dir = StarTools.get_data_dir("astrbot_plugin_jx3api") + # api数据文件 + self.api_file = Path(__file__).parent / "api_config.json" + logger.info("jx3api插件初始化完成") async def initialize(self): """可选择实现异步的插件初始化方法，当实例化该插件类之后会自动调用该方法。""" + #创建类实例 + self.api_data = APIClient() + logger.info("jx3api插件创建实例完成") + async def message_str(self, event: AstrMessageEvent): + """可选择实现异步的消息预处理方法，在消息分发到具体命令前会调用该方法。""" + # 预处理消息内容 + message_str = event.message_str.strip() + return message_str.split() @filter.command("剑三日常") async def jx3_richang(self, event: AstrMessageEvent): @@ -32,25 +48,23 @@ class Jx3ApiPlugin(Star): custom_url = "https://www.jx3api.com/data/active/calendar" # 接口参数 params = { - "server": "眉间雪", # 默认服务器 - "num": 0 # 默认当天 + "server": "眉间雪", # 服务器 + "num": 0 # 当天 } - # 获取消息内容 - message_str = event.message_str.strip() - parts = message_str.split() + parts = await self.message_str(event) # 解析消息内容 if len(parts) > 1: - params["server"] = parts[1] # 第二个参数为服务器 + params["server"] = parts[1] if len(parts) > 2: try: - params["num"] = int(parts[2]) # 第三个参数为日期偏移 + params["num"] = int(parts[2]) except ValueError: - params["num"] = 0 # 参数为日期偏移 + params["num"] = 0 # 获取数据 - data = api_data_get(custom_url, params,"data") + data = await self.api_data.request(self.api_file['jx3_richang'], params,"data") if not data: yield event.plain_result("获取获取接口信息失败，请稍后再试")