async def time_convert(resoucetime=None):
    if not resoucetime:
        return
    utc_time = datetime.datetime.strptime(resoucetime, "%Y-%m-%dT%H:%M:%SZ").replace(tzinfo=datetime.timezone.utc)
    beijing_time = utc_time.astimezone(datetime.timezone(datetime.timedelta(hours=8)))
    return beijing_time.strftime("%Y-%m-%d %H:%M:%S")

async def baidu_sms_kafka_consumer(ns={}):
    consumer = BaiduKafKaConsumer({
        # 接入点
        'bootstrap.servers': '120.48.10.223:9095,180.76.96.108:9095,180.76.147.36:9095',
        # 接入协议
        'security.protocol': 'SASL_SSL',
       'ssl.endpoint.identification.algorithm': 'none',
        # 证书文件路径
        'ssl.ca.location': 'baidu_kafka_ca.pem',
        # SASL 机制
        'sasl.mechanism': 'SCRAM-SHA-512',
        # SASL 用户名
        'sasl.username': 'kaiyuanyun',
        # SASL 用户密码
        'sasl.password': 'Kyy250609#',
        # 消费组id
        'group.id': 'kaiyuanyun_msg_group',
        'auto.offset.reset': 'latest',
        'fetch.message.max.bytes': '1024*512',
    })

    # 订阅的主题名称
    consumer.subscribe(['kaiyuanyun_msg_topic'])

    for i in range(30):
        msg = consumer.poll(0.1)  # 单次轮询获取消息

        if msg is None:
            continue
        elif msg.error():
            print(f"消费者错误: {msg.error()}")
        else:
            try:
                # 解析消息内容为字典（避免变量名冲突）
                msg_data_ori = json.loads(msg.value().decode('utf-8'))
                msg_data = msg_data_ori['messages'][0]
                messageid = msg_data.get('id')
                taskid = msg_data.get('taskId')

                filename = 'baidu_kafka_msg.txt'
                # 检查文件是否存在，不存在则创建
                if not os.path.exists(filename):
                    with open(filename, 'w', encoding='utf-8') as f:
                        print(f"文件不存在，已创建 {filename}")

                # 读取文件内容进行检查
                with open(filename, 'r', encoding='utf-8') as f:
                    content = f.read()

                if messageid in content:
                    print(f"文件中已存在 '{messageid}'，跳过写入")
                    continue
                else:
                    # 追加写入目标内容
                    with open(filename, 'a', encoding='utf-8') as f:
                        f.write(messageid + '\n')
                        print(f"已写入 '{messageid}' 到文件")

                db = DBPools()
                async with db.sqlorContext('kboss') as sor:
                    # 1. 去重检查（messageid和taskid）
                    exist_msg = await sor.R('baidu_kafka_msg', {'messageid': messageid, 'taskid': taskid})
                    if exist_msg:
                        print(f"消息id {messageid} 已存在，跳过处理")
                        consumer.close()
                        return

                    # 2. 构建小写key的ns字典（完整映射所有字段）
                    ns_msg = {
                        'id': uuid(),
                        'messageid': messageid,
                        'taskid': taskid,
                        'userid': msg_data.get('userId'),
                        'accountid': msg_data.get('accountId'),
                        'usertype': msg_data.get('userType'),
                        'receiverid': msg_data.get('receiverId'),
                        'contentvar': msg_data.get('contentVar'),
                        'sendchannel': msg_data.get('sendChannel'),
                        'content': msg_data.get('content'),
                        'messagetemplateid': msg_data.get('messageTemplateId'),
                        'isvirtualstore': msg_data.get('isVirtualStore'),
                        'channelmessageid': msg_data.get('channelMessageId'),
                        'channelstatus': msg_data.get('channelStatus'),
                        'majorcategory': msg_data.get('majorCategory'),
                        'minorcategory': msg_data.get('minorCategory'),
                        'status': msg_data.get('status'),
                        'createtime': await time_convert(msg_data.get('createTime')) if msg_data.get('createTime') else None,
                        'updatetime': await time_convert(msg_data.get('updateTime')) if msg_data.get('updateTime') else None,
                        'expiretime': await time_convert(msg_data.get('expireTime')) if msg_data.get('expireTime') else None,
                        'windowtime': await time_convert(msg_data.get('windowTime')) if msg_data.get('windowTime') else None,
                        'valid': msg_data.get('valid'),
                        'complete': msg_data.get('complete'),
                        'disturbhold': msg_data.get('disturbHold'),
                        'sendcomplete': msg_data.get('sendComplete')
                    }

                    # 3. 执行存库
                    await sor.C('baidu_kafka_msg', ns_msg)
                    print(f"消息id {messageid} 存储成功")

                    # 4. 触发短信发送（当sendChannel为MOBILE时）
                    send_channel = msg_data.get('sendChannel')
                    if send_channel == 'MOBILE':
                        msg_content = msg_data.get('content')

                        # 判断验证码类短信 | 通知类短信
                        if '验证码' in msg_content:
                            sms_stype = '百度kafka普通验证码'
                        else:
                            sms_stype = '百度kafka普通通知'

                        # 查询用户手机号
                        account_id = msg_data.get('accountId')
                        user_local_id_li = await sor.R('baidu_users', {'baidu_id': account_id})
                        user_local_id = user_local_id_li[0]['user_id']
                        user_mobile_li = await sor.R('users', {'id': user_local_id, 'del_flg': '0'})
                        mobile = user_mobile_li[0]['mobile']

                        # 调用短信发送接口
                        kyy_send_status = await send_vcode(mobile, sms_stype, {'content': msg_content})
                        if kyy_send_status.get('status'):
                            await sor.U('baidu_kafka_msg', {'id': ns_msg['id'], 'kyysendstatus': '1'})
                        print(f"已触发短信发送至 {mobile}")

            except json.JSONDecodeError:
                print("错误：消息内容非有效JSON")
                return {
                    'status': False,
                    'msg': '错误：消息内容非有效JSON'
                }
            except Exception as e:
                print(f"处理异常: {str(e)}")
                return {
                    'status': False,
                    'msg': f"处理异常: {str(e)}"
                }

    consumer.close()  # 确保消费者关闭
    return {
        'status': True,
        'msg': '获取信息执行结束'
    }

ret = await baidu_sms_kafka_consumer(params_kw)
return ret