From cd49f2b0d09f763034afd8ec88bd9166583674bc Mon Sep 17 00:00:00 2001 From: qsc Date: Fri, 26 Sep 2025 00:40:06 +0800 Subject: [PATCH] 11 --- core/jx3_data.py | 186 ++++++++++++++++++++++++++------------------- core/sql_data.py | 191 ++++++++++++++++++++--------------------------- 2 files changed, 190 insertions(+), 187 deletions(-) diff --git a/core/jx3_data.py b/core/jx3_data.py index 2e88800..cea35c5 100644 --- a/core/jx3_data.py +++ b/core/jx3_data.py @@ -185,7 +185,6 @@ def jx3_data_jiaoyihang(inserver="眉间雪", inname="武技殊影图"): return result_items -#物价查询 def jx3_data_wujia(inname="秃盒"): """ 获取剑三外观物品价格数据 @@ -194,89 +193,124 @@ def jx3_data_wujia(inname="秃盒"): inname: 物品名称 Returns: - list: 合并后的数据，包含价格和名称信息 + dict: 包含价格和名称信息的字典 """ + # 初始化返回数据结构 datas = { - "code": 0, - "msg": "未获取数据", - "data": {} - } - #获取所查询物品的id和官方名称 - idname = sql_data_select(inname) - if not idname: - datas["code"] = 201 - datas["msg"] = "未找到该外观信息" - return - datas["code"] = 1 - datas["msg"] = "获取外观名称ID完成" - datas["data"]["showName"] = idname.get("showName","未知物品") - datas["data"]["searchId"] = idname.get("searchId","未知ID") - datas["data"]["Name"] = inname - - #查询爱剑三获取外观数据 - aj3_url = "https://www.aijx3.cn/api2/aijx3-wj/goods/getGoodsDetail" - aj3_params = { - "goodsName":datas["data"]["showName"], + "code": 0, + "msg": "未获取数据", + "data": {} } - - dataaj3 = api_data_post(aj3_url, aj3_params,"data") - datas["data"]["goodsDesc"] = dataaj3.get("goodsDesc","无描述") - datas["data"]["publishTime"] = dataaj3.get("publishTime","无数据") - datas["data"]["priceNum"] = dataaj3.get("priceNum",0) - datas["data"]["imgs"] = dataaj3.get("imgs",[])[0] - datas["data"]["goodsId"] = dataaj3.get("goodsId","无数据") - datas["code"] = 2 - datas["msg"] = "获取外观数据完成" - # 查询万宝楼数据 - wbl_url = "https://www.aijx3.cn/api2/aijx3-wblwg/record/queryByCondition" - wbl_params = { - "tradeStatus": "3", - "accoSeq": "", - "orderMode": 1, # 按时间降序 - "orderBy":"price_num", - "searchId":[idname['searchId']], - "current":1, - "size":10 - } + try: + # 1. 获取物品ID和名称 + idname = sql_data_select(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 - #第一次查询万宝楼数据，获取公示数据 - datas["data"]["wblgs"] = [] - datawblgs = api_data_post(wbl_url, wbl_params,"data") - #处理万宝楼公示数据 - for record in datawblgs.get("records",[]): - # 转换时间戳为可读格式 - dt = datetime.fromtimestamp(record.get("replyTime",0)/1000) - record_info = { - "priceNum": record.get("priceNum",0), - "belongQf2": record.get("belongQf2","无数据"), - "replyTime": dt.strftime("%Y-%m-%d %H:%M:%S"), - "discountRate": record.get("discountRate",0.0), + +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", "无数据") } - datas["data"]["wblgs"].append(record_info) + except Exception as e: + logger.error(f"获取爱剑三数据出错: {e}") + return None - #数据更新 - datas["code"] = 3 - datas["msg"] = "获取万宝楼公示数据完成" - #第二次查询万宝楼数据，获取在售数据 - wbl_params["tradeStatus"] = "5" - datas["data"]["wblzs"] = [] - datawblzs = api_data_post(wbl_url, wbl_params,"data") - #处理万宝楼在售数据 - for record in datawblzs.get("records",[]): - # 转换时间戳为可读格式 - dt = datetime.fromtimestamp(record.get("replyTime",0)/1000) - record_info = { - "priceNum": record.get("priceNum",0), - "belongQf2": record.get("belongQf2","无数据"), - "replyTime": dt.strftime("%Y-%m-%d %H:%M:%S"), - "discountRate": record.get("discountRate",0.0), +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 } - datas["data"]["wblzs"].append(record_info) + + # 获取公示数据 + 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 - #数据更新 - datas["code"] = 200 - datas["msg"] = "获取万宝楼在售数据完成" - return datas \ No newline at end of file +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 \ No newline at end of file diff --git a/core/sql_data.py b/core/sql_data.py index 694a298..96febca 100644 --- a/core/sql_data.py +++ b/core/sql_data.py @@ -4,6 +4,17 @@ from astrbot.api import logger from .api_data import api_data_get, api_data_post +#连接数据库配置 +db_config = { + 'host': '154.201.70.116', + 'port': 3306, + 'user': 'asrtbot', + 'password': 'qsc123456', + 'database': 'asrtbot', + 'charset': 'utf8mb4', + 'cursorclass': pymysql.cursors.DictCursor #返回字典格式数据 +} + # 提取所有 dataModels 中的数据 def extract_data_models(source_data): extracted_data = [] @@ -12,127 +23,85 @@ def extract_data_models(source_data): extracted_data.extend(category["dataModels"]) return extracted_data - +# 获取并存储搜索数据 def sql_data_searchdata(): - - # 接口URL - custom_url = "https://www.aijx3.cn/api2/aijx3-wblwg/basedata/getSearchData" - # 接口参数 - params = { - - } - - # 连接数据库 - connection = pymysql.connect( - host='45.205.31.132', # 数据库主机地址 - port=5211, # 数据库端口 - user='asrtbot', # 数据库用户名 - password='qsc123456', # 数据库密码 - database='asrtbot', # 数据库名 - charset='utf8mb4' # 字符编码 - ) + # 接口配置 + custom_url = "https://www.aijx3.cn/api2/aijx3-wblwg/basedata/getSearchData" + params = {} - # 获取数据 try: - source_data = api_data_post(custom_url,params,"data") - + # 获取数据 + source_data = api_data_post(custom_url, params, "data") if not source_data: - test = "获取数据失败或数据为空" - return + return "获取数据失败或数据为空" - except Exception as e: - test = f"获取数据时出错: {e}" - - #处理数据 - extracted_data = extract_data_models(source_data) - if not source_data: - test = "未提取到数据" - return - - # 插入数据到数据库 - try: - with connection.cursor() as cursor: - # 清空表数据 - cursor.execute("TRUNCATE TABLE searchdata") - - # 准备 SQL 插入语句 - sql = """ - INSERT INTO searchdata - (typeName, name, showName, picUrl, searchId, searchDescType) - VALUES (%s, %s, %s, %s, %s, %s) - """ - - # 准备批量数据 - values_list = [] - for item in extracted_data: - values_list.append(( - item['typeName'], - item['name'], - item['showName'], - item['picUrl'], - item['searchId'], - item['searchDescType'] - )) + # 处理数据 + extracted_data = extract_data_models(source_data) + if not extracted_data: # 修正：应该是extracted_data而不是source_data + return "未提取到数据" + + # 连接数据库并插入数据 + with pymysql.connect(**db_config) as connection: + with connection.cursor() as cursor: + # 清空表数据 + cursor.execute("TRUNCATE TABLE searchdata") - # 批量插入 - cursor.executemany(sql, values_list) - connection.commit() - test = f"成功批量插入 {len(extracted_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 + ] + + # 批量插入 + cursor.executemany(sql, values_list) + connection.commit() + + return f"成功批量插入 {len(extracted_data)} 条数据！" + + except pymysql.Error as e: + return f"数据库操作失败: {e}" except Exception as e: - test = f"插入数据时出错: {e}" - connection.rollback() - - finally: - connection.close() - - return test - + return f"操作失败: {e}" +# 根据搜索字符串查询匹配的数据 def sql_data_select(search_string): - - # 连接数据库 - connection = pymysql.connect( - host='45.205.31.132', # 数据库主机地址 - port=5211, # 数据库端口 - user='asrtbot', # 数据库用户名 - password='qsc123456', # 数据库密码 - database='asrtbot', # 数据库名 - charset='utf8mb4' # 字符编码 - ) - - results = [] - - # 插入数据到数据库 + """ + 根据搜索字符串查询匹配的数据 + + Args: + search_string: 搜索字符串 + + Returns: + 匹配的第一条记录，如果没有匹配则返回None + """ try: - with connection.cursor() as cursor: - # 准备 SQL 插入语句 - sql = """ - SELECT searchId, showName - FROM searchdata - WHERE name = %s OR showName = %s - """ - - # 添加通配符 % 到搜索字符串的两端 - #search_pattern = f"%{search_string}%" - - # 执行查询 - cursor.execute(sql, (search_string, search_string)) - - # 获取所有匹配的结果 - rows = cursor.fetchall() - - # 将结果转换为字典列表 - for row in rows: - results.append({ - "searchId": row[0], - "showName": row[1] - }) + with pymysql.connect(**db_config) as connection: + with connection.cursor() as cursor: + sql = """ + SELECT searchId, showName, name + FROM searchdata + WHERE name = %s OR showName = %s + LIMIT 1 + """ + params = (search_string, search_string) + cursor.execute(sql, params) + return cursor.fetchone() + except Exception as e: logger.error(f"查询数据时出错: {e}") - - finally: - connection.close() - - return results[0] \ No newline at end of file + return None \ No newline at end of file