tts_app/voice_api/views.py

import copy
import logging
import os
import time
import traceback
import uuid
from io import BytesIO

import librosa
import numpy as np
from flask import request, jsonify, make_response, send_file, Blueprint
from werkzeug.utils import secure_filename

from contants import config
# from gpt_sovits.utils import load_audio
from logger import logger
from contants import ModelType
from tts_app.voice_api.auth import require_api_key
from tts_app.model_manager import model_manager, tts_manager
from tts_app.voice_api.utils import *
from utils.data_utils import check_is_none

voice_api = Blueprint("voice_api", __name__)


def get_param(request_data, key, default, data_type=None):
    if key == "segment_size" and "max" in request_data:
        logger.warning(
            "The 'max' parameter is deprecated and will be phased out in the future. Please use 'segment_size' instead.")
        return get_param(request_data, "max", default, data_type)

    value = request_data.get(key, "")

    if data_type:
        try:
            value = data_type(value)
        except:
            value = default

    if value == "":
        value = default

    return value


def extract_filename_and_directory(path):
    filename = os.path.basename(path)
    directory = os.path.dirname(path)
    directory_name = os.path.basename(directory)
    if not directory:  # 如果文件所在文件夹为空（即在根目录）
        return filename
    else:
        return directory_name + "/" + filename


def update_default_params(state):
    model_type = state["model_type"]
    if model_type == ModelType.VITS:
        config_dict = config.vits_config.asdict()
    elif model_type == ModelType.W2V2_VITS:
        config_dict = config.w2v2_vits_config.asdict()
    elif model_type == ModelType.HUBERT_VITS:
        config_dict = config.hubert_vits_config.asdict()
    elif model_type == ModelType.BERT_VITS2:
        config_dict = config.bert_vits2_config.asdict()
    elif model_type == ModelType.GPT_SOVITS:
        config_dict = config.gpt_sovits_config.asdict()

    for key, value in config_dict.items():
        if key not in state or value is None:
            state[key] = value
    return state


@voice_api.route('/default_parameter', methods=["GET", "POST"])
def default_parameter():
    gpt_sovits_config = copy.deepcopy(config.gpt_sovits_config.asdict())
    for preset_name, preset in gpt_sovits_config["presets"].items():
        if not check_is_none(preset["refer_wav_path"]):
            preset["refer_wav_path"] = extract_filename_and_directory(preset["refer_wav_path"])

    data = {"vits_config": config.vits_config.asdict(),
            "w2v2_vits_config": config.w2v2_vits_config.asdict(),
            "hubert_vits_config": config.hubert_vits_config.asdict(),
            "bert_vits2_config": config.bert_vits2_config.asdict(),
            "gpt_sovits_config": gpt_sovits_config
            }
    return jsonify(data)


@voice_api.route('/speakers', methods=["GET", "POST"])
def voice_speakers_api():
    return jsonify(model_manager.voice_speakers)


