AiOpsWechatiPad/wechat/biz.py

import threading

from common import kafka_helper,redis_helper,utils

import json,time,re,random,os
from common.log import logger, log_exception 
from datetime import datetime


from wechat import gewe_chat



def wx_messages_process_callback(agent_tel,message):
    try:
        # print(f'手机号 {agent_tel}')
        wxchat = gewe_chat.wxchat
        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_tel = content_data.get("agent_tel", None)
        
        if msg_type_data == 'group-sending':
            process_group_sending(wxchat, content_data, agent_tel)
            
    except json.JSONDecodeError as e:
        print(f"JSON解码错误: {e}, 消息内容: {message}")
    except Exception as e:
        print(f"处理消息时发生错误: {e}, 消息内容: {message}")

    

def process_group_sending_v0(wxchat:gewe_chat.GeWeChatCom, content_data, agent_tel:str):
    # 获取登录信息
    hash_key = f"__AI_OPS_WX__:LOGININFO:{agent_tel}"
    logininfo = redis_helper.redis_helper.get_hash(hash_key)
    
    if not logininfo:
        logger.warning(f"未找到 {agent_tel} 的登录信息")
        return
    
    token_id = logininfo.get('tokenId')
    app_id = logininfo.get('appId')
    agent_wxid = logininfo.get('wxid')
    
     
    # 获取联系人列表并计算交集
    hash_key = f"__AI_OPS_WX__:CONTACTS_BRIEF:{agent_wxid}"
    cache_friend_wxids_str=redis_helper.redis_helper.get_hash_field(hash_key,"data")
    cache_friend_wxids_list=json.loads(cache_friend_wxids_str)  if cache_friend_wxids_str else []
    cache_friend_wxids=[f["userName"] for f in cache_friend_wxids_list]


    wxid_contact_list_content_data = [c['wxid'] for c in content_data.get("contact_list", [])]
    intersection_wxids = list(set(cache_friend_wxids) & set(wxid_contact_list_content_data))
    
    # 发送消息
    wx_content_list = content_data.get("wx_content", [])
    for wx_content in wx_content_list:
        if wx_content["type"] == "text":
            send_text_message(wxchat, token_id, app_id, agent_wxid, intersection_wxids, wx_content["text"])
        elif wx_content["type"] == "image_url":
            send_image_message(wxchat, token_id, app_id, agent_wxid, intersection_wxids, wx_content.get("image_url", {}).get("url"))
        elif wx_content["type"] == "tts":
            send_tts_message(wxchat, token_id, app_id, agent_wxid, intersection_wxids, wx_content["text"])


