蜜桃av色欲a片精品一区,麻豆aⅴ精品无码一区二区,亚洲人成网站在线播放影院在线,亚洲 素人 字幕 在线 最新

  Cosyvoice-api 代碼分析

發(fā)布日期：2025/5/11 23:47:20      瀏覽量：

        該代碼用于 CosyVoice2 的 api 文件，部署好 cosyVoice 項目后，將該 api.py 文件同 webui.py放在一起，然后執(zhí)行 python api.py。

如果是三方整合包，將 api.py 同 bat 腳本放在一起，然后查找其中python.exe所在的位置，在bat所在當前文件夾地址欄中輸入cmd回車，然后執(zhí)行 目錄/python.exe api.py

如果執(zhí)行時提示module flask not found,請執(zhí)行 python.exe -m pip install flask 安裝

根據(jù)內(nèi)置角色合成文字

• 接口地址: /tts

• 單純將文字合成語音，不進行音色克隆

• 必須設置的參數(shù)：

text:需要合成語音的文字

role: ’中文女’, ’中文男’, ’日語男’, ’粵語女’, ’英文女’, ’英文男’, ’韓語女’ 選擇一個

• 成功返回:wav音頻數(shù)據(jù)

• 示例代碼

data={
    "text":"你好啊親愛的朋友們",
    "reference_audio":"10.wav"
}