@voice_api.route('/', methods=["GET", "POST"])
@voice_api.route('/vits', methods=["GET", "POST"])
@require_api_key
def voice_vits_api():
    try:
        if request.method == "GET":
            request_data = request.args
        elif request.method == "POST":
            content_type = request.headers.get('Content-Type')
            if content_type == 'application/json':
                request_data = request.get_json()
            else:
                request_data = request.form

        text = get_param(request_data, "text", "", str)
        id = get_param(request_data, "id", config.vits_config.id, int)
        format = get_param(request_data, "format", config.vits_config.format, str)
        lang = get_param(request_data, "lang", config.vits_config.lang, str).lower()
        length = get_param(request_data, "length", config.vits_config.length, float)
        noise = get_param(request_data, "noise", config.vits_config.noise, float)
        noisew = get_param(request_data, "noisew", config.vits_config.noisew, float)
        segment_size = get_param(request_data, "segment_size", config.vits_config.segment_size, int)
        use_streaming = get_param(request_data, 'streaming', config.vits_config.use_streaming, bool)
    except Exception as e:
        logger.error(f"[{ModelType.VITS.value}] {e}")
        return make_response("parameter error", 400)

    logger.info(
        f"[{ModelType.VITS.value}] id:{id} format:{format} lang:{lang} length:{length} noise:{noise} noisew:{noisew} segment_size:{segment_size}")
    logger.info(f"[{ModelType.VITS.value}] len:{len(text)} text：{text}")

    if check_is_none(text):
        logger.info(f"[{ModelType.VITS.value}] text is empty")
        return make_response(jsonify({"status": "error", "message": "text is empty"}), 400)

    if check_is_none(id):
        logger.info(f"[{ModelType.VITS.value}] speaker id is empty")
        return make_response(jsonify({"status": "error", "message": "speaker id is empty"}), 400)

    if id < 0 or id >= model_manager.vits_speakers_count:
        logger.info(f"[{ModelType.VITS.value}] speaker id {id} does not exist")
        return make_response(jsonify({"status": "error", "message": f"id {id} does not exist"}), 400)

    # 校验模型是否支持输入的语言
    speaker_lang = model_manager.voice_speakers[ModelType.VITS.value][id].get('lang')
    if lang not in ["auto", "mix"] and len(speaker_lang) > 1 and lang not in speaker_lang:
        logger.info(f"[{ModelType.VITS.value}] lang \"{lang}\" is not in {speaker_lang}")
        return make_response(jsonify({"status": "error", "message": f"lang '{lang}' is not in {speaker_lang}"}),
                             400)

    # 如果配置文件中设置了LANGUAGE_AUTOMATIC_DETECT则强制将speaker_lang设置为LANGUAGE_AUTOMATIC_DETECT
    if (lang_detect := config.language_identification.language_automatic_detect) and isinstance(lang_detect, list):
        speaker_lang = lang_detect

    if use_streaming and format.upper() != "MP3":
        format = "mp3"
        logger.warning("Streaming response only supports MP3 format.")

    fname = f"{str(uuid.uuid1())}.{format}"
    file_type = f"audio/{format}"
    state = {"text": text,
             "id": id,
             "format": format,
             "length": length,
             "noise": noise,
             "noisew": noisew,
             "segment_size": segment_size,
             "lang": lang,
             "speaker_lang": speaker_lang}

    if use_streaming:
        audio = tts_manager.stream_vits_infer(state)
        response = make_response(audio)
        response.headers['Content-Disposition'] = f'attachment; filename={fname}'
        response.headers['Content-Type'] = file_type
        return response
    else:
        t1 = time.time()
        audio = tts_manager.vits_infer(state)
        t2 = time.time()
        logger.info(f"[{ModelType.VITS.value}] finish in {(t2 - t1):.2f}s")

        if config.system.cache_audio:
            logger.debug(f"[{ModelType.VITS.value}] {fname}")
            path = os.path.join(config.system.cache_path, fname)
            save_audio(audio.getvalue(), path)

        return send_file(path_or_file=audio, mimetype=file_type, download_name=fname)


@voice_api.route('/hubert-vits', methods=["POST"])
@require_api_key
def voice_hubert_api():
    if request.method == "POST":
        try:
            voice = request.files['upload']
            id = get_param(request.form, "id", config.hubert_vits_config.id, int)
            format = get_param(request.form, "format", config.hubert_vits_config.format)
            length = get_param(request.form, "length", config.hubert_vits_config.length, float)
            noise = get_param(request.form, "noise", config.hubert_vits_config.noise, float)
            noisew = get_param(request.form, "noisew", config.hubert_vits_config.noisew, float)
            use_streaming = get_param(request.form, 'streaming', False, bool)
        except Exception as e:
            logger.error(f"[{ModelType.HUBERT_VITS.value}] {e}")
            return make_response("parameter error", 400)

    logger.info(
        f"[{ModelType.HUBERT_VITS.value}] id:{id} format:{format} length:{length} noise:{noise} noisew:{noisew}")

    fname = secure_filename(str(uuid.uuid1()) + "." + voice.filename.split(".")[1])
    voice.save(os.path.join(config.system.upload_folder, fname))

    if check_is_none(id):
        logger.info(f"[{ModelType.HUBERT_VITS.value}] speaker id is empty")
        return make_response(jsonify({"status": "error", "message": "speaker id is empty"}), 400)

    if id < 0 or id >= model_manager.hubert_speakers_count:
        logger.info(f"[{ModelType.HUBERT_VITS.value}] speaker id {id} does not exist")
        return make_response(jsonify({"status": "error", "message": f"id {id} does not exist"}), 400)

    file_type = f"audio/{format}"
    task = {"id": id,
            "format": format,
            "length": length,
            "noise": noise,
            "noisew": noisew,
            "audio_path": os.path.join(config.system.upload_folder, fname)}

    t1 = time.time()
    audio = tts_manager.hubert_vits_infer(task)
    t2 = time.time()
    logger.info(f"[{ModelType.HUBERT_VITS.value}] finish in {(t2 - t1):.2f}s")

    if config.system.cache_audio:
        logger.debug(f"[{ModelType.HUBERT_VITS.value}] {fname}")
        path = os.path.join(config.system.cache_path, fname)
        save_audio(audio.getvalue(), path)

    if use_streaming:
        audio = tts_manager.generate_audio_chunks(audio)
        response = make_response(audio)
        response.headers['Content-Disposition'] = f'attachment; filename={fname}'
        response.headers['Content-Type'] = file_type
        return response
    else:
        return send_file(path_or_file=audio, mimetype=file_type, download_name=fname)


