import aiohttp import asyncio import json import base64 import io import json import os import threading import time import uuid,random from fastapi import FastAPI, Depends from common.log import logger from common.singleton import singleton from services.kafka_service import KafkaService from fastapi import Request from common.utils import * @singleton class BizService(): def __init__(self,app:FastAPI): if not hasattr(self, 'initialized'): #self.kafka_service =kafka_service # 获取 KafkaService 单例 self.kafka_service =app.state.kafka_service self.wxchat=app.state.gewe_service self.redis_service=app.state.redis_service self.initialized = True def setup_handlers(self): """设置默认的消息处理器""" # 这里可以添加业务逻辑 # 注册默认处理器 self.kafka_service.add_handler( self.kafka_service.consumer_topic, self.ops_messages_process_handler ) async def ops_messages_process_handler(self, message: str): """消息处理器""" #print(f"BizService handling message: {message}") try: 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": await self.login_handler_async(content_data) elif msg_type_data == 'group-sending': print(f'处理消息类型group-sending') await self.group_sending_handler_async(content_data) elif msg_type_data == 'login_wx_captch_code': pass else: print(f'kakfa 未处理息类型 {msg_type_data}') except Exception as e: print(f"处理消息时发生错误: {e}, 消息内容: {message}") async def login_handler_async(self, content_data: dict): tel=content_data.get('tel', '18733438393') token_id=content_data.get('token_id', 'c50b7d57-2efa-4a53-8c11-104a06d1e1fa') region_id=content_data.get('region_id', '440000') agent_token_id=content_data.get('agent_token_id', 'sk-fAOIdANeGXjWKW5mFybnsNZZGYU2lFLmqVY9rVFaFmjiOaWt3tcWMi') loginfo= await self.wxchat.get_login_info_from_cache_async(tel,token_id,region_id,agent_token_id) print(loginfo) status=loginfo.get('status','0') if status=='1': logger.info(f'手机号{tel},wx_token{token_id} 已经微信登录,终止登录流程') return async def group_sending_handler_async(self,content_data: dict): agent_tel=content_data.get('agent_tel', '18733438393') hash_key = f"__AI_OPS_WX__:LOGININFO:{agent_tel}" logininfo = await self.redis_service.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=await self.redis_service.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 = await self.redis_service.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", []) self.wxchat.forward_video_aeskey = '' self.wxchat.forward_video_cdnvideourl = '' self.wxchat.forward_video_length = 0 for intersection_wxid in intersection_wxids: for wx_content in wx_content_list: if wx_content["type"] == "text": await self.send_text_message_async(token_id, app_id, agent_wxid, [intersection_wxid], wx_content["text"]) elif wx_content["type"] == "image_url": await self.send_image_messagae_sync(token_id, app_id, agent_wxid, [intersection_wxid], wx_content.get("image_url", {}).get("url")) elif wx_content["type"] == "tts": await self.send_tts_message(token_id, app_id, agent_wxid, [intersection_wxid], wx_content["text"]) elif wx_content["type"] == "file": await self.send_file_message(token_id, app_id, agent_wxid, [intersection_wxid], wx_content.get("file_url", {}).get("url")) async def send_text_message_async(self, token_id, app_id, agent_wxid, intersection_wxids, text): for t in intersection_wxids: # 发送文本消息 ret,ret_msg,res = await self.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 = dialogue_message(agent_wxid, t, input_wx_content_dialogue_message) await self.kafka_service.send_message_async(input_message) logger.info("发送对话 %s", input_message) # 等待随机时间 await asyncio.sleep(random.uniform(5, 15)) async def send_image_messagae_sync(self,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 = await self.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 !="": # 转发图片 ret,ret_msg,res = await self.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 = await self.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 = dialogue_message(agent_wxid, t, wx_content_dialogue_message) await self.kafka_service.send_message_async(input_message) logger.info("发送对话 %s", input_message) # 等待随机时间 await asyncio.sleep(random.uniform(5, 15)) async def send_tts_message(self, token_id, app_id, agent_wxid, intersection_wxids, text): voice_during,voice_url=wx_voice(text) for t in intersection_wxids: # 发送送语音消息 if voice_url: ret,ret_msg,res = await self.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 = dialogue_message(agent_wxid, t, input_wx_content_dialogue_message) await self.kafka_service.send_message_async(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}】出错')) # 等待随机时间 await asyncio.sleep(random.uniform(5, 15)) async def send_file_message(self,token_id, app_id, agent_wxid, intersection_wxids, file_url): parsed_url = urlparse(file_url) path = parsed_url.path # 从路径中提取文件名 filename = path.split('/')[-1] # 获取扩展名 _, ext = os.path.splitext(filename) if ext == '.mp4': await self.send_video_message(self, token_id, app_id, agent_wxid, intersection_wxids, file_url) else: await self.send_other_file_message(self, token_id, app_id, agent_wxid, intersection_wxids, file_url) #time.sleep(random.uniform(5, 15)) async def send_video_message(self, token_id, app_id, agent_wxid, intersection_wxids, file_url): for t in intersection_wxids: # 发送视频消息 parsed_url = urlparse(file_url) filename = os.path.basename(parsed_url.path) tmp_file_path = os.path.join(os.getcwd(),'tmp', filename) # 拼接完整路径 thumbnail_path=tmp_file_path.replace('.mp4','.jpg') if self.wxchat.forward_video_aeskey == '': video_thumb_url,video_duration =download_video_and_get_thumbnail(file_url,thumbnail_path) print(f'视频缩略图 {video_thumb_url} 时长 {video_duration}') ret,ret_msg,res = await self.wxchat.post_video(token_id, app_id, t, file_url,video_thumb_url,video_duration) if ret==200: self.wxchat.forward_video_aeskey = res["aesKey"] self.wxchat.forward_video_cdnvideourl = res["cdnThumbUrl"] self.wxchat.forward_video_length = res["length"] else: ret,ret_msg,res = await self.wxchat.forward_video(token_id, app_id, t, self.wxchat.forward_video_aeskey, self.wxchat.forward_video_cdnvideourl, self.wxchat.forward_video_length) print('转发视频') if ret==200: logger.info(f'{agent_wxid} 向 {t} 发送视频【{file_url}】{ret_msg}') # 构造对话消息并发送到 Kafka input_wx_content_dialogue_message = [{"type": "file", "file_url": {"url": file_url}}] input_message = dialogue_message(agent_wxid, t, input_wx_content_dialogue_message) await self.kafka_service.send_message_async(input_message) logger.info("发送对话 %s", input_message) else: logger.warning((f'{agent_wxid} 向 {t} 发送视频【{file_url}】{ret_msg}')) # 等待随机时间 await asyncio.sleep(random.uniform(5, 15)) async def send_other_file_message(self, token_id, app_id, agent_wxid, intersection_wxids, file_url): print('send_otherfile_message')