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

11

Browse Source
This commit is contained in:
qsc
2025-09-30 02:04:18 +08:00
parent 3959bf4f27
commit d8dac64b1d
6 changed files with 174 additions and 144 deletions
Download Patch File Download Diff File Expand all files Collapse all files
api_config.json
+19
View File
@@ -86,5 +86,24 @@
"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"]
}
}
}
core/class_reqsest.py
+43
View File
@@ -181,6 +181,49 @@ class APIClient:
return data.get(outdata, {})
async def all_pages(self,http, api_url, params_data = None, outdata: str = "",listdata: str = "list", max_pages: int = 10):
"""
分页获取所有数据
Args:
http: 请求方法 ('GET' 或 'POST')
api_url: API地址
params_data: 参数字典，如 {"name": "万花"}
outdata: 返回数据中要提取的字段
max_pages: 最大页数限制，0表示不限制
listdata: 返回数据中包含列表数据的字段名，默认为"list"
Returns:
成功时返回所有页数据的列表，失败返回None
"""
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[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):
"""兼容原有函数的POST请求"""
core/cless_jx3.py
+44 -3
View File
@@ -4,7 +4,7 @@ from astrbot.api import logger
from .class_reqsest import APIClient
from .cless_mysql import AsyncMySQL
from .function_basic import load_template, extract_field
from .function_basic import load_template, extract_field,flatten_field,extract_fields
class JX3Function:
def __init__(self, api_config,db: AsyncMySQL ):
@@ -219,7 +219,7 @@ class JX3Function:
return return_data
# 提取数据
try:
extracted_data = extract_field(data, "dataModels")
extracted_data = flatten_field(data, "dataModels")
except FileNotFoundError as e:
logger.error(f"提取数据失败: {e}")
return_data["msg"] = "提取指定数据字段失败"
@@ -381,4 +381,45 @@ class JX3Function:
logger.error(f"处理万宝楼记录出错: {e}")
continue
return processed
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
return_data["data"]=data
return_data["code"] = 200
return return_data
core/function_basic.py
+49 -9
View File
@@ -24,6 +24,54 @@ def load_template(template_name):
return f.read()
def flatten_field(data_list, field_name):
"""
提取并扁平化指定字段的值
从一个字典列表中，收集指定字段的所有值。
如果字段的值是列表，则会被展开（扁平化）加入结果；
如果字段的值是单个元素，则直接加入结果。
Args:
data_list (list[dict]): 数据列表，每个元素是字典
field_name (str): 要提取的字段名
Returns:
list: 扁平化后的字段值列表
"""
extracted_data = []
for item in data_list:
if field_name in item and item[field_name]:
# 如果是 list，展开
if isinstance(item[field_name], list):
extracted_data.extend(item[field_name])
else:
extracted_data.append(item[field_name])
return extracted_data
def extract_fields(data_list, fields):
"""
从字典列表中提取多个字段
Args:
data_list (list[dict]): 包含字典的列表
fields (list[str]): 要提取的字段名列表
Returns:
list[dict]: 只包含指定字段的新字典列表
"""
result = []
try:
for item in data_list:
extracted = {field: item.get(field) for field in fields}
result.append(extracted)
except Exception as e:
print(f"提取字段时出错: {e}")
return []
return result
def extract_field(data_list, field_name):
"""
从列表中的字典提取指定字段的所有值
@@ -35,12 +83,4 @@ def extract_field(data_list, field_name):
Returns:
list: 提取出来的字段值列表
"""
extracted_data = []
for item in data_list:
if field_name in item and item[field_name]:
# 如果是 list，展开
if isinstance(item[field_name], list):
extracted_data.extend(item[field_name])
else:
extracted_data.append(item[field_name])
return extracted_data
return [item[field_name] for item in data_list if field_name in item]
core/jx3_data.py
+1 -131
View File
@@ -182,134 +182,4 @@ def jx3_data_jiaoyihang(inserver="眉间雪", inname="武技殊影图"):
result_items.sort(key=lambda x: x["avg_price"])
return result_items
def jx3_data_wujia(inname="秃盒"):
"""
获取剑三外观物品价格数据
Args:
inname: 物品名称
Returns:
dict: 包含价格和名称信息的字典
"""
# 初始化返回数据结构
datas = {
"code": 0,
"msg": "未获取数据",
"data": {}
}
try:
# 1. 获取物品ID和名称
idname = api_data_get(inname)
if not idname:
datas.update({"code": 201, "msg": "未找到该外观信息"})
return datas
# 更新基础信息
datas.update({
"code": 1,
"msg": "获取外观名称ID完成",
"data": {
"showName": idname.get("showName", "未知物品"),
"searchId": idname.get("searchId", "未知ID"),
"Name": inname
}
})
# 2. 查询爱剑三获取外观数据
aj3_data = get_aj3_goods_data(datas["data"]["showName"])
if aj3_data:
datas["data"].update(aj3_data)
datas.update({"code": 2, "msg": "获取外观数据完成"})
# 3. 查询万宝楼数据（公示和在售）
wbl_data = get_wbl_data(idname['searchId'])
if wbl_data:
datas["data"].update(wbl_data)
datas.update({"code": 200, "msg": "获取万宝楼数据完成"})
except Exception as e:
logger.error(f"物价查询出错: {e}")
datas.update({"code": 500, "msg": f"查询过程中出错: {e}"})
return datas
def get_aj3_goods_data(goods_name):
"""获取爱剑三商品数据"""
try:
aj3_url = "https://www.aijx3.cn/api2/aijx3-wj/goods/getGoodsDetail"
aj3_params = {"goodsName": goods_name}
dataaj3 = api_data_post(aj3_url, aj3_params, "data")
if not dataaj3:
return None
imgs = dataaj3.get("imgs", [])
return {
"goodsDesc": dataaj3.get("goodsDesc", "无描述"),
"publishTime": dataaj3.get("publishTime", "无数据"),
"priceNum": dataaj3.get("priceNum", 0),
"imgs": imgs[0] if imgs else "",
"goodsId": dataaj3.get("goodsId", "无数据")
}
except Exception as e:
logger.error(f"获取爱剑三数据出错: {e}")
return None
def get_wbl_data(search_id):
"""获取万宝楼数据（公示和在售）"""
try:
wbl_url = "https://www.aijx3.cn/api2/aijx3-wblwg/record/queryByCondition"
base_params = {
"accoSeq": "",
"orderMode": 1,
"orderBy": "price_num",
"searchId": [search_id],
"current": 1,
"size": 10
}
# 获取公示数据
gs_params = base_params.copy()
gs_params["tradeStatus"] = "3"
datawblgs = api_data_post(wbl_url, gs_params, "data")
# 获取在售数据
zs_params = base_params.copy()
zs_params["tradeStatus"] = "5"
datawblzs = api_data_post(wbl_url, zs_params, "data")
return {
"wblgs": process_wbl_records(datawblgs.get("records", [])),
"wblzs": process_wbl_records(datawblzs.get("records", []))
}
except Exception as e:
logger.error(f"获取万宝楼数据出错: {e}")
return None
def process_wbl_records(records):
"""处理万宝楼记录数据"""
processed = []
for record in records:
try:
# 转换时间戳为可读格式
timestamp = record.get("replyTime", 0)
dt = datetime.fromtimestamp(timestamp / 1000) if timestamp else datetime.now()
processed.append({
"priceNum": record.get("priceNum", 0),
"belongQf2": record.get("belongQf2", "无数据"),
"replyTime": dt.strftime("%Y-%m-%d %H:%M:%S"),
"discountRate": record.get("discountRate", 0.0),
})
except Exception as e:
logger.error(f"处理万宝楼记录出错: {e}")
continue
return processed
main.py
+18 -1
View File
@@ -8,7 +8,7 @@ from astrbot.api import logger
from astrbot.api import AstrBotConfig
from .core.function_basic import load_template
from .core.jx3_data import jx3_data_jiaoyihang,jx3_data_wujia
from .core.jx3_data import jx3_data_jiaoyihang
from .core.cless_mysql import AsyncMySQL
from .core.cless_jx3 import JX3Function
@@ -165,6 +165,23 @@ class Jx3ApiPlugin(Star):
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)
yield event.plain_result(f"{data['data']}")
else:
yield event.plain_result(data["msg"])
return
except Exception as e:
logger.error(f"功能函数执行错误: {e}")
yield event.plain_result("猪脑过载，请稍后再试")
@filter.permission_type(filter.PermissionType.ADMIN)
@jx3.command("外观数据同步")
async def jx3_SearchData(self, event: AstrMessageEvent):