@voice_api.route('/w2v2-vits', methods=["GET", "POST"])
@require_api_key
def voice_w2v2_api():
    try:
        if request.method == "GET":
            request_data = request.args
        elif request.method == "POST":
            content_type = request.headers.get('Content-Type')
            if content_type == 'application/json':
                request_data = request.get_json()
            else:
                request_data = request.form

        text = get_param(request_data, "text", "", str)
        id = get_param(request_data, "id", config.w2v2_vits_config.id, int)
        format = get_param(request_data, "format", config.w2v2_vits_config.format, str)
        lang = get_param(request_data, "lang", config.w2v2_vits_config.lang, str).lower()
        length = get_param(request_data, "length", config.w2v2_vits_config.length, float)
        noise = get_param(request_data, "noise", config.w2v2_vits_config.noise, float)
        noisew = get_param(request_data, "noisew", config.w2v2_vits_config.noisew, float)
        segment_size = get_param(request_data, "segment_size", config.w2v2_vits_config.segment_size, int)
        emotion = get_param(request_data, "emotion", config.w2v2_vits_config.emotion, int)
        emotion_reference = get_param(request_data, "emotion_reference", None, str)
        use_streaming = get_param(request_data, 'streaming', False, bool)
    except Exception as e:
        logger.error(f"[{ModelType.W2V2_VITS.value}] {e}")
        return make_response(f"parameter error", 400)

    logger.info(f"[{ModelType.W2V2_VITS.value}] id:{id} format:{format} lang:{lang} "
                f"length:{length} noise:{noise} noisew:{noisew} emotion:{emotion} segment_size:{segment_size}")
    logger.info(f"[{ModelType.W2V2_VITS.value}] len:{len(text)} text：{text}")

    if check_is_none(text):
        logger.info(f"[{ModelType.W2V2_VITS.value}] text is empty")
        return make_response(jsonify({"status": "error", "message": "text is empty"}), 400)

    if check_is_none(id):
        logger.info(f"[{ModelType.W2V2_VITS.value}] speaker id is empty")
        return make_response(jsonify({"status": "error", "message": "speaker id is empty"}), 400)

    if id < 0 or id >= model_manager.w2v2_speakers_count:
        logger.info(f"[{ModelType.W2V2_VITS.value}] speaker id {id} does not exist")
        return make_response(jsonify({"status": "error", "message": f"id {id} does not exist"}), 400)

    # 校验模型是否支持输入的语言
    speaker_lang = model_manager.voice_speakers[ModelType.W2V2_VITS.value][id].get('lang')
    if lang not in ["auto", "mix"] and len(speaker_lang) > 1 and lang not in speaker_lang:
        logger.info(f"[{ModelType.W2V2_VITS.value}] lang \"{lang}\" is not in {speaker_lang}")
        return make_response(jsonify({"status": "error", "message": f"lang '{lang}' is not in {speaker_lang}"}),
                             400)

    # 如果配置文件中设置了LANGUAGE_AUTOMATIC_DETECT则强制将speaker_lang设置为LANGUAGE_AUTOMATIC_DETECT
    if (lang_detect := config.language_identification.language_automatic_detect) and isinstance(lang_detect, list):
        speaker_lang = lang_detect

    if use_streaming and format.upper() != "MP3":
        format = "mp3"
        logger.warning("Streaming response only supports MP3 format.")

    fname = f"{str(uuid.uuid1())}.{format}"
    file_type = f"audio/{format}"
    task = {"text": text,
            "id": id,
            "format": format,
            "length": length,
            "noise": noise,
            "noisew": noisew,
            "segment_size": segment_size,
            "lang": lang,
            "emotion": emotion,
            "emotion_reference": emotion_reference,
            "speaker_lang": speaker_lang}

    t1 = time.time()
    audio = tts_manager.w2v2_vits_infer(task)
    t2 = time.time()
    logger.info(f"[{ModelType.W2V2_VITS.value}] finish in {(t2 - t1):.2f}s")

    if config.system.cache_audio:
        logger.debug(f"[{ModelType.W2V2_VITS.value}] {fname}")
        path = os.path.join(config.system.cache_path, fname)
        save_audio(audio.getvalue(), path)

    if use_streaming:
        audio = tts_manager.generate_audio_chunks(audio)
        response = make_response(audio)
        response.headers['Content-Disposition'] = f'attachment; filename={fname}'
        response.headers['Content-Type'] = file_type
        return response
    else:
        return send_file(path_or_file=audio, mimetype=file_type, download_name=fname)


