ai-on-wechat/channel/wechat/wechat_channel.py

# encoding:utf-8

"""
wechat channel
"""

import io
import json
import os
import threading
import time
import requests

from bridge.context import *
from bridge.reply import *
from channel.chat_channel import ChatChannel
from channel import chat_channel
from channel.wechat.wechat_message import *
from common.expired_dict import ExpiredDict
from common.log import logger
from common.singleton import singleton
from common.time_check import time_checker
from common.utils import convert_webp_to_png
from config import conf, get_appdata_dir
from lib import itchat
from lib.itchat.content import *
from urllib.parse import urlparse


import threading

from common import kafka_helper, redis_helper

from confluent_kafka import Consumer, KafkaException
import json,time,re
import pickle
from datetime import datetime
import oss2


# from common.kafka_client import KafkaClient


@itchat.msg_register([TEXT, VOICE, PICTURE, NOTE, ATTACHMENT, SHARING])
def handler_single_msg(msg):
    try:
        cmsg = WechatMessage(msg, False)
    except NotImplementedError as e:
        logger.debug("[WX]single message {} skipped: {}".format(msg["MsgId"], e))
        return None
    WechatChannel().handle_single(cmsg)
    return None


@itchat.msg_register([TEXT, VOICE, PICTURE, NOTE, ATTACHMENT, SHARING], isGroupChat=True)
def handler_group_msg(msg):
    try:
        cmsg = WechatMessage(msg, True)
    except NotImplementedError as e:
        logger.debug("[WX]group message {} skipped: {}".format(msg["MsgId"], e))
        return None
    WechatChannel().handle_group(cmsg)
    return None


def _check(func):
    def wrapper(self, cmsg: ChatMessage):
        msgId = cmsg.msg_id
        if msgId in self.receivedMsgs:
            logger.info("Wechat message {} already received, ignore".format(msgId))
            return
        self.receivedMsgs[msgId] = True
        create_time = cmsg.create_time  # 消息时间戳
        if conf().get("hot_reload") == True and int(create_time) < int(time.time()) - 60:  # 跳过1分钟前的历史消息
            logger.debug("[WX]history message {} skipped".format(msgId))
            return
        if cmsg.my_msg and not cmsg.is_group:
            logger.debug("[WX]my message {} skipped".format(msgId))
            return
        return func(self, cmsg)

    return wrapper


# 可用的二维码生成接口
# https://api.qrserver.com/v1/create-qr-code/?size=400×400&data=https://www.abc.com
# https://api.isoyu.com/qr/?m=1&e=L&p=20&url=https://www.abc.com
def qrCallback(uuid, status, qrcode):
    # logger.debug("qrCallback: {} {}".format(uuid,status))
    if status == "0":
        try:
            from PIL import Image

            img = Image.open(io.BytesIO(qrcode))
            _thread = threading.Thread(target=img.show, args=("QRCode",))
            _thread.setDaemon(True)
            _thread.start()
        except Exception as e:
            pass

        import qrcode

        url = f"https://login.weixin.qq.com/l/{uuid}"

        qr_api1 = "https://api.isoyu.com/qr/?m=1&e=L&p=20&url={}".format(url)
        qr_api2 = "https://api.qrserver.com/v1/create-qr-code/?size=400×400&data={}".format(url)
        qr_api3 = "https://api.pwmqr.com/qrcode/create/?url={}".format(url)
        qr_api4 = "https://my.tv.sohu.com/user/a/wvideo/getQRCode.do?text={}".format(url)
        print("You can also scan QRCode in any website below:")
        print(qr_api3)
        print(qr_api4)
        print(qr_api2)
        print(qr_api1)
        _send_qr_code([qr_api3, qr_api4, qr_api2, qr_api1])
        qr = qrcode.QRCode(border=1)
        qr.add_data(url)
        qr.make(fit=True)
        qr.print_ascii(invert=True)


