import threading from common import kafka_helper,redis_helper,utils import json,time,re,random,os from common.log import logger, log_exception 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(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') # 获取联系人列表并计算交集 # contacts_list = wxchat.fetch_contacts_list(token_id, app_id) # contacts_list = wxchat.fetch_contacts_list(token_id, app_id) # friend_wxids = contacts_list['friends'] hash_key = f"__AI_OPS_WX__:CONTACTS_BRIEF:{agent_wxid}" friend_wxids_str=redis_helper.redis_helper.get_hash_field(hash_key,"data") friend_wxids_list=json.loads(friend_wxids_str) if friend_wxids_str else [] friend_wxids=[f["userName"] for f in friend_wxids_list] print(friend_wxids) wxid_contact_list_content_data = [c['wxid'] for c in content_data.get("contact_list", [])] intersection_wxids = list(set(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 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.daemon = True # 设置为守护线程，应用退出时会自动结束 consumer_thread.start()