response=requests.post(f’http://127.0.0.1:9933/tts’,data=data,timeout=3600)

同語言克隆音色合成

• 地址：/clone_eq

參考音頻發(fā)音語言和需要合成的文字語言一致，例如參考音頻是中文發(fā)音，同時需要根據(jù)該音頻將中文文本合成為語音

• 必須設置參數(shù):

text： 需要合成語音的文字

reference_audio：需要克隆音色的參考音頻

reference_text：參考音頻對應的文字內(nèi)容 參考音頻相對于 api.py 的路徑，例如引用1.wav，該文件和api.py在同一文件夾內(nèi)，則填寫 1.wav

• 成功返回:wav數(shù)據(jù)

• 示例代碼

data={
    "text":"你好啊親愛的朋友們。",
    "reference_audio":"10.wav",
    "reference_text":"希望你過的比我更好喲。"
}

response=requests.post(f’http://127.0.0.1:9933/tts’,data=data,timeout=3600)

不同語言音色克隆:

• 地址： /cone

參考音頻發(fā)音語言和需要合成的文字語言不一致，例如需要根據(jù)中文發(fā)音的參考音頻，將一段英文文本合成為語音。

• 必須設置參數(shù):

text： 需要合成語音的文字

reference_audio：需要克隆音色的參考音頻 參考音頻相對于 api.py 的路徑，例如引用1.wav，該文件和api.py在同一文件夾內(nèi)，則填寫 1.wav

• 成功返回:wav數(shù)據(jù)

• 示例代碼

data={
    "text":"親友からの誕生日プレゼントを遠くから受け取り、思いがけないサプライズと深い祝福に、私の心は甘い喜びで満たされた！。",
    "reference_audio":"10.wav"
}

response=requests.post(f’http://127.0.0.1:9933/tts’,data=data,timeout=3600)

兼容openai tts

• 接口地址 /v1/audio/speech
• 請求方法 POST
• 請求類型 Content-Type: application/json
• 請求參數(shù) input: 要合成的文字 model: 固定 tts-1, 兼容openai參數(shù)，實際未使用 speed: 語速，默認1.0 reponse_format：返回格式，固定wav音頻數(shù)據(jù) voice: 僅用于文字合成時，取其一 ’中文女’, ’中文男’, ’日語男’, ’粵語女’, ’英文女’, ’英文男’, ’韓語女’

用于克隆時，填寫引用的參考音頻相對于 api.py 的路徑，例如引用1.wav，該文件和api.py在同一文件夾內(nèi)，則填寫 1.wav

• 示例代碼

from openai import OpenAI

client = OpenAI(api_key=’12314’, base_url=’http://127.0.0.1:9933/v1’)
with  client.audio.speech.with_streaming_response.create(
                    model=’tts-1’,
                    voice=’中文女’,
                    input=’你好啊，親愛的朋友們’,
                    speed=1.0                    
                ) as response:
    with open(’./test.wav’, ’wb’) as f:
       for chunk in response.iter_bytes():
            f.write(chunk)

	

####################################################

import os,time,sys
from pathlib import Path
root_dir=Path(__file__).parent.as_posix()

# ffmpeg
if sys.platform == ’win32’:
    os.environ[’PATH’] = root_dir + f’;{root_dir}\\ffmpeg;’ + os.environ[’PATH’]+f’;{root_dir}/third_party/Matcha-TTS’
else:
    os.environ[’PATH’] = root_dir + f’:{root_dir}/ffmpeg:’ + os.environ[’PATH’]
    os.environ[’PYTHONPATH’] = os.environ.get(’PYTHONPATH’, ’’) + ’:third_party/Matcha-TTS’
sys.path.append(f’{root_dir}/third_party/Matcha-TTS’)
tmp_dir=Path(f’{root_dir}/tmp’).as_posix()
logs_dir=Path(f’{root_dir}/logs’).as_posix()
os.makedirs(tmp_dir,exist_ok=True)
os.makedirs(logs_dir,exist_ok=True)

from flask import Flask, request, render_template, jsonify,  send_from_directory,send_file,Response, stream_with_context,make_response,send_file
import logging
from logging.handlers import RotatingFileHandler
import subprocess
import shutil
import datetime
from cosyvoice.cli.cosyvoice import CosyVoice, CosyVoice2
from cosyvoice.utils.file_utils import load_wav


import torchaudio,torch
from pathlib import Path
import base64


# 下載模型
from modelscope import snapshot_download
snapshot_download(’iic/CosyVoice2-0.5B’, local_dir=’pretrained_models/CosyVoice2-0.5B’)
snapshot_download(’iic/CosyVoice-300M-SFT’, local_dir=’pretrained_models/CosyVoice-300M-SFT’)



’’’
app logs
’’’
# 配置日志
# 禁用 Werkzeug 默認的日志處理器
log = logging.getLogger(’werkzeug’)
log.handlers[:] = []
log.setLevel(logging.WARNING)

root_log = logging.getLogger()  # Flask的根日志記錄器
root_log.handlers = []
root_log.setLevel(logging.WARNING)

app = Flask(__name__, 
    static_folder=root_dir+’/tmp’, 
    static_url_path=’/tmp’)

app.logger.setLevel(logging.WARNING) 
# 創(chuàng)建 RotatingFileHandler 對象，設置寫入的文件路徑和大小限制
file_handler = RotatingFileHandler(logs_dir+f’/{datetime.datetime.now().strftime("%Y%m%d")}.log’, maxBytes=1024 * 1024, backupCount=5)
# 創(chuàng)建日志的格式
formatter = logging.Formatter(’%(asctime)s - %(name)s - %(levelname)s - %(message)s’)
# 設置文件處理器的級別和格式
file_handler.setLevel(logging.WARNING)
file_handler.setFormatter(formatter)
# 將文件處理器添加到日志記錄器中
app.logger.addHandler(file_handler)



sft_model = None
tts_model = None 

VOICE_LIST=[’中文女’, ’中文男’, ’日語男’, ’粵語女’, ’英文女’, ’英文男’, ’韓語女’]





def base64_to_wav(encoded_str, output_path):
    if not encoded_str:
        raise ValueError("Base64 encoded string is empty.")

    # 將base64編碼的字符串解碼為字節(jié)
    wav_bytes = base64.b64decode(encoded_str)

    # 檢查輸出路徑是否存在，如果不存在則創(chuàng)建
    Path(output_path).parent.mkdir(parents=True, exist_ok=True)

    # 將解碼后的字節(jié)寫入文件
    with open(output_path, "wb") as wav_file:
        wav_file.write(wav_bytes)

    print(f"WAV file has been saved to {output_path}")


# 獲取請求參數(shù)
def get_params(req):
    params={
        "text":"",
        "lang":"",
        "role":"中文女",
        "reference_audio":None,
        "reference_text":"",
        "speed":1.0
    }
    # 原始字符串
    params[’text’] = req.args.get("text","").strip() or req.form.get("text","").strip()
    
    # 字符串語言代碼
    params[’lang’] = req.args.get("lang","").strip().lower() or req.form.get("lang","").strip().lower()
    # 兼容 ja語言代碼
    if params[’lang’]==’ja’:
        params[’lang’]=’jp’
    elif params[’lang’][:2] == ’zh’:
        # 兼容 zh-cn zh-tw zh-hk
        params[’lang’]=’zh’
    
    # 角色名 
    role = req.args.get("role","").strip() or req.form.get("role",’’)
    if role:
        params[’role’]=role
    
    # 要克隆的音色文件    
    params[’reference_audio’] = req.args.get("reference_audio",None) or req.form.get("reference_audio",None)
    encode=req.args.get(’encode’,’’) or req.form.get(’encode’,’’)
    if  encode==’base64’:
        tmp_name=f’tmp/{time.time()}-clone-{len(params["reference_audio"])}.wav’
        base64_to_wav(params[’reference_audio’],root_dir+’/’+tmp_name)
        params[’reference_audio’]=tmp_name
    # 音色文件對應文本
    params[’reference_text’] = req.args.get("reference_text",’’).strip() or req.form.get("reference_text",’’)
    
    return params


def del_tmp_files(tmp_files: list):
    print(’正在刪除緩存文件...’)
    for f in tmp_files:
        if os.path.exists(f):
            print(’刪除緩存文件:’, f)
            os.remove(f)


# 實際批量合成完畢后連接為一個文件
def batch(tts_type,outname,params):
    global sft_model,tts_model
    if not shutil.which("ffmpeg"):
        raise Exception(’必須安裝 ffmpeg’)    
    prompt_speech_16k=None
    if tts_type!=’tts’:
        if not params[’reference_audio’] or not os.path.exists(f"{root_dir}/{params[’reference_audio’]}"):
            raise Exception(f’參考音頻未傳入或不存在 {params["reference_audio"]}’)
        ref_audio=f"{tmp_dir}/-refaudio-{time.time()}.wav" 
        try:
            subprocess.run(["ffmpeg","-hide_banner", "-ignore_unknown","-y","-i",params[’reference_audio’],"-ar","16000",ref_audio],
                   stdout=subprocess.PIPE,
                   stderr=subprocess.PIPE,
                   encoding="utf-8",
                   check=True,
                   text=True,
                   creationflags=0 if sys.platform != ’win32’ else subprocess.CREATE_NO_WINDOW)
        except Exception as e:
            raise Exception(f’處理參考音頻失敗:{e}’)
        
        prompt_speech_16k = load_wav(ref_audio, 16000)

    text=params[’text’]
    audio_list=[]
    if tts_type==’tts’:
        if sft_model is None:
            sft_model = CosyVoice(’pretrained_models/CosyVoice-300M-SFT’, load_jit=True, load_onnx=False)

        # 僅文字合成語音
        for i, j in enumerate(sft_model.inference_sft(text, params[’role’],stream=False,speed=params[’speed’])):
            audio_list.append(j[’tts_speech’])
            
    elif tts_type==’clone_eq’ and params.get(’reference_text’):
        if tts_model is None:
            tts_model=CosyVoice2(’pretrained_models/CosyVoice2-0.5B’, load_jit=True, load_onnx=False, load_trt=False)

        for i, j in enumerate(tts_model.inference_zero_shot(text,params.get(’reference_text’),prompt_speech_16k, stream=False,speed=params[’speed’])):
            audio_list.append(j[’tts_speech’])

    else:
        if tts_model is None:
            tts_model=CosyVoice2(’pretrained_models/CosyVoice2-0.5B’, load_jit=True, load_onnx=False, load_trt=False)

        for i, j in enumerate(tts_model.inference_cross_lingual(text,prompt_speech_16k, stream=False,speed=params[’speed’])):
            audio_list.append(j[’tts_speech’])
    audio_data = torch.concat(audio_list, dim=1)
    
    # 根據(jù)模型yaml配置設置采樣率
    if tts_type==’tts’:
        torchaudio.save(tmp_dir + ’/’ + outname,audio_data, 22050, format="wav")   
    elif tts_type==’clone_eq’:
        torchaudio.save(tmp_dir + ’/’ + outname,audio_data, 24000, format="wav")   
    else:
        torchaudio.save(tmp_dir + ’/’ + outname,audio_data, 24000, format="wav")    
    
    print(f"音頻文件生成成功：{tmp_dir}/{outname}")
    return tmp_dir + ’/’ + outname


# 單純文字合成語音
@app.route(’/tts’, methods=[’GET’, ’POST’])        
def tts():
    params=get_params(request)
    if not params[’text’]:
        return make_response(jsonify({"code":1,"msg":’缺少待合成的文本’}), 500)  # 設置狀態(tài)碼為500
        
    try:
        # 僅文字合成語音
        outname=f"tts-{datetime.datetime.now().strftime(’%Y%m%d-%H%M%S-’)}.wav"
        outname=batch(tts_type=’tts’,outname=outname,params=params)
    except Exception as e:
        print(e)
        return make_response(jsonify({"code":2,"msg":str(e)}), 500)  # 設置狀態(tài)碼為500
    else:
        return send_file(outname, mimetype=’audio/x-wav’)
    


# 跨語言文字合成語音      
@app.route(’/clone_mul’, methods=[’GET’, ’POST’])        
@app.route(’/clone’, methods=[’GET’, ’POST’])        
def clone():

    try:
        params=get_params(request)
        if not params[’text’]:
            return make_response(jsonify({"code":6,"msg":’缺少待合成的文本’}), 500)  # 設置狀態(tài)碼為500
            
        outname=f"clone-{datetime.datetime.now().strftime(’%Y%m%d-%H%M%S-’)}.wav"
        outname=batch(tts_type=’clone’,outname=outname,params=params)
    except Exception as e:
        return make_response(jsonify({"code":8,"msg":str(e)}), 500)  # 設置狀態(tài)碼為500
    else:
        return send_file(outname, mimetype=’audio/x-wav’)
@app.route(’/clone_eq’, methods=[’GET’, ’POST’])         
def clone_eq():

    try:
        params=get_params(request)
        if not params[’text’]:
            return make_response(jsonify({"code":6,"msg":’缺少待合成的文本’}), 500)  # 設置狀態(tài)碼為500
        if not params[’reference_text’]:
            return make_response(jsonify({"code":6,"msg":’同語言克隆必須傳遞引用文本’}), 500)  # 設置狀態(tài)碼為500
            
        outname=f"clone-{datetime.datetime.now().strftime(’%Y%m%d-%H%M%S-’)}.wav"
        outname=batch(tts_type=’clone_eq’,outname=outname,params=params)
    except Exception as e:
        return make_response(jsonify({"code":8,"msg":str(e)}), 500)  # 設置狀態(tài)碼為500
    else:
        return send_file(outname, mimetype=’audio/x-wav’)
     

@app.route(’/v1/audio/speech’, methods=[’POST’])
def audio_speech():
    """
    兼容 OpenAI /v1/audio/speech API 的接口
    """
    import random

    if not request.is_json:
        return jsonify({"error": "請求必須是 JSON 格式"}), 400

    data = request.get_json()

    # 檢查請求中是否包含必要的參數(shù)
    if ’input’ not in data or ’voice’ not in data:
        return jsonify({"error": "請求缺少必要的參數(shù)： input, voice"}), 400
    

    text = data.get(’input’)
    speed =  float(data.get(’speed’,1.0))
    
    voice = data.get(’voice’,’中文女’)
    params = {}
    params[’text’]=text
    params[’speed’]=speed
    api_name=’tts’
    if voice in VOICE_LIST:
        params[’role’]=voice
    elif Path(voice).exists() or Path(f’{root_dir}/{voice}’).exists():
        api_name=’clone’
        params[’reference_audio’]=voice
    else:
        return jsonify({"error": {"message": f"必須填寫配音角色名或參考音頻路徑", "type": e.__class__.__name__, "param": f’speed={speed},voice={voice},input={text}’, "code": 400}}), 500

    
    filename=f’openai-{len(text)}-{speed}-{time.time()}-{random.randint(1000,99999)}.wav’
    try:
        outname=batch(tts_type=api_name,outname=filename,params=params)
        return send_file(outname, mimetype=’audio/x-wav’)
    except Exception as e:
        return jsonify({"error": {"message": f"{e}", "type": e.__class__.__name__, "param": f’speed={speed},voice={voice},input={text}’, "code": 400}}), 500
         
if __name__==’__main__’:
    host=’127.0.0.1’
    port=50000
    print(f’\n啟動api: http://{host}:{port}\n’)
    try:
        from waitress import serve
    except Exception:
        app.run(host=host, port=port)
    else:
        serve(app,host=host, port=port)