diff --git a/README.md b/README.md index b7b3df6..bf9ab9d 100644 --- a/README.md +++ b/README.md @@ -1,9 +1,46 @@ -# Gok +# 王者数据查询工具 -AstrBot 插件模板 +## 目录 -A template plugin for AstrBot plugin feature +- [王者荣耀数据查询工具](#-王者荣耀数据查询工具) + - [目录](#-目录) + - [简介](#-简介) + - [更新记录](#-更新记录) + - [功能特点](#-功能特点) + - [插件配置](#-插件配置) + - [使用方式](#-使用方式) -# 支持 +## 简介 -[帮助文档](https://astrbot.app) +主要是自用，查询战绩，在游戏中隐藏的战绩也可以查。 + +插件功能绝大部分都是用AI写的，又进行了调试。 + +插件数据依赖应天API提供接口，需要自行支持创建token密钥才可使用。（每天免费10000次调用） + +## 更新记录 + +version: 1.0.0：大部分重要功能已经实现 + +## 功能特点 + +- **处理接口数据**：将接口返回的数据处理成，玩家能读懂的消息发送 +- **HTML渲染图片**：利用astrbot里面的文转图服务，处理HTML页面渲染成图片 + +## 插件配置 + +**指令前缀** + +开启后必须要在指令功能前面加上下面配置的前缀才能正常触发。关闭后直接只有指令功能即可。 + +**ytapi_token** + +前往https://api.t1qq.com 注册并创建Token（免费） + +## 使用方式 + +在聊天中输入 **王者功能** 即可获取插件所有的功能使用说明。 + +在使用 **王者战绩** 和 **王者资料** 两个功能时后面可以直接输入营地ID进行查询，或者输入提前自定义好的角色来查询。 + +指令 **角色查看**、**角色添加**、**角色修改**、**角色删除**、**角色查询** 就是用来操作角色数据的，给王者营地ID起一个别名，方便自己记忆，也方便查询。 \ No newline at end of file diff --git a/_conf_schema.json b/_conf_schema.json index 544b7b4..da62a0d 100644 --- a/_conf_schema.json +++ b/_conf_schema.json @@ -1,3 +1,26 @@ { - + "prefix": { + "description": "指令前缀", + "type": "object", + "items": { + "enable": { + "description": "是否启用", + "type": "bool", + "default": false, + "hint": "是否启用指令前缀检查功能" + }, + "text": { + "description": "前缀内容", + "type": "string", + "default": "王者荣耀", + "hint": "为了避免和其他插件的指令冲突可以增加指令的前缀，为空默认值为王者。" + } + } + }, + "ytapi_token": { + "description": "应天API Token", + "type": "string", + "default": "", + "hint": "前往https://api.t1qq.com注册并创建Token（免费）。" + } } \ No newline at end of file diff --git a/core/WZRYFunction.py b/core/WZRYFunction.py deleted file mode 100644 index 143e2aa..0000000 --- a/core/WZRYFunction.py +++ /dev/null @@ -1,300 +0,0 @@ -from datetime import datetime -import base64 -from astrbot.api import logger - -from .request import APIClient -from .async_mysql import AsyncMySQL -from .function_basic import load_template,extract_fields - -class WZRYFunction: - def __init__(self, api_config,db: AsyncMySQL ): - self.__api = APIClient() - self.__db = db - self.__api_config = api_config - - async def _SELECT_ID(self,name: str): - """ - 查询ID - """ - sql = "SELECT id FROM wzydid WHERE name = %s or id = %s" - sqlid = await self.__db.fetch_one(sql, (name,name)) - return sqlid["id"] if sqlid else None - - async def all_user(self): - """ - 查询所有用户 - """ - sql = "SELECT id,name FROM wzydid" - sqlid = await self.__db.fetch_all(sql) - return_data = "营地ID\t\t昵称\n" - for m in sqlid: - return_data += f"{m['id']}\t{m['name']}\n" - return return_data - - async def add_user(self,id: str ,name: str): - """ - 添加用户 - """ - self_id = await self._SELECT_ID(id) - if self_id is not None: - return "该用户已存在，无需重复添加" - sql = "INSERT INTO wzydid (id,name) VALUES (%s, %s)" - rowcount = await self.__db.execute(sql, (id,name)) - if rowcount > 0: - return f"id:{id}昵称:{name}\n添加成功" - else: - return f"id:{id}昵称:{name}\n添加失败，请稍后再试" - - async def update_user(self,id: str ,name: str): - """ - 更新用户 - """ - self_id = await self._SELECT_ID(id) - if self_id is None: - return "该用户不存在，请先添加用户" - sql = "UPDATE wzydid SET name = %s WHERE id = %s" - rowcount = await self.__db.execute(sql, (name,id)) - if rowcount > 0: - return f"id:{id}昵称:{name}\n更新成功" - else: - return f"id:{id}昵称:{name}\n更新失败，请稍后再试" - - async def delete_user(self,id: str ): - """ - 删除用户 - """ - self_id = await self._SELECT_ID(id) - if self_id is None: - return "该用户不存在，无需删除" - sql = "DELETE FROM wzydid WHERE id = %s" - rowcount = await self.__db.execute(sql, (id,)) - if rowcount > 0: - return f"id:{id}\n删除成功" - else: - return f"id:{id}\n删除失败，请稍后再试" - - async def zhanji(self,name: str ,option: str): - """ - 战绩查询 - """ - return_data = { - "code": 0, - "msg": "功能函数未执行", - "data": {} - } - #在配置文件中获取接口配置 - api_config = self.__api_config["wzry_zhanji"] - # ID查询 - sql_id = await self._SELECT_ID(name) - if sql_id is None: - return_data["msg"] = "未查询到该用户，请确认输入正确的昵称或营地ID" - return return_data - #更新参数 - api_config["params"]["id"] = sql_id - api_config["params"]["option"] = option - # 需要提取的字段 - fields = ["gametime","killcnt","deadcnt","assistcnt","gameresult","mvpcnt","losemvp","mapName", - "oldMasterMatchScore","newMasterMatchScore","usedTime","winNum","failNum","roleJobName","stars","desc", - "gradeGame","heroIcon","godLikeCnt", "firstBlood","hero1TripleKillCnt","hero1UltraKillCnt","hero1RampageCnt","evaluateUrlV3","mvpUrlV3"] - # 处理返回数据 - try: - # 获取数据 - data = await self.__api.get(api_config["url"],api_config["params"],"data") - if not data: - return_data["msg"] = "获取接口信息失败" - return return_data - # 提取字段 - result = extract_fields(data["list"], fields) - result = result[:25] - # 数据处理 - for m in result: - minutes = m["usedTime"] // 60 - seconds = m["usedTime"] % 60 - m["time_str"] = f"{minutes}:{seconds:02d}" - - return_data["data"] = result - except Exception as e: - logger.error(f"处理数据时出错: {e}") - return_data["msg"] = "处理接口返回信息时出错" - # 加载模板 - try: - return_data["temp"] = load_template("wangzhezhanji.html") - except FileNotFoundError as e: - logger.error(f"加载模板失败: {e}") - return_data["msg"] = "系统错误：模板文件不存在" - return return_data - return_data["code"] = 200 - return return_data - - async def ziliao(self, name: str): - return_data = { - "code": 0, - "msg": "功能函数未执行", - "data": {} - } - # 查询 ID - sql_id = await self._SELECT_ID(name) - - if sql_id is None: - return_data["msg"] = "未查询到该用户，请确认输入正确的昵称或营地ID" - return return_data - api_config = self.__api_config["wzry_ziliao"] - api_config["params"]["id"] = sql_id - - try: - data = await self.__api.get(api_config["url"], api_config["params"]) - # 转 base64 - return_data["data"]["img_base64"] = base64.b64encode(data).decode("utf-8") - return_data["code"] = 200 - - except Exception as e: - logger.error(f"处理数据时出错: {e}") - return_data["msg"] = "处理接口返回信息时出错" - - # 加载模板 - try: - return_data["temp"] = load_template("wzry_zl.html") - except FileNotFoundError: - logger.error(f"加载模板失败") - return_data["msg"] = "系统错误：模板文件不存在" - - return return_data - - async def bilei_all(self): - """ - 查询仇人列表 - """ - return_data = { - "code": 0, - "msg": "功能函数未执行", - "data": {} - } - # 查询所有 - try: - sql = "SELECT * FROM wzrybl" - sqlid = await self.__db.fetch_all(sql) - except Exception as e: - logger.error(f"SQL执行错误: {e}") - return_data["msg"] = "处理数据库信息时出错" - return return_data - - # 加载模板 - try: - return_data["temp"] = load_template("bilei.html") - except FileNotFoundError as e: - logger.error(f"加载模板失败: {e}") - return_data["msg"] = "系统错误：模板文件不存在" - return return_data - return_data["data"]["lists"] = sqlid - return_data["code"] = 200 - return return_data - - async def bilei_add(self,name:str, text:str, user:str): - """ - 仇人 添加 - """ - return_data = { - "code": 0, - "msg": "功能函数未执行", - "data": {} - } - # 获取系统时间 - now = datetime.now().strftime("%Y-%m-%d %H:%M:%S") - # 插入数据 - try: - sql = "INSERT INTO wzrybl (name, text, time, user) VALUES (%s, %s, %s, %s)" - await self.__db.execute(sql, (name, text, now, user)) - except Exception as e: - logger.error(f"SQL执行错误: {e}") - return_data["msg"] = "处理数据库信息时出错" - return return_data - return_data["msg"] = "添加成功\n" - return_data["msg"] += "仇人名称："+name+"\n" - return_data["msg"] += "仇人备注："+text+"\n" - return_data["msg"] += "添加时间："+now+"\n" - return_data["msg"] += "记录人："+user+"\n" - return_data["code"] = 200 - return return_data - - async def bilei_select(self,name:str): - """ - 仇人 查找 - """ - return_data = { - "code": 0, - "msg": "功能函数未执行", - "data": {} - } - # 查询数据 - try: - sql = "SELECT * FROM wzrybl WHERE name LIKE %s" - sqlid = await self.__db.fetch_all(sql, ("%"+name+"%",)) - except Exception as e: - logger.error(f"SQL执行错误: {e}") - return_data["msg"] = "处理数据库信息时出错" - return return_data - # 加载模板 - try: - return_data["temp"] = load_template("bilei.html") - except FileNotFoundError as e: - logger.error(f"加载模板失败: {e}") - return_data["msg"] = "系统错误：模板文件不存在" - return return_data - return_data["data"]["lists"] = sqlid - return_data["code"] = 200 - return return_data - - - async def bilei_update(self, id:int, name:str, text:str, user:str): - """ - 仇人 修改 ID 名称 备注 - """ - return_data = { - "code": 0, - "msg": "功能函数未执行", - "data": {} - } - # 获取系统时间 - now = datetime.now().strftime("%Y-%m-%d %H:%M:%S") - # 查询数据 - try: - sql = "UPDATE wzrybl SET name = %s, text = %s, time = %s, user = %s WHERE id = %s" - rowcount = await self.__db.execute(sql, (name, text, now, user, id)) - if rowcount > 0: - return_data["msg"] = "修改成功\n" - return_data["msg"] += "仇人名称："+name+"\n" - return_data["msg"] += "仇人备注："+text+"\n" - return_data["msg"] += "修改时间："+now+"\n" - return_data["msg"] += "记录人："+user+"\n" - return_data["code"] = 200 - else: - return_data["msg"] = "修改失败，未找到对应记录" - except Exception as e: - logger.error(f"SQL执行错误: {e}") - return_data["msg"] = "处理数据库信息时出错" - return return_data - return return_data - - async def bilei_delete(self, id:int): - """ - 仇人 删除 ID - """ - return_data = { - "code": 0, - "msg": "功能函数未执行", - "data": {} - } - # 查询数据 - try: - sql = "DELETE FROM wzrybl WHERE id = %s" - rowcount = await self.__db.execute(sql, (id,)) - if rowcount > 0: - return_data["msg"] = "删除成功\n" - return_data["code"] = 200 - else: - return_data["msg"] = "删除失败，未找到对应记录" - except Exception as e: - logger.error(f"SQL执行错误: {e}") - return_data["msg"] = "处理数据库信息时出错" - return return_data - return return_data \ No newline at end of file diff --git a/core/__pycache__/__init__.cpython-313.pyc b/core/__pycache__/__init__.cpython-313.pyc deleted file mode 100644 index e008624..0000000 Binary files a/core/__pycache__/__init__.cpython-313.pyc and /dev/null differ diff --git a/core/__pycache__/request.cpython-313.pyc b/core/__pycache__/request.cpython-313.pyc deleted file mode 100644 index 506f62b..0000000 Binary files a/core/__pycache__/request.cpython-313.pyc and /dev/null differ diff --git a/core/async_mysql.py b/core/async_mysql.py deleted file mode 100644 index 75a89f1..0000000 --- a/core/async_mysql.py +++ /dev/null @@ -1,90 +0,0 @@ -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/fun_basic.py b/core/fun_basic.py new file mode 100644 index 0000000..87246c5 --- /dev/null +++ b/core/fun_basic.py @@ -0,0 +1,41 @@ +from pathlib import Path +import aiofiles + +async def load_template(template_name: str) -> str: + """ + 异步加载模板内容（非阻塞） + """ + plugin_dir = Path(__file__).parent.parent + template_path = plugin_dir / "templates" / template_name + + if not template_path.exists(): + raise FileNotFoundError(f"模板文件不存在: {template_path}") + + async with aiofiles.open(template_path, "r", encoding="utf-8") as f: + return await f.read() + + +def extract_fields(data_list, fields): + """ + 从字典列表中提取多个字段 + + Args: + data_list (list[dict]): 包含字典的列表 + fields (list[str]): 要提取的字段名列表 + + Returns: + list[dict]: 只包含指定字段的新字典列表 + """ + result = [] + try: + for item in data_list: + extracted = {field: item.get(field) for field in fields} + result.append(extracted) + except Exception as e: + print(f"提取字段时出错: {e}") + return [] + return result + + + + diff --git a/core/function_basic.py b/core/function_basic.py deleted file mode 100644 index 5886593..0000000 --- a/core/function_basic.py +++ /dev/null @@ -1,167 +0,0 @@ -import os -from pathlib import Path -import matplotlib.pyplot as plt -from matplotlib.font_manager import FontProperties -from io import BytesIO -import base64 - -def load_template(template_name): - """ - 从模板文件加载模板内容 - - Args: - template_name: 模板文件名（不带路径） - - Returns: - str: 模板内容 - """ - # 获取模板文件路径 - plugin_dir = Path(__file__).parent.parent - template_path = plugin_dir / "templates" / template_name - - # 检查文件是否存在 - if not template_path.exists(): - raise FileNotFoundError(f"模板文件不存在: {template_path}") - - # 读取模板内容 - with open(template_path, 'r', encoding='utf-8') as f: - return f.read() - - -def flatten_field(data_list, field_name): - """ - 提取并扁平化指定字段的值 - - 从一个字典列表中，收集指定字段的所有值。 - 如果字段的值是列表，则会被展开（扁平化）加入结果； - 如果字段的值是单个元素，则直接加入结果。 - - Args: - data_list (list[dict]): 数据列表，每个元素是字典 - field_name (str): 要提取的字段名 - - Returns: - list: 扁平化后的字段值列表 - """ - extracted_data = [] - for item in data_list: - if field_name in item and item[field_name]: - # 如果是 list，展开 - if isinstance(item[field_name], list): - extracted_data.extend(item[field_name]) - else: - extracted_data.append(item[field_name]) - return extracted_data - - -def extract_fields(data_list, fields): - """ - 从字典列表中提取多个字段 - - Args: - data_list (list[dict]): 包含字典的列表 - fields (list[str]): 要提取的字段名列表 - - Returns: - list[dict]: 只包含指定字段的新字典列表 - """ - result = [] - try: - for item in data_list: - extracted = {field: item.get(field) for field in fields} - result.append(extracted) - except Exception as e: - print(f"提取字段时出错: {e}") - return [] - return result - - -def extract_field(data_list, field_name): - """ - 从列表中的字典提取指定字段的所有值 - - Args: - data_list (list): 数据列表，每个元素是 dict - field_name (str): 要提取的字段名 - - Returns: - list: 提取出来的字段值列表 - """ - return [item[field_name] for item in data_list if field_name in item] - - -def gold_to_string(gold_amount): - """ - 将金钱数值转换为字符串表示形式 - - Args: - gold_amount (int): 金钱数值，单位为铜币 - - Returns: - str: 格式化后的金钱字符串，例如 "1金2银3铜" - """ - if not gold_amount: - return "无价格" - - parts = [] - started = False # 标记是否已经遇到第一个非零位 - - bricks = gold_amount // 100000000 - gold = (gold_amount % 100000000) // 10000 - silver = (gold_amount % 10000) // 100 - copper = gold_amount % 100 - - for value, unit in [(bricks, "砖"), (gold, "金"), (silver, "银"), (copper, "铜")]: - if value != 0: - started = True - if started: - parts.append(f"{value}{unit}") - - return "".join(parts) - - -def plot_line_chart_base64(data, x_field, y_field, title=None, reverse_x=False): - """ - 根据列表数据绘制折线图，返回 base64 图片字符串。 - - :param data: list[dict] 数据列表 - :param x_field: str X轴字段 - :param y_field: str Y轴字段 - :param title: str 图表标题（可选） - :param reverse_x: bool 是否反转 X 轴方向（默认 False：从左往右；True：从右往左） - :return: str base64 图片字符串，可直接放到 中 - """ - if not data: - raise ValueError("数据列表不能为空") - - # 字体设置（支持中文） - plt.rcParams['font.sans-serif'] = ['SimHei'] - plt.rcParams['axes.unicode_minus'] = False - - # 提取 X / Y 数据 - x_values = [str(item.get(x_field, "")) for item in data] - y_values = [float(item.get(y_field, 0)) for item in data] - - # 创建画布 - plt.figure(figsize=(8, 5)) - plt.plot(x_values, y_values, marker='o', color='#4a90e2', linewidth=2) - - # 反转 X 轴 - if reverse_x: - plt.gca().invert_xaxis() - - # 标题与标签 - if title is None: - title = f"{y_field} 折线图" - - plt.grid(True, linestyle='--', alpha=0.5) - plt.xticks(rotation=30) - plt.tight_layout() - - # 转 Base64 - buffer = BytesIO() - plt.savefig(buffer, format='png', dpi=150) - plt.close() - - img_base64 = base64.b64encode(buffer.getvalue()).decode('utf-8') - return f"data:image/png;base64,{img_base64}" \ No newline at end of file diff --git a/core/gok_commands.py b/core/gok_commands.py new file mode 100644 index 0000000..7e8a04e --- /dev/null +++ b/core/gok_commands.py @@ -0,0 +1,110 @@ +# pyright: reportArgumentType=false +# pyright: reportAttributeAccessIssue=false +# pyright: reportIndexIssue=false +# pyright: reportOptionalMemberAccess=false +import json +import shutil +import pathlib +import inspect +from pathlib import Path +from typing import Union +import asyncio + +from astrbot.api.event import filter, AstrMessageEvent, MessageEventResult, MessageChain +from astrbot.api.star import Context, Star, register, StarTools +from astrbot.api import logger +from astrbot.api import AstrBotConfig +import astrbot.api.message_components as Comp +from astrbot.core.utils.session_waiter import ( + SessionController, + session_waiter, +) + +from .gok_data import GOKServer + + +class GOKCommands(Star): + def __init__(self, gok_data:GOKServer): + self.gokfun = gok_data + + + async def plain_msg(self, event: AstrMessageEvent, action): + """最终将数据整理成文本发送""" + data= await action() + try: + if data["code"] == 200: + await event.send( event.plain_result(data["data"])) + else: + await event.send(event.plain_result(data["msg"])) + except Exception as e: + logger.error(f"功能函数执行错误: {e}") + await event.send(event.plain_result("猪脑过载，请稍后再试")) + + + async def T2I_image_msg(self, event: AstrMessageEvent, action): + """最终将数据渲染成图片发送""" + data = await action() + try: + if data["code"] == 200: + url = await self.html_render(data["temp"], data["data"], options={}) + await event.send(event.image_result(url)) + else: + await event.send(event.plain_result(data["msg"])) + + except Exception as e: + logger.error(f"功能函数执行错误: {e}") + await event.send(event.plain_result("猪脑过载，请稍后再试")) + + + async def image_msg(self, event: AstrMessageEvent, action): + """最终将数据整理成图片发送""" + data = await action() + try: + if data["code"] == 200: + await event.send(event.image_result(data["data"])) + else: + await event.send(event.plain_result(data["msg"])) + + except Exception as e: + logger.error(f"功能函数执行错误: {e}") + await event.send(event.plain_result("猪脑过载，请稍后再试")) + + + async def gok_helps(self, event: AstrMessageEvent): + """王者功能""" + return await self.T2I_image_msg(event, self.gokfun.helps) + + + async def gok_zhanji(self, event: AstrMessageEvent,name: str,option:str = 0): + """王者战绩""" + return await self.T2I_image_msg(event, lambda: self.gokfun.zhanji(name ,option)) + + + async def gok_ziliao(self, event: AstrMessageEvent,name: str): + """王者资料""" + return await self.T2I_image_msg(event, lambda: self.gokfun.ziliao(name)) + + + async def gok_user_all(self, event: AstrMessageEvent): + """角色查看""" + return await self.T2I_image_msg(event, self.gokfun.all) + + + async def gok_user_add(self, event: AstrMessageEvent, gokid: int, name: str): + """角色添加 王者营地ID 名称""" + return await self.plain_msg(event, lambda: self.gokfun.add(gokid,name)) + + + async def gok_user_update(self, event: AstrMessageEvent, gokid: int, name: str): + """角色修改 王者营地ID 名称""" + return await self.plain_msg(event, lambda: self.gokfun.update(gokid,name)) + + + async def gok_user_delete(self, event: AstrMessageEvent, gokid:int): + """角色删除 王者营地ID""" + return await self.plain_msg(event, lambda: self.gokfun.delete(gokid)) + + + async def gok_user_select(self, event: AstrMessageEvent, gokid): + """角色查询 王者营地ID""" + return await self.T2I_image_msg(event, lambda: self.gokfun.select(gokid)) \ No newline at end of file diff --git a/core/gok_data.py b/core/gok_data.py new file mode 100644 index 0000000..618a2a7 --- /dev/null +++ b/core/gok_data.py @@ -0,0 +1,448 @@ +# pyright: reportArgumentType=false +# pyright: reportAttributeAccessIssue=false +# pyright: reportIndexIssue=false +# pyright: reportOptionalMemberAccess=false +# pyright: reportCallIssue=false + +from datetime import datetime +from typing import Dict, Any, Optional, List, Union +import base64 + +from astrbot.api import logger +from astrbot.api import AstrBotConfig + +from .request import APIClient +from .sqlite import AsyncSQLiteDB +from .fun_basic import load_template,extract_fields + +class GOKServer: + def __init__(self, api_config, config:AstrBotConfig, sqlite:AsyncSQLiteDB ): + self._api = APIClient() + # 引用API配置文件 + self._api_config = api_config + # 引用插件配置文件 + self._config = config + # 引用数据库类 + self._sql_db = sqlite + + # 获取配置中的 Token + self.token = self._config.get("ytapi_token", "") + if self.token == "": + logger.info("获取配置token失败，请正确填写token,否则部分功能无法正常使用") + else: + logger.debug(f"获取配置token成功。{self.token}") + + + async def close(self): + """释放底层 APIClient 资源""" + if self._api: + await self._api.close() + self._api = None + + + 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 helps(self) -> Dict[str, Any]: + """功能""" + return_data = self._init_return_data() + + # 加载模板 + try: + return_data["temp"] = await load_template("helps.html") + except FileNotFoundError as e: + logger.error(f"加载模板失败: {e}") + return_data["msg"] = "系统错误：模板文件不存在" + return return_data + + return_data["code"] = 200 + + return return_data + + + async def add(self,gokid: int, name: str) -> Dict[str, Any]: + """角色添加 王者营地ID 角色""" + return_data = self._init_return_data() + + # 添加数据 + try: + await self._sql_db.insert( + "users", + { + "gokid": gokid, + "name": name, + } + ) + + except FileNotFoundError as e: + logger.error(f"添加角色失败: {e}") + return_data["msg"] = "添加角色失败" + return return_data + + return_data["data"] = ( + "角色添加成功\n" + f"王者营地ID：{gokid}\n" + f"角色名称：{name}\n" + ) + + return_data["code"] = 200 + + return return_data + + + async def all(self) -> Dict[str, Any]: + """角色查看""" + return_data = self._init_return_data() + + + # 查询数据 + try: + data = await self._sql_db.select_all("users") + except FileNotFoundError as e: + logger.error(f"查看角色失败: {e}") + return_data["msg"] = "查看角色失败" + return return_data + + if not data: + return_data["msg"] = "未找到角色数据" + return return_data + + + # 加载模板 + try: + return_data["temp"] = await load_template("jueshe.html") + except FileNotFoundError as e: + logger.error(f"加载模板失败: {e}") + return_data["msg"] = "系统错误：模板文件不存在" + return return_data + + # 数据处理 + return_data["data"]["lists"] = data + + return_data["code"] = 200 + + return return_data + + + async def select(self, name) -> Dict[str, Any]: + """角色查询 名称""" + return_data = self._init_return_data() + + # 判断输入是否为整数 + try: + int(name) + if int(name) >=100000000: + like = "gokid LIKE ?" + else: + raise Exception("id数据异常") + except (ValueError, TypeError, Exception): + like = "name LIKE ?" + + # 模糊拼接 + like_name = f"%{name}%" + + # 查询数据 + try: + data = await self._sql_db.select_all( + "users", + like, + (like_name,) + ) + except FileNotFoundError as e: + logger.error(f"查询角色失败: {e}") + return_data["msg"] = "查询角色失败" + return return_data + + if not data: + return_data["msg"] = "未查询到角色数据" + return return_data + + + # 加载模板 + try: + return_data["temp"] = await load_template("jueshe.html") + except FileNotFoundError as e: + logger.error(f"加载模板失败: {e}") + return_data["msg"] = "系统错误：模板文件不存在" + return return_data + + # 数据处理 + return_data["data"]["lists"] = data + + return_data["code"] = 200 + + return return_data + + + async def update(self, gokid:int, name: str) -> Dict[str, Any]: + """角色修改 王者营地ID 角色""" + return_data = self._init_return_data() + + data = await self._sql_db.select_one( + "users", + "gokid=?", + (gokid,) + ) + + if not data: + return_data["msg"] = "没有当前ID" + return return_data + + + # 修改数据 + try: + await self._sql_db.update( + "users", + { + "name": name, + }, + "gokid=?", + (gokid,) + ) + + except FileNotFoundError as e: + logger.error(f"避雷修改失败: {e}") + return_data["msg"] = "避雷修改失败" + return return_data + + return_data["data"] = ( + "角色修改成功\n" + f"王者营地ID：{gokid}\n" + f"角色名称：{name}\n" + ) + + return_data["code"] = 200 + + return return_data + + + async def delete(self, gokid:int) -> Dict[str, Any]: + """角色删除 王者营地ID""" + return_data = self._init_return_data() + + data = await self._sql_db.select_one( + "users", + "gokid=?", + (gokid,) + ) + + if not data: + return_data["msg"] = "没有当前ID" + return return_data + + # 删除 + try: + await self._sql_db.delete( + "users", + "gokid=?", + (gokid,) + ) + + except FileNotFoundError as e: + logger.error(f"角色删除失败: {e}") + return_data["msg"] = "角色删除失败" + return return_data + + return_data["data"] = f"角色删除成功。王者营地ID：{gokid}" + + return_data["code"] = 200 + + return return_data + + + async def get_gokid(self,name: str): + # 判断输入是否为整数 + try: + # 直接返回输入 + int(name) + if int(name) >=100000000: + gokid = name + return gokid + else: + raise Exception("输入不是ID") + except (ValueError, TypeError, Exception): + # 查询数据 + try: + data = await self._sql_db.select_one( + "users", + "name=?", + (name,) + ) + + if not data: + gokid = None + return gokid + + gokid = data['gokid'] + return gokid + except FileNotFoundError as e: + logger.error(f"查询角色失败: {e}") + gokid = None + return gokid + + + async def zhanji(self,name: str ,option: str): + """ + 战绩查询 + """ + return_data = self._init_return_data() + + # 获取配置中的 Token + token = self._config.get("ytapi_token", "") + if token == "": + return_data["msg"] = "系统未配置API访问Token" + return return_data + + # ID查询 + gokid = await self.get_gokid(name) + + if not gokid : + return_data["msg"] = "未查询到该用户，请确认输入正确的角色或营地ID" + return return_data + + #更新参数 + params = {"id": gokid, "option": option, "key": token} + + # 需要提取的字段 + fields = ["gametime","killcnt","deadcnt","assistcnt","gameresult","mvpcnt","losemvp","mapName", + "oldMasterMatchScore","newMasterMatchScore","usedTime","winNum","failNum","roleJobName","stars","desc", + "gradeGame","heroIcon","godLikeCnt", "firstBlood","hero1TripleKillCnt","hero1UltraKillCnt","hero1RampageCnt","evaluateUrlV3","mvpUrlV3"] + + # 获取数据 + data: Optional[List[Dict[str, Any]]] = await self._base_request("gok_zhanji", "GET", params=params) + if not data: + return_data["msg"] = "获取接口信息失败" + return return_data + + # 处理返回数据 + try: + + # 提取字段 + result = extract_fields(data['list'], fields) + result = result[:25] + # 数据处理 + for m in result: + minutes = m["usedTime"] // 60 + seconds = m["usedTime"] % 60 + m["time_str"] = f"{minutes}:{seconds:02d}" + + return_data["data"]["data"] = result + except Exception as e: + logger.error(f"处理数据时出错: {e}") + return_data["msg"] = "处理接口返回信息时出错" + return return_data + + # 加载模板 + try: + return_data["temp"] = await load_template("wangzhezhanji.html") + except FileNotFoundError as e: + logger.error(f"加载模板失败: {e}") + return_data["msg"] = "系统错误：模板文件不存在" + return return_data + + return_data["code"] = 200 + + return return_data + + + async def ziliao(self, name: str): + return_data = self._init_return_data() + # 获取配置中的 Token + token = self._config.get("ytapi_token", "") + if token == "": + return_data["msg"] = "系统未配置API访问Token" + return return_data + + # ID查询 + gokid = await self.get_gokid(name) + + if not gokid : + return_data["msg"] = "未查询到该用户，请确认输入正确的角色或营地ID" + return return_data + + #更新参数 + params = {"id": gokid, "key": token} + + # 获取数据 + data: Optional[List[Dict[str, Any]]] = await self._base_request("gok_ziliao", "GET", params=params, out_key="") + logger.debug(data) + + if not data: + return_data["msg"] = "获取接口信息失败" + return return_data + + try: + # 转 base64 + return_data["data"]["img_base64"] = base64.b64encode(data).decode("utf-8") + except Exception as e: + logger.error(f"处理数据时出错: {e}") + return_data["msg"] = "处理接口返回信息时出错" + return return_data + + # 加载模板 + try: + return_data["temp"] = await load_template("wzry_zl.html") + except FileNotFoundError: + logger.error(f"加载模板失败") + return_data["msg"] = "系统错误：模板文件不存在" + return return_data + + return_data["code"] = 200 + + return return_data + + diff --git a/core/request.py b/core/request.py index 7b43000..e204057 100644 --- a/core/request.py +++ b/core/request.py @@ -1,9 +1,10 @@ # core/request.py import json import aiohttp +import asyncio from typing import Optional, Dict, Any, Union, List - from aiohttp import ClientTimeout, ClientSession + from astrbot.api import logger class APIClient: @@ -32,6 +33,7 @@ class APIClient: """关闭 Session""" if self._session and not self._session.closed: await self._session.close() + self._session = None async def __aenter__(self): await self.get_session() @@ -74,23 +76,20 @@ class APIClient: """处理响应：自动识别二进制或JSON""" try: logger.debug(f"响应状态: {response.status}") - response.raise_for_status() # 如果是 4xx/5xx 直接抛出异常 + response.raise_for_status() 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) + loop = asyncio.get_running_loop() + data = await loop.run_in_executor(None, json.loads, text) except json.JSONDecodeError: logger.error(f"无法解析响应为 JSON。原始内容: {text[:100]}...") return None diff --git a/core/sqlite.py b/core/sqlite.py new file mode 100644 index 0000000..8f342ff --- /dev/null +++ b/core/sqlite.py @@ -0,0 +1,71 @@ +# pyright: reportOptionalMemberAccess=false + +import aiosqlite +from typing import Any, Dict, List, Optional, Tuple + + +class AsyncSQLiteDB: + def __init__(self, db_path: str = "data.db"): + self.db_path = db_path + self.conn: Optional[aiosqlite.Connection] = None + + # ====================== + # 生命周期 + # ====================== + + async def connect(self): + 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() + + # ====================== + # 基础执行 + # ====================== + + async def execute(self, sql: str, params: Tuple = ()): + async with self.conn.execute(sql, params): + await self.conn.commit() + + async def fetch_one(self, sql: str, params: Tuple = ()) -> Optional[Dict[str, Any]]: + async with self.conn.execute(sql, params) as cursor: + row = await cursor.fetchone() + return dict(row) if row else None + + async def fetch_all(self, sql: str, params: Tuple = ()) -> List[Dict[str, Any]]: + async with self.conn.execute(sql, params) as cursor: + rows = await cursor.fetchall() + return [dict(r) for r in rows] + + # ====================== + # CRUD + # ====================== + + async def insert(self, table: str, data: Dict[str, Any]): + keys = ", ".join(data.keys()) + placeholders = ", ".join(["?"] * len(data)) + sql = f"INSERT INTO {table} ({keys}) VALUES ({placeholders})" + await self.execute(sql, tuple(data.values())) + + async def update(self, table: str, data: Dict[str, Any], where: str, params: Tuple): + set_clause = ", ".join([f"{k}=?" for k in data.keys()]) + sql = f"UPDATE {table} SET {set_clause} WHERE {where}" + await self.execute(sql, tuple(data.values()) + params) + + async def delete(self, table: str, where: str, params: Tuple): + sql = f"DELETE FROM {table} WHERE {where}" + await self.execute(sql, params) + + async def select_one(self, table: str, where: str = "", params: Tuple = ()): + sql = f"SELECT * FROM {table}" + if where: + sql += f" WHERE {where}" + return await self.fetch_one(sql, params) + + async def select_all(self, table: str, where: str = "", params: Tuple = ()): + sql = f"SELECT * FROM {table}" + if where: + sql += f" WHERE {where}" + return await self.fetch_all(sql, params) \ No newline at end of file diff --git a/core/test.py b/core/test.py deleted file mode 100644 index ab2efbd..0000000 --- a/core/test.py +++ /dev/null @@ -1,14 +0,0 @@ -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 diff --git a/api_config.json b/data/api_config.json similarity index 64% rename from api_config.json rename to data/api_config.json index 78917f8..0c39dbe 100644 --- a/api_config.json +++ b/data/api_config.json @@ -1,21 +1,21 @@ { - "wzry_zhanji":{ + "gok_zhanji":{ "url":"https://api.t1qq.com/api/tool/wzrr/morebattle", "method":"GET", "description":"王者荣耀战绩查询", "params":{ - "key": "vBpEzoiC9z5A9c9Nn83IhLn6M9", - "id": "489048724", + "key": "", + "id": "", "option": "1" } }, - "wzry_ziliao":{ + "gok_ziliao":{ "url":"https://api.t1qq.com/api/tool/wzrr/ydtp", "method":"GET", "description":"王者荣耀资料查询", "params":{ - "key": "vBpEzoiC9z5A9c9Nn83IhLn6M9", - "id": "489048724" + "key": "", + "id": "" } } } \ No newline at end of file diff --git a/logo.png b/logo.png new file mode 100644 index 0000000..b7c3cb2 Binary files /dev/null and b/logo.png differ diff --git a/main.py b/main.py index 8202b43..e02f5f6 100644 --- a/main.py +++ b/main.py @@ -1,218 +1,195 @@ +# pyright: reportOptionalMemberAccess=false +# pyright: reportCallIssue=false +# pyright: reportArgumentType=false + import json -import asyncio +import inspect from pathlib import Path -from datetime import datetime + from astrbot.api.event import filter, AstrMessageEvent, MessageEventResult, MessageChain from astrbot.api.star import Context, Star, register, StarTools from astrbot.api import logger from astrbot.api import AstrBotConfig - -from .core.async_mysql import AsyncMySQL -from .core.WZRYFunction import WZRYFunction - +from .core.sqlite import AsyncSQLiteDB +from .core.gok_data import GOKServer +from .core.gok_commands import GOKCommands @register("astrbot_plugin_gok", - "飞翔大野猪", - "通过接口接口获取王者荣耀游戏数据", + "fxdyz", + "通过接口获取王者荣耀游戏数据", "1.0.0", - "https://github.com/qsc20001102/astrbot_plugin_gok" + "https://github.com/qsc20001102/astrbot_plugin_gok.git" ) class GokApiPlugin(Star): def __init__(self, context: Context, config: AstrBotConfig): super().__init__(context) #获取配置 self.conf = config + # 本地数据存储路径 self.local_data_dir = StarTools.get_data_dir("astrbot_plugin_gok") - # api数据文件 - self.api_file_path = Path(__file__).parent / "api_config.json" - # 读取文件内容 + + # SQLite本地路径 + self.sqlite_path = Path(self.local_data_dir) /"sqlite.db" + logger.info(f"SQLite数据文件路径：{self.sqlite_path}") + + # 读取API配置文件 + self.api_file_path = Path(__file__).parent / "data" / "api_config.json" with open(self.api_file_path, 'r', encoding='utf-8') as f: self.api_config = json.load(f) + # 初始化数据 - logger.info("王者荣耀插件初始化完成") + # 声明指令集 + self.command_map = {} + # 指令前缀功能 + self.prefix_en = self.conf.get("prefix").get("enable") + self.prefix_text = self.conf.get("prefix").get("text") + if not self.prefix_text: + self.prefix_text = "王者" + if self.prefix_en: + logger.info(f"已启用指令前缀功能，前缀为：{self.prefix_text}") + else: + logger.info(f"未启用指令前缀功能。") + + logger.info("GOK 插件初始化完成") async def initialize(self): """可选择实现异步的插件初始化方法，当实例化该插件类之后会自动调用该方法。""" - # 数据库配置 - db_config = { - 'host': '38.12.28.24', - 'port': 3306, - 'user': 'asrtbot', - 'password': 'qsc102341', - 'db': 'asrtbot', - 'charset': 'utf8mb4', - 'autocommit': True - } - #创建类实例 - self.db = AsyncMySQL(db_config) - self.wzry = WZRYFunction(self.api_config,self.db) - # 周期函数调用 - - - logger.info("王者荣耀插件创建实例完成") - - - @filter.command_group("王者") - def wz(self): - pass - - - @wz.command("列表") - async def wz_yingdilist(self, event: AstrMessageEvent): - """王者 用户列表""" try: - data = await self.wzry.all_user() - logger.info(f"王者荣耀营地列表查询结果{data}") - yield event.plain_result(data) + # sqlite 实例化 + self.sql_db = AsyncSQLiteDB(self.sqlite_path) + await self.sql_db.connect() + await self.sql_db.execute(""" + CREATE TABLE IF NOT EXISTS users( + gokid INTEGER, + name TEXT + ) + """) + # 王者功能 实例化 + self.gokfun = GOKServer(self.api_config, self.conf, self.sql_db) + # 发送消息实例化 + self.gokmag = GOKCommands(self.gokfun) except Exception as e: - logger.error(f"功能函数执行错误: {e}") - yield event.plain_result("猪脑过载，请稍后再试") + logger.error(f"功能模块初始化失败: {e}") + raise - @wz.command("添加") - async def wz_adduesr(self, event: AstrMessageEvent,id: str,name: str): - """王者 添加用户""" - try: - data = await self.wzry.add_user(id,name) - yield event.plain_result(data) - except Exception as e: - logger.error(f"功能函数执行错误: {e}") - yield event.plain_result("猪脑过载，请稍后再试") - - @wz.command("修改") - async def wz_updateuser(self, event: AstrMessageEvent,id: str,name: str): - """王者 修改用户""" - try: - data = await self.wzry.update_user(id,name) - yield event.plain_result(data) - except Exception as e: - logger.error(f"功能函数执行错误: {e}") - yield event.plain_result("猪脑过载，请稍后再试") - - @wz.command("删除") - async def wz_deleteuser(self, event: AstrMessageEvent,id: str): - """王者 删除用户""" - try: - data = await self.wzry.delete_user(id) - yield event.plain_result(data) - except Exception as e: - logger.error(f"功能函数执行错误: {e}") - yield event.plain_result("猪脑过载，请稍后再试") - - @wz.command("战绩") - async def wz_zhanji(self, event: AstrMessageEvent,name: str = "飞翔大野猪",option: str = "0"): - """王者 战绩 营地ID 对局类型""" - - try: - data = await self.wzry.zhanji(name,option) - # logger.info(f"王者荣耀战绩查询结果{data}") - #prompt = f"{data['data']}\n" - #prompt += f"以上数据是获取到的王者荣耀战绩信息，请根据这些数据生成一两句话的战绩总结，突出玩家的优势和特点，语言风格轻松幽默，适合在游戏群内分享。" - if data["code"] == 200: - url = await self.html_render(data["temp"],{"data":data["data"]}, options={}) - yield event.image_result(url) - else: - yield event.plain_result(data["msg"]) - return - #provider_id = await self.context.get_current_chat_provider_id(umo=event.unified_msg_origin) - #llm_resp = await self.context.llm_generate(chat_provider_id=provider_id, prompt=prompt,) - #out = llm_resp.completion_text - #yield event.plain_result(f"{out}") - except Exception as e: - logger.error(f"功能函数执行错误: {e}") - yield event.plain_result("猪脑过载，请稍后再试") - - - @wz.command("资料") - async def wz_ziliao(self, event: AstrMessageEvent,name: str = "489048724"): - """王者 战绩 营地ID 对局类型""" - try: - data = await self.wzry.ziliao(name) - logger.info(f"输出结果{data}") - if data["code"] == 200: - url = await self.html_render(data["temp"], data["data"], options={}) - yield event.image_result(url) - else: - yield event.plain_result(data["msg"]) - return - except Exception as e: - logger.error(f"功能函数执行错误: {e}") - yield event.plain_result("猪脑过载，请稍后再试") - - @filter.command_group("仇人") - def bilei(self): - pass - - @bilei.command("列表") - async def bilei_list(self, event: AstrMessageEvent,): - """仇人 列表""" - try: - data = await self.wzry.bilei_all() - if data["code"] == 200: - url = await self.html_render(data["temp"], data["data"], options={}) - yield event.image_result(url) - else: - yield event.plain_result(data["msg"]) - return - except Exception as e: - logger.error(f"功能函数执行错误: {e}") - yield event.plain_result("猪脑过载，请稍后再试") - - - @bilei.command("添加") - async def bilei_add(self, event: AstrMessageEvent,name: str, text: str): - """仇人 添加 名称 备注""" - try: - data = await self.wzry.bilei_add(name, text, event.get_sender_name()) - yield event.plain_result(data["msg"]) - except Exception as e: - logger.error(f"功能函数执行错误: {e}") - yield event.plain_result("猪脑过载，请稍后再试") - - - @bilei.command("查找") - async def bilei_select(self, event: AstrMessageEvent,name: str): - """仇人 查找 名称（模糊查找）""" - try: - data = await self.wzry.bilei_select(name) - if data["code"] == 200: - url = await self.html_render(data["temp"], data["data"], options={}) - yield event.image_result(url) - else: - yield event.plain_result(data["msg"]) - return - except Exception as e: - logger.error(f"功能函数执行错误: {e}") - yield event.plain_result("猪脑过载，请稍后再试") - - - @bilei.command("修改") - async def bilei_update(self, event: AstrMessageEvent,id:int, name: str, text: str): - """仇人 修改 ID 名称 备注""" - try: - data = await self.wzry.bilei_update(id, name, text, event.get_sender_name()) - yield event.plain_result(data["msg"]) - except Exception as e: - logger.error(f"功能函数执行错误: {e}") - yield event.plain_result("猪脑过载，请稍后再试") - - @bilei.command("删除") - async def bilei_delete(self, event: AstrMessageEvent,id:int): - """仇人 删除 ID """ - try: - data = await self.wzry.bilei_delete(id) - yield event.plain_result(data["msg"]) - except Exception as e: - logger.error(f"功能函数执行错误: {e}") - yield event.plain_result("猪脑过载，请稍后再试") + # 指令集 + self.ini_command_map() + logger.info("GOK 异步插件初始化完成") async def terminate(self): """可选择实现异步的插件销毁方法，当插件被卸载/停用时会调用。""" - await self.db.close_pool() - # 后台进程销毁 - logger.info("王者荣耀插件已卸载/停用") \ No newline at end of file + if self.gokfun: + await self.gokfun.close() + self.gokfun = None + + if self.sql_db: + await self.sql_db.close() + self.sql_db = None + + logger.info("GOK 插件已卸载/停用") + + + def parse_message(self, text: str) -> list[str] | None: + """消息解析""" + text = text.strip() + if not text: + return None + + # 前缀模式 + if self.prefix_en: + prefix = self.prefix_text + if text.startswith(prefix): + text = text[len(prefix):].strip() + else: + # 非前缀消息，直接忽略 + return None + + return text.split() + + + async def _call_with_auto_args(self, handler, event: AstrMessageEvent, args: list[str]): + """指令执行函数""" + sig = inspect.signature(handler) + params = list(sig.parameters.values()) + + call_args = [] + arg_index = 0 + + for p in params: + if p.name == "self": + continue + + if p.name == "event": + call_args.append(event) + continue + + if arg_index < len(args): + raw = args[arg_index] + arg_index += 1 + try: + if p.annotation is int: + call_args.append(int(raw)) + elif p.annotation is float: + call_args.append(float(raw)) + else: + call_args.append(raw) + except Exception: + call_args.append(p.default) + else: + if p.default is not inspect._empty: + call_args.append(p.default) + else: + raise ValueError(f"缺少参数: {p.name}") + + # 只允许 coroutine + return await handler(*call_args) + + + @filter.event_message_type(filter.EventMessageType.ALL) + async def on_all_message(self, event: AstrMessageEvent): + """解析所有消息""" + if not self.command_map: + logger.debug("插件尚未初始化完成，忽略消息") + return + parts = self.parse_message(event.message_str) + if not parts: + logger.debug("未触发指令，忽略消息") + return + + cmd, *args = parts + handler = self.command_map.get(cmd) + if not handler: + logger.debug("指令函数为空，忽略消息") + return + + try: + event.stop_event() + ret = await self._call_with_auto_args(handler, event, args) + if ret is not None: + yield ret + except Exception as e: + logger.exception(f"指令执行失败: {cmd}, error={e}") + yield event.plain_result("参数错误或执行失败") + + + def ini_command_map(self): + """初始化指令集""" + self.command_map = { + "王者功能": self.gokmag.gok_helps, + "王者战绩": self.gokmag.gok_zhanji, + "王者资料": self.gokmag.gok_ziliao, + "角色查看": self.gokmag.gok_user_all, + "角色添加": self.gokmag.gok_user_add, + "角色修改": self.gokmag.gok_user_update, + "角色删除": self.gokmag.gok_user_delete, + "角色查询": self.gokmag.gok_user_select + } + + + diff --git a/metadata.yaml b/metadata.yaml index bd7f119..9bd270e 100644 --- a/metadata.yaml +++ b/metadata.yaml @@ -1,6 +1,6 @@ -name: astrbot_plugin_gok # 这是你的插件的唯一识别名。 -display_name: 王者荣耀数据查询工具 # 插件的显示名称 -desc: 王者荣耀的战绩、资料等内容的查询 # 插件简短描述 -version: v1.0.0 # 插件版本号。格式：v1.1.1 或者 v1.1 -author: 飞翔大野猪 # 作者 -repo: https://github.com/qsc20001102/astrbot_plugin_gok # 插件的仓库地址 +name: astrbot_plugin_gok +display_name: 王者荣耀游戏数据查询工具 +desc: 王者荣耀的战绩、资料等内容的查询 +version: v1.0.0 +author: 飞翔大野猪 +repo: https://github.com/qsc20001102/astrbot_plugin_gok diff --git a/requirements.txt b/requirements.txt index 3132c6a..aee3d98 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,2 +1 @@ -aiomysql -matplotlib \ No newline at end of file +aiosqlite \ No newline at end of file diff --git a/templates/helps.html b/templates/helps.html new file mode 100644 index 0000000..ff0d7ed --- /dev/null +++ b/templates/helps.html @@ -0,0 +1,157 @@ + + + + + 王者插件功能 + + + + +
+ +
+

