diff --git a/.gitignore b/.gitignore deleted file mode 100644 index b7faf40..0000000 --- a/.gitignore +++ /dev/null @@ -1,207 +0,0 @@ -# Byte-compiled / optimized / DLL files -__pycache__/ -*.py[codz] -*$py.class - -# C extensions -*.so - -# Distribution / packaging -.Python -build/ -develop-eggs/ -dist/ -downloads/ -eggs/ -.eggs/ -lib/ -lib64/ -parts/ -sdist/ -var/ -wheels/ -share/python-wheels/ -*.egg-info/ -.installed.cfg -*.egg -MANIFEST - -# PyInstaller -# Usually these files are written by a python script from a template -# before PyInstaller builds the exe, so as to inject date/other infos into it. -*.manifest -*.spec - -# Installer logs -pip-log.txt -pip-delete-this-directory.txt - -# Unit test / coverage reports -htmlcov/ -.tox/ -.nox/ -.coverage -.coverage.* -.cache -nosetests.xml -coverage.xml -*.cover -*.py.cover -.hypothesis/ -.pytest_cache/ -cover/ - -# Translations -*.mo -*.pot - -# Django stuff: -*.log -local_settings.py -db.sqlite3 -db.sqlite3-journal - -# Flask stuff: -instance/ -.webassets-cache - -# Scrapy stuff: -.scrapy - -# Sphinx documentation -docs/_build/ - -# PyBuilder -.pybuilder/ -target/ - -# Jupyter Notebook -.ipynb_checkpoints - -# IPython -profile_default/ -ipython_config.py - -# pyenv -# For a library or package, you might want to ignore these files since the code is -# intended to run in multiple environments; otherwise, check them in: -# .python-version - -# pipenv -# According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control. -# However, in case of collaboration, if having platform-specific dependencies or dependencies -# having no cross-platform support, pipenv may install dependencies that don't work, or not -# install all needed dependencies. -#Pipfile.lock - -# UV -# Similar to Pipfile.lock, it is generally recommended to include uv.lock in version control. -# This is especially recommended for binary packages to ensure reproducibility, and is more -# commonly ignored for libraries. -#uv.lock - -# poetry -# Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control. -# This is especially recommended for binary packages to ensure reproducibility, and is more -# commonly ignored for libraries. -# https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control -#poetry.lock -#poetry.toml - -# pdm -# Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control. -# pdm recommends including project-wide configuration in pdm.toml, but excluding .pdm-python. -# https://pdm-project.org/en/latest/usage/project/#working-with-version-control -#pdm.lock -#pdm.toml -.pdm-python -.pdm-build/ - -# pixi -# Similar to Pipfile.lock, it is generally recommended to include pixi.lock in version control. -#pixi.lock -# Pixi creates a virtual environment in the .pixi directory, just like venv module creates one -# in the .venv directory. It is recommended not to include this directory in version control. -.pixi - -# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm -__pypackages__/ - -# Celery stuff -celerybeat-schedule -celerybeat.pid - -# SageMath parsed files -*.sage.py - -# Environments -.env -.envrc -.venv -env/ -venv/ -ENV/ -env.bak/ -venv.bak/ - -# Spyder project settings -.spyderproject -.spyproject - -# Rope project settings -.ropeproject - -# mkdocs documentation -/site - -# mypy -.mypy_cache/ -.dmypy.json -dmypy.json - -# Pyre type checker -.pyre/ - -# pytype static type analyzer -.pytype/ - -# Cython debug symbols -cython_debug/ - -# PyCharm -# JetBrains specific template is maintained in a separate JetBrains.gitignore that can -# be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore -# and can be added to the global gitignore or merged into this file. For a more nuclear -# option (not recommended) you can uncomment the following to ignore the entire idea folder. -#.idea/ - -# Abstra -# Abstra is an AI-powered process automation framework. -# Ignore directories containing user credentials, local state, and settings. -# Learn more at https://abstra.io/docs -.abstra/ - -# Visual Studio Code -# Visual Studio Code specific template is maintained in a separate VisualStudioCode.gitignore -# that can be found at https://github.com/github/gitignore/blob/main/Global/VisualStudioCode.gitignore -# and can be added to the global gitignore or merged into this file. However, if you prefer, -# you could uncomment the following to ignore the entire vscode folder -# .vscode/ - -# Ruff stuff: -.ruff_cache/ - -# PyPI configuration file -.pypirc - -# Cursor -# Cursor is an AI-powered code editor. `.cursorignore` specifies files/directories to -# exclude from AI features like autocomplete and code analysis. Recommended for sensitive data -# refer to https://docs.cursor.com/context/ignore-files -.cursorignore -.cursorindexingignore - -# Marimo -marimo/_static/ -marimo/_lsp/ -__marimo__/ diff --git a/README.md b/README.md index cfca54a..b7b3df6 100644 --- a/README.md +++ b/README.md @@ -1,4 +1,4 @@ -# JX3API +# Gok AstrBot 插件模板 diff --git a/api_config.json b/api_config.json index a73e0cd..78917f8 100644 --- a/api_config.json +++ b/api_config.json @@ -1,132 +1,4 @@ { - "jx3_richang":{ - "url":"https://www.jx3api.com/data/active/calendar", - "method":"GET", - "description":"获取剑网三日常任务", - "params":{ - "server": "眉间雪", - "num": 0 - } - }, - "jx3_shaohua":{ - "url":"https://www.jx3api.com/data/saohua/random", - "method":"GET", - "description":"获取剑网三骚话", - "params":{ - "name": "万花" - } - }, - "jx3_jigai":{ - "url":"https://www.jx3api.com/data/skills/records", - "method":"GET", - "description":"获取剑网三技能记录", - "params":{ - } - }, - "jx3_kaifu":{ - "url":"https://www.jx3api.com/data/server/check", - "method":"GET", - "description":"剑网三开服查询", - "params":{ - "server": "梦江南" - } - }, - "aijx3_shapan":{ - "url":"https://www.jianxiachaguan.cn/api2/aijx3-jxcg/game/get-sand-table-img", - "method":"POST", - "description":"获取剑网三沙盘状态", - "params":{ - "serverName": "眉间雪" - } - }, - "aijx3_jinjia":{ - "url":"https://www.jianxiachaguan.cn/api2/aijx3-jxcg/game/get-gold", - "method":"POST", - "description":"获取剑网三某区服金价", - "params":{ - "serverName": "眉间雪" - } - }, - "aijx3_qiyu":{ - "url":"https://www.jianxiachaguan.cn/api2/aijx3-jxcg/game/get-adventure-record", - "method":"POST", - "description":"获取剑网三某区服奇遇触发记录", - "params":{ - "adventureName": "阴阳两界", - "serverName": "眉间雪" - } - }, - "aijx3_SearchData":{ - "url":"https://www.aijx3.cn/api2/aijx3-wblwg/basedata/getSearchData", - "method":"POST", - "description":"获取爱剑网三所有外观数据", - "params":{ - } - }, - "jx3box_exterior":{ - "url":"https://node.jx3box.com/api/node/v2/exterior/list", - "method":"GET", - "description":"魔盒中模糊查找剑网三的外观数据", - "params":{ - "keyword": "秃盒", - "tradable": "0", - "list_type": "0" - } - }, - "aijx3_GoodsDetail":{ - "url":"https://www.aijx3.cn/api2/aijx3-wj/goods/getGoodsDetail", - "method":"POST", - "description":"在爱剑网三中获取外观详细数据", - "params":{ - "goodsName": "秃盒" - } - }, - "aijx3_wblwg":{ - "url":"https://www.aijx3.cn/api2/aijx3-wblwg/record/queryByCondition", - "method":"POST", - "description":"在爱剑网三中获取外观详细数据", - "params":{ - "accoSeq": "", - "orderMode": 1, - "orderBy": "price_num", - "searchId": [48356], - "current": 1, - "size": 10, - "tradeStatus": "" - } - }, - "jx3box_item":{ - "url":"https://node.jx3box.com/api/node/item/search", - "method":"GET", - "description":"魔盒中模糊查找剑网三的物品数据", - "params":{ - "keyword": "守缺式", - "page": "1", - "per": "50" - } - }, - "jx3box_itemprice":{ - "url":"https://next2.jx3box.com/api/item-price/list", - "method":"GET", - "description":"交易行中在售物品数据", - "params":{ - "server": "眉间雪", - "itemIds": ["5_76650"] - } - }, - "wbl_waiguan":{ - "url":"https://trade-api.seasunwbl.com/api/buyer/goods/list", - "method":"GET", - "description":"万宝楼官方交易数据", - "params":{ - "filter[state]": "0", - "filter[role_appearance]": "浮屠明音礼盒", - "page": "1", - "size": "10", - "goods_type": "3", - "sort[price]": "1" - } - }, "wzry_zhanji":{ "url":"https://api.t1qq.com/api/tool/wzrr/morebattle", "method":"GET", diff --git a/core/APIClient.py b/core/APIClient.py deleted file mode 100644 index f7a3edb..0000000 --- a/core/APIClient.py +++ /dev/null @@ -1,150 +0,0 @@ -# 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) diff --git a/core/AsyncMySQL.py b/core/AsyncMySQL.py deleted file mode 100644 index 255fae6..0000000 --- a/core/AsyncMySQL.py +++ /dev/null @@ -1,90 +0,0 @@ -import aiomysql - - -class AsyncMySQL: - def __init__(self, db_config: dict): - self.db_config = db_config - self.pool = None - - async def init_pool(self): - """初始化连接池""" - if self.pool is None: - self.pool = await aiomysql.create_pool(**self.db_config) - - async def close_pool(self): - """关闭连接池""" - if self.pool: - self.pool.close() - await self.pool.wait_closed() - self.pool = None - - async def fetch_one(self, sql: str, params=None): - """查询单条数据""" - await self.init_pool() - async with self.pool.acquire() as conn: - 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())) diff --git a/core/JX3Function.py b/core/JX3Function.py deleted file mode 100644 index 9b77648..0000000 --- a/core/JX3Function.py +++ /dev/null @@ -1,536 +0,0 @@ -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 diff --git a/core/WZRYFunction.py b/core/WZRYFunction.py index 10b697d..1c3bbe2 100644 --- a/core/WZRYFunction.py +++ b/core/WZRYFunction.py @@ -2,9 +2,9 @@ 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 +from .request import APIClient +from .async_mysql import AsyncMySQL +from .function_basic import load_template,extract_fields class WZRYFunction: def __init__(self, api_config,db: AsyncMySQL ): diff --git a/core/__pycache__/__init__.cpython-313.pyc b/core/__pycache__/__init__.cpython-313.pyc new file mode 100644 index 0000000..e008624 Binary files /dev/null and b/core/__pycache__/__init__.cpython-313.pyc differ diff --git a/core/__pycache__/request.cpython-313.pyc b/core/__pycache__/request.cpython-313.pyc new file mode 100644 index 0000000..506f62b Binary files /dev/null and b/core/__pycache__/request.cpython-313.pyc differ diff --git a/core/jx3_service.py b/core/jx3_service.py deleted file mode 100644 index b45743b..0000000 --- a/core/jx3_service.py +++ /dev/null @@ -1,593 +0,0 @@ -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 diff --git a/main.py b/main.py index 358f21f..2dec3b6 100644 --- a/main.py +++ b/main.py @@ -9,24 +9,23 @@ from astrbot.api import logger from astrbot.api import AstrBotConfig from .core.async_mysql import AsyncMySQL -from .core.jx3_service import JX3Service from .core.WZRYFunction import WZRYFunction -@register("astrbot_plugin_jx3api", - "fxdyz", +@register("astrbot_plugin_gok", + "飞翔大野猪", "通过接口调用剑网三API接口获取游戏数据", "1.0.0", - "https://github.com/qsc20001102/astrbot_plugin_jx3api.git" + "https://github.com/qsc20001102/astrbot_plugin_gok" ) -class Jx3ApiPlugin(Star): +class GokApiPlugin(Star): def __init__(self, context: Context, config: AstrBotConfig): super().__init__(context) #获取配置 self.conf = config # 本地数据存储路径 - self.local_data_dir = StarTools.get_data_dir("astrbot_plugin_jx3api") + self.local_data_dir = StarTools.get_data_dir("astrbot_plugin_gok") # api数据文件 self.api_file_path = Path(__file__).parent / "api_config.json" # 读取文件内容 @@ -34,7 +33,6 @@ class Jx3ApiPlugin(Star): self.api_config = json.load(f) # 初始化数据 self.inidata() - self.kf_task = asyncio.create_task(self.cycle_kaifjiankong()) logger.info("jx3api插件初始化完成") @@ -52,7 +50,6 @@ class Jx3ApiPlugin(Star): } #创建类实例 self.db = AsyncMySQL(db_config) - self.jx3fun = JX3Service(self.api_config,self.db) self.wzry = WZRYFunction(self.api_config,self.db) # 周期函数调用 @@ -64,229 +61,6 @@ class Jx3ApiPlugin(Star): self.test_server = False logger.info("数据初始化完成") - async def cycle_kaifjiankong(self): - """开服监控后台程序""" - # 获取配置信息 - time_int = self.conf.get("jx3_kfjk_time", 10) - self.kfjk_en = self.conf.get("jx3_kfjk_en", True) - self.kfjk_umos = self.conf.get("jx3_kfjk_list", []) - self.kfjk_servername = self.conf.get("jx3_kfjk_server", "梦江南") - kfjk_test = self.conf.get("jx3_kfjk_test", False) - - self.kfjk_server_state = True # 上一次查询的状态 - self.kfjk_server_state_new = False # 最新查询的状态 - - if self.kfjk_en: - logger.info(f"开服监控功能开启") - while self.kfjk_en: - data = await self.jx3fun.kaifu(self.kfjk_servername) - if kfjk_test: - self.kfjk_server_state_new = self.test_server - else: - self.kfjk_server_state_new = data["status"] - # logger.info(f"开服监控功能循环中,上次询问服务器状态{self.kfjk_server_state},本次询问的服务器状态{self.kfjk_server_state_new}") - if self.kfjk_server_state != self.kfjk_server_state_new: - logger.info(f"开服监控功能循环中,上次询问服务器状态{self.kfjk_server_state},本次询问的服务器状态{self.kfjk_server_state_new}") - if self.kfjk_server_state and not self.kfjk_server_state_new: - message_chain = MessageChain().message(f"{self.kfjk_servername}服务器已关闭\n休息一会把,开服了喊你！") - if self.kfjk_server_state_new and not self.kfjk_server_state: - message_chain = MessageChain().message(f"{self.kfjk_servername}服务器已开启\n快冲！快冲！") - if self.kfjk_umos: - for umo in self.kfjk_umos: - await self.context.send_message(umo, message_chain) - self.kfjk_server_state = self.kfjk_server_state_new - await asyncio.sleep(time_int) - if not self.kfjk_en: - # 销毁进程 - self.kf_task.cancel() - logger.info(f"开服监控功能关闭") - - @filter.command_group("剑三") - def jx3(self): - pass - - @jx3.command("日常") - async def jx3_richang(self, event: AstrMessageEvent,server: str = "梦江南",num: int = 0): - """剑三 日常 服务器 天数""" - try: - data= await self.jx3fun.richang(server,num) - if data["code"] == 200: - yield event.plain_result(data["data"]) - else: - yield event.plain_result("msg") - return - except Exception as e: - logger.error(f"功能函数执行错误: {e}") - yield event.plain_result("猪脑过载，请稍后再试") - - @jx3.command("开服") - async def jx3_kaifu(self, event: AstrMessageEvent,server: str = "梦江南"): - """剑三 日常 服务器 天数""" - try: - data= await self.jx3fun.kaifu(server) - if data["code"] == 200: - yield event.plain_result(data["data"]) - else: - yield event.plain_result("msg") - return - except Exception as e: - logger.error(f"功能函数执行错误: {e}") - yield event.plain_result("猪脑过载，请稍后再试") - - @jx3.command("沙盘") - async def jx3_shapan(self, event: AstrMessageEvent,server: str = "梦江南"): - """剑三 沙盘 服务器""" - try: - data= await self.jx3fun.shapan(server) - if data["code"] == 200: - yield event.image_result(data["data"]) - else: - yield event.plain_result(data["msg"]) - return - except Exception as e: - logger.error(f"功能函数执行错误: {e}") - yield event.plain_result("猪脑过载，请稍后再试") - - - @jx3.command("骚话") - async def jx3_shaohua(self, event: AstrMessageEvent,): - """剑三 骚话""" - try: - data= await self.jx3fun.shaohua() - if data["code"] == 200: - yield event.plain_result(data["data"]) - else: - yield event.plain_result(data["msg"]) - return - except Exception as e: - logger.error(f"功能函数执行错误: {e}") - yield event.plain_result("猪脑过载，请稍后再试") - - - @jx3.command("技改") - async def jx3_jigai(self, event: AstrMessageEvent,): - """剑三 技改""" - try: - data= await self.jx3fun.jigai() - if data["code"] == 200: - yield event.plain_result(data["data"]) - else: - yield event.plain_result(data["msg"]) - return - except Exception as e: - logger.error(f"功能函数执行错误: {e}") - yield event.plain_result("猪脑过载，请稍后再试") - - - @jx3.command("金价") - async def jx3_jinjia(self, event: AstrMessageEvent,server: str = "梦江南"): - """剑三 金价 服务器""" - try: - data= await self.jx3fun.jinjia(server) - if data["code"] == 200: - url = await self.html_render(data["temp"], data["data"], options={}) - yield event.image_result(url) - else: - yield event.plain_result(data["msg"]) - return - except Exception as e: - logger.error(f"功能函数执行错误: {e}") - yield event.plain_result("猪脑过载，请稍后再试") - - - @jx3.command("奇遇") - async def jx3_qiyu(self, event: AstrMessageEvent,adventureName: str = "阴阳两界", serverName: str = "梦江南"): - """剑三 奇遇 奇遇名称 服务器""" - try: - data= await self.jx3fun.qiyu(adventureName,serverName) - if data["code"] == 200: - url = await self.html_render(data["temp"], data["data"], options={}) - yield event.image_result(url) - else: - yield event.plain_result(data["msg"]) - return - except Exception as e: - logger.error(f"功能函数执行错误: {e}") - yield event.plain_result("猪脑过载，请稍后再试") - - - @jx3.command("物价") - async def jx3_wujia(self, event: AstrMessageEvent,Name: str = "秃盒"): - """剑三 外观名称""" - try: - data=await self.jx3fun.wujia(Name) - if data["code"] == 200: - url = await self.html_render(data["temp"], data["data"], options={}) - yield event.image_result(url) - else: - yield event.plain_result(data["msg"]) - return - except Exception as e: - logger.error(f"功能函数执行错误: {e}") - yield event.plain_result("猪脑过载，请稍后再试") - - - @jx3.command("交易行") - async def jx3_jiaoyihang(self, event: AstrMessageEvent,Name: str = "守缺式",server: str = "梦江南"): - """剑三 外观名称""" - try: - data=await self.jx3fun.jiaoyihang(Name,server) - if data["code"] == 200: - url = await self.html_render(data["temp"], data["data"], options={}) - yield event.image_result(url) - - else: - yield event.plain_result(data["msg"]) - return - except Exception as e: - logger.error(f"功能函数执行错误: {e}") - yield event.plain_result("猪脑过载，请稍后再试") - - @jx3.command("开服监控") - async def jx3_kaifhujiank(self, event: AstrMessageEvent,en: str = "状态"): - """剑三 开服监控""" - if en == "开启": - umo_set = set(self.kfjk_umos) - umo_set.add(event.unified_msg_origin) - self.kfjk_umos = list(umo_set) - yield event.plain_result(f"开服监控已开启") - elif en =="关闭": - try: - self.kfjk_umos.remove(event.unified_msg_origin) - yield event.plain_result(f"开服监控已关闭") - except ValueError as e: - yield event.plain_result(f"开服监控已关闭") - elif en == "状态": - yield event.plain_result(f"开服监控后台状态：{self.kfjk_en}\n监控服务器：{self.kfjk_servername}\n上次询问服务器状态{self.kfjk_server_state}\n本次询问的服务器状态{self.kfjk_server_state_new}\n推送会话列表：\n{self.kfjk_umos}") - else: - yield event.plain_result("开服监控指令错误，请输入 开启/关闭") - - - @filter.permission_type(filter.PermissionType.ADMIN) - @jx3.command("外观数据同步") - async def jx3_SearchData(self, event: AstrMessageEvent): - """剑三 外观数据同步""" - try: - data=await self.jx3fun.SearchData() - if data["code"] == 200: - yield event.plain_result(data["msg"]) - else: - yield event.plain_result(data["msg"]) - return - except Exception as e: - logger.error(f"功能函数执行错误: {e}") - yield event.plain_result("猪脑过载，请稍后再试") - - @filter.command("测试值") - async def jx3_testvar(self, event: AstrMessageEvent,var: str = "0"): - """测试程序""" - if var == "1": - self.test_server = True - if var == "0": - self.test_server = False - - yield event.plain_result(f"测试值:{self.test_server}") - @filter.command_group("王者") def wz(self): diff --git a/metadata.yaml b/metadata.yaml index 28603d4..f46e081 100644 --- a/metadata.yaml +++ b/metadata.yaml @@ -1,5 +1,5 @@ -name: jx3api # 这是你的插件的唯一识别名。 -desc: 通过接口调用剑网三API接口获取游戏数据 # 插件简短描述 +name: astrbot_plugin_gok # 这是你的插件的唯一识别名。 +desc: 王者荣耀的战绩、资料等内容的查询 # 插件简短描述 version: v1.0.0 # 插件版本号。格式：v1.1.1 或者 v1.1 -author: fxdyz # 作者 -repo: https://github.com/qsc20001102/astrbot_plugin_jx3api # 插件的仓库地址 +author: 飞翔大野猪 # 作者 +repo: https://github.com/qsc20001102/astrbot_plugin_gok # 插件的仓库地址 diff --git a/templates/jiaoyihang.html b/templates/jiaoyihang.html deleted file mode 100644 index 4c4a416..0000000 --- a/templates/jiaoyihang.html +++ /dev/null @@ -1,30 +0,0 @@ -
-

