2python之用户配置合并开发及域名变数组解析

写在前面的话

孩子能成。昨天，嫂子联系我说，要给堂哥和一个叔上号，在嫂子的配合下，顺利上了两个号。第二天，需要重新登。我联系堂哥，是堂哥孩子接的，他说：叔，我爸出去了，等一会回来了，我联系您，今天最好上完，我下午要走了。我想着：过年时，我给堂哥孩子发200压岁钱，堂哥给我转400，因为我两孩子，正好趁此机会，补偿一下，给侄子转100块钱，我说下午你走时，买点饮料喝。侄子回我说：叔，不收了，我的钱够花的，钱给两个小弟弟买东西吃吧。问了下他的学业，他说：我是学土木的，准备往上走走，对象先不谈，以学业为重，今年大三，明年该考研了，叔放心，一定给它弄上。我说：嫂子哥这么年挺辛苦，你好好干，他们指望着你呢，但不要有压力。侄子说：不会不会，我的调节能力还是可以的，但学习靠自己，不管咋样，我都得考上。侄子的态度我是很赞赏的，祝他成功吧。

[142+200]-------->底部有张生活照片(头条号运营：大家想全托管上号的联系我哦，每天让你得个地铁钱，微信号: qhz_toutiao)

一、用户配置合并开发（三级）

描述：要把配置文件与租户数据库配置合并，当前处理到，从数据库中取数据。

开工：

第一步：切分支到租户开发专用分支(四级)

20250630周一时间段：15:29-17:00

切到专用分支，截图如下：

图2a-1

注：接下来，就是数据获取开发。

第二步：数据获取（四级）

20250630周一时间段：15:44-17:00

获取数据函数如下：

def fetch_tenant_config_from_mysql() -> Dict[str, Any]:
    """
    从MySQL数据库获取租户配置

    Args:
        host: MySQL主机地址
        port: MySQL端口
        user: 用户名
        password: 密码
        database: 数据库名
        table: 配置表名(默认为tenant_config)

    Returns:
        包含配置的字典
    """
    host = os.environ.get('MYSQL_HOST', host)
    port = int(os.environ.get('MYSQL_PORT', port))
    user = os.environ.get('MYSQL_USER', user)
    password = os.environ.get('MYSQL_PASSWORD', password)
    database = os.environ.get('MYSQL_DATABASE', database)
    table = 'tenant'
    try:
        connection = pymysql.connect(
            host=host,
            port=port,
            user=user,
            password=password,
            database=database,
            cursorclass=DictCursor
        )

        with connection.cursor() as cursor:
            # 查询最新配置(假设表中有version或update_time字段)
            sql = f"SELECT * FROM {table} ORDER BY version DESC LIMIT 1"
            cursor.execute(sql)
            result = cursor.fetchone()

            if not result:
                logger.warning("No tenant configuration found in database")
                return {}

            return result

    except pymysql.Error as e:
        logger.error(f"Error fetching tenant config from MySQL: {e}")
        raise
    finally:
        if 'connection' in locals() and connection.open:
            connection.close()

注：运行一下，看看效果，用测试用例multi_chat_choose。

第三步：跑测试用例(四级)

20250630周一时间段：16:14-17:00

跑测试用例，发现测试用例出不来，有错误，看一下，数据库没连上，现在可以了，如下：

图2a-2

注：测试用例如下：

