dm/ai-slibra-assistant/util/utils.py

#!/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:
            for answer in answers:
                if answer.get("code") in ["1.20", "1.10", "1.00"]:
                    asr = answer.get("option", [{}])[-1].get("asr", '')
                    code = answer.get("code")
                    opt = answer.get("option", {})
                    if (code == "1.10" 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("嗯，那个我们家是碧水兰庭的"))