def process_group_sending(wxchat:gewe_chat.GeWeChatCom, content_data, agent_tel:str):
    # 获取登录信息
    hash_key = f"__AI_OPS_WX__:LOGININFO:{agent_tel}"
    logininfo = redis_helper.redis_helper.get_hash(hash_key)
    
    if not logininfo:
        logger.warning(f"未找到 {agent_tel} 的登录信息")
        return
    
    token_id = logininfo.get('tokenId')
    app_id = logininfo.get('appId')
    agent_wxid = logininfo.get('wxid')
    
     
    # 获取联系人列表并计算交集
    hash_key = f"__AI_OPS_WX__:CONTACTS_BRIEF:{agent_wxid}"
    cache_friend_wxids_str=redis_helper.redis_helper.get_hash_field(hash_key,"data")
    cache_friend_wxids_list=json.loads(cache_friend_wxids_str)  if cache_friend_wxids_str else []
    cache_friend_wxids=[f["userName"] for f in cache_friend_wxids_list]


    # 获取群交集
    hash_key = f"__AI_OPS_WX__:GROUPS_INFO:{agent_wxid}"
    cache_chatrooms = redis_helper.redis_helper.get_hash(hash_key)
    cache_chatroom_ids=cache_chatrooms.keys()



    wxid_contact_list_content_data = [c['wxid'] for c in content_data.get("contact_list", [])]
    intersection_friend_wxids = list(set(cache_friend_wxids) & set(wxid_contact_list_content_data))
    intersection_chatroom_ids = list(set(cache_chatroom_ids) & set(wxid_contact_list_content_data))

    intersection_wxids=intersection_friend_wxids+intersection_chatroom_ids
    
    # 发送消息
    wx_content_list = content_data.get("wx_content", [])
    # for wx_content in wx_content_list:
    #     if wx_content["type"] == "text":
    #         send_text_message(wxchat, token_id, app_id, agent_wxid, intersection_wxids, wx_content["text"])
    #     elif wx_content["type"] == "image_url":
    #         send_image_message(wxchat, token_id, app_id, agent_wxid, intersection_wxids, wx_content.get("image_url", {}).get("url"))
    #     elif wx_content["type"] == "tts":
    #         send_tts_message(wxchat, token_id, app_id, agent_wxid, intersection_wxids, wx_content["text"])
    
    for intersection_wxid in intersection_wxids:
        for wx_content in wx_content_list:
            if wx_content["type"] == "text":
                send_text_message(wxchat, token_id, app_id, agent_wxid, [intersection_wxid], wx_content["text"])
            elif wx_content["type"] == "image_url":
                send_image_message(wxchat, token_id, app_id, agent_wxid, [intersection_wxid], wx_content.get("image_url", {}).get("url"))
            elif wx_content["type"] == "tts":
                send_tts_message(wxchat, token_id, app_id, agent_wxid, [intersection_wxid], wx_content["text"])



def send_text_message(wxchat:gewe_chat.GeWeChatCom, token_id, app_id, agent_wxid, intersection_wxids, text):
    for t in intersection_wxids:
        # 发送文本消息
        ret,ret_msg,res = wxchat.post_text(token_id, app_id, t, text)
        logger.info(f'{agent_wxid} 向 {t} 发送文字【{text}】')
        
        # 构造对话消息并发送到 Kafka
        input_wx_content_dialogue_message = [{"type": "text", "text": text}]
        input_message = utils.dialogue_message(agent_wxid, t, input_wx_content_dialogue_message)
        kafka_helper.kafka_client.produce_message(input_message)
        logger.info("发送对话 %s", input_message)
        
        # 等待随机时间
        time.sleep(random.uniform(5, 15))

def send_image_message(wxchat:gewe_chat.GeWeChatCom, token_id, app_id, agent_wxid, intersection_wxids, image_url):

    aeskey, cdnthumburl, cdnthumblength, cdnthumbheight, cdnthumbwidth, length, md5 = "", "", 0, 0, 0, 0, ""
    
    for t in intersection_wxids:
        if t == intersection_wxids[0]:
            # 发送图片
            ret,ret_msg,res = wxchat.post_image(token_id, app_id, t, image_url)
            if ret==200:
                aeskey = res["aesKey"]
                cdnthumburl = res["fileId"]
                cdnthumblength = res["cdnThumbLength"]
                cdnthumbheight = res["height"]
                cdnthumbwidth = res["width"]
                length = res["length"]
                md5 = res["md5"]
                logger.info(f'{agent_wxid} 向 {t} 发送图片【{image_url}】{ret_msg}')
            else:
                logger.warning(f'{agent_wxid} 向 {t} 发送图片【{image_url}】{ret_msg}')

        else:
            if aeskey !="":
                # 转发图片
                res,ret,ret_msg= wxchat.forward_image(token_id, app_id, t, aeskey, cdnthumburl, cdnthumblength, cdnthumbheight, cdnthumbwidth, length, md5)
                logger.info(f'{agent_wxid} 向 {t} 转发图片【{image_url}】{ret_msg}')
            else:
                # 发送图片
                ret,ret_msg,res = wxchat.post_image(token_id, app_id, t, image_url)
                if ret==200:
                    aeskey = res["aesKey"]
                    cdnthumburl = res["fileId"]
                    cdnthumblength = res["cdnThumbLength"]
                    cdnthumbheight = res["height"]
                    cdnthumbwidth = res["width"]
                    length = res["length"]
                    md5 = res["md5"]
                    logger.info(f'{agent_wxid} 向 {t} 发送图片【{image_url}】{ret_msg}')
                else:
                    logger.warning(f'{agent_wxid} 向 {t} 发送图片【{image_url}】{ret_msg}')
        
        # 构造对话消息并发送到 Kafka
        wx_content_dialogue_message = [{"type": "image_url", "image_url": {"url": image_url}}]
        input_message = utils.dialogue_message(agent_wxid, t, wx_content_dialogue_message)
        kafka_helper.kafka_client.produce_message(input_message)
        logger.info("发送对话 %s", input_message)
        
        # 等待随机时间
        time.sleep(random.uniform(5, 15))