剑网3交易行价格查询

-

服务器: {{ server }}

- -
- - - - - - - - - - - {% for item in items %} - - - - - - - {% endfor %} - -
物品名称最低价总数量数据更新时间
- icon - {{ item.Name }} - {{ item.AvgPrice }}{{ item.SampleSize }}{{ item.Date }}
-
-
diff --git a/templates/jinjia.html b/templates/jinjia.html deleted file mode 100644 index 2b4c44c..0000000 --- a/templates/jinjia.html +++ /dev/null @@ -1,33 +0,0 @@ -
-

剑网三金价

- -
- - - - - {% for item in items %} - - {% endfor %} - - - - - - - {% for item in items %} - - {% endfor %} - - - - - {% for item in items %} - - {% endfor %} - - -
时间{{ item.date }}
万宝楼{{ item.priceWanbaolou }}
贴吧{{ item.priceTieba }}
-
- -
diff --git a/templates/qiyuliebiao.html b/templates/qiyuliebiao.html deleted file mode 100644 index 7356f8f..0000000 --- a/templates/qiyuliebiao.html +++ /dev/null @@ -1,34 +0,0 @@ -
-

剑网3奇遇查询

-

奇遇名称：{{ qiyuname }}　服务器：{{ server }}

- -
- - - - - - - - - - - {% for item in items %} - - - - - - - {% endfor %} - -
奇遇名称服务器角色名称触发时间
{{ item.adventureName }}{{ server }}{{ item.gameName }}{{ item.time }}
-
- - -
- 奇遇趋势图 -
-
diff --git a/templates/wujia.html b/templates/wujia.html deleted file mode 100644 index 07f7689..0000000 --- a/templates/wujia.html +++ /dev/null @@ -1,65 +0,0 @@ -
- - -

{{ showName }} ({{ goodsAlias }})

- {{ showName }} -

{{ goodsDesc }}

-

发布时间: {{ publishTime.split('T')[0] if publishTime else '未知' }}

-

参考价格: {{ priceNum }} 元

- - -
- - -
-
万宝楼公示
-
- - - - - - - - - - {% for item in wblgs %} - - - - - - {% endfor %} - -
剩余时间服务器价格
{{ item.replyTime }}{{ item.belongQf2 }}{{ item.priceNum }}元
-
-
- - -
-
万宝楼在售
-
- - - - - - - - - - {% for item in wblzs %} - - - - - - {% endfor %} - -
剩余时间服务器价格
{{ item.replyTime }}{{ item.belongQf2 }}{{ item.priceNum }}元
-
-
- -
- -