11

core/APIClient.py

@@ -0,0 +1,150 @@
+# core/request.py
+import aiohttp
+import json
+from aiohttp import ClientTimeout
+from astrbot.api import logger
+
+class APIClient:
+    """
+    API客户端类，支持GET/POST请求，同时兼容JSON和二进制数据
+    """
+    
+    def __init__(self, base_timeout=10, ssl_verify=False):
+        self.base_timeout = base_timeout
+        self.ssl_verify = ssl_verify
+
+    async def _make_request(self, method, url, params_data=None):
+        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:
+                if method.upper() == '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':
+                    headers = {'Content-Type': 'application/json'}
+                    async with session.post(url, json=params_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 Exception as e:
+            logger.error(f"未知错误: {e}")
+            return None
+
+    async def _handle_response(self, response):
+        """
+        处理响应：支持 JSON 和二进制数据
+        """
+        try:
+            logger.debug(f"响应状态: {response.status}")
+            response.raise_for_status()
+            content_type = response.headers.get('Content-Type', '')
+
+            if 'image' in content_type or 'octet-stream' in content_type:
+                # 返回二进制数据
+                data = await response.read()
+                return data
+
+            # 尝试解析为 JSON
+            try:
+                data = await response.json(content_type=None)
+            except Exception:
+                text = await response.text()
+                logger.debug(f"原始文本响应: {text}")
+                try:
+                    data = json.loads(text)
+                except json.JSONDecodeError:
+                    logger.error("无法解析为 JSON 数据")
+                    return None
+
+            logger.debug(f"响应数据: {data}")
+            return self._check_response_data(data)
+        except aiohttp.ClientError as e:
+            logger.error(f"HTTP错误: {e}")
+            return None
+        except Exception as e:
+            logger.error(f"未知错误: {e}")
+            return None
+
+    def _check_response_data(self, data):
+        """
+        检查 API 返回 JSON 数据
+        """
+        if isinstance(data, str):
+            try:
+                data = json.loads(data)
+            except json.JSONDecodeError:
+                logger.error("响应数据是无效的JSON字符串")
+                return None
+
+        if data and isinstance(data, dict) and 'code' in data:
+            if data.get('code') not in [200, "0", 0, 1]:
+                logger.error(f"API返回错误:{data.get('code', '未知状态')} {data.get('msg', '未知错误')}")
+                return None
+        elif not data:
+            logger.error("API返回空数据")
+            return None
+
+        return data
+
+    async def post(self, api_url, params_data=None, outdata=None):
+        data = await self._make_request('POST', api_url, params_data)
+        if data is None:
+            return None
+        if isinstance(data, bytes):
+            return data
+        if not outdata:
+            return data
+        return data.get(outdata, {})
+
+    async def get(self, api_url, params_data=None, outdata=None):
+        data = await self._make_request('GET', api_url, params_data)
+        if data is None:
+            return None
+        if isinstance(data, bytes):
+            return data
+        if not outdata:
+            return data
+        return data.get(outdata, {})
+
+    async def all_pages(self, http, api_url, params_data=None, outdata: str = "", listdata: str = "list", max_pages: int = 10):
+        all_data = []
+        current_page = 1
+        while True:
+            params = params_data.copy() if params_data else {}
+            params["page"] = str(current_page)
+
+            if http.upper() == "POST":
+                data = await self.post(api_url, params, outdata)
+            else:
+                data = await self.get(api_url, params, outdata)
+
+            if not data or isinstance(data, bytes):
+                # 二进制数据或者空数据，不分页
+                break
+
+            if not data.get(listdata):
+                break
+
+            all_data.extend(data[listdata])
+
+            if max_pages and current_page >= max_pages:
+                break
+
+            current_page += 1
+            logger.info(f"已获取第 {current_page} 页数据")
+        return all_data
+
+# 保持原接口兼容
+async def api_data_post(api_url, params_data=None, outdata=None):
+    client = APIClient()
+    return await client.post(api_url, params_data, outdata)
+
+async def api_data_get(api_url, params_data=None, outdata=None):
+    client = APIClient()
+    return await client.get(api_url, params_data, outdata)

core/AsyncMySQL.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:
+            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:
+            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:
+            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:
+            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:
+            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()))