def send_tts_message(wxchat:gewe_chat.GeWeChatCom, token_id, app_id, agent_wxid, intersection_wxids, text):
    
    voice_during,voice_url=utils.wx_voice(text)
    for t in intersection_wxids:
        # 发送送语音消息
        if voice_url:
            ret,ret_msg,res = wxchat.post_voice(token_id, app_id, t, voice_url,voice_during)
            if ret==200:
                logger.info(f'{agent_wxid} 向 {t} 发送语音文本【{text}】{ret_msg}')
                
                # 构造对话消息并发送到 Kafka
                input_wx_content_dialogue_message = [{"type": "text", "text": text}]
                input_message = utils.dialogue_message(agent_wxid, t, input_wx_content_dialogue_message)
                kafka_helper.kafka_client.produce_message(input_message)
                logger.info("发送对话 %s", input_message)
            else:
                logger.warning((f'{agent_wxid} 向 {t} 发送语音文本【{text}】{ret_msg}'))
        else:
            logger.warning((f'{agent_wxid} 向 {t} 发送语音文本【{text}】出错'))
        
        # 等待随机时间
        time.sleep(random.uniform(5, 15))

def clean_json_string(json_str):

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

# 启动 Kafka 消费者线程
def start_kafka_consumer_thread():
    agent_tel=os.environ.get('tel', '18029274615')
    # consumer_thread = threading.Thread(target=kafka_helper.kafka_client.consume_messages, args=(agent_tel,wx_messages_process_callback,))
    consumer_thread = threading.Thread(target=kafka_helper.kafka_client.consume_messages, args=(ops_messages_process,))
    consumer_thread.daemon = True  # 设置为守护线程，应用退出时会自动结束
    consumer_thread.start()


def login_or_reconnect(wxchat:gewe_chat.GeWeChatCom, token_id, app_id, hash_key, is_reconnect=False, max_retries=5):
    """
    封装微信登录或重连的逻辑
    """
    agent_tel=hash_key.split(":")[-1]
    retry_count = 0
    while retry_count < max_retries:
        retry_count += 1
        if is_reconnect:
            logger.info("尝试重连...")
        else:
            logger.info("获取二维码进行登录...")
        
        qr_code = wxchat.get_login_qr_code(token_id, app_id)
        base64_string = qr_code.get('qrImgBase64')
        uuid = qr_code.get('uuid')

        app_id = app_id or qr_code.get('appId')
        start_time = time.time()
        qr_code_urls= wxchat.qrCallback(uuid, base64_string)
        # 构造 Kafka 消息发送二维码
        k_message=utils.login_qrcode_message(token_id,agent_tel,base64_string,qr_code_urls)
        kafka_helper.kafka_client.produce_message(k_message)
        while True:
            # 如果登录超时，重新获取二维码
            if time.time() - start_time > 150:  #150 秒 二维码失效
                break
            res = wxchat.check_login(token_id, app_id, uuid)
            flag = res.get('status')
            # 构造 Kafka 消息发送登录状态
            # todo
            if flag == 2:
                logger.info(f"登录成功: {res}")
                login_info = res.get('loginInfo', {})

                login_info.update({'appId': app_id, 'uuid': uuid, 'tokenId': token_id,'status': 1})
                # cache_login_info=redis_helper.redis_helper.get_hash(hash_key)

                # if 'appId' in cache_login_info:
                #     login_info.update({"reg_time":datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")[:-3],"login_time":datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")[:-3]})   
                # else:
                #     login_info.update({"login_time":datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")[:-3]})
                cleaned_login_info = {k: (v if v is not None else '') for k, v in login_info.items()}
                redis_helper.redis_helper.set_hash(hash_key, cleaned_login_info)
                return login_info
            time.sleep(5)
    logger.error(f"登录失败，二维码生成 {max_retries} 次")

