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

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
@Time    : 2024/10/17 14:36
@File    : request_utils.py.py
@Desc    : 
"""
#import sys
#sys.path.append("/Users/caoyongchuang/PycharmProjects/predict/ai-slibra-assistant")
import re
import os
import shutil
import json
import time

from config import get_logger
import requests
import base64
from database import *
from util import timetic, norm_community
import uuid
SERVE_HOST = os.environ.get("SERVE_HOST", "192.168.100.159")
VOICE_DIR = os.environ.get("VOICE_DIR","/root/aibot/dm/voice")


logger = get_logger()

def nlg_service(uid, bid, node_name, choose_speech):
    return ''


def get_voice(content, is_local, silence_duration):
    # appid = "1226203350"
    # access_token = "mMPNJ4WbSVL6NHQKvn4qllYfwv1G4X5w"
    # 企业
    appid = "8661018433"
    access_token ="KkTVKPD9kY-i27Hr9gXtFQ4zqY7nrhJl"
    cluster = "volcano_tts"
    voice_type = "BV700_V2_streaming"
    host = "openspeech.bytedance.com"
    api_url = f"https://{host}/api/v1/tts"
    header = {"Authorization": f"Bearer;{access_token}"}
    #DEFAULT_FILE_UPLOAD_URL = 'http://10.0.0.28:8080/qiniuyun/upLoadImage'
    DEFAULT_FILE_UPLOAD_URL = 'http://192.168.9.54:8080/qiniuyun/upLoadImage'
    rid = str(uuid.uuid4())
    request_json = {
        "app": {
            "appid": appid,
            "token": "access_token",
            "cluster": cluster
        },
        "user": {
            "uid": "388808087185088"
        },
        "audio": {
            "voice_type": voice_type,
            #"encoding": "mp3",
            "encoding": "wav",
            "rate": "16000",
            "speed_ratio": 0.9,
            "volume_ratio": 1.0,
            "pitch_ratio": 1.0,
            "emotion": "customer_service",
            "language": "zh"
        },
        "request": {
            "reqid": rid,
            "text": content,
            "text_type": "plain",
            "operation": "query",
            "with_frontend": 1,
            "frontend_type": "unitTson",
            "silence_duration":silence_duration
        }
    }
    try:
        resp = requests.post(api_url, json.dumps(request_json), headers=header)
        rfile = f"{rid}.wav"
        path, file = f"{VOICE_DIR}/{rfile}", f"../voice/{rfile}"
        if "data" in resp.json():
            data = resp.json()["data"]
            file_to_save = open(file, "wb")
            file_to_save.write(base64.b64decode(data))
            if is_local:
                if os.path.exists(file):
                    logger.info(f"voice local file ::session_id={rid}, res={file}")
                    return path
            else:
                files = {'file': open(file, 'rb')}
                response = requests.post(DEFAULT_FILE_UPLOAD_URL, files=files)
                if response.ok:
                    result = json.loads(response.text)
                    url = result.get('data')
                    if os.path.exists(file):
                        os.system(f"rm -fr {file}")
                    logger.info(f"voice upload_plot::session_id={rid}, res={url}")
                    return url
    except Exception as e:
        logger.info(f"voice generate 错误{e}")


@timetic
def voice_service(content, local=False, silence_duration="125"):
    encode_b = content.encode("utf-8")
    key = base64.b64encode(encode_b)
    name = "voice_url"
    retry,num =3,1
    while num <= retry:
        try:
            url = r.hget(name, key)
            #url=''
            if url and os.path.exists(url):
                logger.info(f"获取voice url成功:{content}")
                return url
            else:
                url = get_voice(content, local, silence_duration)
                r.hset(name, key, url)
                r.expire(name, 3600 * 24*7)
                return url
        except Exception as e:
            logger.info(f"get voice url {num}缓存错误{e}")
            num +=1

@timetic
def intent_service(node_name, asr, bid, code, uid, sessionid):
    param = json.dumps(dict(nodeId=code,
                            userId=uid,
                            sessionId=sessionid,
                            taskId=bid,
                            query=asr,
                            nodeName = node_name
                            ), ensure_ascii=False)
    try:
        #ip= "10.0.0.24"
        res = requests.post(f"http://{SERVE_HOST}:50072/intention",
                            param.encode("UTF-8"),
                            headers={'Content-Type': 'application/json;charset=utf-8'},
                            timeout=8)
        content = json.loads(res.text)
        logger.info(f"intent service:{content}")
        return [content]
    except Exception as e:
        logger.error(f"intent服务异常:query:{asr},uid:{uid},session:{sessionid}:{e}")
        return []

@timetic
def business_service(session_id, uid, code, tools,  asr):
    def getContent(contents, tools):
        if tools in ["water_info", "water_loc_info"]:
            mess = '您查询的小区'
            neighbourhoodName = [item.get('neighbourhoodName') for item in contents]
            reason = ["因"+item.get('reason') for item in contents]
            timeBegin = ["于"+item.get('timeBegin') + "停水" for item in contents]
            timeEnd = [item.get('timeEnd') for item in contents]
            conclusions = [f"预计{time}恢复供水" if time is not None else "暂未确定恢复时间" for time in timeEnd]
            nums = len(neighbourhoodName)
            mess +=";".join([",".join(i) for i in zip(neighbourhoodName, reason, timeBegin, conclusions)])
            mess +=",是否已解决您的问题？"
        elif tools in ["fee_info", "fee_user_info"]:
            mess = "您账户截止"
            statisticsTime = [content.get("statisticsTime", '') for content in contents]
            waterFees = [round(float(content.get("waterFees", 0)), 2) for content in contents]
            meterAmount = [round(float(content.get("meterAmount", 0)), 2) for content in contents]
            mess = mess + statisticsTime[0] + "您的抄表表数为"+str(meterAmount[0]) +"欠费金额为" + str(waterFees[0])+ "元。如需详细查询缴费和水量情况可以登陆佳木斯供水公众号，如需人工查询请拨打824/--/777/--/6，还有什么可以帮你？."
        elif tools in ["user_info", "user_phone_info"]:
            neighbour, cardNo = [content.get("neighbourhoodName") for content in contents], [content.get("userNo") for content in contents]
            nums = len(neighbour)
            mess = f"根据您的手机号查询到{nums}个小区，" + ";".join(map(lambda x: ",".join(x), zip(neighbour, ["户号是"]* nums, cardNo)))+ "。解决轻按1， 未解决请安2."
        elif tools in ["meter_owner_phone", "meter_owner_neighbour"]:
            loc_phone = [[content['neighbourhoodName'], content['meterReaderPhone']] for content in contents]
            mess = f"您查询的小区{loc_phone[0][0]}，抄表员电话是{loc_phone[0][1]}；抄表员电话是{loc_phone[0][1]}, 重听请说再说一次，是否已解决您的问题？"
        else:
            mess = ''
        return mess
    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""""$%&'()*+,，-./:;<=>?@[\]^_`{|}~。？?!""")
    if tools in ["water_loc_info", "fee_user_info", "user_phone_info", "meter_owner_neighbour"] and len(asr)==0:
        return [{"title": "NO", "isFaq": False, "faqContent": '', "asr": asr, "businessContent": ''}]
    # parse water_loc_info
    if tools in ["water_loc_info", "meter_owner_neighbour", "water_info", "meter_owner_phone"]:
        asr = norm_community(asr)
    if tools in ["water_info", "meter_owner_phone"]:
        newtools = "water_loc_info" if tools == "water_info" else "meter_owner_neighbour"
        param = json.dumps(dict(nodeId=code,
                                userId=uid,
                                sessionId=session_id,
                                asrText=asr,
                                method=newtools
                                ), ensure_ascii=False)
        try:
            # 192.168.40.21
            res = requests.post(f"http://{SERVE_HOST}:8001/bigModel/queryBusinessInfo",
                                param.encode("UTF-8"),
                                headers={'Content-Type': 'application/json;charset=utf-8'},
                                timeout=30)
            resp = json.loads(res.text)
            logger.info(f"bussiness:{resp}, tools:{tools}, session:{session_id}")
            if resp['code'] == "0":
                content = resp['data'].get("contents")
                title = "NO" if content is None or len(content) == 0 else "YES"
                businessContent = getContent(content, tools) if title == "YES" else ''
                opt = [
                    {"title": title, "isFaq": False, "faqContent": '', "asr": asr, "businessContent": businessContent}]
                logger.info(f"code:{code},uid:{uid}, tools:{tools},asr:{asr}, opt:{opt}")
                return opt
        except Exception as e:
            logger.info(f"bussion service服务异常:session:{session_id}, tools:{tools},uid:{uid}:{e}")
    param = json.dumps(dict(nodeId=code,
                            userId=uid,
                            sessionId=session_id,
                            asrText=asr,
                            method=tools
                            ), ensure_ascii=False)
    try:
        # 192.168.40.21
        res = requests.post(f"http://{SERVE_HOST}:8001/bigModel/queryBusinessInfo",
                            param.encode("UTF-8"),
                            headers={'Content-Type': 'application/json;charset=utf-8'},
                            timeout=30)
        resp = json.loads(res.text)
        logger.info(f"bussiness:{resp}, tools:{tools}, session:{session_id}")
        if resp['code'] == "0":
            content = resp['data'].get("contents")
            title = "NO" if content is None or len(content)==0  else "YES"
            businessContent = getContent(content, tools) if title == "YES" else ''
            opt = [{"title": title, "isFaq": False, "faqContent": '', "asr": asr, "businessContent": businessContent}]
            logger.info(f"code:{code},uid:{uid}, tools:{tools},asr:{asr}, opt:{opt}")
            return opt
        else:
            return [{"title": "NO", "isFaq": False, "faqContent": '', "asr": asr, "businessContent": ''}]
    except Exception as e:
        logger.info(f"bussion service服务异常:session:{session_id}, tools:{tools},uid:{ uid}:{e}")
        return [{"title": "NO", "isFaq": False, "faqContent": '', "asr": asr, "businessContent":''}]


@timetic
def aibot_service(ip ="192.168.100.159",port="40072", nodeId="start", userId='no', sessionId='1', taskId='10001', asrText='是', ext='', recordId=''):
    param = json.dumps(dict(nodeId=nodeId,
                            userId=userId,
                            sessionId=sessionId,
                            taskId=taskId,
                            asrText=asrText,
                            ext=ext,
                            recordId=recordId
                            ), ensure_ascii=False)
    try:
        res = requests.post(f"http://{ip}:{port}/botservice",
                            param.encode("UTF-8"),
                            headers={'Content-Type': 'application/json;charset=utf-8'},
                            timeout=30)
        resp = json.loads(res.text)
        if resp['code'] == 0:
            content = resp['data']
            logger.info(f"aibot: {resp}")
            return resp
    except Exception as e:
        logger.error(f"Ai bot服务异常:{nodeId}, {taskId},{ userId}:{e}")
        return



if __name__ == "__main__":
    text="欢迎致电“佳木斯龙江环保供水服务热线”。我们最新推出智能语音服务，说话就能查询业务， 抢先体验请按1， 传统服务请按2"
    print(voice_service(text))
    #intent_service("1", "没听清", "2200", "1", "10", "1")
    #aibot_service()