11

core/async_mysql.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: # 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()))

core/jx3_service.py

@@ -0,0 +1,593 @@
+from datetime import datetime
+from typing import Dict, Any, Optional, List, Union
+
+from astrbot.api import logger
+
+from .request import APIClient
+from .AsyncMySQL import AsyncMySQL
+from .function_basic import load_template,flatten_field,extract_fields,gold_to_string
+
+class JX3Service:
+    def __init__(self, api_config,db: AsyncMySQL ):
+        self._api = APIClient()
+        self._db = db
+        self._api_config = api_config
+
+
+    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 richang(self,server: str, num: int = 0) -> Dict[str, Any]:
+        """日常活动"""
+        return_data = self._init_return_data()
+
+        # 1. 构造请求参数
+        params = {"server": server, "num": num}
+
+        # 2. 调用基础请求
+        data: Optional[Dict[str, Any]] = await self._base_request(
+            "jx3_richang", "GET", params=params
+        )
+
+        if not data:
+            return_data["msg"] = "获取接口信息失败"
+            return return_data
+    
+        # 3. 处理返回数据
+        try:
+            # 格式化字符串，利用字典的 get 方法提供默认值
+            result_msg = (
+                f"{server}\n{data.get('date', '未知日期')}-星期{data.get('week', '未知')}\n"
+                f"大战：{data.get('war', '无')}\n"
+                f"战场：{data.get('battle', '无')}\n"
+                f"阵营：{data.get('orecar', '无')}\n"
+                f"宗门：{data.get('school', '无')}\n"
+                f"驰援：{data.get('rescue', '无')}\n"
+                f"画像：{data.get('draw', '无')}\n"
+            )
+            
+            # 安全地处理列表索引
+            luck = data.get('luck', [None, None, None])
+            luck_msg = f"[宠物福缘]：\n{luck[0] or '无'},{luck[1] or '无'},{luck[2] or '无'}\n"
+            card = data.get('card', [None, None, None])
+            card_msg = f"[家园声望·加倍道具]：\n{card[0] or '无'},{card[1] or '无'},{card[2] or '无'}\n"
+            team = data.get('team', [None, None, None])
+            team_msg = f"[武林通鉴·公共任务]：\n{team[0] or '无'}\n[武林通鉴·团队秘境]：\n{team[2] or '无'}\n"
+
+            return_data["data"] = result_msg + luck_msg + card_msg + team_msg
+            return_data["code"] = 200
+        except Exception as e:
+            logger.error(f"richang 数据处理时出错: {e}")
+            return_data["msg"] = "处理接口返回信息时出错"
+
+        return return_data
+
+
+    async def shapan(self, server: str ) -> Dict[str, Any]:
+        """区服沙盘"""
+        return_data = self._init_return_data()
+        
+        # 1. 构造请求参数
+        params = {"serverName": server}
+        
+        # 2. 调用基础请求
+        data: Optional[Dict[str, Any]] = await self._base_request(
+            "aijx3_shapan", "POST", params=params
+        )
+        
+        if not data:
+            return_data["msg"] = "获取接口信息失败"
+            return return_data
+            
+        # 3. 处理返回数据 (直接提取图片 URL)
+        pic_url = data.get("picUrl")
+        if pic_url:
+            return_data["data"] = pic_url
+            return_data["code"] = 200
+        else:
+            return_data["msg"] = "接口未返回图片URL"
+            
+        return return_data
+    
+    async def kaifu(self, server: str) -> Dict[str, Any]:
+        """开服状态查询"""
+        return_data = self._init_return_data()
+        
+        # 1. 构造请求参数
+        params = {"server": server}
+        
+        # 2. 调用基础请求
+        data: Optional[Dict[str, Union[int, str]]] = await self._base_request(
+            "jx3_kaifu", "GET", params=params
+        )
+        
+        if not data:
+            return_data["msg"] = "获取接口信息失败"
+            return return_data
+            
+        # 3. 处理返回数据
+        try:
+            status = data.get("status", 0)
+            timestamp = data.get("time", 0)
+            
+            status_time = datetime.fromtimestamp(float(timestamp)).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"kaifu 数据处理时出错: {e}")
+            return_data["msg"] = "处理接口返回信息时出错"
+            
+        return return_data
+
+    async def shaohua(self) -> Dict[str, Any]:
+        """骚话"""
+        return_data = self._init_return_data()
+        
+        # 因为没有参数，所以 params=None
+        data: Optional[Dict[str, Any]] = await self._base_request("jx3_shaohua", "GET") 
+        
+        if not data:
+            return_data["msg"] = "获取接口信息失败"
+            return return_data
+            
+        text = data.get("text")
+        if text:
+            return_data["data"] = text
+            return_data["code"] = 200
+        else:
+            return_data["msg"] = "接口未返回文本"
+            
+        return return_data
+
+
+    async def jigai(self) -> Dict[str, Any]:
+        """技改记录"""
+        return_data = self._init_return_data()
+        
+        # 提取字段可能返回列表
+        data: Optional[List[Dict[str, Any]]] = await self._base_request("jx3_jigai", "GET")
+        
+        if not data or not isinstance(data, list):
+            return_data["msg"] = "获取接口信息失败或数据格式错误"
+            return return_data
+        
+        try:
+            result_msg = "剑网三最近技改\n"
+            # 仅展示前1条，避免消息过长
+            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"jigai 数据处理时出错: {e}")
+            return_data["msg"] = "处理接口返回信息时出错"
+            
+        return return_data
+    
+
+    async def jinjia(self, server: str) -> Dict[str, Any]:
+        """区服金价"""
+        return_data = self._init_return_data()
+        
+        params = {"serverName": server}
+        data_list: Optional[List[Dict[str, Any]]] = await self._base_request("aijx3_jinjia", "POST", params=params)
+        
+        if not data_list or not isinstance(data_list, list):
+            return_data["msg"] = "获取接口信息失败或数据格式错误"
+            return return_data
+            
+        # 安全处理列表切片
+        display_data = data_list[:15]
+        
+        # 加载模板
+        try:
+            return_data["temp"] = load_template("jinjia.html")
+        except FileNotFoundError as e:
+            logger.error(f"加载模板失败: {e}")
+            return_data["msg"] = "系统错误：模板文件不存在"
+            return return_data
+            
+        # 准备模板渲染数据
+        try:
+            return_data["data"] = {
+                "items": display_data,
+                "server": server,
+                "update_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
+            } 
+            return_data["code"] = 200
+        except Exception as e:
+            logger.error(f"jinjia 模板数据准备失败: {e}")
+            return_data["msg"] = "系统错误：模板渲染数据准备失败"
+            
+        return return_data
+
+
+    async def qiyu(self, adventureName: str = "阴阳两界", serverName: str = "眉间雪") -> Dict[str, Any]:
+        """区服奇遇"""
+        return_data = self._init_return_data()
+        
+        params = {"adventureName": adventureName, "serverName": serverName}
+        data_list: Optional[List[Dict[str, Any]]] = await self._base_request("aijx3_qiyu", "POST", params=params)
+        
+        if not data_list or not isinstance(data_list, list):
+            return_data["msg"] = "获取接口信息失败或数据格式错误"
+            return return_data
+            
+        # 格式化时间
+        for item in data_list:
+            timestamp = item.get("time")
+            if timestamp and isinstance(timestamp, (int, float)):
+                # 修复时间戳：原代码显示这里是毫秒级，除以 1000
+                item["time"] = datetime.fromtimestamp(timestamp / 1000).strftime("%Y-%m-%d %H:%M:%S")
+            else:
+                item["time"] = "未知时间" # 确保即使 time 字段缺失也不会报错
+                
+        # 加载模板
+        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_list,
+                "server": serverName,
+                "update_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
+                "qiyuname": adventureName
+            }
+            return_data["code"] = 200
+        except Exception as e:
+            logger.error(f"qiyu 模板数据准备失败: {e}")
+            return_data["msg"] = "系统错误：模板渲染数据准备失败"
+            
+        return return_data
+    
+
+    async def SearchData(self) -> Dict[str, Any]:
+        """外观数据插入数据库"""
+        return_data = self._init_return_data()
+        
+        # 获取数据
+        data: Optional[List[Dict[str, Any]]] = await self._base_request("aijx3_SearchData", "POST")
+        
+        if not data:
+            return_data["msg"] = "获取接口信息失败"
+            return return_data
+            
+        # 提取数据
+        try:
+            extracted_data = flatten_field(data, "dataModels")
+        except Exception 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.get('typeName'),
+                item.get('name'),
+                item.get('showName'),
+                item.get('picUrl'),
+                item.get('searchId'),
+                item.get('searchDescType')
+            )
+            for item in extracted_data 
+        ]
+        
+        # 插入数据
+        try:
+            await self._db.truncate_table("searchdata")
+            await self._db.executemany(sql, values_list)
+        except Exception as e:
+            logger.error(f"数据插入失败: {e}")
+            return_data["msg"] = "数据插入数据库失败"
+            return return_data
+            
+        return_data["msg"] = f"成功批量插入 {len(values_list)} 条数据！"
+        return_data["code"] = 200
+        return return_data
+
+
+    async def wujia(self, Name: str) -> Dict[str, Any]:
+        """物价查询"""
+        return_data = self._init_return_data()
+        
+        # 1. 查询魔盒获取外观名称 (jx3box_exterior)
+        params_search = {"keyword": Name}
+        search_data: Optional[Dict[str, Any]] = await self._base_request("jx3box_exterior", "GET", params=params_search)
+        
+        showName = ""
+        searchId = None
+        
+        # 2. 确定外观名称和 ID
+        if search_data and search_data.get("total", 0) > 0 and search_data.get("list"):
+            # 从 API 获取名称
+            showName = search_data["list"][0].get("name", "")
+            
+            # 进一步从 DB 查找 ID
+            try:
+                sql = "SELECT searchId FROM searchdata WHERE showName=%s"
+                sqldata = await self._db.fetch_one(sql, (showName,)) 
+                searchId = sqldata["searchId"] if sqldata else None
+            except Exception as e:
+                logger.warning(f"从 DB 获取外观ID错误: {e}") # 警告而非错误，因为可能数据库同步失败
+                
+        elif search_data and search_data.get("total", 0) == 0:
+            # 从 DB 查找名称和 ID
+            try:
+                sql = "SELECT showName, searchId FROM searchdata WHERE name = %s OR showName = %s"
+                sqldata = await self._db.fetch_one(sql, (Name, Name)) 
+                if sqldata:
+                    showName = sqldata["showName"]
+                    searchId = sqldata["searchId"]
+                else:
+                    raise ValueError("DB did not find data.")
+            except Exception as e:
+                logger.error(f"获取外观信息错误: {e}")
+                return_data["msg"] = "未找到该外观信息"
+                return return_data
+        
+        if not showName or not searchId:
+            return_data["msg"] = "无法确定外观名称或ID"
+            return return_data
+
+        return_data["data"]["showName"] = showName
+        return_data["data"]["searchId"] = searchId
+
+        # 3. 获取外观详细数据 (aijx3_GoodsDetail)
+        params_detail = {"goodsName": showName}
+        detail_data: Optional[Dict[str, Any]] = await self._base_request("aijx3_GoodsDetail", "POST", params=params_detail)
+        
+        if not detail_data:
+            return_data["msg"] = "获取外观详细数据失败"
+            return return_data
+            
+        # 提取外观详细数据
+        try:
+            imgs = detail_data.get("imgs", [])
+            return_data["data"].update({
+                "goodsDesc": detail_data.get("goodsDesc", "无描述"),
+                "publishTime": detail_data.get("publishTime", "无价格"),
+                "priceNum": detail_data.get("priceNum", 0),
+                "goodsId": detail_data.get("goodsId", "无数据"),
+                "imgs": imgs[0] if imgs and isinstance(imgs, list) else "",
+                "goodsAlias": detail_data.get("goodsAlias", "无别名"),
+            })
+        except Exception as e:
+            logger.error(f"提取外观详细数据失败: {e}")
+            return_data["msg"] = "提取外观详细数据失败"
+            return return_data
+
+        # 4. 查询万宝楼数据（公示和在售）
+        wbl_data = await self._get_wbl_data(showName)
+        if wbl_data:
+            return_data["data"]["wblgs"] = wbl_data["wblgs"]
+            return_data["data"]["wblzs"] = wbl_data["wblzs"]
+            return_data["msg"] = "获取万宝楼数据完成"
+        else:
+            logger.warning("万宝楼数据获取失败")
+            
+        # 5. 加载模板
+        try:
+            return_data["temp"] = load_template("wujia.html")
+            return_data["code"] = 200
+        except FileNotFoundError as e:
+            logger.error(f"加载模板失败: {e}")
+            return_data["msg"] = "系统错误：模板文件不存在"
+            return return_data 
+            
+        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) -> Dict[str, Any]:
+        """区服交易行"""
+        return_data = self._init_return_data()
+
+        # 1. 查找物品 ID 列表 (多页)
+        api_config_search = self._api_config["jx3box_item"]
+        search_params = api_config_search.get("params", {}).copy()
+        search_params["keyword"] = Name
+        
+        # 使用 all_pages 获取所有数据
+        all_items: List[Dict[str, Any]] = await self._api.all_pages(
+            "GET", 
+            api_config_search["url"], 
+            params_data=search_params, # 对应到 APIClient.all_pages
+            out_key="data", 
+            list_key="data"
+        )
+        
+        if not all_items:
+            return_data["msg"] = "未找到该物品"
+            return return_data
+            
+        # 2. 提取 ID、IconID 和 Name
+        fields = ["IconID", "Name", "id"]
+        # 安全提取字段，并确保 id 存在且是数字
+        result = [extract_fields([item], fields)[0] for item in all_items if item.get("id") is not None]
+        
+        lists_id = [str(item.get("id")) for item in result if item.get("id") is not None]
+        strlists_id = ",".join(lists_id)
+        
+        if not strlists_id:
+            return_data["msg"] = "未找到有效的物品ID"
+            return return_data
+
+        # 3. 查询物品价格 (jx3box_itemprice)
+        api_config_price = self._api_config["jx3box_itemprice"]
+        price_params = api_config_price.get("params", {}).copy()
+        price_params["itemIds"] = strlists_id
+        price_params["server"] = server
+        
+        price_data: Optional[Dict[str, Dict[str, Any]]] = await self._api.get(
+            api_config_price["url"], params=price_params, out_key="data"
+        )
+        
+        if not price_data:
+            return_data["msg"] = "未找到在售物品"
+            return return_data
+        
+        # 4. 合并数据和格式化
+        fieldsjyh = ["ItemId", "SampleSize", "LowestPrice", "AvgPrice", "Date"]
+        resultjyh = [{f: v[f] for f in fieldsjyh} for v in price_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"
+
+        # 5. 模板渲染
+        try:
+            return_data["temp"] = load_template("jiaoyihang.html")
+            return_data["data"] = {
+                "items": resultjyh,
+                "server": server,
+                "update_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+            } 
+            return_data["code"] = 200
+        except FileNotFoundError as e:
+            logger.error(f"加载模板失败: {e}")
+            return_data["msg"] = "系统错误：模板文件不存在"
+        except Exception as e:
+            logger.error(f"模板渲染数据准备失败: {e}")
+            return_data["msg"] = "系统错误：数据处理失败"
+
+        return return_data

core/request.py

@@ -0,0 +1,192 @@
+# core/request.py
+import json
+import aiohttp
+from typing import Optional, Dict, Any, Union, List
+
+from aiohttp import ClientTimeout, ClientSession
+from astrbot.api import logger
+
+class APIClient:
+    """
+    API客户端类
+    
+    优化说明：
+    1. 复用 aiohttp.ClientSession 以提高性能。
+    2. 增加类型提示 (Type Hints)。
+    3. 支持异步上下文管理器 (Async Context Manager)。
+    """
+
+    def __init__(self, base_timeout: int = 10, ssl_verify: bool = False):
+        self.base_timeout = base_timeout
+        self.ssl_verify = ssl_verify
+        self._session: Optional[ClientSession] = None
+
+    async def get_session(self) -> ClientSession:
+        """获取或创建单例 Session"""
+        if self._session is None or self._session.closed:
+            timeout = ClientTimeout(total=self.base_timeout)
+            self._session = ClientSession(timeout=timeout)
+        return self._session
+
+    async def close(self):
+        """关闭 Session"""
+        if self._session and not self._session.closed:
+            await self._session.close()
+
+    async def __aenter__(self):
+        await self.get_session()
+        return self
+
+    async def __aexit__(self, exc_type, exc_val, exc_tb):
+        await self.close()
+
+    async def _request(self, method: str, url: str, params: Optional[Dict] = None, json_data: Optional[Dict] = None) -> Any:
+        """
+        统一的内部请求处理方法
+        """
+        session = await self.get_session()
+        method = method.upper()
+        
+        # 记录日志
+        logger.debug(f"发起 {method} 请求: {url}")
+        if params: logger.debug(f"Query参数: {params}")
+        if json_data: logger.debug(f"Body数据: {json_data}")
+
+        try:
+            # aiohttp 会自动处理 json=json_data 时的 Content-Type
+            async with session.request(
+                method=method,
+                url=url,
+                params=params,
+                json=json_data,
+                ssl=self.ssl_verify
+            ) as response:
+                return await self._handle_response(response)
+                
+        except aiohttp.ClientError as e:
+            logger.error(f"网络请求出错 ({method} {url}): {e}")
+            return None
+        except Exception as e:
+            logger.error(f"未知错误 ({method} {url}): {e}")
+            return None
+
+    async def _handle_response(self, response: aiohttp.ClientResponse) -> Any:
+        """处理响应：自动识别二进制或JSON"""
+        try:
+            logger.debug(f"响应状态: {response.status}")
+            response.raise_for_status() # 如果是 4xx/5xx 直接抛出异常
+
+            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)
+                except json.JSONDecodeError:
+                    logger.error(f"无法解析响应为 JSON。原始内容: {text[:100]}...")
+                    return None
+
+            logger.debug(f"响应数据: {data}")
+            return self._validate_api_payload(data)
+
+        except aiohttp.ClientError as e:
+            logger.error(f"HTTP响应错误: {e}")
+            return None
+
+    def _validate_api_payload(self, data: Any) -> Any:
+        """校验业务层面的 JSON 数据结构"""
+        if not data:
+            logger.error("API返回空数据")
+            return None
+
+        # 如果返回的是 JSON 字符串而非对象，再次解析
+        if isinstance(data, str):
+            try:
+                data = json.loads(data)
+            except json.JSONDecodeError:
+                return None
+        
+        if isinstance(data, dict) and 'code' in data:
+            # 兼容多种成功状态码：200, "0", 0, 1
+            code = data.get('code')
+            if code not in [200, "0", 0, 1]:
+                msg = data.get('msg') or data.get('message', '未知错误')
+                logger.error(f"API业务报错: code={code}, msg={msg}")
+                return None
+        
+        return data
+
+    async def get(self, url: str, params: Optional[Dict] = None, out_key: Optional[str] = None) -> Any:
+        """GET 请求封装"""
+        data = await self._request('GET', url, params=params)
+        return self._extract_data(data, out_key)
+
+    async def post(self, url: str, data: Optional[Dict] = None, out_key: Optional[str] = None) -> Any:
+        """POST 请求封装 (默认发送 JSON)"""
+        data = await self._request('POST', url, json_data=data)
+        return self._extract_data(data, out_key)
+
+    def _extract_data(self, data: Any, key: Optional[str]) -> Any:
+        """辅助方法：从结果中提取指定字段"""
+        if data is None:
+            return None
+        if isinstance(data, bytes):
+            return data
+        if key and isinstance(data, dict):
+            return data.get(key, {})
+        return data
+
+    async def all_pages(
+        self, 
+        method: str, 
+        url: str, 
+        params_data: Optional[Dict] = None, 
+        out_key: str = "", 
+        list_key: str = "list", 
+        max_pages: int = 10
+    ) -> List[Any]:
+        """
+        分页获取所有数据
+        :param method: GET 或 POST
+        :param list_key: 列表数据在 JSON 中的字段名，如 'data' 或 'list'
+        """
+        all_data = []
+        current_page = 1
+        params = params_data.copy() if params_data else {}
+
+        while True:
+            params["page"] = str(current_page)
+            
+            if method.upper() == "POST":
+                data = await self.post(url, data=params, out_key=out_key)
+            else:
+                data = await self.get(url, params=params, out_key=out_key)
+
+            # 终止条件判断
+            if not data or isinstance(data, bytes):
+                break
+            
+            # 如果 data 是列表本身（有些API直接返回列表）
+            page_items = data if isinstance(data, list) else data.get(list_key)
+            
+            if not page_items:
+                break
+
+            all_data.extend(page_items)
+
+            if current_page >= max_pages:
+                break
+
+            current_page += 1
+            logger.info(f"已获取第 {current_page} 页数据")
+
+        return all_data

main.py

@@ -8,8 +8,8 @@ from astrbot.api.star import Context, Star, register, StarTools
 from astrbot.api import logger
 from astrbot.api import AstrBotConfig
 
-from .core.AsyncMySQL import AsyncMySQL
+from .core.async_mysql import AsyncMySQL
-from .core.JX3Function import JX3Function
+from .core.jx3_service import JX3Service
 from .core.WZRYFunction import WZRYFunction
 
 
@@ -52,7 +52,7 @@ class Jx3ApiPlugin(Star):
         }        
         #创建类实例
         self.db = AsyncMySQL(db_config)
-        self.jx3fun = JX3Function(self.api_config,self.db)
+        self.jx3fun = JX3Service(self.api_config,self.db)
         self.wzry = WZRYFunction(self.api_config,self.db)
         # 周期函数调用
     

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)