11

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,  
+        "code": 0,
-            "msg": "未获取数据",  
+        "msg": "未获取数据",
-            "data": {}  
+        "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"],
     }
     
-    dataaj3 = api_data_post(aj3_url, aj3_params,"data")
+    try:
-    datas["data"]["goodsDesc"] = dataaj3.get("goodsDesc","无描述")
+        # 1. 获取物品ID和名称
-    datas["data"]["publishTime"] = dataaj3.get("publishTime","无数据")
+        idname = sql_data_select(inname)
-    datas["data"]["priceNum"] = dataaj3.get("priceNum",0)
+        if not idname:
-    datas["data"]["imgs"] = dataaj3.get("imgs",[])[0]
+            datas.update({"code": 201, "msg": "未找到该外观信息"})
-    datas["data"]["goodsId"] = dataaj3.get("goodsId","无数据")
+            return datas
-    datas["code"] = 2
-    datas["msg"] = "获取外观数据完成"
             
-    # 查询万宝楼数据
+        # 更新基础信息
-    wbl_url = "https://www.aijx3.cn/api2/aijx3-wblwg/record/queryByCondition"
+        datas.update({
-    wbl_params = {
+            "code": 1,
-        "tradeStatus": "3",
+            "msg": "获取外观名称ID完成",
-        "accoSeq": "",
+            "data": {
-        "orderMode": 1,  # 按时间降序
+                "showName": idname.get("showName", "未知物品"),
-        "orderBy":"price_num",
+                "searchId": idname.get("searchId", "未知ID"),
-        "searchId":[idname['searchId']],
+                "Name": inname
-        "current":1,
+            }
-        "size":10
+        })
-    }
         
-    #第一次查询万宝楼数据，获取公示数据
+        # 2. 查询爱剑三获取外观数据
-    datas["data"]["wblgs"] = []
+        aj3_data = get_aj3_goods_data(datas["data"]["showName"])
-    datawblgs = api_data_post(wbl_url, wbl_params,"data")
+        if aj3_data:
-    #处理万宝楼公示数据
+            datas["data"].update(aj3_data)
-    for record in datawblgs.get("records",[]):
+            datas.update({"code": 2, "msg": "获取外观数据完成"})
-        # 转换时间戳为可读格式
-        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),
-        }
-        datas["data"]["wblgs"].append(record_info)
         
-    #数据更新
+        # 3. 查询万宝楼数据（公示和在售）
-    datas["code"] = 3
+        wbl_data = get_wbl_data(idname['searchId'])
-    datas["msg"] = "获取万宝楼公示数据完成"
+        if wbl_data:
+            datas["data"].update(wbl_data)
+            datas.update({"code": 200, "msg": "获取万宝楼数据完成"})
             
-    #第二次查询万宝楼数据，获取在售数据
+    except Exception as e:
-    wbl_params["tradeStatus"] = "5"
+        logger.error(f"物价查询出错: {e}")
-    datas["data"]["wblzs"] = []
+        datas.update({"code": 500, "msg": f"查询过程中出错: {e}"})
-    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),
-        }
-        datas["data"]["wblzs"].append(record_info)
-
-        #数据更新
-        datas["code"] = 200
-        datas["msg"] = "获取万宝楼在售数据完成"
     
     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

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 = {}
-    params = {
     
-    }
-
-    # 连接数据库
-    connection = pymysql.connect(
-        host='45.205.31.132',      # 数据库主机地址
-        port=5211,                 # 数据库端口
-        user='asrtbot',  # 数据库用户名
-        password='qsc123456',  # 数据库密码
-        database='asrtbot',  # 数据库名
-        charset='utf8mb4'      # 字符编码
-        )
-    
-    # 获取数据
     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
         
+        # 处理数据
+        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")
+                
+                # 准备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}"
+        return 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']
-                ))
-                
-            # 批量插入
-            cursor.executemany(sql, values_list)
-            connection.commit()
-            test = f"成功批量插入 {len(extracted_data)} 条数据！"
-
-    except Exception as e:
-        test = f"插入数据时出错: {e}"
-        connection.rollback()
-
-    finally:
-        connection.close()
-
-    return test
-
 
+# 根据搜索字符串查询匹配的数据
 def sql_data_select(search_string):
+    """
+    根据搜索字符串查询匹配的数据
     
-    # 连接数据库
+    Args:
-    connection = pymysql.connect(
+        search_string: 搜索字符串
-        host='45.205.31.132',      # 数据库主机地址
-        port=5211,                 # 数据库端口
-        user='asrtbot',  # 数据库用户名
-        password='qsc123456',  # 数据库密码
-        database='asrtbot',  # 数据库名
-        charset='utf8mb4'      # 字符编码
-        )
         
-    results = []
+    Returns:
-
+        匹配的第一条记录，如果没有匹配则返回None
-    # 插入数据到数据库
+    """
     try:
-        with connection.cursor() as cursor:
+        with pymysql.connect(**db_config) as connection:
-            # 准备 SQL 插入语句
+            with connection.cursor() as cursor:               
-            sql = """
+                sql = """
-            SELECT searchId, showName 
+                SELECT searchId, showName, name
-            FROM searchdata 
+                FROM searchdata 
-            WHERE name = %s OR showName = %s
+                WHERE name = %s OR showName = %s
-            """
+                LIMIT 1
+                """
+                params = (search_string, search_string)
 
-            # 添加通配符 % 到搜索字符串的两端
+                cursor.execute(sql, params)
-            #search_pattern = f"%{search_string}%"
+                return cursor.fetchone()
-
-            # 执行查询
-            cursor.execute(sql, (search_string, search_string))
-
-            # 获取所有匹配的结果
-            rows = cursor.fetchall()
-
-            # 将结果转换为字典列表
-            for row in rows:
-                results.append({
-                    "searchId": row[0],
-                    "showName": row[1]
-                })
                 
     except Exception as e:
         logger.error(f"查询数据时出错: {e}")
-
+        return None
-    finally:
-        connection.close()
-
-    return results[0]