之前存在106.52上的仓库被重新初始化了，同时增加了测试账号的兼容,测试账号hxbtest001,...hxbtest005

api/apps/sdk/dale_extra.py

@@ -286,21 +286,26 @@ def upload_file(tenant_id,mesum_id):
 
         if parsed_json_res.get('success') is True:
             parsed_json_data = parsed_json_res.get('data')
-            matchedAntiqueArray = parsed_json_data.get('antique').split(';') # 识别出的文物的数组，中间以';'分割，可能有多个
-            if len(matchedAntiqueArray) ==1: # 只有一个匹配项，直接返回
+            matchedAntiqueLabelArray = parsed_json_data.get('antique').split(';') # 识别出的文物的数组，中间以';'分割，可能有多个
+            if len(matchedAntiqueLabelArray) ==1: # 只有一个匹配项
                 logging.info(f"识别完成 得到1个，{parsed_json_data.get('antique')} {labels_with_id} ")
                 for item in labels_with_id:
-                    if item['label'] == parsed_json_data.get('antique'):
+                    if is_base_same(item['label'],parsed_json_data.get('antique')):
                         parsed_json_data['id'] = item.get('id')
+                        antique = {'label': item.get('label')+' '+item.get('category',''),'id':item.get('id')}
+                        matchedArray.append(antique)
+                if len(matchedArray) > 1:
+                    parsed_json_data['matchedArray'] = matchedArray
             else: # 有多个匹配项，需要进行多个匹配
-                for label in matchedAntiqueArray[:5]:
+                for label in matchedAntiqueLabelArray[:5]:
                     antique= {'label':label}
                     for item in labels_with_id:
                         if item['label'] == label:
                             antique['id'] = item.get('id')
                     matchedArray.append(antique)
-            if len(matchedArray) > 0:
-                parsed_json_data['matchedArray'] = matchedArray
+
+                if len(matchedArray) > 0:
+                    parsed_json_data['matchedArray'] = matchedArray
         logging.info(f"{parsed_json_data}")
         return jsonify({'message': 'File uploaded successfully','text': message.content,
                         'data': parsed_json_data}), 200
