#!/usr/bin/env python # -*- coding: utf-8 -*- # import sys # import json from typing import (Any, List, Text, Dict ) from scene import Kernel, Msg, Scene from config import (get_logger, FILE_PATH, START, END, HANGUP, TRANSFER, EMPTY, UNK, CONTINUE, MAX_TIMES_ANSWER, GENERATED, FIXED, # MOUDLES, FAQ, PURSUE) # from importlib import reload from botloader import Speech from util import (get_robot_speeches ,speech_main_contents ,get_next_code_with_track ,business_service ,intent_service ,nlg_service ) import re # # # from io import StringIO # from func import get_next_code_with_track logger = get_logger("taskbased") # 预加载话术 speeches = Speech.get_speech(FILE_PATH) class RobotService(Kernel): def __init__(self, **kwargs): self.code = kwargs.get("code", "start") self.times = kwargs.get("times", 1) self.answer = kwargs.get("answer", []) self.robot_speech_status = kwargs.get("robot_speech_status", None) self.robot_speech_interrupt = kwargs.get("robot_speech_interrupt", None) self.robot_speech = kwargs.get("robot_speech", '') # 机器人话术 self.status = kwargs.get("status", None) # 电话场景通话状态 def next(self, bid, uid, session_id, asr) -> Msg: # 进入下一节点 questions = speeches[bid] if self.code == END: # others bot 跳转结果页 return Msg(END) elif self.code == START: code = list(questions)[0] topic = questions[code] else: topic = questions[self.code] self.code = topic['id'] timeout = topic["waitTime"] inputType = topic['inputType'] if not topic['tools']: return Msg(code=self.code, question=self.robot_speech, interruptable=self.robot_speech_interrupt, wait_time=timeout, action=self.status, inputType= inputType ) else: tools = topic["tools"] option = business_service(session_id, uid, self.code, tools, asr) msg = Msg(code=self.code, option=option, session=session_id, asr=asr ) self.parse(msg, bid, uid, session_id) return self.next(bid, uid, session_id, msg.asr) def parse(self, msg, bid, uid, session_id) -> None: # 根据用户反馈解析数据 questions = speeches[bid] self.answer.append(dict(code=msg.code, question=self.robot_speech, option=msg.option, nodeName=questions[msg.code]['nodeName']) ) logger.info("userId: {}, code: {}, question:{}, user_asr:{}, answer:{}" \ .format(uid, msg.code, self.robot_speech, msg.option[-1]['asr'], msg.option[-1]['title'], ) ) # 当前节点话术状态选择 robot_speech, robot_speech_status, robot_speech_interrupt = get_robot_speeches(msg, bid, uid, questions) if msg.code == "101.00" and self.robot_speech and msg.option[-1]['title'] == "未听清": self.robot_speech = self.robot_speech else: self.robot_speech = robot_speech if robot_speech else self.robot_speech self.robot_speech_status = robot_speech_status self.robot_speech_interrupt = robot_speech_interrupt self.code = msg.code self.judge_over(questions) if msg.code != END: self.code = get_next_code_with_track(uid, msg.code, msg.option[-1]['title'], questions) if len(questions[self.code]['options']) <= 1 or not \ questions[self.code]['options'][0]['title']: if questions[self.code]['nodeName'] in ["转人工", "操作故障转人工"] or self.code in ["99.01", "99.00"]: self.status = TRANSFER else: self.status = HANGUP robot_speech, self.robot_speech_interrupt = speech_main_contents(uid, bid, self.code, questions, msg.option) else: self.status = CONTINUE if self.code != msg.code or msg.option[-1]['title'] == "FAQ": self.times = 1 else: self.times +=1 robot_speech, self.robot_speech_interrupt = speech_main_contents(uid, bid, self.code, questions, msg.option) if msg.code == "101.00" and self.robot_speech and msg.option[-1]['title'] == "未听清": self.robot_speech = self.robot_speech elif robot_speech: self.robot_speech = robot_speech @staticmethod def get_next_node(code: Text = None, option: Text = None, _questions: Dict[Any, Any] = None): """ @param code: node id @param option: 响应结果 @param _questions: 话术对象 @return: """ options = _questions[code]['options'] for cell in options: if option == cell['title']: return cell['next'] return code def judge_over(self, _questions: Dict[Any, Any] = None): # 判断结束(根据目前话术版本，以【节点跳转逻辑】为依据) options = _questions[self.code]['options'] if len(options) == 1 and not options[0]['title']: self.code = END class Dialog(Scene): def __init__(self, _case=None): if _case: self.case=_case else: self.case = RobotService() def dialogue(self, msg, bid, uid, session_id) -> Msg: # 驱动对话机器人跳转 # 会话开始 logger.info(f"节点:{msg.code}, 话术:{msg.question}, 用户asr:{msg.asr}, 用户意图:{msg.option}") if msg.code == "start": self.case.parse(msg, bid, uid, session_id) return self.case.next(bid, uid, session_id, msg.asr) # 触发下一条问答 self.case.parse(msg, bid, uid, session_id) return self.case.next(bid, uid, session_id, msg.asr) def get_current_code(self, bid=None): # 当前节点code if bid not in speeches: logger.info("({}) is not exist".format(bid)) questions = speeches[bid] if self.case.code == START: return list(questions)[0] return self.case.code def get_current_content(self, bid=None, uid=None): # 当前播报话术 from util import nlg_service questions = speeches[bid] cur_code = self.get_current_code(bid) topic = questions[cur_code] node_name = topic['nodeName'] main_question = topic['mainContent'] main_type = topic['mainType'] main_interrupt = topic['mainInterrupt'] timeout = topic["waitTime"] inputType = topic["inputType"] next_act = HANGUP if self.case.code == END else CONTINUE if main_type == GENERATED: # TODO 生成话术 self.case.robot_speech = nlg_service(uid, bid, node_name, 0) # self.case.robot_speech = main_question elif main_type == FIXED: self.case.robot_speech = main_question cur = self.case.robot_speech return Msg(code=cur_code, question=self.case.robot_speech, interruptable=main_interrupt, wait_time=timeout, action=next_act, inputType=inputType ) def mapping_semantics(self, code: Text = None, asr: Text = None, bid: Text = None, uid: Text = None, sessionid: Text = None): """ """ pattern = r'DTMF(.*?)DTMF' matches = re.findall(pattern, asr, re.DOTALL) if matches: asr = re.sub("[()]", "", matches[-1]) else: asr = asr.split("###")[-1] asr = asr.strip(r""""$%&'()*+,，-./:;<=>?@[\]^_`{|}~。？?!""") # 有按键，提取按键内容 questions = speeches[bid] res = [] options = questions[code]['options'] node_name = questions[code]['nodeName'] max_times = questions[code]['maxTimes'] main_type = questions[code]['mainType'] if not self.case.robot_speech: if main_type == GENERATED: self.case.robot_speech = nlg_service(uid, bid, node_name, 0) elif main_type == FIXED: self.case.robot_speech = questions[code]['mainContent'] for cell in options: if asr in cell['commonReply']: tmp = dict(title=cell['title'], asr=asr, isFaq=False, faqContent='') res.append(tmp) if not res and code not in ["2.10", "3.10", "4.10", "2.40", "3.40", "4.40", "5.10", "5.40"]: intent_res = intent_service(node_name, asr, bid, code, uid, sessionid) logger.info(f"intent_res:{intent_res}") if intent_res: for cell in options: for intent in intent_res: if cell['title'] == intent['intent'] or (cell['title'] == 'FAQ' and intent['isFaq']): res.append(dict(title=cell['title'], asr=asr, isFaq=True if intent['isFaq'] else False, faqContent=intent['ans'] if intent['isFaq'] else '', firstclass = intent.get('class') if intent.get('class') else '', subclass = intent.get('subclass') if intent.get('subclass') else '', ) ) logger.info("语义模型映射结果:{}".format(res)) return res if not res: res.append(dict(title=UNK, asr=asr, isFaq=False, faqContent='')) else: return res if (self.case.times >= int(max_times) + 1) and int(max_times) != 0: res.append(dict(title=MAX_TIMES_ANSWER, asr=asr, isFaq=False, faqContent='')) return res