@voice_api.route('/conversion', methods=["POST"])
@voice_api.route('/vits/conversion', methods=["POST"])
@require_api_key
def vits_voice_conversion_api():
    if request.method == "POST":
        try:
            voice = request.files['upload']
            original_id = get_param(request.form, "original_id", 0, int)
            target_id = get_param(request.form, "target_id", 0, int)
            format = get_param(request.form, "format", voice.filename.split(".")[1], str)
            use_streaming = get_param(request.form, 'streaming', False, bool)
        except Exception as e:
            logger.error(f"[vits_voice_convertsion] {e}")
            return make_response("parameter error", 400)

        logger.info(f"[vits_voice_convertsion] orginal_id:{original_id} target_id:{target_id}")
        fname = secure_filename(str(uuid.uuid1()) + "." + voice.filename.split(".")[1])
        audio_path = os.path.join(config.system.upload_folder, fname)
        voice.save(audio_path)
        file_type = f"audio/{format}"
        state = {"audio_path": audio_path,
                 "original_id": original_id,
                 "target_id": target_id,
                 "format": format}

        t1 = time.time()
        audio = tts_manager.vits_voice_conversion(state)
        t2 = time.time()
        logger.info(f"[Voice conversion] finish in {(t2 - t1):.2f}s")

        if config.system.cache_audio:
            logger.debug(f"[Voice conversion] {fname}")
            path = os.path.join(config.system.cache_path, fname)
            save_audio(audio.getvalue(), path)

        if use_streaming:
            audio = tts_manager.generate_audio_chunks(audio)
            response = make_response(audio)
            response.headers['Content-Disposition'] = f'attachment; filename={fname}'
            response.headers['Content-Type'] = file_type
            return response
        else:
            return send_file(path_or_file=audio, mimetype=file_type, download_name=fname)


@voice_api.route('/ssml', methods=["POST"])
@require_api_key
def ssml_api():
    try:
        content_type = request.headers.get('Content-Type')
        if content_type == 'application/json':
            request_data = request.get_json()
        else:
            request_data = request.form

        ssml = request_data.get("ssml")
    except Exception as e:
        logger.info(f"[ssml] {e}")
        return make_response(jsonify({"status": "error", "message": f"parameter error"}), 400)

    logger.debug(ssml)
    voice_tasks, format = tts_manager.parse_ssml(ssml)
    fname = f"{str(uuid.uuid1())}.{format}"
    file_type = f"audio/{format}"

    t1 = time.time()
    audio = tts_manager.process_ssml_infer_task(voice_tasks, format)
    t2 = time.time()
    logger.info(f"[ssml] finish in {(t2 - t1):.2f}s")

    if config.system.cache_audio:
        logger.debug(f"[ssml] {fname}")
        path = os.path.join(config.system.cache_path, fname)
        save_audio(audio.getvalue(), path)

    return send_file(path_or_file=audio, mimetype=file_type, download_name=fname)


