From 9f38a7627be309a4befb09ea48f1fba128393a77 Mon Sep 17 00:00:00 2001 From: qsc20001102 Date: Fri, 28 Nov 2025 17:56:50 +0800 Subject: [PATCH] 11 --- core/cless_mysql.py | 31 ++++++++++++++++-- core/cless_wzry.py | 80 ++++++++++++++++++++++++++++++++++++++++++--- main.py | 50 +++++++++++++++++++++++++--- 3 files changed, 150 insertions(+), 11 deletions(-) diff --git a/core/cless_mysql.py b/core/cless_mysql.py index 9adeb17..255fae6 100644 --- a/core/cless_mysql.py +++ b/core/cless_mysql.py @@ -35,7 +35,7 @@ class AsyncMySQL: return await cursor.fetchall() async def execute(self, sql: str, params=None): - """执行单条 SQL（insert/update/delete）""" + """执行 SQL（insert/update/delete）""" await self.init_pool() async with self.pool.acquire() as conn: async with conn.cursor() as cursor: @@ -53,7 +53,7 @@ class AsyncMySQL: return cursor.rowcount async def truncate_table(self, table_name: str): - """清空指定表的内容""" + """清空指定表""" await self.init_pool() async with self.pool.acquire() as conn: async with conn.cursor() as cursor: @@ -61,3 +61,30 @@ class AsyncMySQL: 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())) diff --git a/core/cless_wzry.py b/core/cless_wzry.py index 127c537..aaab454 100644 --- a/core/cless_wzry.py +++ b/core/cless_wzry.py @@ -12,7 +12,68 @@ class WZRYFunction: self.__db = db self.__api_config = api_config - async def zhanji(self,id: str ,option: str): + 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): """ 战绩查询 """ @@ -23,9 +84,15 @@ class WZRYFunction: } #在配置文件中获取接口配置 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"] = id + 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"] @@ -37,7 +104,7 @@ class WZRYFunction: return_data["msg"] = "获取接口信息失败" return return_data result = extract_fields(data["list"], fields) - result = result[:15] + result = result[:20] # 数据处理 for m in result: minutes = m["usedTime"] // 60 @@ -77,7 +144,7 @@ class WZRYFunction: return_data["code"] = 200 return return_data - async def ziliao(self,id: str): + async def ziliao(self,name: str): """ 资料查询 """ @@ -86,10 +153,13 @@ class WZRYFunction: "msg": "功能函数未执行", "data": {} } + #id查询 + sql = "SELECT id FROM wzydid WHERE name = %s or id = %s" + sqlid = await self.__db.fetch_one(sql, (name,name)) #在配置文件中获取接口配置 api_config = self.__api_config["wzry_ziliao"] #更新参数 - api_config["params"]["id"] = id + api_config["params"]["id"] = sqlid["id"] # 处理返回数据 try: # 获取数据 diff --git a/main.py b/main.py index f885e25..eefdf8c 100644 --- a/main.py +++ b/main.py @@ -292,11 +292,53 @@ class Jx3ApiPlugin(Star): 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) + except Exception as e: + logger.error(f"功能函数执行错误: {e}") + yield event.plain_result("猪脑过载，请稍后再试") + + @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,ID: str = "489048724",option: str = "1"): + async def wz_zhanji(self, event: AstrMessageEvent,name: str = "飞翔大野猪",option: str = "0"): """王者 战绩 营地ID 对局类型""" try: - data = await self.wzry.zhanji(ID,option) + data = await self.wzry.zhanji(name,option) # logger.info(f"王者荣耀战绩查询结果{data}") if data["code"] == 200: url = await self.html_render(data["temp"],{"data": data["data"]}, options={}) @@ -309,10 +351,10 @@ class Jx3ApiPlugin(Star): yield event.plain_result("猪脑过载，请稍后再试") @wz.command("资料") - async def wz_ziliao(self, event: AstrMessageEvent,ID: str = "489048724"): + async def wz_ziliao(self, event: AstrMessageEvent,name: str = "489048724"): """王者 战绩 营地ID 对局类型""" try: - data = await self.wzry.ziliao(ID) + data = await self.wzry.ziliao(name) # logger.info(f"输出结果{data}") if data["code"] == 200: url = await self.html_render(data["temp"], data["data"], options={})