@singleton
class WechatChannel(ChatChannel):
    NOT_SUPPORT_REPLYTYPE = []

    def __init__(self):
        super().__init__()
        self.receivedMsgs = ExpiredDict(conf().get("expires_in_seconds", 3600))
        self.auto_login_times = 0

    def startup(self):
        try:
            itchat.instance.receivingRetryCount = 600  # 修改断线超时时间
            # login by scan QRCode
            hotReload = conf().get("hot_reload", False)
            status_path = os.path.join(get_appdata_dir(), "itchat.pkl")
            # with open(status_path, 'rb') as file:
            #     data = pickle.load(file)
            # logger.info(data)
            itchat.auto_login(
                enableCmdQR=2,
                hotReload=hotReload,
                statusStorageDir=status_path,
                qrCallback=qrCallback,
                exitCallback=self.exitCallback,
                loginCallback=self.loginCallback
            )
            self.user_id = itchat.instance.storageClass.userName
            self.name = itchat.instance.storageClass.nickName
            logger.info("Wechat login success, user_id: {}, nickname: {}".format(self.user_id, self.name))
            
            
            

             # 创建一个线程来运行 consume_messages
            # kafka_thread = threading.Thread(target=consume_messages, args=('47.116.67.214:9092', 'ai-test-group', topic,self.name))
            # kafka_client=KafkaClient()
            # kafka_thread = threading.Thread(target=consume_wx_messages, args=('47.116.67.214:9092', 'ai-test-group', topic,self.name))
            kafka_thread = threading.Thread(target=kafka_helper.kafka_client.consume_messages, args=(wx_messages_process_callback, self.name))

            kafka_thread.start()
            logger.info("启动kafka")

            # 好友定时同步
            agent_nickname=self.name
            friend_thread =threading.Thread(target=hourly_change_save_friends, args=(agent_nickname,))
            friend_thread.start()

            # 立刻同步
            agent_info=fetch_agent_info(agent_nickname)
            agent_tel=agent_info.get("agent_tel",None)
            # friends=itchat.get_contact(update=True)[1:]
            friends=itchat.get_friends(update=True)[1:]
            # logger.info("~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~")
            # logger.info(friends)
            # logger.info("~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~")
            save_friends_to_redis(agent_tel,agent_nickname, friends)
            logger.info("启动好友同步")
            # start message listener

            logger.info("启动itchat")
            itchat.run()
            
            # 运行kafka
            # 

        except Exception as e:
            logger.exception(e)

    def exitCallback(self):
        print('主动退出')
        try:
            from common.linkai_client import chat_client
            if chat_client.client_id and conf().get("use_linkai"):
                print('退出')
                _send_logout()
                time.sleep(2)
                self.auto_login_times += 1
                if self.auto_login_times < 100:
                    chat_channel.handler_pool._shutdown = False
                    self.startup()
        except Exception as e:
            pass

    def loginCallback(self):
        logger.debug("Login success")
        print('登录成功')
        # 同步
        _send_login_success()

    # handle_* 系列函数处理收到的消息后构造Context，然后传入produce函数中处理Context和发送回复
    # Context包含了消息的所有信息，包括以下属性
    #   type 消息类型, 包括TEXT、VOICE、IMAGE_CREATE
    #   content 消息内容，如果是TEXT类型，content就是文本内容，如果是VOICE类型，content就是语音文件名，如果是IMAGE_CREATE类型，content就是图片生成命令
    #   kwargs 附加参数字典，包含以下的key：
    #        session_id: 会话id
    #        isgroup: 是否是群聊
    #        receiver: 需要回复的对象
    #        msg: ChatMessage消息对象
    #        origin_ctype: 原始消息类型，语音转文字后，私聊时如果匹配前缀失败，会根据初始消息是否是语音来放宽触发规则
    #        desire_rtype: 希望回复类型，默认是文本回复，设置为ReplyType.VOICE是语音回复
    @time_checker
    @_check
    def handle_single(self, cmsg: ChatMessage):
        # filter system message
        if cmsg.other_user_id in ["weixin"]:
            return
        if cmsg.ctype == ContextType.VOICE:
            if conf().get("speech_recognition") != True:
                return
            logger.debug("[WX]receive voice msg: {}".format(cmsg.content))
        elif cmsg.ctype == ContextType.IMAGE:
            logger.debug("[WX]receive image msg: {}".format(cmsg.content))
            # print(cmsg.content)
            file_path = cmsg.content
            logger.info(f"on_handle_context: 获取到图片路径 {file_path}")
            oss_access_key_id="LTAI5tRTG6pLhTpKACJYoPR5"
            oss_access_key_secret="E7dMzeeMxq4VQvLg7Tq7uKf3XWpYfN"
            oss_endpoint="http://oss-cn-shanghai.aliyuncs.com"
            oss_bucket_name="cow-agent"
            oss_image_url = upload_oss(oss_access_key_id, oss_access_key_secret, oss_endpoint, oss_bucket_name, file_path, f'cow/{os.path.basename(file_path)}')
            print(f"oss_image_url:{oss_image_url}")
            input_content = oss_image_url
            input_from_user_nickname = cmsg.from_user_nickname
            input_to_user_nickname = cmsg.to_user_nickname 

            input_wx_content_dialogue_message=[{"type": "image_url", "image_url": {"url": input_content}}]
            input_message=dialogue_message(input_from_user_nickname,input_to_user_nickname,input_wx_content_dialogue_message)
            kafka_helper.kafka_client.produce_message(input_message)
            logger.info("发送对话 %s",input_message)

        elif cmsg.ctype == ContextType.PATPAT:
            logger.debug("[WX]receive patpat msg: {}".format(cmsg.content))
        elif cmsg.ctype == ContextType.TEXT:
            logger.debug("[WX]receive text msg: {}, cmsg={}".format(json.dumps(cmsg._rawmsg, ensure_ascii=False), cmsg))
            # content = cmsg.content  # 消息内容
            # from_user_nickname = cmsg.from_user_nickname  # 发送方昵称
            # to_user_nickname = cmsg.to_user_nickname  # 接收方昵称

            # wx_content_dialogue_message=[{"type": "text", "text": content}]
            # message=dialogue_message(from_user_nickname,to_user_nickname,wx_content_dialogue_message)
            # kafka_helper.kafka_client.produce_message(message)
            # logger.info("发送对话 %s", json.dumps(message, ensure_ascii=False))

            input_content = cmsg.content
            input_from_user_nickname = cmsg.from_user_nickname
            input_to_user_nickname = cmsg.to_user_nickname 

            input_wx_content_dialogue_message=[{"type": "text", "text": input_content}]
            input_message=dialogue_message(input_from_user_nickname,input_to_user_nickname,input_wx_content_dialogue_message)
            kafka_helper.kafka_client.produce_message(input_message)
            logger.info("发送对话 %s",input_message)


        else:
            logger.debug("[WX]receive msg: {}, cmsg={}".format(cmsg.content, cmsg))
        context = self._compose_context(cmsg.ctype, cmsg.content, isgroup=False, msg=cmsg)
        if context:
            self.produce(context)

    @time_checker
    @_check
    def handle_group(self, cmsg: ChatMessage):
        if cmsg.ctype == ContextType.VOICE:
            if conf().get("group_speech_recognition") != True:
                return
            logger.debug("[WX]receive voice for group msg: {}".format(cmsg.content))
        elif cmsg.ctype == ContextType.IMAGE:
            logger.debug("[WX]receive image for group msg: {}".format(cmsg.content))
        elif cmsg.ctype in [ContextType.JOIN_GROUP, ContextType.PATPAT, ContextType.ACCEPT_FRIEND, ContextType.EXIT_GROUP]:
            logger.debug("[WX]receive note msg: {}".format(cmsg.content))
        elif cmsg.ctype == ContextType.TEXT:
            # logger.debug("[WX]receive group msg: {}, cmsg={}".format(json.dumps(cmsg._rawmsg, ensure_ascii=False), cmsg))
            pass
        elif cmsg.ctype == ContextType.FILE:
            logger.debug(f"[WX]receive attachment msg, file_name={cmsg.content}")
        else:
            logger.debug("[WX]receive group msg: {}".format(cmsg.content))
        context = self._compose_context(cmsg.ctype, cmsg.content, isgroup=True, msg=cmsg, no_need_at=conf().get("no_need_at", False))
        if context:
            self.produce(context)

    # 统一的发送函数，每个Channel自行实现，根据reply的type字段发送不同类型的消息
    def send(self, reply: Reply, context: Context):
        receiver = context["receiver"]
        if reply.type == ReplyType.TEXT:
            itchat.send(reply.content, toUserName=receiver)
            logger.info("[WX] sendMsg={}, receiver={}".format(reply, receiver))
            # logger.info(context)
            # logger.info(context["msg"])
            # // 发送kafka
            # msg=context["msg"]
            msg: ChatMessage = context["msg"]
            # content=msg["content"]
            
           
            is_group=msg.is_group
            if not is_group:
                # print(f'响应：{content}')
                # 用户输入
                # input_content=msg.content
                # input_from_user_nickname=msg.from_user_nickname
                # input_to_user_nickname=msg.to_user_nickname

                # input_wx_content_dialogue_message=[{"type": "text", "text": input_content}]
                # input_message=dialogue_message(input_from_user_nickname,input_to_user_nickname,input_wx_content_dialogue_message)
                # kafka_helper.kafka_client.produce_message(input_message)
                # logger.info("发送对话 %s", json.dumps(input_message, separators=(',', ':'), ensure_ascii=False))

                # 响应用户
                output_content=reply.content
                output_from_user_nickname=msg.to_user_nickname # 回复翻转
                output_to_user_nickname=msg.from_user_nickname # 回复翻转

                output_wx_content_dialogue_message=[{"type": "text", "text": output_content}]
                output_message=dialogue_message(output_from_user_nickname,output_to_user_nickname,output_wx_content_dialogue_message)
                kafka_helper.kafka_client.produce_message(output_message)
                logger.info("发送对话 %s", output_message)
        elif reply.type == ReplyType.ERROR or reply.type == ReplyType.INFO:
            itchat.send(reply.content, toUserName=receiver)
            logger.info("[WX] sendMsg={}, receiver={}".format(reply, receiver))
        elif reply.type == ReplyType.VOICE:
            itchat.send_file(reply.content, toUserName=receiver)
            logger.info("[WX] sendFile={}, receiver={}".format(reply.content, receiver))
        elif reply.type == ReplyType.IMAGE_URL:  # 从网络下载图片
            img_url = reply.content
            logger.debug(f"[WX] start download image, img_url={img_url}")
            pic_res = requests.get(img_url, stream=True)
            image_storage = io.BytesIO()
            size = 0
            for block in pic_res.iter_content(1024):
                size += len(block)
                image_storage.write(block)
            logger.info(f"[WX] download image success, size={size}, img_url={img_url}")
            image_storage.seek(0)
            if ".webp" in img_url:
                try:
                    image_storage = convert_webp_to_png(image_storage)
                except Exception as e:
                    logger.error(f"Failed to convert image: {e}")
                    return
            itchat.send_image(image_storage, toUserName=receiver)
            logger.info("[WX] sendImage url={}, receiver={}".format(img_url, receiver))
        elif reply.type == ReplyType.IMAGE:  # 从文件读取图片
            image_storage = reply.content
            image_storage.seek(0)
            itchat.send_image(image_storage, toUserName=receiver)
            logger.info("[WX] sendImage, receiver={}".format(receiver))
        elif reply.type == ReplyType.FILE:  # 新增文件回复类型
            file_storage = reply.content
            itchat.send_file(file_storage, toUserName=receiver)
            logger.info("[WX] sendFile, receiver={}".format(receiver))


            # msg: ChatMessage = context["msg"]
            # # content=msg["content"]
            
           
            # is_group=msg.is_group
            # if not is_group:
            #     # print(f'响应：{content}')
            #     # 用户输入
            #     # input_content=msg.content
            #     # input_from_user_nickname=msg.from_user_nickname
            #     # input_to_user_nickname=msg.to_user_nickname

            #     # input_wx_content_dialogue_message=[{"type": "text", "text": input_content}]
            #     # input_message=dialogue_message(input_from_user_nickname,input_to_user_nickname,input_wx_content_dialogue_message)
            #     # kafka_helper.kafka_client.produce_message(input_message)
            #     # logger.info("发送对话 %s", json.dumps(input_message, separators=(',', ':'), ensure_ascii=False))

            #     # 响应用户
            #     output_content=reply.content
            #     output_from_user_nickname=msg.to_user_nickname # 回复翻转
            #     output_to_user_nickname=msg.from_user_nickname # 回复翻转

            #     output_wx_content_dialogue_message=[{"type": "file", "text": output_content}]
            #     output_message=dialogue_message(output_from_user_nickname,output_to_user_nickname,output_wx_content_dialogue_message)
            #     kafka_helper.kafka_client.produce_message(output_message)
            #     logger.info("发送对话 %s", output_message)
            

        elif reply.type == ReplyType.VIDEO:  # 新增视频回复类型
            video_storage = reply.content
            itchat.send_video(video_storage, toUserName=receiver)
            logger.info("[WX] sendFile, receiver={}".format(receiver))
        elif reply.type == ReplyType.VIDEO_URL:  # 新增视频URL回复类型
            video_url = reply.content
            logger.debug(f"[WX] start download video, video_url={video_url}")
            video_res = requests.get(video_url, stream=True)
            video_storage = io.BytesIO()
            size = 0
            for block in video_res.iter_content(1024):
                size += len(block)
                video_storage.write(block)
            logger.info(f"[WX] download video success, size={size}, video_url={video_url}")
            video_storage.seek(0)
            itchat.send_video(video_storage, toUserName=receiver)
            logger.info("[WX] sendVideo url={}, receiver={}".format(video_url, receiver))

