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

112

Browse Source
This commit is contained in:
qsc20001102
2025-10-03 17:08:11 +08:00
parent 2871a0721e
commit c42f37d46a
5 changed files with 173 additions and 313 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core/cless_jx3.py
+20 -7
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,flatten_field,extract_fields,gold_to_string
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 ):
@@ -141,17 +141,26 @@ class JX3Function:
return return_data
# 加载模板
try:
return_data["temp"] = load_template("jinjia.html")
return_data["temp"] = load_template("temp_test.html")
except FileNotFoundError as e:
logger.error(f"加载模板失败: {e}")
return_data["msg"] = "系统错误：模板文件不存在"
return return_data
try:
chart_base64 = plot_line_chart_base64(data, "date", "priceWanbaolou", "万宝楼金价走势",True)
logger.info(f"生成折线图成功: {chart_base64}")
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")
"update_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
"img": chart_base64
}
except Exception as e:
logger.error(f"处理数据时出错: {e}")
@@ -252,7 +261,8 @@ class JX3Function:
return_data["msg"] = f"成功批量插入 {len(extracted_data)} 条数据！"
return_data["code"] = 200
return return_data
async def wujia(self,Name: str = "秃盒"):
return_data = {
"code": 0,
@@ -338,7 +348,8 @@ class JX3Function:
return return_data
return_data["code"] = 200
return return_data
async def __get_wbl_data(self,search_id):
"""获取万宝楼数据（公示和在售）"""
try:
@@ -360,7 +371,8 @@ class JX3Function:
except Exception as e:
logger.error(f"获取万宝楼数据出错: {e}")
return None
async def __process_wbl_records(self,records):
"""处理万宝楼记录数据"""
processed = []
@@ -381,7 +393,8 @@ class JX3Function:
continue
return processed
async def jiaoyihang(self,Name: str = "守缺式",server: str = "梦江南"):
return_data = {
"code": 0,
core/function_basic.py
+52 -1
View File
@@ -1,5 +1,9 @@
import os
from pathlib import Path
import matplotlib.pyplot as plt
from matplotlib.font_manager import FontProperties
from io import BytesIO
import base64
def load_template(template_name):
"""
@@ -113,4 +117,51 @@ def gold_to_string(gold_amount):
if started:
parts.append(f"{value}{unit}")
return "".join(parts)
return "".join(parts)
def plot_line_chart_base64(data, x_field, y_field, title=None, reverse_x=False):
"""
根据列表数据绘制折线图，返回 base64 图片字符串。
:param data: list[dict] 数据列表
:param x_field: str X轴字段
:param y_field: str Y轴字段
:param title: str 图表标题（可选）
:param reverse_x: bool 是否反转 X 轴方向（默认 False：从左往右；True：从右往左）
:return: str base64 图片字符串，可直接放到 <img src="..."> 中
"""
if not data:
raise ValueError("数据列表不能为空")
# 字体设置（支持中文）
plt.rcParams['font.sans-serif'] = ['SimHei']
plt.rcParams['axes.unicode_minus'] = False
# 提取 X / Y 数据
x_values = [str(item.get(x_field, "")) for item in data]
y_values = [float(item.get(y_field, 0)) for item in data]
# 创建画布
plt.figure(figsize=(8, 5))
plt.plot(x_values, y_values, marker='o', color='#4a90e2', linewidth=2)
# 反转 X 轴
if reverse_x:
plt.gca().invert_xaxis()
# 标题与标签
if title is None:
title = f"{y_field} 折线图"
plt.grid(True, linestyle='--', alpha=0.5)
plt.xticks(rotation=30)
plt.tight_layout()
# 转 Base64
buffer = BytesIO()
plt.savefig(buffer, format='png', dpi=150)
plt.close()
img_base64 = base64.b64encode(buffer.getvalue()).decode('utf-8')
return f"data:image/png;base64,{img_base64}"
core/jx3_data.py
-185
View File
@@ -1,185 +0,0 @@
# core/jx3jiaoyihang.py
from astrbot.api import logger
from urllib.parse import quote
from datetime import datetime
from .class_reqsest import api_data_get, api_data_post
def fetch_all_pages(base_url, initial_params, max_pages=None):
"""
获取所有分页数据
Args:
base_url: API基础URL
initial_params: 初始请求参数
max_pages: 最大页数限制（可选）
Returns:
list: 所有页面的数据列表
"""
all_data = []
current_page = 1
while True:
# 设置当前页码
params = initial_params.copy()
params["page"] = str(current_page)
# 请求数据
data = api_data_get(base_url, params)
if not data or "list" not in data:
break
# 添加当前页数据到总列表
all_data.extend(data["list"])
# 检查是否还有更多页面
total_pages = data.get("pages", 1)
if current_page >= total_pages:
break
# 检查是否达到最大页数限制
if max_pages and current_page >= max_pages:
break
current_page += 1
return all_data
def merge_item_data_by_itemid(price_data, item_list_data):
"""
根据ItemId和id字段将物品名称信息合并到价格数据中
Args:
price_data: 第一组数据，包含价格信息的字典
item_list_data: 第二组数据，可以是字典（包含"list"键）或直接是物品列表
Returns:
dict: 合并后的数据，包含价格和名称信息
"""
# 提取价格数据中的物品信息
price_items = price_data.get("data", {})
# 处理不同类型的item_list_data输入
if isinstance(item_list_data, dict) and "list" in item_list_data:
# 如果是字典且包含"list"键
item_list = item_list_data.get("list", [])
elif isinstance(item_list_data, list):
# 如果直接是列表
item_list = item_list_data
else:
# 其他情况，使用空列表
item_list = []
# 创建一个字典用于快速查找物品名称，使用id作为键
item_name_map = {}
for item in item_list:
item_id = item.get("id")
item_name = item.get("Name")
if item_id and item_name:
item_name_map[item_id] = item_name
# 创建一个新的结果字典
merged_data = {"code": price_data.get("code", 0), "msg": price_data.get("msg", ""), "data": {}}
# 遍历价格数据，添加名称信息
for item_id, price_info in price_items.items():
# 复制价格信息
merged_item = price_info.copy()
# 使用ItemId字段查找对应的名称
item_id_to_match = merged_item.get("ItemId")
if item_id_to_match and item_id_to_match in item_name_map:
merged_item["Name"] = item_name_map[item_id_to_match]
else:
merged_item["Name"] = "未知物品"
# 添加到结果中
merged_data["data"][item_id] = merged_item
return merged_data
#交易行数据查询函数
def jx3_data_jiaoyihang(inserver="眉间雪", inname="武技殊影图"):
"""
获取剑三交易行某区某物品价格数据
Args:
inserver: 第一组数据，服务器名称
inname: 物品名称
Returns:
list: 合并后的数据，包含价格和名称信息
"""
# 第一步：获取所有物品列表数据（处理分页）
custom_url = f"https://node.jx3box.com/item_merged/name/{quote(inname)}"
initial_params = {
"client": "std",
"strict": "0",
"per": "50" # 每页数量，可以根据需要调整
}
# 获取所有页面的物品数据
all_items = fetch_all_pages(custom_url, initial_params)
if not all_items:
#logger.error(f"未找到物品: {inname}")
return "未找到改物品"
# 提取所有ID
ids = [item.get("id") for item in all_items if item.get("id")]
if not ids:
#logger.error(f"物品 {inname} 没有有效的ID")
return "未找到改物品"
# 返回格式化结果
stringids = ",".join(ids)
logger.info(f"搜索到物品: {inname}, 共找到 {len(ids)} 个物品\nIDs: {stringids}")
# 第二步：获取价格数据
price_url = "https://next2.jx3box.com/api/item-price/list"
price_params = {
"server": inserver,
"itemIds": stringids
}
price_data = api_data_get(price_url, price_params)
if not price_data or "data" not in price_data:
return "无交易行数据"
if not price_data["data"] or (isinstance(price_data["data"], dict) and not price_data["data"]) or (isinstance(price_data["data"], list) and not price_data["data"]):
return "无交易行数据"
# 第三步：合并数据
merged_data = merge_item_data_by_itemid(price_data, {"list": all_items})
# 处理合并后的数据
result_items = []
for item_id, item_info in merged_data.get("data", {}).items():
# 格式化价格（假设价格是以铜钱为单位，转换为金）
lowest_price = item_info.get("LowestPrice", 0)
bricks = lowest_price // 100000000 if lowest_price else 0
gold = (lowest_price % 100000000) // 10000 if lowest_price else 0
silver = (lowest_price % 10000) // 100 if lowest_price else 0
copper = lowest_price % 100 if lowest_price else 0
price_str = f"{bricks}砖{gold}金{silver}银{copper}铜" if lowest_price else "无价格"
result_items.append({
"name": item_info.get("Name", "未知物品"),
"price": price_str,
"avg_price": item_info.get("AvgPrice", 0),
"sample_size": item_info.get("SampleSize", 0)
})
# 按价格排序
result_items.sort(key=lambda x: x["avg_price"])
return result_items