@voice_api.route('/dimension-emotion', methods=["POST"])
@require_api_key
def dimensional_emotion_api():
    if request.method == "POST":
        try:
            audio = request.files['upload']
        except Exception as e:
            logger.error(f"[dimensional_emotion] {e}")
            return make_response("parameter error", 400)

    content = BytesIO(audio.read())

    file_type = "application/octet-stream; charset=ascii"
    fname = os.path.splitext(audio.filename)[0] + ".npy"
    emotion_npy = tts_manager.get_dimensional_emotion_npy(content)
    return send_file(path_or_file=emotion_npy, mimetype=file_type, download_name=fname)


@voice_api.route('/bert-vits2', methods=["GET", "POST"])
@require_api_key
def voice_bert_vits2_api():
    try:
        if request.method == "GET":
            request_data = request.args
        elif request.method == "POST":
            content_type = request.headers.get('Content-Type')
            if content_type == 'application/json':
                request_data = request.get_json()
            else:
                request_data = request.form

        text = get_param(request_data, "text", "", str)
        id = get_param(request_data, "id", config.bert_vits2_config.id, int)
        format = get_param(request_data, "format", config.bert_vits2_config.format, str)
        lang = get_param(request_data, "lang", config.bert_vits2_config.lang, str).lower()
        length = get_param(request_data, "length", config.bert_vits2_config.length, float)
        # length_zh = get_param(request_data, "length_zh", config.bert_vits2_config.length_zh, float)
        # length_ja = get_param(request_data, "length_ja", config.bert_vits2_config.length_ja, float)
        # length_en = get_param(request_data, "length_en", config.bert_vits2_config.length_en, float)
        noise = get_param(request_data, "noise", config.bert_vits2_config.noise, float)
        noisew = get_param(request_data, "noisew", config.bert_vits2_config.noisew, float)
        sdp_ratio = get_param(request_data, "sdp_ratio", config.bert_vits2_config.sdp_ratio, float)
        segment_size = get_param(request_data, "segment_size", config.bert_vits2_config.segment_size, int)
        use_streaming = get_param(request_data, 'streaming', config.bert_vits2_config.use_streaming, bool)
        emotion = get_param(request_data, 'emotion', config.bert_vits2_config.emotion, int)
        reference_audio = request.files.get("reference_audio", None)
        text_prompt = get_param(request_data, 'text_prompt', config.bert_vits2_config.text_prompt, str)
        style_text = get_param(request_data, 'style_text', config.bert_vits2_config.style_text, str)
        style_weight = get_param(request_data, 'style_weight', config.bert_vits2_config.style_weight, float)
    except Exception as e:
        logger.error(f"[{ModelType.BERT_VITS2.value}] {e}")
        return make_response("parameter error", 400)

    # logger.info(
    #     f"[{ModelType.BERT_VITS2.value}] id:{id} format:{format} lang:{lang} length:{length} noise:{noise} noisew:{noisew} sdp_ratio:{sdp_ratio} segment_size:{segment_size}"
    #     f" length_zh:{length_zh} length_ja:{length_ja} length_en:{length_en}")

    logger.info(
        f"[{ModelType.BERT_VITS2.value}] id:{id} format:{format} lang:{lang} length:{length} noise:{noise} noisew:{noisew} sdp_ratio:{sdp_ratio} segment_size:{segment_size} streaming:{use_streaming}")
    logger.info(f"[{ModelType.BERT_VITS2.value}] len:{len(text)} text：{text}")
    if reference_audio:
        logger.info(f"[{ModelType.BERT_VITS2.value}] reference_audio:{reference_audio.filename}")
    elif emotion:
        logger.info(f"[{ModelType.BERT_VITS2.value}] emotion:{emotion}")
    elif text_prompt:
        logger.info(f"[{ModelType.BERT_VITS2.value}] text_prompt:{text_prompt}")
    elif style_text:
        logger.info(f"[{ModelType.BERT_VITS2.value}] style_text:{style_text} style_weight:{style_weight}")

    if check_is_none(text):
        logger.info(f"[{ModelType.BERT_VITS2.value}] text is empty")
        return make_response(jsonify({"status": "error", "message": "text is empty"}), 400)

    if check_is_none(id):
        logger.info(f"[{ModelType.BERT_VITS2.value}] speaker id is empty")
        return make_response(jsonify({"status": "error", "message": "speaker id is empty"}), 400)

    if id < 0 or id >= model_manager.bert_vits2_speakers_count:
        logger.info(f"[{ModelType.BERT_VITS2.value}] speaker id {id} does not exist")
        return make_response(jsonify({"status": "error", "message": f"id {id} does not exist"}), 400)

    if emotion and (emotion < 0 or emotion > 9):
        logger.info(f"[{ModelType.BERT_VITS2.value}] emotion {emotion} out of the range 0-9")
        return make_response(jsonify({"status": "error", "message": f"emotion {emotion} out of the range 0-9"}), 400)

    # 校验模型是否支持输入的语言
    speaker_lang = model_manager.voice_speakers[ModelType.BERT_VITS2.value][id].get('lang')
    if lang not in ["auto", "mix"] and len(speaker_lang) > 1 and lang not in speaker_lang:
        logger.info(f"[{ModelType.BERT_VITS2.value}] lang \"{lang}\" is not in {speaker_lang}")
        return make_response(jsonify({"status": "error", "message": f"lang '{lang}' is not in {speaker_lang}"}),
                             400)

    # 如果配置文件中设置了LANGUAGE_AUTOMATIC_DETECT则强制将speaker_lang设置为LANGUAGE_AUTOMATIC_DETECT
    if (lang_detect := config.language_identification.language_automatic_detect) and isinstance(lang_detect, list):
        speaker_lang = lang_detect

    if use_streaming and format.upper() != "MP3":
        format = "mp3"
        logger.warning("Streaming response only supports MP3 format.")

    fname = f"{str(uuid.uuid1())}.{format}"
    file_type = f"audio/{format}"
    state = {"text": text,
             "id": id,
             "format": format,
             "length": length,
             # "length_zh": length_zh,
             # "length_ja": length_ja,
             # "length_en": length_en,
             "noise": noise,
             "noisew": noisew,
             "sdp_ratio": sdp_ratio,
             "segment_size": segment_size,
             "lang": lang,
             "speaker_lang": speaker_lang,
             "emotion": emotion,
             "reference_audio": reference_audio,
             "text_prompt": text_prompt,
             "style_text": style_text,
             "style_weight": style_weight,
             }

    if use_streaming:
        # audio = tts_manager.stream_bert_vits2_infer(state)
        audio = tts_manager.stream_bert_vits2_infer(state)
        response = make_response(audio)
        response.headers['Content-Disposition'] = f'attachment; filename={fname}'
        response.headers['Content-Type'] = file_type
        return response
    else:
        t1 = time.time()
        # audio = tts_manager.bert_vits2_infer(state)
        audio = tts_manager.bert_vits2_infer(state)
        t2 = time.time()
        logger.info(f"[{ModelType.BERT_VITS2.value}] finish in {(t2 - t1):.2f}s")

    if config.system.cache_audio:
        logger.debug(f"[{ModelType.BERT_VITS2.value}] {fname}")
        path = os.path.join(config.system.cache_path, fname)
        save_audio(audio.getvalue(), path)

    return send_file(path_or_file=audio, mimetype=file_type, download_name=fname)