def _send_login_success():
    try:
        from common.linkai_client import chat_client
        if chat_client.client_id:
            chat_client.send_login_success()

    except Exception as e:
        pass


def _send_logout():
    try:
        from common.linkai_client import chat_client
        if chat_client.client_id:
            chat_client.send_logout()
    except Exception as e:
        pass


def _send_qr_code(qrcode_list: list):
    try:
        from common.linkai_client import chat_client
        if chat_client.client_id:
            chat_client.send_qrcode(qrcode_list)
    except Exception as e:
        pass


def clean_json_string(json_str):

    # 删除所有控制字符（非打印字符），包括换行符、回车符等
    return re.sub(r'[\x00-\x1f\x7f]', '', json_str)

def save_friends_to_redis(agent_tel,agent_nickname, friends):
    contact_list = []
    for friend in friends:
        friend_data = {
            "UserName": friend.UserName,
            "NickName": friend.NickName,
            "Signature": friend.Signature,
            "Province": friend.Province,
            "City": friend.City,
            "Sex": str(friend.Sex),  # 性别可转换为字符串存储
            "Alias": friend.Alias
        }
        contact_list.append(friend_data)  # 将每个朋友的信息加入到列表中

    agent_contact_list = {
        "AgentTel":agent_tel,
        "agent_nick_name": agent_nickname,
        "contact_list": contact_list  # 将朋友列表添加到字典中
    }
    
    # 将联系人信息保存到 Redis，使用一个合适的 key
    hash_key = f"__AI_OPS_WX__:CONTACTLIST"
    redis_helper.redis_helper.update_hash_field(hash_key, agent_tel, json.dumps(agent_contact_list, ensure_ascii=False))  # 设置有效期为 600 秒