def test_multi_chat_choose(client,login_header):
    log.info("test_multi_chat_choose-108")
    '''
    测试多轮对话-包括三轮对话场景====加入了模拟登录
    主要测新版意图识别
    '''

    ########调用接口
    json_data = {    
        "conversation_id": '605f2736370511f0bf8e75026f04980c',
        "question": '我有售后问题要解决，心情很烦躁',
    }

    #############小租户：
    json_data = {    
        "conversation_id": '4cbbfc624d8111f08be70ac66fac3d9b',
        "question": '我家狗拉肚子',
    }

    #     ###测试专家诊断中的深度思考
    # json_data = {    
    #     "conversation_id": '4cbbfc624d8111f08be70ac66fac3d9b',
    #     "question": '结束基础问询，请直接给出结果',
    #     "deep_think": True,  # 是否深度思考
    #     "enable_search": False,  # 是否启用联网搜索
    # }


    ##### 调用多轮对话接口
    url = f"/v1/conversation/multi_chat_choose"
    resp = client.post(
        url,
        json=json_data,
        headers={
            **login_header    # 携带 JWT（如果有）
        }
    )
    if not 200 <= resp.status_code < 300:
        raise Exception(f"GET {url} status_code {resp.status_code}.")

    # received_data = []
    for chunk in resp.iter_encoded():
        answer = chunk.decode('utf-8').strip()
        log.info(f"\n\n answer: {answer} \n\n")   

注：跑一下，看看效果，效果如下：

图2a-3

注：现在有配置没找到，打日志看一下。

第四步：打日志追踪(四级)

20250630周一时间段：17:27-19:00

打个断点，看下数据获取到没有。修改后，能看到数据，如下：

[{
'domain': '["pet.bitzero.cn"]',
'tenant_name': 'ragflow‘s Kingdom',
'app': '{"welcome":"我是您的专业宠物医生，很高兴为您服务","background_image":"/static/login-background.117236ea.webp","logo":"/static/logos1.76a7cb8a.png","is_open_pet_archives":1}',
'user_intention': '{"enabled":true,"llm_prompts":{"role":"宠物医生","force_complete_flag":"结束基础问询，请直接给出结果","finish_flag":"question_finished","wait_expert_reply":"您已回复完成临床问询，正在分析病情..."},"intention_file":"conf/test_pet/test_pet_intention.yaml"}',
'llm': '',
'kb': None,
'dialog': None,
'use_kb': '["7d19a176807611efb0f80242ac120006"]'
}, {
'domain': '["baili.bitzero.cn"]',
'tenant_name': 'baili@baili.com‘s Kingdom',
'app': '{"welcome":"我是您的专业宠粮助手，很高兴为您服务","background_image":"/static/login-background.117236ea.webp","logo":"/static/logos1.76a7cb8a.png","is_open_pet_archives":0,"chat_title":"知识问答"}',
'user_intention': '{\r\n\t"enabled": true,\r\n\t"role": "宠粮专家",\r\n\t"intention_file": "conf/baili/baili_intention.yaml"\r\n}',
'llm': '{"role_setting":""}',
'kb': '{"extract_extend_entities":true,"invalid_types":[]}',
'dialog': '{"intention_llm":"qwen-plus","online_search_llm":"qwen-plus"}',
'use_kb': '["7d19a176807611efb0f80242ac120006"]'
}]

注：截图如下：

图2a-4

注：接下来，再次跑测试用例，看下报错不，运行结果报错，主要是对域名变数组的解析。

第五步：域名变数组解析（四级）

20250630周一时间段：18:10-19:00

这个要打日志看下怎么改解析函数了，现在zero4被关闭了，需要在连下zero2上的环境。建文件夹后，再次运行如下：

图2a-5

注：这次没报错，是会话没找到，换一下会话id，再次跑一遍，如下：

图2a-6

注：这个跑了一遍是没问题的，我觉得原因是旧的tenant_conf.yaml配置文件没去掉，去掉再跑一遍试试，效果如下：

图2a-7

注：发现也是可以的，接下来，把数据换成从数据库里拿。

第六步：数据从数据库里拿(四级)

20250630周一时间段：18:49-19:00

从数据库里拿到的数据如下：

