from datetime import datetime from typing import Dict, Any, Optional, List, Union from astrbot.api import logger from astrbot.api import AstrBotConfig from .request import APIClient from .aiosqlite import AsyncSQLite from .function_basic import load_template,flatten_field,extract_fields,gold_to_string class JX3Service: def __init__(self, api_config,db: AsyncSQLite,config:AstrBotConfig): self._api = APIClient() self._db = db self._api_config = api_config self._config = 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 helps(self) -> Dict[str, Any]: """帮助""" return_data = self._init_return_data() # 加载模板 try: return_data["temp"] = 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 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 ) logger.info(f"richang 接口返回数据: {data}") 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', []) luck_msg = f"[宠物福缘]：\n{', '.join(luck)}\n" card = data.get('card', []) card_msg = f"[家园声望·加倍道具]：\n{', '.join(card)}\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 xinwei(self) -> Dict[str, Any]: """新闻资讯""" return_data = self._init_return_data() # 提取字段可能返回列表 data: Optional[List[Dict[str, Any]]] = await self._base_request("jx3_xinweng", "GET") if not data or not isinstance(data, list): return_data["msg"] = "获取接口信息失败或数据格式错误" return return_data try: # result = data[0] return_data["status"] = result.get('id') 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('date', '未知时间')}\n" result_msg += f"链接：{item.get('url', '无链接')}\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, limit:str) -> Dict[str, Any]: """区服金价""" return_data = self._init_return_data() # 获取配置中的 Token token = self._config.get("jx3api_token", "") if token == "": return_data["msg"] = "系统未配置API访问Token" return return_data params = {"server": server, "limit": limit, "token": token} data_list: Optional[List[Dict[str, Any]]] = await self._base_request("jx3_jinjia", "GET", params=params) if not data_list or not isinstance(data_list, list): return_data["msg"] = "获取接口信息失败或数据格式错误" return return_data # 加载模板 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"] = data_list 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 (?, ?, ?, ?, ?, ?) """ # 增强数据提取的安全性 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.clear_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, server:str) -> Dict[str, Any]: """物价查询""" return_data = self._init_return_data() # 获取配置中的 Token token = self._config.get("jx3api_token", "") if token == "": return_data["msg"] = "系统未配置API访问Token" return return_data # 2. 确定外观名称和 ID params_search = {"name": Name,"token": token, "server": server} search_data: Optional[Dict[str, Any]] = await self._base_request("jx3_wujia", "GET", params=params_search) if not search_data: return_data["msg"] = "未找到改外观" return return_data return_data["data"] = search_data # 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 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 async def jueshemingpian(self, server: str, name:str ) -> Dict[str, Any]: """角色名片""" return_data = self._init_return_data() # 获取配置中的 Token token = self._config.get("jx3api_token", "") if token == "": return_data["msg"] = "系统未配置API访问Token" return return_data # 1. 构造请求参数 params = {"server": server, "name": name,"token": token} # 2. 调用基础请求 data: Optional[Dict[str, Any]] = await self._base_request( "jx3_jieshemingpian", "GET", params=params ) if not data: return_data["msg"] = "未找到该角色" return return_data # 3. 处理返回数据 (直接提取图片 URL) return_data["data"] = data return_data["code"] = 200 return return_data async def shuijimingpian(self, force: str, body:str, server:str) -> Dict[str, Any]: """随机名片""" return_data = self._init_return_data() # 获取配置中的 Token token = self._config.get("jx3api_token", "") if token == "": return_data["msg"] = "系统未配置API访问Token" return return_data # 1. 构造请求参数 params = {"server": server, "body": body, "force":force, "token": token} # 2. 调用基础请求 data: Optional[Dict[str, Any]] = await self._base_request( "jx3_shuijimingpian", "GET", params=params ) if not data: return_data["msg"] = "获取接口信息失败" return return_data # 3. 处理返回数据 (直接提取图片 URL) return_data["data"] = data return_data["code"] = 200 return return_data async def yanhuachaxun(self, server: str, name:str ) -> Dict[str, Any]: """烟花查询""" return_data = self._init_return_data() # 获取配置中的 Token token = self._config.get("jx3api_token", "") if token == "": return_data["msg"] = "系统未配置API访问Token" return return_data # 1. 构造请求参数 params = {"server": server, "name": name,"token": token} # 2. 调用基础请求 data: Optional[Dict[str, Any]] = await self._base_request( "jx3_yanhuachaxun", "GET", params=params ) if not data: return_data["msg"] = "获取接口信息失败" return return_data # 3. 处理返回数据 (直接提取图片 URL) # 格式化时间 for item in data: timestamp = item.get("time") if timestamp and isinstance(timestamp, (int, float)): # 修复时间戳：原代码显示这里是毫秒级，除以 1000 item["time"] = datetime.fromtimestamp(timestamp).strftime("%Y-%m-%d %H:%M:%S") else: item["time"] = "未知时间" # 确保即使 time 字段缺失也不会报错 # 4. 加载模板 try: return_data["temp"] = load_template("yanhuan.html") except FileNotFoundError as e: logger.error(f"加载模板失败: {e}") return_data["msg"] = "系统错误：模板文件不存在" return return_data return_data["data"]["list"] = data return_data["code"] = 200 return return_data async def dilujilu(self, server: str) -> Dict[str, Any]: """的卢记录""" return_data = self._init_return_data() # 获取配置中的 Token token = self._config.get("jx3api_token", "") if token == "": return_data["msg"] = "系统未配置API访问Token" return return_data # 1. 构造请求参数 params = {"server": server, "token": token} # 2. 调用基础请求 data: Optional[Dict[str, Any]] = await self._base_request( "jx3_dilujilu", "GET", params=params ) if not data: return_data["msg"] = "获取接口信息失败" return return_data # 3. 处理返回数据 for item in data: item["refresh_time"] = datetime.fromtimestamp(item["refresh_time"]).strftime("%Y-%m-%d %H:%M:%S") item["capture_time"] = datetime.fromtimestamp(item["capture_time"]).strftime("%Y-%m-%d %H:%M:%S") item["auction_time"] = datetime.fromtimestamp(item["auction_time"]).strftime("%Y-%m-%d %H:%M:%S") # 4. 加载模板 try: return_data["temp"] = load_template("dilujilu.html") except FileNotFoundError as e: logger.error(f"加载模板失败: {e}") return_data["msg"] = "系统错误：模板文件不存在" return return_data return_data["data"]["list"] = data return_data["code"] = 200 return return_data async def tuanduizhaomu(self, server: str, keyword: str) -> Dict[str, Any]: """团队招募""" return_data = self._init_return_data() # 获取配置中的 Token token = self._config.get("jx3api_token", "") if token == "": return_data["msg"] = "系统未配置API访问Token" return return_data # 1. 构造请求参数 params = {"server": server, "keyword": keyword, "token": token} # 2. 调用基础请求 data: Optional[Dict[str, Any]] = await self._base_request( "jx3_tuanduizhaomu", "GET", params=params ) if not data: return_data["msg"] = "获取接口信息失败" return return_data # 3. 处理返回数据 for item in data["data"]: item["createTime"] = datetime.fromtimestamp(item["createTime"]).strftime("%Y-%m-%d %H:%M:%S") item["number"] = f"{item['number']}/{item['maxNumber']}" # 4. 加载模板 try: return_data["temp"] = load_template("tuanduizhaomu.html") except FileNotFoundError as e: logger.error(f"加载模板失败: {e}") return_data["msg"] = "系统错误：模板文件不存在" return return_data return_data["data"]["list"] = data["data"] return_data["code"] = 200 return return_data async def zhanji(self, name: str, server:str, mode:str) -> Dict[str, Any]: """战绩+名片""" return_data = self._init_return_data() # 获取配置中的 Token token = self._config.get("jx3api_token", "") if token == "": return_data["msg"] = "系统未配置API访问Token" return return_data # 1. 构造请求参数 params = {"server": server, "name":name, "mode":mode, "token": token} # 2. 调用基础请求 data: Optional[Dict[str, Any]] = await self._base_request( "jx3_zhanji", "GET", params=params ) if not data: return_data["msg"] = "查询角色战绩失败" return return_data logger.info("战绩获取完成") # 角色名片获取 datamp = await self.jueshemingpian(server,name) if datamp["code"] == 200: data["showAvatar"] = datamp['data']['showAvatar'] logger.info("名片获取完成") else: data["showAvatar"] = "" logger.info("名片获取失败") # 4. 加载模板 try: return_data["temp"] = load_template("zhanji.html") except FileNotFoundError as e: logger.error(f"加载模板失败: {e}") return_data["msg"] = "系统错误：模板文件不存在" return return_data return_data["data"] = data return_data["code"] = 200 return return_data async def juesheqiyu(self, name: str, server: str) -> Dict[str, Any]: """角色奇遇""" return_data = self._init_return_data() # 获取配置中的 Token token = self._config.get("jx3api_token", "") if token == "": return_data["msg"] = "系统未配置API访问Token" return return_data # 1. 构造请求参数 params = {"server": server, "name": name, "token": token} # 2. 调用基础请求 data: Optional[Dict[str, Any]] = await self._base_request( "jx3_qiyu", "GET", params=params ) if not data: return_data["msg"] = "未找到改角色奇遇信息" return return_data # 3. 处理返回数据 try: return_data["data"]["ptqy"] = [] return_data["data"]["jsqy"] = [] return_data["data"]["cwqy"] = [] for item in data: item["time"] = datetime.fromtimestamp(item["time"]).strftime("%Y-%m-%d %H:%M:%S") if item["level"] == 1: return_data["data"]["ptqy"].append(item) if item["level"] == 2: return_data["data"]["jsqy"].append(item) if item["level"] == 3: return_data["data"]["cwqy"].append(item) except Exception as e: logger.error(f"处理返回数据失败: {e}") return_data["msg"] = "处理返回数据失败" # 4. 加载模板 try: return_data["temp"] = load_template("juesheqiyu.html") except FileNotFoundError as e: logger.error(f"加载模板失败: {e}") return_data["msg"] = "系统错误：模板文件不存在" return return_data return_data["code"] = 200 return return_data