def fetch_and_save_contacts(wxchat:gewe_chat.GeWeChatCom, token_id, app_id, hash_key):
    """
    获取联系人列表并保存到缓存
    """
    ret,msg,contacts_list = wxchat.fetch_contacts_list(token_id, app_id)
    friend_wxids = contacts_list['friends'][3:]  # 可以调整截取范围
    wxid = redis_helper.redis_helper.get_hash_field(hash_key, 'wxid')
    
    wxchat.save_contacts_brief_to_cache(token_id, app_id, wxid, friend_wxids)
    print(f'微信ID {wxid} 登录APPID {app_id} 成功，联系人已保存')


def wx_login(wxchat:gewe_chat.GeWeChatCom,tel,token_id):
        hash_key = f"__AI_OPS_WX__:LOGININFO:{tel}"
        login_info = redis_helper.redis_helper.get_hash(hash_key)

        if not login_info:
            login_info = login_or_reconnect(wxchat, token_id, '', hash_key)
        else:
            app_id = login_info.get('appId')
            token_id = login_info.get('tokenId')
            wxid= login_info.get('wxid')
            # 检查是否已经登录
            is_online = wxchat.check_online(token_id, app_id)
            if is_online:
                logger.info(f'微信ID {wxid} 在APPID {app_id} 已经在线')
            else:
                # 尝试重连
                res = wxchat.reconnection(token_id, app_id)
                if res.get('ret') == 200:
                    logger.info(f'微信ID {wxid} 在APPID {app_id} 重连成功')
                else:
                    print("重连失败，重新登录...")
                    login_info = login_or_reconnect(wxchat, token_id, app_id, hash_key, is_reconnect=True)

        fetch_and_save_contacts(wxchat, token_id, login_info.get('appId'), hash_key)    

def ops_messages_process(message):
    try:
        wxchat = gewe_chat.wxchat
        #print(message)
        # logger.info(f"接收到kafka消息: {json.dumps(message, separators=(',', ':'), ensure_ascii=False)}")  
        logger.info(f"接收到kafka消息: {json.dumps(json.loads(message), ensure_ascii=False)}")
        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", {})  
        if msg_type_data=="login":

            tel=content_data.get('tel', '18029274615')
            token_id=content_data.get('token_id', 'f828cb3c-1039-489f-b9ae-7494d1778a15')

            #wx_login(wxchat,tel,token_id)
            thread = threading.Thread(target=wx_login, args=(wxchat,tel,token_id,))
            thread.daemon = True 
            thread.start()
        elif msg_type_data == 'group-sending':
            agent_tel=content_data.get('agent_tel', '18029274615')
            # 使用线程处理
            #wx_messages_process_callback(agent_tel,message)

            thread = threading.Thread(target=wx_messages_process_callback, args=(agent_tel,message,))
            thread.daemon = True 
            thread.start()
        else:
            print(f'未处理息类型 {msg_type_data}')

    except Exception as e:
        print(f"处理消息时发生错误: {e}, 消息内容: {message}")

        
# 启动
# def consumer_start_up_thread():
#     print('启动')
#     try:
#         print('启动')
#     except Exception as e:
#         print(f"处理消息时发生错误: {e}, 消息内容: {message}")