@voice_api.route('/gpt-sovits', methods=["GET", "POST"])
@require_api_key
def voice_gpt_sovits_api():
    try:
        if request.method == "GET":
            request_data = request.args
        elif request.method == "POST":
            content_type = request.headers.get('Content-Type')
            if content_type == 'application/json':
                request_data = request.get_json()
            else:
                request_data = request.form

        text = get_param(request_data, "text", "", str)
        id = get_param(request_data, "id", config.gpt_sovits_config.id, int)
        lang = get_param(request_data, "lang", config.gpt_sovits_config.lang, str)
        format = get_param(request_data, "format", config.gpt_sovits_config.format, str)
        segment_size = get_param(request_data, "segment_size", config.gpt_sovits_config.segment_size, int)
        reference_audio = request.files.get("reference_audio", None)
        preset = get_param(request_data, "preset", None, str)
        # refer_wav_path = get_param(request_data, "refer_wav_path",
        #                            config.gpt_sovits_config.presets.get("default").refer_wav_path, str)
        prompt_text = get_param(request_data, "prompt_text", None, str)
        prompt_lang = get_param(request_data, "prompt_lang", None, str)
        top_k = get_param(request_data, "top_k", config.gpt_sovits_config.top_k, int)
        top_p = get_param(request_data, "top_p", config.gpt_sovits_config.top_p, float)
        temperature = get_param(request_data, "temperature", config.gpt_sovits_config.temperature, float)
        use_streaming = get_param(request_data, 'streaming', config.gpt_sovits_config.use_streaming, bool)
        batch_size = get_param(request_data, 'batch_size', config.gpt_sovits_config.batch_size, int)
        speed_factor = get_param(request_data, 'speed', config.gpt_sovits_config.speed, float)
    except Exception as e:
        logger.error(f"[{ModelType.GPT_SOVITS.value}] {e}")
        return make_response("parameter error", 400)

    logger.info(
        f"[{ModelType.GPT_SOVITS.value}] id:{id} format:{format} lang:{lang} segment_size:{segment_size} top_k:{top_k} top_p:{top_p} temperature:{temperature} streaming:{use_streaming}")
    logger.info(
        f"[{ModelType.GPT_SOVITS.value}] batch_size:{batch_size} speed_factor:{speed_factor}")
    logger.info(f"[{ModelType.GPT_SOVITS.value}] len:{len(text)} text：{text}")

    if check_is_none(text):
        logger.info(f"[{ModelType.GPT_SOVITS.value}] text is empty")
        return make_response(jsonify({"status": "error", "message": "text is empty"}), 400)

    if check_is_none(id):
        logger.info(f"[{ModelType.GPT_SOVITS.value}] speaker id is empty")
        return make_response(jsonify({"status": "error", "message": "speaker id is empty"}), 400)

    if id < 0 or id >= model_manager.gpt_sovits_speakers_count:
        logger.info(f"[{ModelType.GPT_SOVITS.value}] speaker id {id} does not exist")
        return make_response(jsonify({"status": "error", "message": f"id {id} does not exist"}), 400)

    # 校验模型是否支持输入的语言
    speaker_lang = model_manager.voice_speakers[ModelType.GPT_SOVITS.value][id].get('lang')
    if lang not in ["auto", "mix"] and len(speaker_lang) > 1 and lang not in speaker_lang:
        logger.info(f"[{ModelType.GPT_SOVITS.value}] lang \"{lang}\" is not in {speaker_lang}")
        return make_response(jsonify({"status": "error", "message": f"lang '{lang}' is not in {speaker_lang}"}),
                             400)

    # 如果配置文件中设置了LANGUAGE_AUTOMATIC_DETECT则强制将speaker_lang设置为LANGUAGE_AUTOMATIC_DETECT
    if (lang_detect := config.language_identification.language_automatic_detect) and isinstance(lang_detect, list):
        speaker_lang = lang_detect

    logger.info(
        f"[{ModelType.GPT_SOVITS.value}] prompt_text:{prompt_text} prompt_lang:{prompt_lang} ")

    if use_streaming and format.upper() != "MP3":
        format = "mp3"
        logger.warning("Streaming response only supports MP3 format.")

    fname = f"{str(uuid.uuid1())}.{format}"
    file_type = f"audio/{format}"
    state = {"text": text,
             "id": id,
             "format": format,
             "segment_size": segment_size,
             "lang": lang,
             "speaker_lang": speaker_lang,
             "reference_audio": reference_audio,
             # "reference_audio_sr": reference_audio_sr,
             "prompt_text": prompt_text,
             "prompt_lang": prompt_lang,
             "top_k": top_k,
             "top_p": top_p,
             "temperature": temperature,
             "preset": preset,
             "batch_size": batch_size,
             "speed_factor": speed_factor
             }

    if use_streaming:
        audio = tts_manager.stream_gpt_sovits_infer(state)
        response = make_response(audio)
        response.headers['Content-Disposition'] = f'attachment; filename={fname}'
        response.headers['Content-Type'] = file_type
        return response
    else:
        t1 = time.time()
        audio = tts_manager.gpt_sovits_infer(state)
        t2 = time.time()
        logger.info(f"[{ModelType.GPT_SOVITS.value}] finish in {(t2 - t1):.2f}s")

    if config.system.cache_audio:
        logger.debug(f"[{ModelType.GPT_SOVITS.value}] {fname}")
        path = os.path.join(config.system.cache_path, fname)
        save_audio(audio.getvalue(), path)

    return send_file(path_or_file=audio, mimetype=file_type, download_name=fname)


