from flask_restful import Resource, reqparse from flask import jsonify,request import requests,json,re from wechat import gewe_chat from voice.ali.ali_voice import AliVoice from common import utils,redis_helper,memory,kafka_helper from common.log import logger import xml.etree.ElementTree as ET import threading,time import os from voice import audio_convert timeout_duration = 8.0 class MessagesResource(Resource): def __init__(self): self.parser = reqparse.RequestParser() def post(self): msg = request.get_json() logger.info(f"收到微信回调消息: {msg}") type_name =msg.get("TypeName") app_id = msg.get("Appid") # token_id = "f828cb3c-1039-489f-b9ae-7494d1778a15" token_id=get_token_id_by_app_id(app_id) if token_id=="": logger.warning('找不到登录信息，不处理') return jsonify({"message": "收到微信回调消息"}) wxid = msg.get("Wxid",'') wx_config = gewe_chat.wxchat.get_wxchat_config_from_cache(wxid) if not bool(wx_config.get("agentEnabled",False)): logger.info('智能体未启用，不处理') return jsonify({"message": "收到微信回调消息"}) if type_name=='AddMsg': #wxid = msg.get("Wxid") msg_data = msg.get("Data") msg_type = msg_data.get("MsgType",None) from_wxid = msg_data["FromUserName"]["string"] to_wxid = msg_data["ToUserName"]["string"] msg_push_content=msg_data.get("PushContent") #群发 #wx_config = gewe_chat.wxchat.get_wxchat_config_from_cache(wxid) handlers = { 1: handle_text, 3: handle_image, 34: handle_voice, 49: handle_xml, 37: handle_add_friend_notice, 10002: handle_10002_msg } # (扫码进群情况)判断受否是群聊,并添加到通信录 if check_chatroom(from_wxid) or check_chatroom(to_wxid): logger.info('群信息') chatroom_id=from_wxid ret,msg,data=gewe_chat.wxchat.save_contract_list(token_id,app_id,chatroom_id,3) logger.info(f'保存到通讯录 chatroom_id {chatroom_id} {msg}') gewe_chat.wxchat.update_group_info_to_cache(token_id,app_id,wxid,chatroom_id) handlers[1]=handle_text_group handlers[3]=handle_image_group handlers[34]=handle_voice_group handler = handlers.get(msg_type) if handler: return handler(token_id,app_id, wxid,msg_data,from_wxid, to_wxid) else: logger.warning(f"微信回调消息类型 {msg_type} 未处理") elif type_name=='ModContacts': ''' 好友通过验证及好友资料变更的通知消息 ''' wxid = msg.get("Wxid") msg_data = msg.get("Data") handle_mod_contacts(token_id,app_id,wxid,msg_data) elif type_name=="DelContacts": ''' 删除好友通知/退出群聊 ''' msg_data = msg.get("Data") username=msg_data["UserName"]["string"] if check_chatroom(username): logger.info('退出群聊') wxid = msg.get("Wxid") chatroom_id=username redis_helper.redis_helper.delete_hash_field(f'__AI_OPS_WX__:GROUPS_INFO:{wxid}',chatroom_id) logger.info(f'清除 chatroom_id{chatroom_id} 数据') else: logger.info('删除好友通知') elif type_name=="Offline": ''' 已经离线 ''' wxid = msg.get("Wxid") logger.warning(f'微信ID {wxid}在设备{app_id}已经离线') k,r=get_login_info_by_app_id(app_id) print(k) redis_helper.redis_helper.update_hash_field(k,'status',0) redis_helper.redis_helper.update_hash_field(k,'modify_at',int(time.time())) else: logger.warning(f"未知消息类型") return jsonify({"message": "收到微信回调消息"}) def handle_text(token_id,app_id, wxid,msg_data,from_wxid, to_wxid): ''' 私聊文本消息 ''' msg_content=msg_data["Content"]["string"] if wxid == from_wxid: #手动发送消息 logger.info("Active message sending detected") gewe_chat.wxchat.save_contacts_brief_to_cache(token_id,app_id,wxid,[to_wxid]) callback_to_user=msg_data["ToUserName"]["string"] input_wx_content_dialogue_message=[{"type": "text", "text": msg_content}] input_message=utils.dialogue_message(from_wxid,to_wxid,input_wx_content_dialogue_message) kafka_helper.kafka_client.produce_message(input_message) logger.info("发送对话 %s",input_message) else: callback_to_user=msg_data["FromUserName"]["string"] # hash_key = f'__AI_OPS_WX__:MESSAGES:{wxid}:{callback_to_user}' # prompt={"role": "user", "content": [{ # "type": "text", # "text": msg_content # }]} # messages_to_send=gewe_chat.wxchat.save_session_messages_to_cache(hash_key, prompt) # # 收到的对话 # input_wx_content_dialogue_message=[{"type": "text", "text": msg_content}] # input_message=utils.dialogue_message(callback_to_user,wxid,input_wx_content_dialogue_message) # kafka_helper.kafka_client.produce_message(input_message) # logger.info("发送对话 %s",input_message) # cache_data = memory.USER_INTERACTIVE_CACHE.get(wxid) # if cache_data and cache_data.get('interactive') : # o=get_first_char_if_digit(msg_content) # if o is not None: # userSelectOptions=cache_data.get('options') # if o < len(userSelectOptions): # o=o-1 # msg_content=userSelectOptions[o].get("value") # messages_to_send=[{"role": "user", "content": msg_content}] # else: # messages_to_send=[{"role": "user", "content": msg_content}] # else: # messages_to_send=[{"role": "user", "content": msg_content}] # res=fast_gpt_api(messages_to_send,wxid,callback_to_user) # reply_content=res["choices"][0]["message"]["content"] # description = '' # userSelectOptions = [] # if isinstance(reply_content, list) and any(item.get("type") == "interactive" for item in reply_content): # for item in reply_content: # if item["type"] == "interactive" and item["interactive"]["type"] == "userSelect": # params = item["interactive"]["params"] # description = params.get("description") # userSelectOptions = params.get("userSelectOptions", []) # values_string = "\n".join(option["value"] for option in userSelectOptions) # if description is not None: # memory.USER_INTERACTIVE_CACHE[wxid] = { # "interactive":True, # "options": userSelectOptions, # } # reply_content=description + '------------------------------\n'+values_string # elif isinstance(reply_content, list) and any(item.get("type") == "text" for item in reply_content): # memory.USER_INTERACTIVE_CACHE[wxid] = { # "interactive":False # } # text='' # for item in reply_content: # if item["type"] == "text": # text=item["text"]["content"] # if text=='': # # 去除上次上一轮对话再次请求 # cache_messages_str=redis_helper.redis_helper.get_hash_field(hash_key,"data") # cache_messages = json.loads(cache_messages_str) if cache_messages_str else [] # if len(cache_messages) >= 3: # cache_messages = cache_messages[:-3] # redis_helper.redis_helper.update_hash_field(hash_key,"data",json.dumps(cache_messages,ensure_ascii=False)) # messages_to_send=gewe_chat.wxchat.save_session_messages_to_cache(hash_key, prompt) # res=fast_gpt_api(messages_to_send,wxid,callback_to_user) # reply_content=res["choices"][0]["message"]["content"] # if isinstance(reply_content, list) : # reply_content=reply_content[0].get('text').get("content") # else: # reply_content=text # else: # memory.USER_INTERACTIVE_CACHE[wxid] = { # "interactive":False # } # reply_content=res["choices"][0]["message"]["content"] # gewe_chat.wxchat.post_text(token_id,app_id,callback_to_user,reply_content) # gewe_chat.wxchat.save_session_messages_to_cache(hash_key, {"role": "assistant", "content": reply_content}) # # 回复的对话 # input_wx_content_dialogue_message=[{"type": "text", "text": reply_content}] # input_message=utils.dialogue_message(wxid,callback_to_user,input_wx_content_dialogue_message,True) # kafka_helper.kafka_client.produce_message(input_message) # logger.info("发送对话 %s",input_message) # 创建并启动任务线程，将参数传递给 ai_chat_text 函数 task_thread = threading.Thread( target=ai_chat_text, args=(token_id,app_id,wxid,msg_data,msg_content) ) task_thread.start() # 设置定时器，1秒后检查任务是否超时。这里需要使用 lambda 来传递参数 timeout_timer = threading.Timer( timeout_duration, lambda:check_timeout(task_thread, token_id, wxid,app_id, callback_to_user) ) timeout_timer.start() # 等待任务线程完成 task_thread.join() # 取消定时器 timeout_timer.cancel() def check_timeout( task_thread:threading.Thread, token_id,wxid, app_id, callback_to_user): if task_thread.is_alive(): print(f"任务运行时间超过{timeout_duration}秒，token_id={token_id}, app_id={app_id}, callback_to_user={callback_to_user}") wx_config = gewe_chat.wxchat.get_wxchat_config_from_cache(wxid) if bool(wx_config.get("chatWaitingMsgEnabled",True)): gewe_chat.wxchat.post_text(token_id,app_id,callback_to_user,"亲，我正在组织回复的信息，请稍等一会") def ai_chat_text(token_id,app_id,wxid,msg_data,msg_content): start_time = time.time() # 记录任务开始时间 callback_to_user=msg_data["FromUserName"]["string"] hash_key = f'__AI_OPS_WX__:MESSAGES:{wxid}:{callback_to_user}' prompt={"role": "user", "content": [{ "type": "text", "text": msg_content }]} messages_to_send=gewe_chat.wxchat.save_session_messages_to_cache(hash_key, prompt) # 收到的对话 input_wx_content_dialogue_message=[{"type": "text", "text": msg_content}] input_message=utils.dialogue_message(callback_to_user,wxid,input_wx_content_dialogue_message) kafka_helper.kafka_client.produce_message(input_message) logger.info("发送对话 %s",input_message) cache_data = memory.USER_INTERACTIVE_CACHE.get(wxid) if cache_data and cache_data.get('interactive') : o=get_first_char_if_digit(msg_content) if o is not None: userSelectOptions=cache_data.get('options') if o < len(userSelectOptions): o=o-1 msg_content=userSelectOptions[o].get("value") messages_to_send=[{"role": "user", "content": msg_content}] else: messages_to_send=[{"role": "user", "content": msg_content}] else: messages_to_send=[{"role": "user", "content": msg_content}] res=fast_gpt_api(messages_to_send,wxid,callback_to_user) reply_content=res["choices"][0]["message"]["content"] description = '' userSelectOptions = [] if isinstance(reply_content, list) and any(item.get("type") == "interactive" for item in reply_content): for item in reply_content: if item["type"] == "interactive" and item["interactive"]["type"] == "userSelect": params = item["interactive"]["params"] description = params.get("description") userSelectOptions = params.get("userSelectOptions", []) values_string = "\n".join(option["value"] for option in userSelectOptions) if description is not None: memory.USER_INTERACTIVE_CACHE[wxid] = { "interactive":True, "options": userSelectOptions, } reply_content=description + '------------------------------\n'+values_string elif isinstance(reply_content, list) and any(item.get("type") == "text" for item in reply_content): memory.USER_INTERACTIVE_CACHE[wxid] = { "interactive":False } text='' for item in reply_content: if item["type"] == "text": text=item["text"]["content"] if text=='': # 去除上次上一轮对话再次请求 cache_messages_str=redis_helper.redis_helper.get_hash_field(hash_key,"data") cache_messages = json.loads(cache_messages_str) if cache_messages_str else [] if len(cache_messages) >= 3: cache_messages = cache_messages[:-3] redis_helper.redis_helper.update_hash_field(hash_key,"data",json.dumps(cache_messages,ensure_ascii=False)) messages_to_send=gewe_chat.wxchat.save_session_messages_to_cache(hash_key, prompt) res=fast_gpt_api(messages_to_send,wxid,callback_to_user) reply_content=res["choices"][0]["message"]["content"] if isinstance(reply_content, list) : reply_content=reply_content[0].get('text').get("content") else: reply_content=text else: memory.USER_INTERACTIVE_CACHE[wxid] = { "interactive":False } reply_content=res["choices"][0]["message"]["content"] gewe_chat.wxchat.post_text(token_id,app_id,callback_to_user,reply_content) gewe_chat.wxchat.save_session_messages_to_cache(hash_key, {"role": "assistant", "content": reply_content}) # 回复的对话 input_wx_content_dialogue_message=[{"type": "text", "text": reply_content}] input_message=utils.dialogue_message(wxid,callback_to_user,input_wx_content_dialogue_message,True) kafka_helper.kafka_client.produce_message(input_message) logger.info("发送对话 %s",input_message) end_time = time.time() # 记录任务结束时间 execution_time = end_time - start_time # 计算执行时间 logger.info(f"AI回答任务完成，耗时 {execution_time:.2f} 秒") def handle_text_group(token_id,app_id, wxid,msg_data,from_wxid, to_wxid): ''' 群聊文本消息 ''' msg_content=msg_data["Content"]["string"] msg_push_content=msg_data.get("PushContent") k,login_info=get_login_info_by_app_id(app_id) nickname=login_info.get("nickName") if wxid == from_wxid: #手动发送消息 logger.info("Active message sending detected") gewe_chat.wxchat.save_contacts_brief_to_cache(token_id,app_id,wxid,[to_wxid]) callback_to_user=msg_data["ToUserName"]["string"] input_wx_content_dialogue_message=[{"type": "text", "text": msg_content}] input_message=utils.dialogue_message(from_wxid,to_wxid,input_wx_content_dialogue_message) kafka_helper.kafka_client.produce_message(input_message) logger.info("发送对话 %s",input_message) else: c = gewe_chat.wxchat.get_wxchat_config_from_cache(wxid) chatroom_id_white_list = c.get("chatroomIdWhiteList", []) if not chatroom_id_white_list: logger.info('白名单为空或未定义,不处理') return if from_wxid not in chatroom_id_white_list: logger.info('群ID不在白名单中,不处理') return if '在群聊中@了你' in msg_push_content or '@'+nickname in msg_push_content: callback_to_user=msg_data["FromUserName"]["string"] hash_key = f'__AI_OPS_WX__:MESSAGES:{wxid}:{callback_to_user}' prompt={"role": "user", "content": [{ "type": "text", "text": msg_content }]} messages_to_send=gewe_chat.wxchat.save_session_messages_to_cache(hash_key, prompt) # 收到的对话 input_wx_content_dialogue_message=[{"type": "text", "text": msg_content}] input_message=utils.dialogue_message(callback_to_user,wxid,input_wx_content_dialogue_message) kafka_helper.kafka_client.produce_message(input_message) logger.info("发送对话 %s",input_message) cache_data = memory.USER_INTERACTIVE_CACHE.get(wxid) if cache_data and cache_data.get('interactive') : o=get_first_char_if_digit(msg_content) if o is not None: userSelectOptions=cache_data.get('options') if o < len(userSelectOptions): o=o-1 msg_content=userSelectOptions[o].get("value") messages_to_send=[{"role": "user", "content": msg_content}] else: messages_to_send=[{"role": "user", "content": msg_content}] else: messages_to_send=[{"role": "user", "content": msg_content}] res=fast_gpt_api(messages_to_send,wxid,callback_to_user) reply_content=res["choices"][0]["message"]["content"] description = '' userSelectOptions = [] if isinstance(reply_content, list) and any(item.get("type") == "interactive" for item in reply_content): for item in reply_content: if item["type"] == "interactive" and item["interactive"]["type"] == "userSelect": params = item["interactive"]["params"] description = params.get("description") userSelectOptions = params.get("userSelectOptions", []) values_string = "\n".join(option["value"] for option in userSelectOptions) if description is not None: memory.USER_INTERACTIVE_CACHE[wxid] = { "interactive":True, "options": userSelectOptions, } reply_content=description + '------------------------------\n'+values_string elif isinstance(reply_content, list) and any(item.get("type") == "text" for item in reply_content): memory.USER_INTERACTIVE_CACHE[wxid] = { "interactive":False } text='' for item in reply_content: if item["type"] == "text": text=item["text"]["content"] if text=='': # 去除上次上一轮对话再次请求 cache_messages_str=redis_helper.redis_helper.get_hash_field(hash_key,"data") cache_messages = json.loads(cache_messages_str) if cache_messages_str else [] if len(cache_messages) >= 3: cache_messages = cache_messages[:-3] redis_helper.redis_helper.update_hash_field(hash_key,"data",json.dumps(cache_messages,ensure_ascii=False)) messages_to_send=gewe_chat.wxchat.save_session_messages_to_cache(hash_key, prompt) res=fast_gpt_api(messages_to_send,wxid,callback_to_user) reply_content=res["choices"][0]["message"]["content"] else: reply_content=text else: memory.USER_INTERACTIVE_CACHE[wxid] = { "interactive":False } reply_content=res["choices"][0]["message"]["content"] reply_content='@'+extract_nickname(msg_push_content) + reply_content gewe_chat.wxchat.post_text(token_id,app_id,callback_to_user,reply_content) gewe_chat.wxchat.save_session_messages_to_cache(hash_key, {"role": "assistant", "content": reply_content}) # 回复的对话 input_wx_content_dialogue_message=[{"type": "text", "text": reply_content}] input_message=utils.dialogue_message(wxid,callback_to_user,input_wx_content_dialogue_message,True) kafka_helper.kafka_client.produce_message(input_message) logger.info("发送对话 %s",input_message) else: logger.info('群聊公开消息') callback_to_user=msg_data["FromUserName"]["string"] dialogue_message=[{"type": "text", "text": msg_content}] input_message=utils.dialogue_message(callback_to_user,wxid,dialogue_message) kafka_helper.kafka_client.produce_message(input_message) logger.info("发送对话 %s",input_message) return def handle_image(token_id,app_id, wxid,msg_data,from_wxid, to_wxid): ''' 图片消息 ''' msg_content=msg_data["Content"]["string"] callback_to_user=from_wxid hash_key = f'__AI_OPS_WX__:MESSAGES:{wxid}:{callback_to_user}' wx_img_url=gewe_chat.wxchat.download_image_msg(token_id,app_id,msg_content) oss_access_key_id="LTAI5tRTG6pLhTpKACJYoPR5" oss_access_key_secret="E7dMzeeMxq4VQvLg7Tq7uKf3XWpYfN" oss_endpoint="http://oss-cn-shanghai.aliyuncs.com" oss_bucket_name="cow-agent" oss_prefix="cow" img_url=utils.upload_oss(oss_access_key_id, oss_access_key_secret, oss_endpoint, oss_bucket_name, wx_img_url, oss_prefix) prompt={ "role": "user", "content": [{ "type": "image_url", "image_url": {"url": img_url} }] } gewe_chat.wxchat.save_session_messages_to_cache(hash_key, prompt) gewe_chat.wxchat.post_text(token_id,app_id,callback_to_user,'已经上传了图片，有什么可以为您服务') logger.info(f"上传图片 URL: {img_url}") wx_content_dialogue_message=[{"type": "image_url", "image_url": {"url": img_url}}] input_message=utils.dialogue_message(wxid,callback_to_user,wx_content_dialogue_message) kafka_helper.kafka_client.produce_message(input_message) logger.info("发送对话 %s",input_message) def handle_image_group(token_id,app_id, wxid,msg_data,from_wxid, to_wxid): logger.info('群聊图片消息') def handle_voice(token_id,app_id, wxid,msg_data,from_wxid, to_wxid): ''' 语音消息 ''' callback_to_user=from_wxid msg_content=msg_data["Content"]["string"] msg_id=msg_data["MsgId"] file_url=gewe_chat.wxchat.download_audio_msg(token_id,app_id,msg_id,msg_content) react_silk_path=utils.save_to_local_from_url(file_url) react_wav_path = os.path.splitext(react_silk_path)[0] + ".wav" audio_convert.any_to_wav(react_silk_path,react_wav_path) react_voice_text=AliVoice().voiceToText(react_wav_path) os.remove(react_silk_path) os.remove(react_wav_path) hash_key = f'__AI_OPS_WX__:MESSAGES:{wxid}:{callback_to_user}' messages=gewe_chat.wxchat.save_session_messages_to_cache(hash_key, {"role": "user", "content": react_voice_text}) ai_res=fast_gpt_api(messages,wxid,callback_to_user) ai_res_content=ai_res["choices"][0]["message"]["content"] has_url=contains_url(ai_res_content) if not has_url: voice_during,voice_url=utils.wx_voice(ai_res_content) if voice_during < 60 * 1000: ret,ret_msg,res=gewe_chat.wxchat.post_voice(token_id,app_id,callback_to_user,voice_url,voice_during) else: ret,ret_msg,res=gewe_chat.wxchat.post_text(token_id,app_id,callback_to_user,ai_res_content) logger.warning(f'回应语音消息长度 {voice_during/1000}秒，超过60秒，转为文本回复') if ret==200: logger.info((f'{wxid} 向 {callback_to_user} 发送语音文本【{ai_res_content}】{ret_msg}')) else: logger.warning((f'{wxid} 向 {callback_to_user} 发送语音文本【{ai_res_content}】{ret_msg}')) ret,ret_msg,res=gewe_chat.wxchat.post_text(token_id,app_id,callback_to_user,ai_res_content) logger.info((f'{wxid} 向 {callback_to_user} 发送文本【{ai_res_content}】{ret_msg}')) else: logger.info(f"回复内容包含网址，不发送语音,回复文字内容：{ai_res_content}") ret,ret_msg,res=gewe_chat.wxchat.post_text(token_id,app_id,callback_to_user,ai_res_content) gewe_chat.wxchat.save_session_messages_to_cache(hash_key, {"role": "assistant", "content": ai_res_content}) # 构造对话消息并发送到 Kafka input_wx_content_dialogue_message = [{"type": "text", "text": ai_res_content}] input_message = utils.dialogue_message(wxid, callback_to_user, input_wx_content_dialogue_message,True) kafka_helper.kafka_client.produce_message(input_message) logger.info("发送对话 %s", input_message) def handle_voice_group(token_id,app_id, wxid,msg_data,from_wxid, to_wxid): logger.info('语音消息') def handle_xml(token_id,app_id, wxid,msg_data,from_wxid, to_wxid): ''' 处理xml ''' msg_content_xml=msg_data["Content"]["string"] root = ET.fromstring(msg_content_xml) type_value = root.find(".//appmsg/type").text handlers = { 57: handle_xml_reference, } handler = handlers.get(type_value) if handler: return handler(token_id,app_id, wxid,msg_data,from_wxid, to_wxid) elif "邀请你加入了群聊" in msg_content_xml: # 邀请加入群聊 logger.warning(f"xml消息 {type_value} 邀请你加入了群聊.todo") else: print(f"xml消息 {type_value} 未解析") def handle_xml_reference(token_id,app_id, wxid,msg_data,from_wxid, to_wxid): ''' 判断此类消息的逻辑：$.Data.MsgType=49 并且 解析$.Data.Content.string中的xml msg.appmsg.type=57 ''' callback_to_user=from_wxid hash_key = f'__AI_OPS_WX__:MESSAGES:{wxid}:{callback_to_user}' msg_content= msg_data["PushContent"] prompt={"role": "user", "content": [{ "type": "text", "text": msg_content }]} # 收到的对话 messages_to_send=gewe_chat.wxchat.save_session_messages_to_cache(hash_key, prompt) input_wx_content_dialogue_message=[{"type": "text", "text": msg_content}] input_message=utils.dialogue_message(callback_to_user,wxid,input_wx_content_dialogue_message) kafka_helper.kafka_client.produce_message(input_message) logger.info("发送对话 %s",input_message) # 回复的对话 res=fast_gpt_api(messages_to_send,wxid,callback_to_user) reply_content=res["choices"][0]["message"]["content"] input_wx_content_dialogue_message=[{"type": "text", "text": reply_content}] input_message=utils.dialogue_message(wxid,callback_to_user,input_wx_content_dialogue_message,True) kafka_helper.kafka_client.produce_message(input_message) logger.info("发送对话 %s",input_message) gewe_chat.wxchat.save_session_messages_to_cache(hash_key, {"role": "assistant", "content": reply_content}) gewe_chat.wxchat.post_text(token_id,app_id,callback_to_user,reply_content) def handle_add_friend_notice(token_id,app_id, wxid,msg_data,from_wxid, to_wxid): ''' 好友添加请求通知 ''' logger.info('好友添加请求通知') msg_content_xml=msg_data["Content"]["string"] root = ET.fromstring(msg_content_xml) msg_content = root.attrib.get('content', None) v3= root.attrib.get('encryptusername', None) v4= root.attrib.get('ticket', None) scene=root.attrib.get('scene', None) to_contact_wxid=root.attrib.get('fromusername', None) wxid=msg_data["ToUserName"]["string"] # 自动同意好友 # print(v3) # print(v4) # print(scene) # print(msg_content) # 操作类型，2添加好友 3同意好友 4拒绝好友 #option=2 option=3 reply_add_contact_contact="亲，我是你的美丽顾问" ret,ret_msg=gewe_chat.wxchat.add_contacts(token_id,app_id,scene,option,v3,v4,reply_add_contact_contact) if ret==200: logger.info('自动添加好友成功') # 好友发送的文字 hash_key = f'__AI_OPS_WX__:MESSAGES:{wxid}:{to_contact_wxid}' prompt={"role": "user", "content": [{"type": "text","text": msg_content}]} messages_to_send=gewe_chat.wxchat.save_session_messages_to_cache(hash_key, prompt) input_wx_content_dialogue_message=[{"type": "text", "text": msg_content}] input_message=utils.dialogue_message(to_contact_wxid,wxid,input_wx_content_dialogue_message) kafka_helper.kafka_client.produce_message(input_message) logger.info("发送对话 %s",input_message) callback_to_user=to_contact_wxid res=fast_gpt_api(messages_to_send,wxid,callback_to_user) reply_content=res["choices"][0]["message"]["content"] #保存好友信息 gewe_chat.wxchat.save_contacts_brief_to_cache(token_id,app_id, wxid,[to_contact_wxid]) # 保存到缓存 gewe_chat.wxchat.save_session_messages_to_cache(hash_key, {"role": "assistant", "content": reply_content}) # 发送信息 gewe_chat.wxchat.post_text(token_id,app_id, to_contact_wxid,reply_content) # 发送到kafka input_wx_content_dialogue_message=[{"type": "text", "text": reply_content}] input_message=utils.dialogue_message(wxid,to_contact_wxid,input_wx_content_dialogue_message,True) kafka_helper.kafka_client.produce_message(input_message) logger.info("发送对话 %s",input_message) else: logger.warning("添加好友失败") def handle_10002_msg(token_id,app_id, wxid,msg_data,from_wxid, to_wxid): ''' 群聊邀请 撤回消息 拍一拍消息 地理位置 踢出群聊通知 解散群聊通知 发布群公告 ''' msg_content_xml=msg_data["Content"]["string"] # 群聊邀请 if '邀请你加入了群聊' in msg_content_xml and check_chatroom(msg_data["FromUserName"]["string"]): chatroom_id=msg_data["FromUserName"]["string"] ret,msg,data=gewe_chat.wxchat.save_contract_list(token_id,app_id,chatroom_id,3) logger.info(f'群聊邀请,保存到通讯录 chatroom_id {chatroom_id} {msg}') gewe_chat.wxchat.update_group_info_to_cache(token_id,app_id,wxid,chatroom_id) if '移出了群聊' in msg_content_xml and 'sysmsgtemplate' in msg_content_xml : chatroom_id=msg_data["FromUserName"]["string"] ret,msg,data=gewe_chat.wxchat.save_contract_list(token_id,app_id,chatroom_id,2) logger.info(f'踢出群聊,移除从通讯录 chatroom_id {chatroom_id} {msg}') redis_helper.redis_helper.delete_hash_field(f'__AI_OPS_WX__:GROUPS_INFO:{wxid}',chatroom_id) logger.info(f'清除 chatroom_id{chatroom_id} 数据') if '已解散该群聊' in msg_content_xml and 'sysmsgtemplate' in msg_content_xml : chatroom_id=msg_data["FromUserName"]["string"] ret,msg,data=gewe_chat.wxchat.save_contract_list(token_id,app_id,chatroom_id,2) logger.info(f'解散群聊,移除从通讯录 chatroom_id {chatroom_id} {msg}') redis_helper.redis_helper.delete_hash_field(f'__AI_OPS_WX__:GROUPS_INFO:{wxid}',chatroom_id) logger.info(f'清除 chatroom_id{chatroom_id} 数据') print('撤回消息,拍一拍消息,地理位置') def handle_mod_contacts(token_id,app_id,wxid,msg_data): ''' 好友通过验证及好友资料变更的通知消息 ''' logger.info('好友通过验证及好友资料变更的通知消息') if not check_chatroom(msg_data["UserName"]["string"]): contact_wxid = msg_data["UserName"]["string"] # 更新好友信息 # 检查好友关系，不是好友则删除 ret,msg,check_relation=gewe_chat.wxchat.check_relation(token_id, app_id,[contact_wxid]) first_item = check_relation[0] check_relation_status=first_item.get('relation') logger.info(f'{wxid} 好友 {contact_wxid} 关系检查：{check_relation_status}') if check_relation_status != 0: gewe_chat.wxchat.delete_contacts_brief_from_cache(wxid, [contact_wxid]) logger.info(f'好友关系异常：{check_relation_status}，删除好友 {contact_wxid} 信息') # ret,msg,contacts_list = gewe_chat.wxchat.fetch_contacts_list(token_id, app_id) # friend_wxids = contacts_list['friends'][3:] # 可以调整截取范围 # print(friend_wxids) # #friend_wxids.remove('weixin') #gewe_chat.wxchat.save_contacts_brief_to_cache(token_id, app_id, wxid, friend_wxids) else: logger.info('群聊好友通过验证及好友资料变更的通知消息') def get_messages_from_cache(hash_key,object:object)->list: ''' 对话列表 ''' messages=redis_helper.redis_helper.get_hash(hash_key) wxid=hash_key.split(':')[-1] if not messages: messages=[{"role": "system", "content": ""}] messages.append(object) redis_helper.redis_helper.set_hash(hash_key,{"data":json.dumps(messages,ensure_ascii=False)},3600) else: messages_str=redis_helper.redis_helper.get_hash_field(hash_key,"data") messages = json.loads(messages_str) if messages_str else [] #判断是否含有图片 last_message = messages[-1] content = last_message.get("content", []) if isinstance(content, list) and content: last_content_type = content[-1].get("type") if last_content_type == 'image_url': content.append(object['content'][0]) messages[-1]['content']=content else: messages.append(object) else: messages.append(object) redis_helper.redis_helper.set_hash(hash_key,{"data":json.dumps(messages,ensure_ascii=False)},3600) return messages def fast_gpt_api(messages:list,wixd:str,friend_wxid:str): c=gewe_chat.wxchat.get_wxchat_config_from_cache(wixd) api_key=c.get('agentTokenId',"sk-jr69ONIehfGKe9JFphuNk4DU5Y5wooHKHhQv7oSnFzVbwCnW65fXO9kvH") print(f'流程key：{api_key}\n') #api_key="sk-jr69ONIehfGKe9JFphuNk4DU5Y5wooHKHhQv7oSnFzVbwCnW65fXO9kvH" #测试 #api_key="sk-uJDBdKmJVb2cmfldGOvlIY6Qx0AzqWMPD3lS1IzgQYzHNOXv9SKNI" #开发2 api_url = "http://106.15.182.218:3000/api/v1/chat/completions" headers = { "Content-Type": "application/json", "Authorization": f"Bearer {api_key}" } session_id=f'{wixd}-{friend_wxid}' data={ "model": "", "messages":messages, "chatId": session_id, "detail": True } #print(json.dumps(data,ensure_ascii=False)) logger.info("[CHATGPT] 请求={}".format(json.dumps(data, ensure_ascii=False))) response = requests.post(url=api_url, headers=headers, data=json.dumps(data), timeout=600) response.raise_for_status() response_data = response.json() logger.info("[CHATGPT] 响应={}".format(json.dumps(response_data, separators=(',', ':'),ensure_ascii=False))) #print(response_data) return response_data def get_token_id_by_app_id(app_id: str) -> str: # 使用 SCAN 避免一次性返回所有的匹配键，逐步扫描 cursor = 0 while True: cursor, login_keys = redis_helper.redis_helper.client.scan(cursor, match='__AI_OPS_WX__:LOGININFO:*') # 批量获取所有键的 hash 数据 for k in login_keys: r = redis_helper.redis_helper.get_hash(k) if r.get("appId") == app_id: return r.get("tokenId", "") # 如果游标为 0，则表示扫描完成 if cursor == 0: break return "" def get_login_info_by_app_id(app_id: str) ->dict: # 使用 SCAN 避免一次性返回所有的匹配键，逐步扫描 cursor = 0 while True: cursor, login_keys = redis_helper.redis_helper.client.scan(cursor, match='__AI_OPS_WX__:LOGININFO:*') # 批量获取所有键的 hash 数据 for k in login_keys: r = redis_helper.redis_helper.get_hash(k) if r.get("appId") == app_id: return k,r # 如果游标为 0，则表示扫描完成 if cursor == 0: break return "" def contains_url(text): # 定义检测网址的正则表达式 url_pattern = re.compile( r'http[s]?://(?:[a-zA-Z]|[0-9]|[$-_@.&+]|[!*\\(\\),]|(?:%[0-9a-fA-F][0-9a-fA-F]))+' ) # 检查字符串是否包含网址 return bool(url_pattern.search(text)) def get_first_char_if_digit(s): if s and s[0].isdigit(): # 判断字符串是否非空且首字符为数字 return int(s[0]) # 返回数字形式 return None # 如果不是数字则返回 None def remove_at_mention_regex(text): # 使用正则表达式去掉“在群聊中@了你” return re.sub(r"在群聊中@了你", "", text) def extract_nickname(text)->str: if "在群聊中@了你" in text: # 如果包含 "在群聊中@了你"，提取其前面的名字 match = re.search(r"^(.*?)在群聊中@了你", text) if match: return match.group(1).strip() elif ": @" in text: # 如果包含 ": @"，提取其前面的名字 return text.split(": @")[0].strip() return '' def check_chatroom(userName): pattern = r'^\d+@chatroom$' if re.match(pattern, userName): return True return False