def hourly_change_save_friends(agent_nickname):
    last_hour = datetime.now().hour  # 获取当前小时
    while True:
        current_hour = datetime.now().hour
        if current_hour != last_hour:  # 检测小时是否变化
            friends=itchat.get_friends(update=True)[1:]

            agent_info=fetch_agent_info(agent_nickname)
            agent_tel=agent_info.get("agent_tel",None)
            save_friends_to_redis(agent_tel,agent_nickname, friends)
            last_hour = current_hour
        time.sleep(1)  # 每秒检查一次

def wx_messages_process_callback(user_nickname,message):
    """
    处理消费到的 Kafka 消息（基础示例）
    :param message: Kafka 消费到的消息内容
    """
    # print(user_nickname)
    # print(f"Processing message: {message}")
    # return True

    msg_content= message
    cleaned_content = clean_json_string(msg_content)
    content=json.loads(cleaned_content)
    data = content.get("data", {})  
    msg_type_data=data.get("msg_type",None)  
    content_data = data.get("content",{})  
    agent_nickname_data=content_data.get("agent_nickname",None)
    agent_tel=content_data.get("agent_tel",None)

    if user_nickname == agent_nickname_data and msg_type_data=='group-sending':
        friends=itchat.get_friends(update=True)[1:]
        contact_list_content_data=content_data.get("contact_list",None)

        # 更新好友缓存
        save_friends_to_redis(agent_tel,agent_nickname_data,friends)

        # 遍历要群发的好友
        for contact in contact_list_content_data:
            nickname = contact.get("nickname",None)
            if(nickname not in [friend['NickName'] for friend in friends]):
                logger.warning(f'微信中没有 {nickname} 的昵称,将不会发送消息')

            for friend in friends:
                if friend.get("NickName",None) == nickname:
                    wx_content_list=content_data.get("wx_content",[])
                    for wx_content in wx_content_list:
                        # 处理文件
                        if wx_content.get("type",None) == 'text':
                            wx_content_text=wx_content.get("text",None)
                            itchat.send(wx_content_text, toUserName=friend.get("UserName",None))
                            logger.info(f"{user_nickname} 向 {nickname} 发送文字【 {wx_content_text} 】")

                             # // 发送kafka
                            wx_content_dialogue_message=[{"type": "text", "text": wx_content_text}]
                            message=dialogue_message(agent_nickname_data,friend.get("NickName",None),wx_content_dialogue_message)
                            kafka_helper.kafka_client.produce_message(message)
                            logger.info("发送对话 %s",message)

                            time.sleep(3)
                        # 处理图片   
                        elif wx_content.get("type",None) == 'image_url':
                            print('发送图片')
                            image_url= wx_content.get("image_url",{})
                            url=image_url.get("url",None)
                            
                            # 网络图片
                            logger.debug(f"[WX] start download image, img_url={url}")
                            pic_res = requests.get(url, stream=True)
                            image_storage = io.BytesIO()
                            size = 0
                            for block in pic_res.iter_content(1024):
                                size += len(block)
                                image_storage.write(block)
                            logger.info(f"[WX] download image success, size={size}, img_url={url}")
                            image_storage.seek(0)
                            if ".webp" in url:
                                try:
                                    image_storage = convert_webp_to_png(image_storage)
                                except Exception as e:
                                    logger.error(f"Failed to convert image: {e}")
                                    return

                            itchat.send_image(image_storage, toUserName=friend.get("UserName",None))
                            logger.info(f"{user_nickname} 向 {nickname} 发送图片【 {url} 】")
                            
                            # // 发送kafka
                            wx_content_dialogue_message=[{"type": "image_url", "image_url": {"url": url}}]
                            message=dialogue_message(agent_nickname_data,friend.get("NickName",None),wx_content_dialogue_message)
                            kafka_helper.kafka_client.produce_message(message)
                            logger.info("发送对话 %s",message)
                            time.sleep(3)
                        #处理文件
                        elif wx_content.get("type",None) == 'file':
                            print('处理文件')
                            file_url= wx_content.get("file_url",{})
                            url=file_url.get("url",None)

                            # 提取路径部分
                            parsed_url = urlparse(url).path

                            # 获取文件名和扩展名
                            filename = os.path.basename(parsed_url)  # 文件名（包含扩展名）
                            name, ext = os.path.splitext(filename)  # 分离文件名和扩展名
                            if ext in ['.pdf']:
                                print('处理PDF文件')

                                tmp_file_path=save_to_local_from_url(url)

                                itchat.send_file(tmp_file_path, toUserName=friend.get("UserName",None))
                                logger.info(f'删除本地{ext}文件: {tmp_file_path}')
                                os.remove(tmp_file_path)
                                logger.info(f"{user_nickname} 向 {nickname} 发送 {ext} 文件【 {url} 】")
                                # // 发送kafka
                                wx_content_dialogue_message=[{"type": "file", "file_url": {"url": url}}]
                                message=dialogue_message(agent_nickname_data,friend.get("NickName",None),wx_content_dialogue_message)
                                kafka_helper.kafka_client.produce_message(message)
                                logger.info("发送对话 %s",message)
                                time.sleep(3)

                            elif ext in ['.mp4']:

                                print('处理MP4文件')
                                tmp_file_path=save_to_local_from_url(url)
                                itchat.send_file(tmp_file_path, toUserName=friend.get("UserName",None))
                                logger.info(f'删除本地{ext}文件: {tmp_file_path}')
                                os.remove(tmp_file_path)
                                logger.info(f"{user_nickname} 向 {nickname} 发送 {ext} 文件【 {url} 】")
                                 # // 发送kafka
                                wx_content_dialogue_message=[{"type": "file", "file_url": {"url": url}}]
                                message=dialogue_message(agent_nickname_data,friend.get("NickName",None),wx_content_dialogue_message)
                                kafka_helper.kafka_client.produce_message(message)
                                logger.info("发送对话 %s",message)
                                time.sleep(3)
                                
                            else:
                                logger.warning(f'暂不支持 {ext} 文件的处理')
                        
        return True
    else:
        return False