@voice_api.route('/reading', methods=["GET", "POST"])
@require_api_key
def voice_reading_api():
    in_state = {}  # interlocutor
    nr_state = {}  # narrator
    state = {}
    try:
        if request.method == "GET":
            request_data = request.args
        elif request.method == "POST":
            content_type = request.headers.get('Content-Type')
            if content_type == 'application/json':
                request_data = request.get_json()
            else:
                request_data = request.form

        in_state["model_type"] = ModelType(
            get_param(request_data, "in_model_type", config.reading_config.interlocutor.model_type, str))
        in_state["id"] = get_param(request_data, "in_id", config.reading_config.interlocutor.id, int)
        in_state["preset"] = get_param(request_data, "in_preset", config.reading_config.interlocutor.preset, str)

        # narrator
        nr_state["model_type"] = ModelType(
            get_param(request_data, "nr_model_type", config.reading_config.narrator.model_type, str))
        nr_state["id"] = get_param(request_data, "nr_id", config.reading_config.narrator.model_type, int)
        nr_state["preset"] = get_param(request_data, "nr_preset", config.reading_config.narrator.preset, str)

        state["text"] = get_param(request_data, "text", "", str)
        state["lang"] = get_param(request_data, "lang", "auto", str)
        state["format"] = get_param(request_data, "format", "wav", str)

    except Exception as e:
        logger.error(f"[Reading] {e}")
        return make_response("parameter error", 400)

    in_state.update(state)
    nr_state.update(state)

    in_state = update_default_params(in_state)
    nr_state = update_default_params(nr_state)

    file_type = f'audio/{state["format"]}'
    fname = f"{str(uuid.uuid1())}.{state['format']}"

    t1 = time.time()
    audio = tts_manager.reading(in_state=in_state, nr_state=nr_state)
    t2 = time.time()
    logger.info(f"[Reading] finish in {(t2 - t1):.2f}s")

    return send_file(path_or_file=audio, mimetype=file_type, download_name=fname)