王者插件功能

+

astrbot_plugin_gok 指令说明与使用示例

+
+ +
+
基础功能
+
+
王者功能
王者功能
+
王者战绩
王者战绩 角色/营地ID
+
王者资料
王者资料 角色/营地ID
+
角色查看
角色查看
+
角色添加
角色添加 营地ID 角色
+
角色修改
角色修改 营地ID 角色
+
角色删除
角色删除 营地ID
+
角色查询
角色查询 营地ID/角色
+
+
+ +
+ 使用提示：
+ 1. 指令里面的角色指的是自定义的查询名称
+ 2. 战绩和资料的指令必须 token 可以使用
+
+ +
+ + \ No newline at end of file diff --git a/templates/bilei.html b/templates/jueshe.html similarity index 71% rename from templates/bilei.html rename to templates/jueshe.html index 6cbe1ed..0e2e448 100644 --- a/templates/bilei.html +++ b/templates/jueshe.html @@ -2,7 +2,7 @@ -王者荣耀 — 仇人列表 +角色查看 - - - -
-

仇人列表

- - - - - - - - - - - - - - {% for m in lists %} - - - - - - - - {% endfor %} - -
ID名称描述时间记录者
{{ m.id }}{{ m.name }}{{ m.text }}{{ m.time }}{{ m.user }}
- -
- - - diff --git a/test.py b/test.py deleted file mode 100644 index ab2efbd..0000000 --- a/test.py +++ /dev/null @@ -1,14 +0,0 @@ -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