@@ -888,6 +893,9 @@ def list_objects(tenant_id,bucket: str, prefix: str = "", recursive: bool = True
     except Exception as e:
         return get_error_data_result(message=f"minio put list objects error {e}")
 
+@manager.route('/get_v1_uuid', methods=['GET'])
+def get_v1_uuid():
+    return get_result(data={"uuid":get_uuid()})
 #------------------------------------------------
 def audio_fade_in(audio_data, fade_length):
     # 假设音频数据是16位单声道PCM
@@ -915,4 +923,28 @@ def parse_markdown_json(json_string):
             # 如果解析失败，返回错误信息
             return {'success': False, 'data': str(e)}
     else:
-        return {'success': False, 'data': 'not a valid markdown json string'}
+        return {'success': False, 'data': 'not a valid markdown json string'}
+
+
+def is_base_same(a, b):
+    """
+    判断两个字符串是否除了尾部数字外完全相同
+
+    参数:
+        a, b: 要比较的两个字符串
+
+    返回:
+        bool: 如果基础部分相同返回True，否则返回False
+        ("陶觚", "陶觚1"),   # 相同
+        ("陶觚1", "陶觚2"),  # 相同
+        ("陶觚", "陶觚12"),  # 相同
+    """
+    # 创建正则表达式模式：匹配基础部分（忽略尾部数字）
+    pattern = re.compile(r'^(.*?)(?:\d*)$')
+
+    # 提取两个字符串的基础部分
+    base_a = pattern.match(a).group(1)
+    base_b = pattern.match(b).group(1)
+
+    # 比较基础部分是否相同
+    return base_a == base_b

api/apps/sdk/session.py

@@ -115,7 +115,7 @@ def update(tenant_id, chat_id, session_id):
 @token_required
 def completion(tenant_id, chat_id):  # chat_id  和 别的文件中的dialog_id  应该是一个意思? cyx 2025-01-25
     req = request.json
-    logging.info(f"/chats/{chat_id}/completions--0 req={req}")  # cyx
+    #logging.info(f"/chats/{chat_id}/completions--0 req={req}")  # cyx
     if not req.get("session_id"):  # session_id  和 别的文件中的conversation_id  应该是一个意思? cyx 2025-01-25
         conv = {
             "id": get_uuid(),
@@ -130,6 +130,17 @@ def completion(tenant_id, chat_id):  # chat_id  和 别的文件中的dialog_id
         session_id = conv.id
     else:
         session_id = req.get("session_id")
+        # 2025 0423 cyx 修改，前端传入的session_id ，但数据库库中可能不存在，需要创建
+        conv_exist = ConversationService.query(id=session_id, dialog_id=chat_id)
+        if not conv_exist: # session_id 的对话在数据库中不存在 # 当 conv_exist 为 None、空列表 []、空元组 () 等时进入此分支
+            conv = {
+                "id": session_id,
+                "dialog_id": chat_id,
+                "name": req.get("name", "New session"),
+                "message": [{"role": "assistant", "content": "Hi! I am your assistant，can I help you?"}]
+            }
+            ConversationService.save(**conv)
+        #-------------------------------------
     if not req.get("question"):
         return get_error_data_result(message="Please input your question.")
 
@@ -148,6 +159,7 @@ def completion(tenant_id, chat_id):  # chat_id  和 别的文件中的dialog_id
     if not conv:
         return get_error_data_result(message="Session does not exist")
     conv = conv[0]
+    #logging.info(f"/chats/{chat_id}/completions--4 history_limit={history_limit}  conv={conv.message}")  # cyx
     if not DialogService.query(id=chat_id, tenant_id=tenant_id, status=StatusEnum.VALID.value):
         return get_error_data_result(message="You do not own the chat")
     msg = []
@@ -172,7 +184,6 @@ def completion(tenant_id, chat_id):  # chat_id  和 别的文件中的dialog_id
                 current_assistant_count += 1
                 continue
         msg.append(m)
-
     message_id = msg[-1].get("id")
     e, dia = DialogService.get_by_id(conv.dialog_id)
     logging.info(f"/chats/{chat_id}/completions req={req}--dale --2 history_limit={history_limit} dia {dia}")  # cyx

api/db/services/antique_service.py

@@ -48,8 +48,8 @@ class MesumAntiqueService(CommonService):
                 # 统一替换中文分号为英文分号，并去除末尾分号
                 if categories_text:
                     categories_text = categories_text.replace("；", ";").rstrip(";")
-                # 分割并清理空格/空值
-                mesum_antique_categories = [dynasty.strip() for dynasty in categories_text.split(";") if dynasty.strip()]
+                    # 分割并清理空格/空值
+                    mesum_antique_categories = [dynasty.strip() for dynasty in categories_text.split(";") if dynasty.strip()]
 
         finally:
             pass
@@ -115,7 +115,8 @@ class MesumAntiqueService(CommonService):
         for obj in query.dicts():
             labels_data.append({
                 'id': obj['id'],
-                'label': obj['label']
+                'label': obj['label'],
+                "category": obj['category'],
             })
 
         return labels_data

api/db/services/dialog_service.py

@@ -625,7 +625,7 @@ def chat(dialog, messages, stream=True, **kwargs):
     fid = None
     llm_id = tmp[0]
     if len(tmp)>1: fid = tmp[1]
-
+    #logging.info(f"dialog_service--0 message={messages}")  # cyx
     llm = LLMService.query(llm_name=llm_id) if not fid else LLMService.query(llm_name=llm_id, fid=fid)
     if not llm:
         llm = TenantLLMService.query(tenant_id=dialog.tenant_id, llm_name=llm_id) if not fid else \
@@ -716,8 +716,8 @@ def chat(dialog, messages, stream=True, **kwargs):
                                         top=dialog.top_k, aggs=False, rerank_mdl=rerank_mdl)
     knowledges = [ck["content_with_weight"] for ck in kbinfos["chunks"]]
     logging.debug( "{}->{}".format(" ".join(questions), "\n->".join(knowledges)))
-    # 打印历史记录
-    # logging.info( "dale-----!!!:{}->{}".format(" ".join(questions), "\n->".join(knowledges)))
+    # 打印查询到的知识库信息
+    #logging.info( "知识库中知识--!!!:{}->{}".format(" ".join(questions), "\n->".join(knowledges)))
     retrieval_tm = timer()
 
     if not knowledges and prompt_config.get("empty_response"):
@@ -736,7 +736,7 @@ def chat(dialog, messages, stream=True, **kwargs):
     assert len(msg) >= 2, f"message_fit_in has bug: {msg}"
     prompt = msg[0]["content"]
     prompt += "\n\n### Query:\n%s" % " ".join(questions)
-
+    #logging.info(f"dialog_service--3 chat msg={msg}")  # cyx
     if "max_tokens" in gen_conf:
         gen_conf["max_tokens"] = min(
             gen_conf["max_tokens"],

asr-monitor-test/app/database.py

@@ -365,9 +365,9 @@ def create_user(data: dict):
     sql = """
     INSERT INTO rag_flow.users_info 
     (user_id, openid, phone, email, token, balance, status, 
-    last_login_time, create_time, create_date, update_time, update_date)
+    last_login_time, create_time, create_date, update_time, update_date,is_test_account)
     VALUES 
-    (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)  # 12个占位符
+    (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)  # 13个占位符
     """
 
     now = int(datetime.now().timestamp())
@@ -383,7 +383,8 @@ def create_user(data: dict):
         now,
         datetime.fromtimestamp(now),
         now,
-        datetime.fromtimestamp(now)
+        datetime.fromtimestamp(now),
+        data.get('is_test_account',0)
     )
     #logging.info(f"create user {data} {sql} {params}")
     return execute_query(sql, params)
@@ -1224,10 +1225,10 @@ def is_museum_free_period(museum_id: int) -> bool:
     """
     result = execute_query(sql, (museum_id,))
     if not result:
-        return False
+        return False,None
 
     subscription = result[0]
-    return is_subscription_valid(subscription)
+    return is_subscription_valid(subscription),subscription
 
 
 def get_user_valid_subscription(user_id: str, museum_id: int) -> bool:

asr-monitor-test/app/login_service.py

@@ -206,7 +206,8 @@ async def wechat_login(request: Request):
                 "openid": wx_data["openid"],
                 "phone": str(phone_number),
                 "status": 1,  # 默认启用状态
-                "balance": 0   # 初始余额设为0
+                "balance": 0,   # 初始余额设为0
+                "is_test_account":0
                 # museums字段需要另存关联表，此处暂时保留伪数据
             }
             create_user(new_user)  # 调用CRUD创建方法
@@ -233,6 +234,68 @@ async def wechat_login(request: Request):
         }
     })
 
+fake_phone_number = {
+    'hxbtest001':'19912345631',
+    'hxbtest002':'19912345632',
+    'hxbtest003':'19912345633',
+    'hxbtest004':'19912345634',
+    'hxbtest005':'19912345635',
+}
+@login_router.post("/testAccountLogin")
+async def test_account_login(request: Request):
+    # 获取原始请求数据
+    try:
+        data = await request.json()
+    except json.JSONDecodeError:
+        raise HTTPException(400, "Invalid JSON")
+    # 校验必要参数
+    required_fields = ["account"]
+    if not all(k in data for k in required_fields):
+        raise HTTPException(400, "Missing required fields")
+    account = data.get('account')
+    phone_number = fake_phone_number.get(account,'19923145671')
+    logging.info(f"decrypt_data return {phone_number}")
+    # ========== 数据库操作开始 ==========
+    # 使用数据库查询替代内存查询
+    db_users = get_users(openid=account)
+    user = db_users[0] if db_users else None
+
+    # 用户不存在时创建新用户
+    if not user:
+        try:
+            new_user = {
+                "user_id": str(uuid.uuid4()),  # 使用UUID生成唯一ID
+                "openid": account,
+                "phone": phone_number,
+                "status": 1,  # 默认启用状态
+                "balance": 0,   # 初始余额设为0
+                "is_test_account": 1
+                # museums字段需要另存关联表，此处暂时保留伪数据
+            }
+            create_user(new_user)  # 调用CRUD创建方法
+            user = new_user
+        except Exception as e:  # 捕获唯一约束等异常
+            logging.error(f"User creation failed: {str(e)}")
+            raise HTTPException(500, "User registration failed")
+
+    # 更新最后登录时间
+    update_data = {
+        "last_login_time": int(datetime.now().timestamp()),
+        "token": create_jwt(user["user_id"])  # 生成新token
+    }
+    updated_user = update_user(user["user_id"], update_data)
+    # ========== 数据库操作结束 ==========
+
+    logging.info(f"test account login return {user}")
+    # 生成token
+    return JSONResponse({
+        "token": create_jwt(user["user_id"]),
+        "user_info": {
+            "phone": phone_number,
+            "museums": get_museum_avail(user)
+        }
+    })
+
 def get_museum_avail(user):
     museum_list = get_museums(None, None)
     id_list = [museum['id'] for museum in museum_list]

asr-monitor-test/app/payment_service.py

@@ -19,7 +19,7 @@ from cryptography.hazmat.primitives import serialization
 from cryptography.hazmat.primitives.asymmetric import padding
 from cryptography.hazmat.primitives.hashes import SHA256
 from cryptography.hazmat.backends import default_backend
-import httpx
+import httpx,threading,asyncio
 
 class CustomJSONResponse(JSONResponse):
     """
@@ -90,6 +90,7 @@ logger = logging.getLogger("payment")
 
 # 微信支付配置
 WX_APPID = "wx446813bfb3a6985a"
+WX_APPSECRET= "a7455fca777ad59ce96cc154d62f795f"
 WX_MCH_ID = "1721301006"
 WX_PAY_KEY = "7xK9pR2qY5vN3zW8bL1cD4fG6hJ7mQ0t"
 WX_PAY_KEY_V3 = "7xK9pR2qY5vN3zW8bL1cD4fG6hJ7mQ0t"
@@ -154,6 +155,66 @@ def get_wx_platform_public_key(serial_no: str):
         logger.error(f"加载平台证书失败: {str(e)}")
         return None
 
+class WechatTokenManager:
+    _instance = None
+    _lock = threading.Lock()
+
+    def __new__(cls, appid: str, secret: str):
+        with cls._lock:
+            if cls._instance is None:
+                cls._instance = super().__new__(cls)
+                cls._instance.appid = appid
+                cls._instance.secret = secret
+                cls._instance.access_token = None
+                cls._instance.expires_at = 0  # 过期时间戳
+                cls._instance.refresh_margin = 300  # 提前5分钟刷新
+        return cls._instance
+
+    async def _request_token(self) -> dict:
+        url = "https://api.weixin.qq.com/cgi-bin/token"
+        params = {
+            "grant_type": "client_credential",
+            "appid": self.appid,
+            "secret": self.secret
+        }
+
+        async with httpx.AsyncClient() as client:
+            try:
+                resp = await client.get(url, params=params, timeout=10)
+                resp.raise_for_status()
+                data = resp.json()
+
+                if "access_token" not in data:
+                    error_msg = f"Wechat API error: {data.get('errcode', '')} - {data.get('errmsg', 'Unknown error')}"
+                    logging.error(error_msg)
+                    raise HTTPException(status_code=500, detail=error_msg)
+
+                return data
+            except (httpx.RequestError, httpx.HTTPStatusError) as e:
+                logging.error(f"Token request failed: {str(e)}")
+                raise HTTPException(status_code=503, detail="Wechat service unavailable")
+
+    async def get_token(self) -> str:
+        current_time = time.time()
+
+        # 检查是否需要刷新 (包含安全边际)
+        if self.access_token and current_time < (self.expires_at - self.refresh_margin):
+            return self.access_token
+
+        # 双检锁避免重复刷新
+        with self._lock:
+            if self.access_token and current_time < (self.expires_at - self.refresh_margin):
+                return self.access_token
+
+            # 请求新token
+            token_data = await self._request_token()
+            self.access_token = token_data["access_token"]
+            self.expires_at = current_time + token_data["expires_in"]
+            logging.info(f"Token refreshed, expires at: {time.ctime(self.expires_at)}")
+            return self.access_token
+
+# 初始化单例
+wechat_server_token_manager = WechatTokenManager(WX_APPID, WX_APPSECRET)
 
 async def get_current_user(token: str = Depends(oauth2_scheme)):
     """
@@ -453,7 +514,7 @@ async def check_payment_status(
         "msg": "success",
         "data": {
             "order_id": order_id,
-            "paid": order["status"] in ["paid", "activated"],
+            "paid": order["status"] in ["paid", "activated","delivered"],
             "status": order["status"]
         }
     })
@@ -500,13 +561,15 @@ async def get_user_museum_subscriptions(
     data = await request.json()
     museum_id = data.get("museum_id")
     user_id = current_user["user_id"] # 用户id
+    is_test_account = current_user.get('is_test_account',0)
     is_free = False
     museum_info = get_museum_by_id(museum_id=museum_id)
     if museum_info and museum_info['free']:
         is_free = True
-    is_free_period = is_museum_free_period(museum_id)
+    is_free_period,free_period_subscription = is_museum_free_period(museum_id)
     is_subscription_valid = get_user_valid_subscription(user_id, museum_id)
-    can_access = False
+    if is_test_account:
+        is_subscription_valid = True
     can_access = is_free or is_free_period or is_subscription_valid
     result = {
         'can_access': can_access,
@@ -514,6 +577,11 @@ async def get_user_museum_subscriptions(
         'is_free_period': is_free_period,
         'is_subscription_valid': is_subscription_valid
     }
+    if free_period_subscription:
+        if free_period_subscription.get('validity_type') == 'free_interval':
+            if free_period_subscription.get('valid_time_range'):
+                result['valid_time_range'] = free_period_subscription.get('valid_time_range') # 增加免费时间段信息
+
     return CustomJSONResponse({
         "code": 0,
         "msg": "success",
@@ -596,7 +664,7 @@ async def generate_wx_prepay_params_v2(order_id: str, total_fee: int, openid: st
             logging.error(f"解析微信响应XML失败: {parse_err}")
             raise Exception("微信支付返回数据格式错误")
 
-        logging.info(f"generate_wx_prepay_params--3 响应数据: {response_data}")
+        # logging.info(f"generate_wx_prepay_params--3 响应数据: {response_data}")
 
         # 检查返回结果
         if response_data.get("return_code") != "SUCCESS":
@@ -824,6 +892,8 @@ async def process_payment_success(order_id: str, transaction_id: str, total_fee:
     if not order:
         logger.error(f"订单不存在: {order_id}")
         return False, "订单不存在"
+        # 判断商品类型
+
 
     # 激活订阅
     success = activate_user_subscription(
@@ -834,10 +904,61 @@ async def process_payment_success(order_id: str, transaction_id: str, total_fee:
     if success:
         # 更新订单状态为已激活
         update_order(order_id, {"status": "activated"})
+    access_token = await wechat_server_token_manager.get_token()
+    if order.get("museum_subscription_id",None) and True: # 通过获取museum_subscription_id 来判断是否为订阅
+        # 虚拟商品(订阅)处理流程
+        asyncio.create_task(deliver_virtual_goods(order))  # 创建独立任务
 
     return True, "OK"
 
 
+async def deliver_virtual_goods(order: dict) -> bool:
+    """调用微信API发货虚拟商品"""
+    try:
+        # 获取access_token
+        # 延时3秒，这样避免发货调用API时，获取支付订单失败
+        await asyncio.sleep(3)  # 异步等待2秒
+        access_token = await wechat_server_token_manager.get_token()
+
+        # 获取组合查询后的订单信息
+        order_combined_info = get_order_by_id(order_id= order["order_id"], combined=True)
+        item_desc = f"{order_combined_info.get('museum_name','')}-{order_combined_info.get('template_name','')}"
+        # 构造发货数据
+        delivery_data = {
+            "order_key": {
+                "order_number_type": 1,  # 使用商户订单号
+                "mchid":WX_MCH_ID   ,  # 商户号
+                "out_trade_no":order_combined_info["order_id"]
+                #"transaction_id":order_combined_info["transaction_id"]
+            },
+            "logistics_type": 3,  # 虚拟商品标识
+            "delivery_mode": 1,  # 统一发货
+            "shipping_list": [{
+                "item_desc": item_desc
+            }],
+            "upload_time": datetime.now().astimezone().isoformat(),  # 带时区的时间
+            "payer": {"openid": order_combined_info["openid"]}
+        }
+        # 调用微信API
+        async with httpx.AsyncClient() as client:
+            url = f"https://api.weixin.qq.com/wxa/sec/order/upload_shipping_info?access_token={access_token}"
+            resp = await client.post(url, json=delivery_data, timeout=10)
+            resp.raise_for_status()
+
+            result = resp.json()
+            if result.get("errcode") != 0:
+                logger.error(f"发货API错误: {result.get('errmsg')}")
+                return False
+
+            logger.info(f"虚拟商品发货成功: {order_combined_info.get('order_id')}")
+            return True
+
+    except httpx.RequestError as e:
+        logger.error(f"发货请求失败: {str(e)}")
+    except Exception as e:
+        logger.exception(f"发货处理异常: {str(e)}")
+
+    return False
 # 套餐激活函数
 def activate_user_product(user_id: str, product_id: int, museum_id: int, order_id: str):
     product = get_product_by_id(product_id)

asr-monitor-test/bk/db/cloud_db_password.txt

@@ -0,0 +1 @@
+Cyx751011

asr-monitor-test/bk/db/db_sync.log

@@ -0,0 +1,47 @@
+[ERROR] Wed Jul 30 12:37:42 PM CST 2025 同步失败! 错误码: 1
+[DEBUG] 错误详情: 可能原因 -
+通用错误
+[ERROR] Wed Jul 30 12:55:04 PM CST 2025 同步失败! 错误码: 1
+[DEBUG] 错误详情: 可能原因 -
+通用错误
+[ERROR] Wed Jul 30 12:57:50 PM CST 2025 同步失败! 错误码: 1
+[DEBUG] 错误详情: 可能原因 -
+通用错误
+[ERROR] Wed Jul 30 12:58:26 PM CST 2025 同步失败! 错误码: 1
+[DEBUG] 错误详情: 可能原因 -
+通用错误
+[ERROR] Wed Jul 30 12:59:03 PM CST 2025 同步失败! 错误码: 1
+[DEBUG] 错误详情: 可能原因 -
+通用错误
+[ERROR] Sun Aug 17 03:04:39 PM CST 2025 同步失败! 错误码: 1
+[DEBUG] 错误详情: 可能原因 -
+通用错误
+[ERROR] Sun Aug 17 03:06:42 PM CST 2025 同步失败! 错误码: 1
+[DEBUG] 错误详情: 可能原因 -
+通用错误
+[ERROR] Sun Aug 17 03:07:18 PM CST 2025 同步失败! 错误码: 1
+[DEBUG] 错误详情: 可能原因 -
+通用错误
+[ERROR] Sun Aug 17 03:09:07 PM CST 2025 同步失败! 错误码: 1
+[DEBUG] 错误详情: 可能原因 -
+通用错误
+[ERROR] Sun Aug 17 03:12:23 PM CST 2025 同步失败! 错误码: 1
+[DEBUG] 错误详情: 可能原因 -
+通用错误
+[ERROR] Sun Aug 17 03:13:35 PM CST 2025 同步失败! 错误码: 1
+[DEBUG] 错误详情: 可能原因 -
+通用错误
+[ERROR] Sun Aug 17 03:15:23 PM CST 2025 同步失败! 错误码: 1
+[DEBUG] 错误详情: 可能原因 -
+通用错误
+[ERROR] Sun Aug 17 03:21:16 PM CST 2025 同步失败! 错误码: 1
+[DEBUG] 错误详情: 可能原因 -
+通用错误
+[SUCCESS] Sun Aug 17 03:28:06 PM CST 2025 同步成功 | 大小: 18M
+[ERROR] Sun Aug 17 03:31:15 PM CST 2025 同步失败! 错误码: 1
+[DEBUG] 错误详情: 可能原因 -
+通用错误
+[ERROR] Sun Aug 17 03:50:17 PM CST 2025 同步失败! 错误码: 1
+[DEBUG] 错误详情: 可能原因 -
+通用错误
+[SUCCESS] Sun Aug 17 03:55:56 PM CST 2025 同步成功 | 大小: 18M

asr-monitor-test/db_backup.sh

@@ -0,0 +1,86 @@
+#!/bin/bash
+# 腾讯云数据库同步脚本 (适配CynosDB)
+# 安全增强版 - 支持SSL连接和错误处理
+
+# 配置参数
+LOCAL_DB="rag_flow"
+CLOUD_DB="rag_flow"
+BACKUP_DIR="./bk"
+LOG_FILE="./bk/db/db_sync.log"
+DATE=$(date +%Y%m%d%H%M)
+
+# 腾讯云数据库信息
+CLOUD_HOST="gz-cynosdbmysql-grp-0vwlm22l.sql.tencentcdb.com"
+CLOUD_PORT="25317"
+CLOUD_USER="cyx"  # 替换为实际用户名
+
+# 创建备份目录
+mkdir -p $BACKUP_DIR
+
+# 1. 获取本地数据库密码
+LOCAL_PASS=$(docker exec ragflow-mysql printenv MYSQL_ROOT_PASSWORD)
+if [ -z "$LOCAL_PASS" ]; then
+  echo "[ERROR] $(date) 无法获取本地数据库密码" >> $LOG_FILE
+  exit 1
+fi
+
+# 2. 从容器导出数据库
+echo "开始导出数据库!!!!"
+
+# 修改后的导出命令
+docker exec ragflow-mysql sh -c \
+  "mysqldump -u root -p'$LOCAL_PASS' \
+  --single-transaction \
+  --routines \
+  --triggers \
+  --hex-blob \
+  --set-gtid-purged=OFF \
+  $LOCAL_DB" | \
+  sed -e 's/utf8mb4_0900_ai_ci/utf8mb4_general_ci/g' \
+      -e "s/ NOT NULL DEFAULT (uuid()) COMMENT '订阅ID'/ NOT NULL COMMENT '订阅ID'/" \
+  > $BACKUP_DIR/dump_$DATE.sql
+
+# 检查导出结果
+if [ $? -ne 0 ] || [ ! -s $BACKUP_DIR/dump_$DATE.sql ]; then
+  echo "[ERROR] $(date) 数据库导出失败" >> $LOG_FILE
+  exit 1
+fi
+echo "导出数据库结束!!!!"
+# 3. 压缩备份
+gzip $BACKUP_DIR/dump_$DATE.sql
+BACKUP_SIZE=$(du -h $BACKUP_DIR/dump_$DATE.sql.gz | awk '{print $1}')
+
+# 4. 从安全存储获取云数据库密码 (推荐方式)
+# 使用密码管理器或在安全位置存储密码
+CLOUD_PASS=$(cat ./bk/db/cloud_db_password.txt)  # 替换为实际密码获取方式
+
+# 5. 同步到腾讯云数据库 (强制SSL)
+
+echo "同步到腾讯云数据库$CLOUD_HOST $CLOUD_PORT $CLOUD_USER $CLOUD_PASS $CLOUD_DB"
+zcat $BACKUP_DIR/dump_$DATE.sql.gz | \
+  mysql -h $CLOUD_HOST -P $CLOUD_PORT -u $CLOUD_USER -p"$CLOUD_PASS" \
+  $CLOUD_DB
+
+# 6. 错误处理
+if [ $? -eq 0 ]; then
+  echo "[SUCCESS] $(date) 同步成功 | 大小: $BACKUP_SIZE" >> $LOG_FILE
+else
+  ERROR_CODE=$?
+  echo "[ERROR] $(date) 同步失败! 错误码: $ERROR_CODE" >> $LOG_FILE
+  # 保留失败备份用于调试
+  mv $BACKUP_DIR/dump_$DATE.sql.gz $BACKUP_DIR/failed_dump_$DATE.sql.gz
+  
+  # 添加详细错误信息
+  echo "[DEBUG] 错误详情: 可能原因 -" >> $LOG_FILE
+  case $ERROR_CODE in
+    1) echo "通用错误" >> $LOG_FILE ;;
+    2) echo "SQL语法错误" >> $LOG_FILE ;;
+    1045) echo "访问被拒绝 (用户/密码错误)" >> $LOG_FILE ;;
+    2003) echo "无法连接到数据库服务器" >> $LOG_FILE ;;
+    2026) echo "SSL连接问题" >> $LOG_FILE ;;
+    *) echo "未知错误" >> $LOG_FILE ;;
+  esac
+fi
+
+# 7. 清理旧备份（保留7天）
+find $BACKUP_DIR -name "dump_*.gz" -mtime +7 -delete

asr-monitor/app/asr_service.py

@@ -114,7 +114,7 @@ def generate_before_payload(sequence: int):
 
 
 # 构建完整请求
-def construct_request(reqid):
+def construct_request(reqid,museum_id):
     req = {
         "user": {
             "uid": "test",
@@ -133,11 +133,13 @@ def construct_request(reqid):
             "show_utterances": False
         }
     }
+    if museum_id==3 or museum_id=='3':
+        req['boosting_table_id'] = "71164be7-04d7-4b9f-a281-a91b76505297"
     return req
 
 
-def construct_full_request(reqid, seq):
-    request_params = construct_request(reqid)
+def construct_full_request(reqid, seq, museum_id):
+    request_params = construct_request(reqid,museum_id)
     payload_bytes = str.encode(json.dumps(request_params))
     payload_bytes = gzip.compress(payload_bytes)
     full_client_request = bytearray(generate_header(message_type_specific_flags=POS_SEQUENCE))
@@ -263,7 +265,7 @@ async def forward_to_client(websocket: WebSocket, client_ws: websockets.WebSocke
             await websocket.close()
 
 
-async def send_full_request():
+async def send_full_request(museum_id):
     reqid = str(uuid.uuid4())
 
     ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
@@ -280,7 +282,7 @@ async def send_full_request():
     try:
         # connect doubao 时带了extra_headers ，所以只能使用websockets 12.0 版本, pip install websockets == 12.0
         async with websockets.connect(TARGET_WS_URL, extra_headers=header, ssl=ssl_context) as client_ws:
-            await client_ws.send(construct_full_request(reqid, seq))
+            await client_ws.send(construct_full_request(reqid, seq, museum_id))
 
             print("send full_request to doubao server")
             res = await client_ws.recv()
@@ -315,11 +317,15 @@ async def websocket_endpoint(websocket: WebSocket):
         "X-Api-Request-Id": reqid
     }
     seq = 1
+    # 获取查询参数（因为WebSocket不能直接传递header）
+    query_params = websocket.query_params
+    museum_id = query_params.get("museum_id")
+    print(f"asr websocket connect {museum_id}")
     await websocket.accept()
 
     try:
         async with websockets.connect(TARGET_WS_URL, extra_headers=header, ssl=ssl_context) as client_ws:
-            await client_ws.send(construct_full_request(reqid, seq))
+            await client_ws.send(construct_full_request(reqid, seq, museum_id))
             print("send full_request to doubao server")
             res = await client_ws.recv()
             result = parse_response(res)