core/JX3Function.py

@@ -0,0 +1,536 @@
+from datetime import datetime
+
+from astrbot.api import logger
+
+from .APIClient import APIClient
+from .AsyncMySQL import AsyncMySQL
+from .function_basic import load_template,extract_field,flatten_field,extract_fields,gold_to_string,plot_line_chart_base64
+
+class JX3Function:
+    def __init__(self, api_config,db: AsyncMySQL ):
+        self.__api = APIClient()
+        self.__db = db
+        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"宠物福缘：{data.get('luck')[0]},{data.get('luck')[1]},{data.get('luck')[2]}\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 kaifu(self,server: str = "梦江南"):
+        """
+        开服状态查询
+        """
+        return_data = {
+            "code": 0,
+            "msg": "功能函数未执行",
+            "data": {}
+        }
+        #在配置文件中获取接口配置
+        api_config = self.__api_config["jx3_kaifu"]
+        #更新参数
+        api_config["params"]["server"] = server
+        # 获取数据
+        data = await self.__api.get(api_config["url"],api_config["params"],"data")         
+        if not data:
+            return_data["msg"] = "获取接口信息失败"
+            return  return_data        
+        # 处理返回数据
+        try:
+            status = data.get("status")
+            status_time = datetime.fromtimestamp(data.get("time")).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"处理数据时出错: {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[: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"处理数据时出错: {e}")
+            return_data["msg"] = "处理接口返回信息时出错"
+        return return_data
+    
+
+    async def jinjia(self, server: str = "眉间雪"):
+        """
+        区服金价
+        """
+        return_data = {
+            "code": 0,
+            "msg": "功能函数未执行",
+            "data": {}
+        }
+        #在配置文件中获取接口配置
+        api_config = self.__api_config["aijx3_jinjia"]
+        #更新参数
+        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
+        chart_data = data
+        data = data[:15]
+        # 加载模板
+        try:
+            return_data["temp"] = load_template("jinjia.html")
+        except FileNotFoundError as e:
+            logger.error(f"加载模板失败: {e}")
+            return_data["msg"] = "系统错误：模板文件不存在"
+            return return_data
+        # 生成折线图
+        try:
+            chart_base64 = plot_line_chart_base64(chart_data, "date", "priceWanbaolou", "万宝楼金价走势",True)
+        except Exception as e:
+            logger.error(f"生成折线图失败: {e}")
+            return_data["msg"] = "系统错误：生成折线图失败"
+            return return_data
+        # 准备模板渲染数据
+        try:
+            
+            return_data["data"] = {
+                "items": data,
+                "server": api_config["params"]["serverName"],
+                "update_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
+                "img": chart_base64
+            } 
+        except Exception as e:
+            logger.error(f"处理数据时出错: {e}")
+            return_data["msg"] = "系统错误：模板渲染数据准备失败"
+            return return_data      
+        return_data["code"] = 200       
+        return return_data
+
+
+    async def qiyu(self, adventureName: str = "阴阳两界", serverName: str = "眉间雪"):
+        """
+        区服奇遇
+        """
+        return_data = {
+            "code": 0,
+            "msg": "功能函数未执行",
+            "data": {}
+        }
+        #在配置文件中获取接口配置
+        api_config = self.__api_config["aijx3_qiyu"]
+        #更新参数
+        api_config["params"]["adventureName"] = adventureName
+        api_config["params"]["serverName"] = serverName
+        # 获取数据
+        data = await self.__api.post(api_config["url"],api_config["params"],"data")
+        if not data:
+            return_data["msg"] = "获取接口信息失败"
+            return  return_data
+        # 格式化时间
+        for item in data:
+            if "time" in item:
+                item["time"] = datetime.fromtimestamp(item["time"]/1000).strftime("%Y-%m-%d %H:%M:%S")
+        # 加载模板
+        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,
+                "server": serverName,
+                "update_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+                ,"qiyuname": adventureName
+            }
+        except Exception as e:
+            logger.error(f"处理数据时出错: {e}")
+            return_data["msg"] = "系统错误：模板渲染数据准备失败"
+            return return_data      
+        return_data["code"] = 200       
+        return return_data
+    
+
+    async def SearchData(self):
+        """
+        外观数据插入数据库
+        """
+        return_data = {
+            "code": 0,
+            "msg": "功能函数未执行",
+            "data": {}
+        }
+        #在配置文件中获取接口配置
+        api_config = self.__api_config["aijx3_SearchData"]
+        # 获取数据
+        data = await self.__api.post(api_config["url"],api_config["params"],"data")
+        if not data:
+            return_data["msg"] = "获取接口信息失败"
+            return  return_data
+        # 提取数据
+        try:
+            extracted_data = flatten_field(data, "dataModels")
+        except FileNotFoundError 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['typeName'],
+                item['name'],
+                item['showName'],
+                item['picUrl'],
+                item['searchId'],
+                item['searchDescType']
+            )
+            for item in extracted_data
+        ]
+        # 插入数据
+        try:
+            await self.__db.truncate_table("searchdata")
+            await self.__db.executemany(sql, values_list)
+        except FileNotFoundError as e:
+            logger.error(f"数据插入失败: {e}")
+            return_data["msg"] = "数据插入数据库失败"
+            return return_data
+        return_data["msg"] = f"成功批量插入 {len(extracted_data)} 条数据！"
+        return_data["code"] = 200       
+        return return_data
+
+
+    async def wujia(self,Name: str = "秃盒"):
+        """
+        物价查询
+        """
+        return_data = {
+            "code": 0,
+            "msg": "功能函数未执行",
+            "data": {}
+        }
+        #在配置文件中获取接口配置
+        api_config = self.__api_config["jx3box_exterior"]
+        #更新参数
+        api_config["params"]["keyword"] = Name
+        # 获取查找信息
+        data = await self.__api.get(api_config["url"],api_config["params"],"data")
+        if data["total"] == 0:
+            try:
+                sql = "SELECT showName, searchId FROM searchdata WHERE name = %s OR showName = %s"
+                sqldata = await self.__db.fetch_one(sql, (Name,Name)) 
+                searchId=sqldata["searchId"]
+                showName=sqldata["showName"]
+                return_data["data"]["searchId"]=searchId
+                return_data["data"]["showName"]=showName
+            except Exception as e:
+                logger.error(f"获取外观信息错误: {e}")
+                return_data["msg"] = "未找到该外观信息"
+                return  return_data
+        else:
+            # 查询魔盒获取外观名称
+            try:  
+                showName=data["list"][0]["name"]
+                return_data["data"]["showName"]=showName
+            except Exception as e:
+                logger.error(f"获取外观名称错误: {e}")
+                return_data["msg"] = "未找到该外观信息"
+                return  return_data
+            # 查询爱剑三获取外观ID
+            try:
+                sql = "SELECT searchId FROM searchdata WHERE showName=%s"
+                sqldata = await self.__db.fetch_one(sql, (return_data["data"]["showName"],))   
+                searchId=sqldata["searchId"]
+                return_data["data"]["searchId"]=searchId
+            except Exception as e:
+                logger.error(f"获取外观ID错误: {e}\n{return_data['data']['showName']}")
+                return_data["msg"] = "未找到该外观信息"
+                return  return_data
+
+        #在配置文件中获取接口配置
+        api_config = self.__api_config["aijx3_GoodsDetail"]
+        #更新参数
+        api_config["params"]["goodsName"] = showName
+        # 获取数据外观详细数据
+        data = await self.__api.post(api_config["url"],api_config["params"],"data")
+        if not data:
+            return_data["msg"] = "获取外观详细数据失败"
+            return  return_data
+        # 提取外观详细数据
+        try:
+            imgs = data.get("imgs", [])
+            return_data["data"]["goodsDesc"]=data.get("goodsDesc", "无描述")
+            return_data["data"]["publishTime"]=data.get("publishTime", "无价格")
+            return_data["data"]["priceNum"]=data.get("priceNum", 0)
+            return_data["data"]["goodsId"]=data.get("goodsId", "无数据")
+            return_data["data"]["imgs"]=imgs[0] if imgs else ""
+            return_data["data"]["goodsAlias"]=data.get("goodsAlias", "无别名")
+        except Exception as e:
+            logger.error(f"提取外观详细数据失败: {e}")
+            return_data["msg"] = "提取外观详细数据失败"
+            return  return_data
+
+        # 查询万宝楼数据（公示和在售）
+        wbl_data = await self.__get_wbl_data(showName)
+        if not wbl_data:
+            return_data["msg"] = "获取万宝楼数据失败"
+            return  return_data
+        if wbl_data:
+            return_data["data"]["wblgs"]=wbl_data["wblgs"]
+            return_data["data"]["wblzs"]=wbl_data["wblzs"]
+            return_data["msg"] = "获取万宝楼数据完成"
+        # 加载模板
+        try:
+            return_data["temp"] = load_template("wujia.html")
+        except FileNotFoundError as e:
+            logger.error(f"加载模板失败: {e}")
+            return_data["msg"] = "系统错误：模板文件不存在"
+            return return_data
+        return_data["code"] = 200   
+        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 = "梦江南"):
+        """
+        区服交易行
+        """
+        return_data = {
+            "code": 0,
+            "msg": "功能函数未执行",
+            "data": {}
+        }
+        #在配置文件中获取接口配置
+        api_config = self.__api_config["jx3box_item"]
+        #更新参数
+        api_config["params"]["keyword"] = Name
+        # 获取多页查找信息
+        data = await self.__api.all_pages("GET",api_config["url"],api_config["params"],"data","data")
+        if not data:
+            logger.error(f"获取多页数据失败")
+            return_data["msg"] = "未找到该物品"
+            return  return_data
+        # 提取指定字段
+        fields = ["IconID", "Name","id"]
+        result = extract_fields(data, fields)
+        if not data:
+            logger.error(f"提取字段失败")
+            return_data["msg"] = "未找到该物品"
+            return  return_data
+        # 提取id列表
+        lists_id = extract_field(result, "id")
+        strlists_id = ",".join(str(x) for x in lists_id)
+        logger.info(f"{strlists_id}")
+        #在配置文件中获取接口配置
+        api_config = self.__api_config["jx3box_itemprice"]
+        #更新参数
+        api_config["params"]["itemIds"] = strlists_id
+        api_config["params"]["server"] = server
+        # 获取数据
+        data = await self.__api.get(api_config["url"],api_config["params"],"data")
+        if not data:
+            return_data["msg"] = "未找到在售物品"
+            return  return_data
+        #提取需要的字段
+        fieldsjyh = ["ItemId", "SampleSize", "LowestPrice", "AvgPrice", "Date"]
+        resultjyh = [{f: v[f] for f in fieldsjyh} for v in 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"
+        # 准备模板渲染数据
+        return_data["data"] = {
+            "items": resultjyh,
+            "server": server,
+            "update_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+        } 
+        # 加载模板
+        try:
+            return_data["temp"] = load_template("jiaoyihang.html")
+        except FileNotFoundError as e:
+            logger.error(f"加载模板失败: {e}")
+            return_data["msg"] = "系统错误：模板文件不存在"
+            return return_data 
+        return_data["code"] = 200
+        return return_data

core/WZRYFunction.py

@@ -0,0 +1,190 @@
+from datetime import datetime
+import base64
+from astrbot.api import logger
+
+from .APIClient import APIClient
+from .AsyncMySQL import AsyncMySQL
+from .function_basic import load_template,extract_field,flatten_field,extract_fields,gold_to_string,plot_line_chart_base64
+
+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"处理数据时出错: {e}")
+            return_data["msg"] = "处理数据库信息时出错"
+            return return_data
+        
+        # 加载模板
+        try:
+            return_data["temp"] = load_template("temp_test.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

core/function_basic.py

@@ -0,0 +1,167 @@
+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 图片字符串，可直接放到 <img src="..."> 中
+    """
+    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}"

core/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)