[{'domain': "['pet.bitzero.cn']", 'tenant_name': 'ragflow‘s Kingdom', 'app': '{"welcome":"我是您的专业宠物医生，很高兴为您服务","background_image":"/static/login-background.117236ea.webp","logo":"/static/logos1.76a7cb8a.png","is_open_pet_archives":1}', 'user_intention': '{"enabled":true,"llm_prompts":{"role":"宠物医生","force_complete_flag":"结束基础问询，请直接给出结果","finish_flag":"question_finished","wait_expert_reply":"您已回复完成临床问询，正在分析病情..."},"intention_file":"conf/test_pet/test_pet_intention.yaml"}', 'llm': '', 'kb': None, 'dialog': None, 'use_kb': '["7d19a176807611efb0f80242ac120006"]'}, {'domain': "['baili.bitzero.cn']", 'tenant_name': 'baili@baili.com‘s Kingdom', 'app': '{"welcome":"我是您的专业宠粮助手，很高兴为您服务","background_image":"/static/login-background.117236ea.webp","logo":"/static/logos1.76a7cb8a.png","is_open_pet_archives":0,"chat_title":"知识问答"}', 'user_intention': '{\r\n\t"enabled": true,\r\n\t"role": "宠粮专家",\r\n\t"intention_file": "conf/baili/baili_intention.yaml"\r\n}', 'llm': '{"role_setting":""}', 'kb': '{"extract_extend_entities":true,"invalid_types":[]}', 'dialog': '{"intention_llm":"qwen-plus","online_search_llm":"qwen-plus"}', 'use_kb': '["7d19a176807611efb0f80242ac120006"]'}]

注：这个数据不是json，怎么转成json，查下ai，方案如下：

写一个转json函数，如下：

def convert_tenant_config_to_json(tenant_configs: List[Dict[str, Any]]) -> str:
    """
    将租户配置数据转换为标准JSON格式字符串

    Args:
        tenant_configs: 租户配置列表

    Returns:
        标准JSON格式字符串
    """
    # 处理特殊字段和空值
    def sanitize_config(config: Dict[str, Any]) -> Dict[str, Any]:
        sanitized = {}
        for key, value in config.items():
            if value == '':
                sanitized[key] = None  # 将空字符串转为null
            elif isinstance(value, str) and value.lower() == 'none':
                sanitized[key] = None
            else:
                sanitized[key] = value
        return sanitized

    # 处理每个租户配置
    processed_configs = []
    for config in tenant_configs:
        processed_config = sanitize_config(config)
        processed_configs.append(processed_config)

    # 转换为JSON字符串
    json_str = json.dumps(
        processed_configs,
        indent=2,
        ensure_ascii=False,  # 保留中文等非ASCII字符
        default=str  # 处理日期等不可序列化对象
    )

    ##把json字符串转换为json对象
    try:
        # 去除字符串中的换行符和缩进
        compact_json = json_str.replace('\n', '').replace(' ', '')

        # 将紧凑的JSON字符串转换为Python对象
        json_obj = json.loads(compact_json)

        return json_obj
    except json.JSONDecodeError as e:
        raise ValueError(f"Invalid JSON string: {e}")

注：一会测一下。

二、头条战果汇报

昨日数据来啦，昨日总收入:593.99块，昨日总播放:352.59万次，软件截图如下：

图2b-1

注：想要全托管运营头条号的联系我，你出账号，我来运营，保你天天有钱花，咨询电话： 17701328814（微信同号），也可以加群先了解一下。

图2b-2

注：个人微信如下，欢迎骚扰。

图2b-3

三、生活照片

拍摄于‎2026‎年‎1‎月‎3‎日，‏‎14:27:28，老妈过生日时拍的，当时二宝三岁三个月。有一句古话说：富不过三代，穷也不过三代。就是说：一个家庭由穷变富需要几代人的努力，父辈这一代，都是普通的种地者，到我这一代，大家都拼了命的努力，不管是我，还是堂哥他们，这些年的发展而言，上述提到的堂哥相对发展的不是太顺利，但昨天和他的孩子聊天，让我看到了希望，感觉自制力太强了，有我当年的影子，但比我还强，我当时也是努力学习，但喜欢胡思乱想，这位侄子说话比较上进，比较干练，比较懂事，觉得将来必成大器，加油，侄子，看好你哦。

图2c-1

《本文完》