#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
@Time : 2024/10/15 16:19
@File : utils.py
@Desc :
"""
import sys
sys.path.append("..")
from datetime import datetime
from functools import wraps
from typing import (Any,
List,
Text,
Dict
)
from config import (
get_logger,
GENERATED,
FIXED,
MOUDLES
)
import pandas as pd
from threading import Thread
import json
logger = get_logger()
from database import Mysql
from pypinyin import pinyin, Style
import jieba
import re
import itertools
from concurrent.futures import ThreadPoolExecutor,as_completed
executor = ThreadPoolExecutor(max_workers=20)
def get_speech_status(bid: Text = None, options: List[Dict[Text, Text]] = None):
"""which speech template to choose"""
res = dict()
module = MOUDLES[bid]
option = ''
if options:
option = options[-1]['title'] # 外呼机器人单选
for key, value in module.items():
if isinstance(value, dict):
if option in value['content']:
res['action'] = key
res['speech_id'] = value['speech_id']
res['speech_type'] = value['speech_type']
res['speech_interrupt'] = value['speech_interrupt']
res['asr'] = options[-1]['asr']
break
return res
def get_robot_speeches(msg, bid, uid, questions: Dict[Any, Any] = None):
"""GET speech and speech status"""
from entity import Status
choose_speech_status = get_speech_status(bid, msg.option)
if not choose_speech_status:
choose_speech = Status.base.value
robot_speech, interrupt = speech_main_contents(uid, bid, msg.code, questions)
interrupt = questions[msg.code]['mainInterrupt']
return robot_speech, choose_speech, interrupt
return None, None, None
def speech_main_contents(uid: Text = None,
bid: Text = None,
code: Text = None,
questions: Dict[Any, Any] = None,
options: List[Dict[Any, Any]] = None):
from util import nlg_service
from entity import Status
def parse_options():
if options:
option = options[-1]
is_faq = option['isFaq']
faq_content = option['faqContent']
if is_faq:
return faq_content
if "businessContent" in option:
return option['businessContent']
return
_robot_speech = ''
speech_type = questions[code]['mainType']
node_name = questions[code]['nodeName']
interrupt = questions[code]['mainInterrupt']
choose_speech = Status.base.value
#logger.info(f"code:{code},speech_type:{speech_type}")
topic = questions[code]
# logger.info(f"topic: {topic}")
_faq_content = parse_options()
if speech_type == GENERATED:
# TODO 生成话术
resp = nlg_service(uid, bid, node_name, choose_speech)
if _faq_content or resp:
_robot_speech = "{}&{}".format(_faq_content, resp) if _faq_content else resp
elif speech_type == FIXED:
content = questions[code]['mainContent']
if content or _faq_content:
_robot_speech = "{}&{}".format(_faq_content, content) if _faq_content else content
return _robot_speech, interrupt
def get_next_code_with_track(uid: Text = None,
code: Text = None,
option: Text = None,
questions: Dict[Any, Any] = None
):
"""
use user trackCode get qus id
@param uid:
@param code:
@param option:
@param questions:
@return:
"""
options = questions[code]['options']
for cell in options:
if option == cell['title']:
next = cell['next']
logger.info(f"uid:{uid}:code from {code} to {next}")
return next
return code
def _async(f):
def wrapper(*args, **kwargs):
thr = Thread(target=f, args=args, kwargs=kwargs)
thr.start()
return wrapper
@_async
def insert_log(bid, uid, session_id, scene):
"""
CREATE TABLE botrecords (
id INT AUTO_INCREMENT PRIMARY KEY,
session VARCHAR(50) unique not null COMMENT '请求id',
req_time DATETIME COMMENT '来电时间',
uid VARCHAR(20) COMMENT '来电手机号',
bid VARCHAR(20) COMMENT '话术id',
intent VARCHAR(20) COMMENT '意图',
contents TEXT comment '内容',
dialog TEXT COMMENT '对话'
);
"""
tmp = json.dumps(scene.case, ensure_ascii=False, default=lambda obj: obj.__dict__)
res = json.loads(tmp)
if res:
answers = res.get("answer")
intent = ''
codes = [i for i in map(lambda x: x.get("code"), answers)]
if "1.20" in codes or "1.10" in codes or "1.00" in codes or "1.30" in codes:
for answer in answers:
if answer.get("code") in ["1.20", "1.10", "1.00", "1.30"]:
asr = answer.get("option", [{}])[-1].get("asr", '')
code = answer.get("code")
opt = answer.get("option", {})
if (code in ["1.10", "1.30"] and asr!="1") or (code == "1.00" and asr not in ["2", "1"]):
if opt:
if opt[-1].get("subclass"):
intent = opt[-1].get("subclass")
elif opt[-1].get("firstclass"):
intent = opt[-1].get("firstclass")
else:
intent =opt[-1].get("title", '')
elif code == "1.20":
tags = {"1": "1_停水咨询", "2":"1_漏水保修", "3":"1_户号查询","4":"1_水费查询", "5":"1_水价咨询","6":"1_水质水价保修", "7":"1_投诉建议", "0":"1_转人工"}
intent = tags.get(asr, "1_其他")
elif code == "1.00" and asr=="2":
intent = "不体验AI服务"
elif code == "1.00" and asr=="1":
intent = asr
filter_ans = filter(lambda x: x.get("code") not in ["2.00", "3.00", "4.00"], answers)
content = [i for i in map(lambda x:[ x.get("question"), x.get("option", [{}])[-1].get("asr", '')], filter_ans)]
content.append((res.get("robot_speech"), ''))
contents = json.dumps({"data": content}, ensure_ascii=False)
req_time = datetime.now().strftime("%Y/%m/%d %H:%M:%S")
mysql = Mysql()
mysql.insert_records([session_id, req_time, uid, bid, intent, contents, json.dumps(res, ensure_ascii=False)])
# records = mysql.get_records(uid)
# logger.info(f"databases:uid:{uid}, {records}")
mysql.close_mysql()
def timetic(func):
@wraps(func)
def wrapper(*args, **kwargs):
start = datetime.now()
results = func(*args, **kwargs)
cost = (datetime.now() - start).total_seconds()
if func.__qualname__ == "botservice":
bot = kwargs.get("reqbot")
sessionId, userId,nodeId = bot.sessionId, bot.userId,bot.nodeId
logger.info("{},session:{},uid:{},nodeid:{} ==> {}s".format(func.__qualname__,sessionId , userId, nodeId, cost))
else:
logger.info("{} ==> {}s".format(func.__qualname__, cost))
return results
return wrapper
def loaddict():
loc = dict()
df = pd.read_excel("../data/location.xlsx", header=0)
loc['zh'] = dict(df[['norm_name', 'name']].values)
loc['pinyin'] = dict(df[['name_pinyin', 'name']].values)
short_val = [(i, 80) for i in df['short_name'].dropna().tolist()]
norm_val = [(i, 100) for i in df['norm_name'].dropna().tolist()]
norm_val.extend(short_val)
loc['total'] = dict(norm_val)
return loc
user_dict= loaddict()
jieba.load_userdict(user_dict['total'])
def norm_community(asr):
def match_loc(comb):
cur_wd = ''.join(comb)
if cur_wd in asr:
if cur_wd in user_dict['zh']:
return user_dict['zh'].get(cur_wd)
term = "|".join([term[0] for term in pinyin(cur_wd, style=Style.NORMAL)])
if term in user_dict['pinyin']:
return user_dict['pinyin'][term]
return None
if asr in user_dict['zh']:
return user_dict['zh'][asr]
text = re.sub(r'[(（）)]', '', asr)
text = "|".join([word[0] for word in pinyin(text, style=Style.NORMAL)])
if text in user_dict['pinyin']:
return user_dict['pinyin'][text]
words = jieba.lcut(asr)
for word in words:
if word in user_dict['zh']:
return user_dict['zh'].get(word)
term = "|".join([term[0] for term in pinyin(word, style=Style.NORMAL)])
if term in user_dict['pinyin']:
return user_dict['pinyin'][term]
for r in range(1, len(words) + 1):
combinations_list = list(itertools.combinations(words, r))
features = [executor.submit(match_loc, combo) for combo in combinations_list]
result =[feature.result() for feature in as_completed(features)]
res = [i for i in filter(lambda x: x is not None, result)]
if len(res) >0:
return res[0]
return asr
if __name__ == "__main__":
print(norm_community("五一村"))