def dialogue_message(nickname_from,nickname_to,wx_content):
    """
    构造消息的 JSON 数据
    :param contents: list，包含多个消息内容，每个内容为字典，如：
                     [{"type": "text", "text": "AAAAAAA"},
                      {"type": "image_url", "image_url": {"url": "https://AAAAA.jpg"}},
                      {"type":"file","file_url":{"url":"https://AAAAA.pdf"}}
                      ]
    :return: JSON 字符串
    """

    # 获取当前时间戳，精确到毫秒
    current_timestamp = int(time.time() * 1000)

    # 获取当前时间，格式化为 "YYYY-MM-DD HH:MM:SS"
    current_time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
    
    # 构造 JSON 数据
    data = {
        "messageId": str(current_timestamp),
        "topic": "topic.aiops.wx",
        "time": current_time,
        "data": {
            "msg_type": "dialogue",
            "content": {
                "nickname_from": nickname_from,
                "nickname_to": nickname_to,
                "wx_content":wx_content
            }
        }
    }

    return json.dumps(data, separators=(',', ':'), ensure_ascii=False)

def fetch_agent_info(agent_nickname):

    if os.environ.get('environment', 'default')=='default':
        return {
                    "agent_nickname": agent_nickname,
                    "agent_tel": "19200137635"
                }

    aiops_api=conf().get("aiops_api")
    # 定义请求URL
    url = f"{aiops_api}/business/Agent/smartinfobyname"
    
    # 定义请求头
    headers = {
        "accept": "*/*",
        "Content-Type": "application/json"
    }
    
    # 定义请求数据
    data = {
        "smartName": agent_nickname
    }
    
    try:
        # 发送POST请求
        response = requests.post(url, headers=headers, data=json.dumps(data))
        
        # 确认响应状态码
        if response.status_code == 200:
            response_data = response.json()
            if response_data.get("code") == 200:
                # 提取 smartName 和 smartPhone
                data = response_data.get("data", {})
                return {
                    "agent_nickname": data.get("smartName"),
                    "agent_tel": data.get("smartPhone")
                }
            else:
                logger.error(f"API 返回错误信息: {response_data.get('msg')}")
                return None
        else:
            logger.error(f"请求失败，状态码：{response.status_code}")
            return None
    except Exception as e:
        logger.error(f"请求出错: {e}")
        return None


