dm/ai-slibra-assistant/api/interface.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
@Time    : 2024/10/11 14:08
@File    : interface.py
@Desc    : 
"""
from typing import Union,Optional
from pydantic import BaseModel
from scene import Msg
import json
from util import voice_service
from entity import Error
from config import get_logger
from starlette.responses import JSONResponse
logger = get_logger()

class reqRobot(BaseModel):
    """对话请求参数
    """
    nodeId:Optional[str]="start"
    userId:str
    sessionId:str
    taskId:str
    asrText:Optional[str] = None
    ext:Optional[str] = None
    recordId:Optional[str] = None


class ChatAction:
    def __init__(self, action_code=None, action_content=None):
        self.action_code = action_code         # normal:正常通话；hang:挂断；transfer：转人工
        self.action_content = action_content   # 动作内容

    def to_json_string(self):
        return json.dumps(self.__dict__, ensure_ascii=False)

    @classmethod
    def from_json(cls, json_data):
        return cls(
            action_code=json_data.get("action_code"),
            action_content=json_data.get("action_content")
        )


class ChatContent:
    def __init__(self, content_type=None, content=None, voice_url=None, voice_content=None):
        self.content_type = content_type   # 播放类型
        self.content = content             # 播放内容
        self.voice_url = voice_url         # 语音地址
        self.voice_content = voice_content # 语音文本

    def to_json_string(self):
        return json.dumps(self.__dict__, ensure_ascii=False)

    @classmethod
    def from_json(cls, json_data):
        return cls(
            content_type=json_data.get("content_type"),
            content=json_data.get("content"),
            voice_url=json_data.get("voice_url"),
            voice_content=json_data.get("voice_content")
        )


class ChatMessage:
    def __init__(self, node_id=None, contents=None, interruptable=None, wait_time=None,
                 action=None, inputType=None):
        self.node_id = node_id   # 节点id
        self.contents = contents if contents is not None else []   # 内容列表
        self.interruptable = interruptable   # 是否可打断
        self.wait_time = wait_time   # 用户静默时长
        self.action = action        # 动作代码
        self.inputType = inputType

    def to_json_string(self):
        return json.dumps({
            "node_id": self.node_id,
            "contents": [content.__dict__ for content in self.contents],
            # "interruptable": self.interruptable,
            "wait_time": self.wait_time,
            "action": self.action.__dict__ if self.action else None,
            "inputType": self.inputType
        }, ensure_ascii=False)

    @classmethod
    def from_json(cls, json_data):
        contents = [ChatContent.from_json(item) for item in json_data.get("contents", [])]
        action = ChatAction.from_json(json_data.get("action", {})) if json_data.get("action") else None
        return cls(
            node_id=json_data.get("node_id"),
            contents=contents,
            interruptable=json_data.get("interruptable"),
            wait_time=json_data.get("wait_time"),
            action=action,
            inputType = json_data.get("inputType")
        )


class ChatResponse:
    def __init__(self, data=None, message=None, code=None):
        self.data = data if data is not None else ChatMessage()
        self.message = message
        self.code = code

    def to_json_string(self):
        return JSONResponse(content={
            "data": json.loads(self.data.to_json_string()),
            "message": self.message,
            "code": self.code
        }, media_type="application/json")

    @classmethod
    def from_json(cls, json_string):
        data = json.loads(json_string)
        response_data = ChatMessage.from_json(data.get("data", {}))
        return cls(
            data=response_data,
            message=data.get("message"),
            code=data.get("code")
        )

    @staticmethod
    def get_content(msg: Msg):
        _contents, code, mess = [], 0, ''
        if msg.question:
            contents = msg.question.split("&")
            if len(contents) >= 1 and contents[0]:
                for content in contents:
                    if not content or len(content)==0:
                        continue
                    if len(contents) >1 and contents[1] and content ==contents[0]:
                        voice_url = voice_service(content, True, 1000)
                    else:
                        voice_url = voice_service(content, True)
                    _contents.append(
                        ChatContent(**{
                        "content_type": 'voice',
                        "content": content,
                        "voice_url": voice_url,
                        "voice_content": ''
                    })
                    )
                return _contents, 0, "success"
            elif len(contents) == 1 and not contents[0]:
                logger.error("话术生成服务异常{}".format(Error.error_nlg.value))
                return _contents, Error.error_nlg.value, "话术生成服务异常"
            elif len(contents) > 1 and not contents[0]:
                logger.error("faq服务异常{}".format(Error.error_faq.value))
                return _contents, Error.error_faq.value, "faq服务异常"
        return _contents, Error.error_nlg.value, "话术服务异常"
    @staticmethod
    def get_action(msg: Msg):
        action_code = ''
        action_content = ''
        if msg.action == 'normal':
            action_code = msg.action
            action_content = '正常通话'
        elif msg.action == 'hang':
            action_code = msg.action
            action_content = '机器人挂断'
        elif msg.action == "transfer":
            action_code = msg.action
            action_content = '转人工'
        return ChatAction(action_code, action_content)
    @classmethod
    def from_msg(cls, msg:Msg):
        contents, code, message = cls.get_content(msg)
        actions = cls.get_action(msg)
        kwargs = {
            "node_id": msg.code,
            "contents": contents,
            # "interruptable": msg.interruptable,
            "wait_time": msg.wait_time,
            "action": actions,
            "inputType": msg.inputType
        }
        data = ChatMessage(**kwargs)
        return cls(
            data= data,
            message= message,
            code = code
        )