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 re
import os
import shutil
import json
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 =1,1
    while num <= retry:
        try:
            url = r.hget(name, key)
            #url=''
            if 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 +=",解决轻按1，未解决请按2"
        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])+ "元。如需详细查询缴费和水量情况可以登陆佳木斯供水公众号，如需人工查询请拨打8247777，解决轻按1， 未解决请按2."
        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]}，重听请说再说一次，解决轻按“1”， 未解决请按“2”."
        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"]:
        asr = norm_community(asr)
    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__":
    print(voice_service("欢迎致电“佳木斯龙江环保供水服务热线”。我们最新推出智能语音服务，说话就能查询业务， 抢先体验请按1， 传统服务请按2"))
    #intent_service("1", "没听清", "2200", "1", "10", "1")
    #aibot_service()