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

ok

Browse Source
This commit is contained in:
qsc20001102
2025-09-22 18:04:12 +08:00
parent f2a1eb90f6
commit ed82671d48
3 changed files with 68 additions and 75 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core/api_data.py
+31 -17
View File
@@ -3,33 +3,37 @@ import requests
import json import json
from astrbot.api import logger from astrbot.api import logger
def api_data(api_url, json_data=None, outdata="data"): def api_data_post(api_url, json_data=None, outdata=None):
""" """
获取数据的POST请求函数 获取数据的POST请求函数
Args: Args:
api_url: API地址 api_url: API地址
json_data: POST请求的JSON数据 json_data: POST请求的JSON数据
outdata: 返回数据中要提取的字段，默认为"data" outdata: 返回数据中要提取的字段
Returns: Returns:
成功时返回outdata字段的数据，失败返回None 成功时返回outdata字段的数据，失败返回None
""" """
try: try:
response = requests.post( response = requests.post(api_url, json=json_data, timeout=10, verify=False)
api_url,
json=json_data, # JSON body数据
timeout=10,
verify=False
)
response.raise_for_status() response.raise_for_status()
data = response.json() data = response.json()
if data.get('code') not in [200, "0"]: # 检查是否有code字段
logger.error(f"API返回错误: {data.get('msg', '未知错误')}") if 'code' in data:
# 有code字段时，检查是否成功
if data.get('code') not in [200, "0", 0]:
logger.error(f"API返回错误:{data.get('code', '未知状态')} {data.get('msg', '未知错误')}")
return None
else:
# 无code字段时，检查数据是否为空
if not data:
logger.error("API返回空数据")
return None return None
if outdata is None or outdata == "":
return data
return data.get(outdata, {}) return data.get(outdata, {})
except requests.exceptions.RequestException as e: except requests.exceptions.RequestException as e:
@@ -39,27 +43,37 @@ def api_data(api_url, json_data=None, outdata="data"):
logger.error(f"JSON解析错误: {e}") logger.error(f"JSON解析错误: {e}")
return None return None
def fetch_jx3_data(api_url=None,outdata="data", **params): def api_data_get(api_url,params=None,outdata=None):
""" """
获取数据的GET请求函数 获取数据的GET请求函数
Args: Args:
api_url: API地址 api_url: API地址
outdata: 返回数据中要提取的字段，默认为"data"
**params: 其他查询参数 params: 其他查询参数
Returns: Returns:
成功时返回outdata字段的数据，失败返回None 成功时返回outdata字段的数据，失败返回None
""" """
try: try:
response = requests.get(api_url, params=params, timeout=10, verify=False) response = requests.get(api_url, params, timeout=10, verify=False)
response.raise_for_status() response.raise_for_status()
data = response.json() data = response.json()
if data.get('code') != 200: # 检查是否有code字段
logger.error(f"API返回错误: {data.get('msg', '未知错误')}") if 'code' in data:
# 有code字段时，检查是否成功
if data.get('code') not in [200, "0", 0]:
logger.error(f"API返回错误:{data.get('code', '未知状态')} {data.get('msg', '未知错误')}")
return None
else:
# 无code字段时，检查数据是否为空
if not data:
logger.error("API返回空数据")
return None return None
if outdata is None or outdata == "":
return data
return data.get(outdata, {}) return data.get(outdata, {})
except requests.exceptions.RequestException as e: except requests.exceptions.RequestException as e:
core/jx3jiaoyihang.py → core/jx3_data.py
+4 -22
View File
@@ -1,26 +1,8 @@
# core/jx3jiaoyihang.py # core/jx3jiaoyihang.py
import requests
import json
from astrbot.api import logger from astrbot.api import logger
from urllib.parse import quote from urllib.parse import quote
from .api_data import api_data_get, api_data_post
#接口函数的实现
def fetch_jx3_data(api_url=None, **params):
# 函数实现
try:
response = requests.get(api_url, params=params, timeout=10, verify=False)
response.raise_for_status()
data = response.json()
return data
except requests.exceptions.RequestException as e:
logger.error(f"请求出错: {e}")
return None
except json.JSONDecodeError as e:
logger.error(f"JSON解析错误: {e}")
return None
def fetch_all_pages(base_url, initial_params, max_pages=None): def fetch_all_pages(base_url, initial_params, max_pages=None):
@@ -44,7 +26,7 @@ def fetch_all_pages(base_url, initial_params, max_pages=None):
params["page"] = str(current_page) params["page"] = str(current_page)
# 请求数据 # 请求数据
data = fetch_jx3_data(base_url, **params) data = api_data_get(base_url, params)
if not data or "list" not in data: if not data or "list" not in data:
break break
@@ -121,7 +103,7 @@ def merge_item_data_by_itemid(price_data, item_list_data):
#交易行数据查询函数 #交易行数据查询函数
def fetch_jx3_jiaoyihang(inserver="眉间雪", inname="武技殊影图"): def jx3_data_jiaoyihang(inserver="眉间雪", inname="武技殊影图"):
# 第一步：获取所有物品列表数据（处理分页） # 第一步：获取所有物品列表数据（处理分页）
custom_url = f"https://node.jx3box.com/item_merged/name/{quote(inname)}" custom_url = f"https://node.jx3box.com/item_merged/name/{quote(inname)}"
@@ -156,7 +138,7 @@ def fetch_jx3_jiaoyihang(inserver="眉间雪", inname="武技殊影图"):
"itemIds": stringids "itemIds": stringids
} }
price_data = fetch_jx3_data(price_url, **price_params) price_data = api_data_get(price_url, price_params)
if not price_data or "data" not in price_data: if not price_data or "data" not in price_data:
return "无交易行数据" return "无交易行数据"
main.py
+29 -32
View File
@@ -5,9 +5,9 @@ from astrbot.api.star import Context, Star, register
from astrbot.api import logger from astrbot.api import logger
from astrbot.api import AstrBotConfig from astrbot.api import AstrBotConfig
from .core.jx3jiaoyihang import fetch_jx3_jiaoyihang
from .core.load_template import load_template from .core.load_template import load_template
from .core.api_data import api_data, fetch_jx3_data from .core.jx3_data import jx3_data_jiaoyihang
from .core.api_data import api_data_get, api_data_post
# 禁用 SSL 警告 # 禁用 SSL 警告
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
@@ -38,7 +38,6 @@ class Jx3ApiPlugin(Star):
# 获取消息内容 # 获取消息内容
message_str = event.message_str.strip() message_str = event.message_str.strip()
parts = message_str.split() parts = message_str.split()
# 解析消息内容 # 解析消息内容
if len(parts) > 1: if len(parts) > 1:
params["server"] = parts[1] # 第二个参数为服务器 params["server"] = parts[1] # 第二个参数为服务器
@@ -48,14 +47,15 @@ class Jx3ApiPlugin(Star):
params["num"] = int(parts[2]) # 第三个参数为日期偏移 params["num"] = int(parts[2]) # 第三个参数为日期偏移
except ValueError: except ValueError:
params["num"] = 0 # 参数为日期偏移 params["num"] = 0 # 参数为日期偏移
# 获取数据 # 获取数据
data = fetch_jx3_data(custom_url, **params) data = api_data_get(custom_url, params,"data")
if not data: if not data:
yield event.plain_result("获取获取接口信息失败，请稍后再试") yield event.plain_result("获取获取接口信息失败，请稍后再试")
return return
# 格式化返回消息 # 处理返回数据
try: try:
# 构建回复消息 # 构建回复消息
result_msg = f"{params['server']}-{data.get('date')}-星期{data.get('week')}\n" result_msg = f"{params['server']}-{data.get('date')}-星期{data.get('week')}\n"
@@ -91,13 +91,13 @@ class Jx3ApiPlugin(Star):
# 解析消息内容 # 解析消息内容
# 获取数据 # 获取数据
data = fetch_jx3_data(custom_url, **params) data = api_data_get(custom_url, params,"data")
if not data: if not data:
yield event.plain_result("获取获取接口信息失败，请稍后再试") yield event.plain_result("获取获取接口信息失败，请稍后再试")
return return
# 格式化返回消息 # 处理返回数据
try: try:
# 构建回复消息 # 构建回复消息
result_msg = f"{data.get('text')}\n" result_msg = f"{data.get('text')}\n"
@@ -124,13 +124,13 @@ class Jx3ApiPlugin(Star):
# 解析消息内容 # 解析消息内容
# 获取数据 # 获取数据
data = fetch_jx3_data(custom_url,**params) data = api_data_get(custom_url,params,"data")
if not data: if not data:
yield event.plain_result("获取获取接口信息失败，请稍后再试") yield event.plain_result("获取获取接口信息失败，请稍后再试")
return return
# 格式化返回消息 # 处理返回数据
try: try:
# 构建回复消息 # 构建回复消息
result_msg = f"剑网三最近技改\n" result_msg = f"剑网三最近技改\n"
@@ -147,8 +147,8 @@ class Jx3ApiPlugin(Star):
@filter.command("剑三交易行") @filter.command("剑三交易行")
async def jx3_jiaoyihang(self, event: AstrMessageEvent): async def jx3_data_jiaoyihang(self, event: AstrMessageEvent):
"""剑三交易行 服务器 物品名称""" """剑三交易行 物品名称 服务器"""
# 接口参数 # 接口参数
params = { params = {
@@ -162,14 +162,14 @@ class Jx3ApiPlugin(Star):
# 解析消息内容 # 解析消息内容
if len(parts) > 1: if len(parts) > 1:
params["server"] = parts[1] # 第二个参数为服务器 params["name"] = parts[1]
if len(parts) > 2: if len(parts) > 2:
params["name"] = parts[2] # 第三个参数为日期偏移 params["server"] = parts[2]
# 获取交易行数据 # 获取交易行数据
try: try:
items_data = fetch_jx3_jiaoyihang(params["server"], params["name"]) items_data = jx3_data_jiaoyihang(params["server"], params["name"])
if not items_data or items_data == "无交易行数据": if not items_data or items_data == "无交易行数据":
yield event.plain_result(f"在服务器【{params['server']}】未找到物品【{params['name']}】的交易行数据") yield event.plain_result(f"在服务器【{params['server']}】未找到物品【{params['name']}】的交易行数据")
@@ -194,11 +194,7 @@ class Jx3ApiPlugin(Star):
"update_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S") "update_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
} }
# 调整渲染图片的尺寸 url = await self.html_render(template_content, render_data, options={})
options = {
}
url = await self.html_render(template_content, render_data, options)
yield event.image_result(url) yield event.image_result(url)
except Exception as e: except Exception as e:
@@ -215,20 +211,22 @@ class Jx3ApiPlugin(Star):
params = { params = {
"serverName": "眉间雪" # 默认服务器 "serverName": "眉间雪" # 默认服务器
} }
# 获取消息内容 # 获取消息内容
message_str = event.message_str.strip() message_str = event.message_str.strip()
parts = message_str.split() parts = message_str.split()
# 解析消息内容 # 解析消息内容
if len(parts) > 1: if len(parts) > 1:
params["serverName"] = parts[1] # 第二个参数为服务器 params["serverName"] = parts[1] # 第二个参数为服务器
# 获取数据 # 获取数据
data = api_data(custom_url,params) data = api_data_post(custom_url,params,"data")
if not data: if not data:
yield event.plain_result("获取获取接口信息失败，请稍后再试") yield event.plain_result("获取获取接口信息失败，请稍后再试")
return return
# 格式化返回消息 # 处理返回数据
try: try:
# 构建回复消息 # 构建回复消息
yield event.image_result(data.get("picUrl")) yield event.image_result(data.get("picUrl"))
@@ -237,6 +235,7 @@ class Jx3ApiPlugin(Star):
logger.error(f"处理数据时出错: {e}") logger.error(f"处理数据时出错: {e}")
yield event.plain_result("处理接口返回信息时出错") yield event.plain_result("处理接口返回信息时出错")
@filter.command("剑三金价") @filter.command("剑三金价")
async def jx3_jinjia(self, event: AstrMessageEvent): async def jx3_jinjia(self, event: AstrMessageEvent):
"""剑三金价 服务器""" """剑三金价 服务器"""
@@ -246,20 +245,22 @@ class Jx3ApiPlugin(Star):
params = { params = {
"serverName": "眉间雪" # 默认服务器 "serverName": "眉间雪" # 默认服务器
} }
# 获取消息内容 # 获取消息内容
message_str = event.message_str.strip() message_str = event.message_str.strip()
parts = message_str.split() parts = message_str.split()
# 解析消息内容 # 解析消息内容
if len(parts) > 1: if len(parts) > 1:
params["serverName"] = parts[1] # 服务器 params["serverName"] = parts[1] # 服务器
# 获取数据 # 获取数据
data = api_data(custom_url,params) data = api_data_post(custom_url,params,"data")
if not data: if not data:
yield event.plain_result("获取获取接口信息失败，请稍后再试") yield event.plain_result("获取获取接口信息失败，请稍后再试")
return return
# 格式化返回消息 # 处理返回数据
try: try:
# 加载模板 # 加载模板
try: try:
@@ -274,11 +275,8 @@ class Jx3ApiPlugin(Star):
"server": params["serverName"], "server": params["serverName"],
"update_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S") "update_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
} }
# 调整渲染图片的尺寸
options = {
}
url = await self.html_render(template_content, render_data, options) url = await self.html_render(template_content, render_data, options={})
yield event.image_result(url) yield event.image_result(url)
except Exception as e: except Exception as e:
@@ -296,6 +294,7 @@ class Jx3ApiPlugin(Star):
"adventureName": "阴阳两界", # 默奇遇 "adventureName": "阴阳两界", # 默奇遇
"serverName": "眉间雪" # 默认服务器 "serverName": "眉间雪" # 默认服务器
} }
# 获取消息内容 # 获取消息内容
message_str = event.message_str.strip() message_str = event.message_str.strip()
parts = message_str.split() parts = message_str.split()
@@ -304,8 +303,9 @@ class Jx3ApiPlugin(Star):
params["adventureName"] = parts[1] # 奇遇名称 params["adventureName"] = parts[1] # 奇遇名称
if len(parts) > 2: if len(parts) > 2:
params["serverName"] = parts[2] # 第二个参数为服务器 params["serverName"] = parts[2] # 第二个参数为服务器
# 获取数据 # 获取数据
data = api_data(custom_url,params) data = api_data_post(custom_url,params,"data")
if not data: if not data:
yield event.plain_result("获取获取接口信息失败，请稍后再试") yield event.plain_result("获取获取接口信息失败，请稍后再试")
@@ -334,11 +334,8 @@ class Jx3ApiPlugin(Star):
"update_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S") "update_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
,"qiyuname": params["adventureName"] ,"qiyuname": params["adventureName"]
} }
# 调整渲染图片的尺寸
options = {
}
url = await self.html_render(template_content, render_data, options) url = await self.html_render(template_content, render_data, options={})
yield event.image_result(url) yield event.image_result(url)
except Exception as e: except Exception as e: