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-12-24 23:58:50 +08:00
parent 40e0ad8fca
commit dbe2aa489c
4 changed files with 127 additions and 175 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core/jx3_service.py
+71 -54
View File
@@ -9,9 +9,8 @@ from .aiosqlite import AsyncSQLite
from .function_basic import load_template,flatten_field,extract_fields,gold_to_string
class JX3Service:
def __init__(self, api_config,db: AsyncSQLite,config:AstrBotConfig):
def __init__(self, api_config, config:AstrBotConfig):
self._api = APIClient()
self._db = db
# 获取API配置文件
self._api_config = api_config
# 获取插件配置文件
@@ -176,6 +175,7 @@ class JX3Service:
return return_data
async def kaifu(self, server: str) -> Dict[str, Any]:
"""开服状态查询"""
return_data = self._init_return_data()
@@ -215,6 +215,7 @@ class JX3Service:
return return_data
async def shaohua(self) -> Dict[str, Any]:
"""骚话"""
return_data = self._init_return_data()
@@ -373,58 +374,6 @@ class JX3Service:
return_data["msg"] = "系统错误：模板渲染数据准备失败"
return return_data
async def SearchData(self) -> Dict[str, Any]:
"""外观数据插入数据库"""
return_data = self._init_return_data()
# 获取数据
data: Optional[List[Dict[str, Any]]] = await self._base_request("aijx3_SearchData", "POST")
if not data:
return_data["msg"] = "获取接口信息失败"
return return_data
# 提取数据
try:
extracted_data = flatten_field(data, "dataModels")
except Exception as e:
logger.error(f"提取数据失败: {e}")
return_data["msg"] = "提取指定数据字段失败"
return return_data
# 准备SQL和批量数据
sql = """
INSERT INTO searchdata
(typeName, name, showName, picUrl, searchId, searchDescType)
VALUES (?, ?, ?, ?, ?, ?)
"""
# 增强数据提取的安全性
values_list = [
(
item.get('typeName'),
item.get('name'),
item.get('showName'),
item.get('picUrl'),
item.get('searchId'),
item.get('searchDescType')
)
for item in extracted_data
]
# 插入数据
try:
await self._db.clear_table("searchdata")
await self._db.executemany(sql, values_list)
except Exception as e:
logger.error(f"数据插入失败: {e}")
return_data["msg"] = "数据插入数据库失败"
return return_data
return_data["msg"] = f"成功批量插入 {len(values_list)} 条数据！"
return_data["code"] = 200
return return_data
async def wujia(self, Name: str, server:str) -> Dict[str, Any]:
@@ -778,4 +727,72 @@ class JX3Service:
return_data["data"]["list"] = data
return_data["code"] = 200
return return_data
async def fuyaojjiutian(self, server: str) -> Dict[str, Any]:
"""扶摇九天"""
return_data = self._init_return_data()
# 1. 构造请求参数
params = {"server": server, "token": self.token}
# 2. 调用基础请求
data: Optional[Dict[str, Any]] = await self._base_request(
"jx3_fuyaojiutian", "GET", params=params
)
if not data:
return_data["msg"] = "获取接口信息失败"
return return_data
# 3. 处理返回数据
try:
data_new = data[0]
data_old = data[1]
result_msg = f"{server}\n"
result_msg += f"上次[扶摇九天]开启时间：\n{datetime.fromtimestamp(data_old['time']).strftime('%Y-%m-%d %H:%M:%S')}\n"
result_msg += f"下次[扶摇九天]开启时间：\n{datetime.fromtimestamp(data_new['time']).strftime('%Y-%m-%d %H:%M:%S')}"
except Exception as e:
logger.error(f"处理返回数据失败: {e}")
return_data["msg"] = "处理返回数据失败"
return_data["data"] = result_msg
return_data["code"] = 200
return return_data
async def shuma(self, server: str) -> Dict[str, Any]:
"""刷马"""
return_data = self._init_return_data()
# 1. 构造请求参数
params = {"server": server, "token": self.token}
# 2. 调用基础请求
data: Optional[Dict[str, Any]] = await self._base_request(
"jx3_shuama", "GET", params=params, out_key=""
)
if not data:
return_data["msg"] = "获取接口信息失败"
return return_data
# 3. 处理返回数据
try:
_data = data["data"]["data"]
result_msg = f"{server}\n"
result_msg += f"黑戈壁：\n{_data['黑戈壁'][0]}\n"
result_msg += f"阴山大草原：\n{_data['阴山大草原'][0]}\n"
result_msg += f"鲲鹏岛：\n{_data['鲲鹏岛'][0]}\n"
result_msg += f"的卢:\n{_data['龙泉府 / 进图（21:10）'][0]}\n"
result_msg += f"赤兔:\n{data['data']['note']}"
except Exception as e:
logger.error(f"处理返回数据失败: {e}")
return_data["msg"] = "处理返回数据失败"
return_data["data"] = result_msg
return_data["code"] = 200
return return_data
data/api_config.json
+19 -99
View File
@@ -39,14 +39,6 @@
"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",
@@ -56,96 +48,6 @@
"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",
"description":"王者荣耀战绩查询",
"params":{
"key": "vBpEzoiC9z5A9c9Nn83IhLn6M9",
"id": "489048724",
"option": "1"
}
},
"wzry_ziliao":{
"url":"https://api.t1qq.com/api/tool/wzrr/ydtp",
"method":"GET",
"description":"王者荣耀资料查询",
"params":{
"key": "vBpEzoiC9z5A9c9Nn83IhLn6M9",
"id": "489048724"
}
},
"jx3_jieshemingpian":{
"url":"https://www.jx3api.com/data/show/card",
"method":"GET",
@@ -225,7 +127,7 @@
"server": "梦江南",
"name": "飞翔大野猪",
"mode": "33",
"ticket": "a7fd656a52d747d7830cfa6d686e6030:qsc20001102:kingsoft::sVpKi7nFKzFlbTwtBMTg2w==",
"ticket": "",
"token":""
}
},
@@ -275,5 +177,23 @@
"name": "玄晶",
"token": ""
}
},
"jx3_fuyaojiutian":{
"url":"https://www.jx3api.com/data/active/next/event",
"method":"GET",
"description":"获取某区服扶摇九天的下次开启时间",
"params":{
"server": "梦江南",
"token": ""
}
},
"jx3_shuama":{
"url":"https://www.jx3api.com/data/horse/ranch",
"method":"GET",
"description":"获取某区服坐骑刷新时间",
"params":{
"server": "梦江南",
"token": ""
}
}
}
data/local_data.db
BIN
View File
Binary file not shown.
main.py
+37 -22
View File
@@ -61,16 +61,17 @@ class Jx3ApiPlugin(Star):
async def initialize(self):
"""可选择实现异步的插件初始化方法，当实例化该插件类之后会自动调用该方法。"""
#创建类实例
self.db = AsyncSQLite(str(self.file_local_data))
self.jx3fun = JX3Service(self.api_config,self.db,self.conf)
self.at = AsyncTask(self.context, self.conf, self.jx3fun)
# self.db = AsyncSQLite(str(self.file_local_data))
try:
self.jx3fun = JX3Service(self.api_config,self.conf)
self.at = AsyncTask(self.context, self.conf, self.jx3fun)
except Exception as e:
logger.error(f"功能示例初始化失败: {e}")
return
# 周期函数调用
self.kf_task = asyncio.create_task(self.at.cycle_kfjk())
self.xw_task = asyncio.create_task(self.at.cycle_xwzx())
# 初始化函数
#data=await self.jx3fun.SearchData()
#logger.info(data["msg"])
logger.info("jx3api异步插件初始化完成")
@@ -290,22 +291,6 @@ class Jx3ApiPlugin(Star):
yield event.plain_result(return_msg)
#@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("猪脑过载，请稍后再试")
@jx3.command("名片")
async def jx3_jueshemingpian(self, event: AstrMessageEvent, name: str = "飞翔大野猪", server: str = None):
"""剑三 名片 角色 服务器"""
@@ -442,6 +427,36 @@ class Jx3ApiPlugin(Star):
yield event.plain_result("猪脑过载，请稍后再试")
@jx3.command("扶摇九天")
async def jx3_fuyaojjiutian(self, event: AstrMessageEvent,server: str = None):
"""剑三 日常 服务器 天数"""
try:
data= await self.jx3fun.fuyaojjiutian(await self.serverdefault(server))
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_shuma(self, event: AstrMessageEvent,server: str = None):
"""剑三 日常 服务器 天数"""
try:
data= await self.jx3fun.shuma(await self.serverdefault(server))
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("猪脑过载，请稍后再试")
async def terminate(self):
"""可选择实现异步的插件销毁方法，当插件被卸载/停用时会调用。"""
# 关闭数据库连接