import json from urllib.parse import urlparse from typing import Dict, Any, Optional import queue import threading import traceback import mmh3 def murmur3_32(player_file): if isinstance(player_file, list): player_file = ','.join(player_file) return abs(mmh3.hash(player_file)) #机器人外呼 class BotChatRequest: def __init__(self, node_id=None, user_id=None, session_id=None, record_id=None, task_id=None, asr_text=None, key_input=None, ext:Dict={}): self.node_id = node_id # 节点id self.user_id = user_id # 用户id self.session_id = session_id # 会话id self.record_id = record_id # 唯一标识 self.task_id = task_id # 机器人任务id self.event_type = None # 1:电话接通,2:用户语音asr结果上传,3:用户按键输入,4:用户挂断电话,5:读取外呼结果(可在外呼过程中调用，不一定在结束之后),6:用户不说话超时,7:TTS或录音文件播放完毕或被打断,8:ASR错误导致挂断电话,9:TTS错误导致挂断电话,10:其它系统错误导致挂断电话 self.asr_text = asr_text # asr识别的文本 self.ext = ext def to_json_string(self): return json.dumps(self.__dict__, ensure_ascii=False) @classmethod def from_json(cls, json_string): data = json.loads(json_string) return cls(**data) # def __repr__(self): # return (f"OutboundRobotRequestParams(node_id={self.node_id}, user_id={self.user_id}, " # f"session_id={self.session_id}, record_id={self.record_id}, task_id={self.task_id}, " # f"event_type={self.event_type}, asr_text={self.asr_text}, key_input={self.key_input})") class ChatAction: def __init__(self, action_code=None, action_content=None): self.action_code = action_code # normal:正常通话；hang:挂断；transfer：转人工 self.action_content = action_content # 动作内容 def to_json_string(self): return json.dumps(self.__dict__, ensure_ascii=False) @classmethod def from_json(cls, json_data): return cls( action_code=json_data.get("action_code"), action_content=json_data.get("action_content") ) class ChatContent: def __init__(self, content_type=None, content=None, voice_url=None, voice_content=None): self.content_type = content_type # 播放类型 self.content = content # 播放内容 self.voice_url = voice_url # 语音地址 self.voice_content = voice_content # 语音文本 def to_json_string(self): return json.dumps(self.__dict__, ensure_ascii=False) @classmethod def from_json(cls, json_data): return cls( content_type=json_data.get("content_type"), content=json_data.get("content"), voice_url=json_data.get("voice_url"), voice_content=json_data.get("voice_content") ) class ChatMessage: def __init__(self, node_id=None, contents=None, interruptable=None, wait_time=None, action=None, talk_time_out=None): self.node_id = node_id # 节点id self.contents = contents if contents is not None else [] # 内容列表 self.interruptable = interruptable # 是否可打断 self.wait_time = wait_time # 用户静默时长 self.action = action # 动作代码 self.talk_time_out = talk_time_out # 用户通话时长 def to_json_string(self): return json.dumps({ "node_id": self.node_id, "contents": [content.__dict__ for content in self.contents], "interruptable": self.interruptable, "wait_time": self.wait_time, "action": self.action.__dict__ if self.action else None, "talk_time_out": self.talk_time_out }, ensure_ascii=False) @classmethod def from_json(cls, json_data): contents = [ChatContent.from_json(item) for item in json_data.get("contents", [])] action = ChatAction.from_json(json_data.get("action", {})) if json_data.get("action") else None return cls( node_id=json_data.get("node_id"), contents=contents, interruptable=json_data.get("interruptable"), wait_time=json_data.get("wait_time"), action=action, talk_time_out=json_data.get("talk_time_out") ) class ChatResponse: def __init__(self, data=None, message=None, code=None): self.data = data if data is not None else ChatMessage() self.message = message self.code = code def to_json_string(self): return json.dumps({ "data": json.loads(self.data.to_json_string()), "message": self.message, "code": self.code }, ensure_ascii=False) @classmethod def from_json(cls, json_string): data = json.loads(json_string) response_data = ChatMessage.from_json(data.get("data", {})) return cls( data=response_data, message=data.get("message"), code=data.get("code") ) class ToTextBotAgent: def __init__(self, event_type, user_asr_text): if not user_asr_text: print("ASR文本为空，终止执行。") return self.request_data = BotChatRequest( node_id="1", user_id="139311", session_id="1", record_id="2", task_id="ceshi", asr_text=user_asr_text ) self.cur_player_file = None self.message_queue = queue.Queue(maxsize=3) # 发送请求并处理响应 self.test_request(self.request_data,event_type) self.wait_time = None def to_hang(self, action_content): print('todo ') def transfer_people(self): print('todo 转人工') def handling_release(self, message: ChatMessage): print('handling_release', message) if message: action = message.action action_code = action.action_code print(f"ActionCode: {action_code}", message) if action_code == 'hang': # 挂断 action_content = action.action_content self.to_hang(action_content) elif action_code == 'transfer': # 转人工 self.transfer_people() elif action_code == 'normal': # 正常通话 self.message_queue.put(message) print(f"成功获取响应: ActionCode: {message}") else: print("文本机器人接口调用失败") # 模拟接口请求返回 def test_request(self, params: BotChatRequest,event_type): print("test_request::params=", params) response_data = { "node_id": "1.0", "contents": [], "interruptable": False, "wait_time": "6", "action": { "action_code": "normal", "action_content": "正常通话" }, "talk_time_out": "" } print("event_type:", event_type) if event_type == '1': response_data['contents'].append({ "content_type": "voice", "content": "", "voice_url": '/code/src/core/voip/test111.wav', "voice_content": "五一北京到上海的高铁票还有吗？" }) elif event_type == '2': response_data['contents'].append({ "content_type": "voice", "content": "", "voice_url": '/code/src/core/voip/test222.wav', "voice_content": "测试第二个录音文件" }) # response = { # "data": response_data, # "message": "success", # "code": 200 # } try: # 转换为JSON字符串并解析 # response_json = json.dumps(response) parsed_response = ChatMessage.from_json(response_data) print(f"Response in test_request: {parsed_response}") # 确认response self.handling_release(parsed_response) except Exception as e: print(f"Error in test_request: {e}") traceback.print_exc() # 打印完整的错误信息 return None def get_player_file(self): try: # player_file = self.call.player_queue.get(block=False) message = self.message_queue.get(block=False) player_file = [item.voice_url for item in message.contents if item.content_type == 'voice'] wait_time = message.wait_time return player_file, wait_time except: pass def send_bot_speaker(self, player_file): if not player_file : return print(type(player_file)) player_id = murmur3_32(player_file) print('send_bot_speaker:', player_id) def ceshi(self): message_queue_size = self.message_queue.qsize() if (message_queue_size > 0 and not self.cur_player_file) : self.cur_player_file, self.wait_time = self.get_player_file() print('cur_player_file:', self.cur_player_file , 'wait_time=', self.wait_time) self.send_bot_speaker(self.cur_player_file) if __name__ == "__main__": # 运行主函数 # asyncio.run(main()) # bot = ToTextBotAgent("1",'sddsd') # bot.ceshi() wait_time =None if wait_time and wait_time != "0": print('wait_time:', wait_time)