def save_to_local_from_url(url):
    '''
    从url保存到本地tmp目录
    '''

    parsed_url = urlparse(url)
     # 从 URL 提取文件名
    filename = os.path.basename(parsed_url.path)
    # tmp_dir = os.path(__file__)  # 获取系统临时目录
    # print(tmp_dir)
    tmp_file_path = os.path.join(os.getcwd(),'tmp', filename)  # 拼接完整路径

    # 检查是否存在同名文件
    if os.path.exists(tmp_file_path):
        logger.info(f"文件已存在，将覆盖：{tmp_file_path}")

    # 下载文件并保存到临时目录
    response = requests.get(url, stream=True)
    with open(tmp_file_path, 'wb') as f:
        for chunk in response.iter_content(chunk_size=1024):
            if chunk:  # 检查是否有内容
                f.write(chunk)

    return tmp_file_path

def upload_oss(access_key_id, access_key_secret, endpoint, bucket_name, local_file_path, oss_file_name, expiration_days=7):
    """
    上传文件到阿里云OSS并设置生命周期规则，同时返回文件的公共访问地址。
    
    :param access_key_id: 阿里云AccessKey ID
    :param access_key_secret: 阿里云AccessKey Secret
    :param endpoint: OSS区域对应的Endpoint
    :param bucket_name: OSS中的Bucket名称
    :param local_file_path: 本地文件路径
    :param oss_file_name: OSS中的文件存储路径
    :param expiration_days: 文件保存天数，默认7天后删除
    :return: 文件的公共访问地址
    """
    
    # 创建Bucket实例
    auth = oss2.Auth(access_key_id, access_key_secret)
    bucket = oss2.Bucket(auth, endpoint, bucket_name)

    ### 1. 设置生命周期规则 ###
    rule_id = f'delete_after_{expiration_days}_days'  # 规则ID
    prefix = oss_file_name.split('/')[0] + '/'  # 设置规则应用的前缀为文件所在目录

    # 定义生命周期规则
    rule = oss2.models.LifecycleRule(rule_id, prefix, status=oss2.models.LifecycleRule.ENABLED,
                                     expiration=oss2.models.LifecycleExpiration(days=expiration_days))

    # 设置Bucket的生命周期
    lifecycle = oss2.models.BucketLifecycle([rule])
    bucket.put_bucket_lifecycle(lifecycle)

    print(f"已设置生命周期规则：文件将在{expiration_days}天后自动删除")

    ### 2. 上传文件到OSS ###
    bucket.put_object_from_file(oss_file_name, local_file_path)

    ### 3. 构建公共访问URL ###
    file_url = f"http://{bucket_name}.{endpoint.replace('http://', '')}/{oss_file_name}"

    print(f"文件上传成功，公共访问地址：{file_url}")
    
    return file_url