qsc20001102/astrbot_plugin_gok
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

11

Browse Source
This commit is contained in:
qsc
2025-12-03 13:56:49 +08:00
parent 73681864b3
commit 2572c1d239
16 changed files with 13 additions and 2105 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
-207
View File
@@ -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__/
README.md
+1 -1
View File
@@ -1,4 +1,4 @@
# JX3API
# Gok
AstrBot 插件模板
api_config.json
-128
View File
@@ -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",
core/APIClient.py
-150
View File
@@ -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)
core/AsyncMySQL.py
-90
View File
@@ -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()))
core/JX3Function.py
-536
View File
@@ -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
core/WZRYFunction.py
+3 -3
View File
@@ -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 ):
core/__pycache__/__init__.cpython-313.pyc
BIN
View File
Binary file not shown.
core/__pycache__/request.cpython-313.pyc
BIN
View File
Binary file not shown.
core/jx3_service.py
-593
View File
@@ -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
main.py
+5 -231
View File
@@ -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):
metadata.yaml
+4 -4
View File
@@ -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 # 插件的仓库地址
templates/jiaoyihang.html
-30
View File
@@ -1,30 +0,0 @@
<div style="width: 1220px; margin: 0 auto;">
<h1 style="text-align: center;">剑网3交易行价格查询</h1>
<p style="text-align: center;">服务器: {{ server }}</p>
<div style="overflow-x: auto; overflow-y: hidden;">
<table style="width: 100%; border-collapse: collapse; margin: 0 auto;">
<thead>
<tr style="height: 50px; text-align: center;">
<th style="border:1px solid #000; padding:4px;">物品名称</th>
<th style="border:1px solid #000; padding:4px;">最低价</th>
<th style="border:1px solid #000; padding:4px;">总数量</th>
<th style="border:1px solid #000; padding:4px;">数据更新时间</th>
</tr>
</thead>
<tbody>
{% for item in items %}
<tr style="height: 45px; text-align: center;">
<td style="border:1px solid #000; padding:4px; text-align: left;">
<img src="{{ item.IconID }}" alt="icon" style="width:20px; height:20px; vertical-align:middle; border-radius:4px; margin-right:6px;">
{{ item.Name }}
</td>
<td style="border:1px solid #000; padding:4px;">{{ item.AvgPrice }}</td>
<td style="border:1px solid #000; padding:4px;">{{ item.SampleSize }}</td>
<td style="border:1px solid #000; padding:4px;">{{ item.Date }}</td>
</tr>
{% endfor %}
</tbody>
</table>
</div>
</div>
templates/jinjia.html
-33
View File
@@ -1,33 +0,0 @@
<div style="width: 1220px; margin: 0 auto;">
<h1 style="text-align: center" id="剑网三金价">剑网三金价</h1>
<div style="overflow-x: auto; overflow-y: hidden;">
<table style="width: 100%; border-collapse: collapse; margin: 0 auto;">
<thead>
<tr style="height: 60px; text-align: center;">
<th style="border:1px solid #000; padding:2px;">时间</th>
{% for item in items %}
<th style="border:1px solid #000; padding:2px;">{{ item.date }}</th>
{% endfor %}
</tr>
</thead>
<tbody>
<tr style="height: 50px; text-align: center;">
<td style="border:1px solid #000; padding:2px;">万宝楼</td>
{% for item in items %}
<td style="border:1px solid #000; padding:2px;">{{ item.priceWanbaolou }}</td>
{% endfor %}
</tr>
<tr style="height: 50px; text-align: center;">
<td style="border:1px solid #000; padding:2px;">贴吧</td>
{% for item in items %}
<td style="border:1px solid #000; padding:2px;">{{ item.priceTieba }}</td>
{% endfor %}
</tr>
</tbody>
</table>
</div>
</div>
templates/qiyuliebiao.html
-34
View File
@@ -1,34 +0,0 @@
<div style="width: 800px; margin: 0 auto;">
<h1 style="text-align: center;" id="剑网三奇遇查询">剑网3奇遇查询</h1>
<p style="text-align: center;">奇遇名称：{{ qiyuname }}　服务器：{{ server }}</p>
<div style="overflow-x: auto; overflow-y: hidden;">
<table style="width: 100%; border-collapse: collapse; margin: 0 auto;">
<thead>
<tr style="height: 50px; text-align: center;">
<th style="border:1px solid #000; padding:4px;">奇遇名称</th>
<th style="border:1px solid #000; padding:4px;">服务器</th>
<th style="border:1px solid #000; padding:4px;">角色名称</th>
<th style="border:1px solid #000; padding:4px;">触发时间</th>
</tr>
</thead>
<tbody>
{% for item in items %}
<tr style="height: 45px; text-align: center;">
<td style="border:1px solid #000; padding:4px;">{{ item.adventureName }}</td>
<td style="border:1px solid #000; padding:4px;">{{ server }}</td>
<td style="border:1px solid #000; padding:4px;">{{ item.gameName }}</td>
<td style="border:1px solid #000; padding:4px;">{{ item.time }}</td>
</tr>
{% endfor %}
</tbody>
</table>
</div>
<!-- 可在这里插入 Base64 图片 -->
<div style="text-align: center; margin-top: 20px;">
<img src="data:image/png;base64, {{ your_base64_string }}"
alt="奇遇趋势图"
style="max-width: 100%; height: auto;" />
</div>
</div>
templates/wujia.html
-65
View File
@@ -1,65 +0,0 @@
<div style="max-width: 800px; margin: 0 auto; font-family: 'Microsoft YaHei', sans-serif; color: #000; text-align: center;">
<!-- 商品基本信息 -->
<h1 style="margin-bottom: 5px;">{{ showName }} ({{ goodsAlias }})</h1>
<img src="{{ imgs }}" alt="{{ showName }}" style="width: 100%; max-width: 600px; margin-bottom: 8px;">
<p style="margin: 4px 0;">{{ goodsDesc }}</p>
<p style="margin: 4px 0;"><strong>发布时间:</strong> {{ publishTime.split('T')[0] if publishTime else '未知' }}</p>
<p style="margin: 4px 0;"><strong>参考价格:</strong> {{ priceNum }} 元</p>
<!-- 双列表格 -->
<div style="display: flex; gap: 10px; margin-top: 15px; flex-wrap: wrap; justify-content: center;">
<!-- 万宝楼公示 -->
<div style="flex: 1; min-width: 300px; border:1px solid #000; overflow:hidden;">
<div style="padding:4px; font-weight:bold; text-align:center;">万宝楼公示</div>
<div style="overflow-x:auto;">
<table style="width:100%; border-collapse:collapse; margin:0 auto; text-align:center; font-size:0.9em;">
<thead>
<tr>
<th style="border:1px solid #000; padding:2px;">剩余时间</th>
<th style="border:1px solid #000; padding:2px;">服务器</th>
<th style="border:1px solid #000; padding:2px;">价格</th>
</tr>
</thead>
<tbody>
{% for item in wblgs %}
<tr>
<td style="border:1px solid #000; padding:2px;">{{ item.replyTime }}</td>
<td style="border:1px solid #000; padding:2px;">{{ item.belongQf2 }}</td>
<td style="border:1px solid #000; padding:2px;">{{ item.priceNum }}元</td>
</tr>
{% endfor %}
</tbody>
</table>
</div>
</div>
<!-- 万宝楼在售 -->
<div style="flex: 1; min-width: 300px; border:1px solid #000; overflow:hidden;">
<div style="padding:4px; font-weight:bold; text-align:center;">万宝楼在售</div>
<div style="overflow-x:auto;">
<table style="width:100%; border-collapse:collapse; margin:0 auto; text-align:center; font-size:0.9em;">
<thead>
<tr>
<th style="border:1px solid #000; padding:2px;">剩余时间</th>
<th style="border:1px solid #000; padding:2px;">服务器</th>
<th style="border:1px solid #000; padding:2px;">价格</th>
</tr>
</thead>
<tbody>
{% for item in wblzs %}
<tr>
<td style="border:1px solid #000; padding:2px;">{{ item.replyTime }}</td>
<td style="border:1px solid #000; padding:2px;">{{ item.belongQf2 }}</td>
<td style="border:1px solid #000; padding:2px;">{{ item.priceNum }}元</td>
</tr>
{% endfor %}
</tbody>
</table>
</div>
</div>
</div>
</div>