@voice_api.route('/check', methods=["GET", "POST"])
def check():
    try:
        if request.method == "GET":
            request_data = request.args
        elif request.method == "POST":
            content_type = request.headers.get('Content-Type')
            if content_type == 'application/json':
                request_data = request.get_json()
            else:
                request_data = request.form

        model_type_str = request_data.get("model_type", request_data.get("model")).upper()
        id = int(request_data.get("id"))
    except Exception as e:
        logger.info(f"[check] {e}")
        return make_response(jsonify({"status": "error", "message": "parameter error"}), 400)

    if check_is_none(model_type_str):
        logger.info(f"[check] model {model_type_str} is empty")
        return make_response(jsonify({"status": "error", "message": "model is empty"}), 400)

    if model_type_str not in ModelType._value2member_map_:
        res = make_response(jsonify({"status": "error", "message": f"model {model_type_str} does not exist"}))
        res.status = 404
        logger.info(f"[check] speaker id {id} error")
        return res

    if check_is_none(id):
        logger.info(f"[check] speaker id is empty")
        return make_response(jsonify({"status": "error", "message": "speaker id is empty"}), 400)

    model_type = ModelType(model_type_str)
    speaker_list = model_manager.voice_speakers[model_type.value]

    if len(speaker_list) == 0:
        logger.info(f"[check] {model_type_str} not loaded")
        return make_response(jsonify({"status": "error", "message": f"{model_type_str} not loaded"}), 400)

    if id < 0 or id >= len(speaker_list):
        logger.info(f"[check] speaker id {id} does not exist")
        return make_response(jsonify({"status": "error", "message": f"id {id} does not exist"}), 400)
    name = str(speaker_list[id]["name"])
    lang = speaker_list[id]["lang"]
    logger.info(f"[check] check id:{id} name:{name} lang:{lang}")

    return make_response(jsonify({"status": "success", "id": id, "name": name, "lang": lang}), 200)