ai_qback/core/api_get_audio.py

# -*- coding: utf-8 -*-
from fastapi import FastAPI, Form, HTTPException, Request, status, APIRouter, Query, Path
from fastapi.responses import StreamingResponse
from tools.loglog import logger
from tools.audio import GetAudio
from core.respone_format import *
import traceback

router = APIRouter()
get_audio = GetAudio()

"""
生成音频tts接口，传递两个参数，word和resp_type； word和resp_type回复设计：0返回oss路径，1 二进制文件，2 url三种；
"""


@router.get("/tts")
def get_tts(word: str = Query(None, max_length=300)):
    try:
        f = get_audio.submit_task(word_or_phrase=word, resp_type=0)
        r = f.result()
        if r:
            return resp_200(data=r)
        return resp_500(message="生成失败")

    except Exception as e:
        logger.error(f"{type(e).__name__},{e}")
        return resp_500(message=f"{type(e).__name__},{e}")


@router.post("/tts")
def get_tts(json_data: dict, request: Request):
    word_or_phrase = json_data["text"]
    resp_type = json_data.get("type")

    if len(word_or_phrase) >= 300:
        logger.error(f"单词或短语过长")
        return resp_400(message="单词或短语过长")
    if resp_type not in [0, 1, 2]:
        logger.error(f"type参数不是012")
        return resp_400(message="type参数不是012")

    try:
        f = get_audio.submit_task(word_or_phrase=word_or_phrase, resp_type=resp_type)
        r = f.result()
        if r and resp_type in [0, 2]:
            return resp_200(data=r)
        if r and resp_type == 1:
            return StreamingResponse(content=r, media_type='audio/mpeg')
        return resp_500(message="生成失败")

    except Exception as e:
        traceback_str = traceback.format_exc()
        logger.error(traceback_str)
        logger.error(f"{type(e).__name__},{e}")
        return resp_500(message=f"{type(e).__name__},{e}")