military-tech-voice3.0/backend/app/routes/tts.py

from flask import Blueprint, jsonify, request, send_file
import requests, os, uuid
import logging
import dashscope
from dashscope.audio.tts_v2 import SpeechSynthesizer

# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# 从环境变量读取配置（如果不存在则使用默认值）
MINIMAX_API_KEY = os.environ.get("MINIMAX_API_KEY", "sk-api-x3IxJC9nlORkopOJ4NOWff0Er4rtDjVrM08O5HOCYnsmBaZ2k1n2HIBLZGAwIDOSTh6bsmC2Nmz_xfyNnEWfhjLTrQPRLLN64Hu89ed0vG6gzwFfE24gfQM")
MINIMAX_VOICE_ID = os.environ.get("MINIMAX_VOICE_ID", "lanhao_military_tech")
COSYVOICE_API_KEY = os.environ.get("COSYVOICE_API_KEY", "sk-3c386742861b4d2da421030fa6254ef6")
COSYVOICE_VOICE_ID = os.environ.get("COSYVOICE_VOICE_ID", "cosyvoice-v3.5-flash-bailian-e859a6e87823419f9c6b3eefb1a97dc3")

# 配置DashScope WebSocket
dashscope.api_key = COSYVOICE_API_KEY
dashscope.base_websocket_api_url = 'wss://dashscope.aliyuncs.com/api-ws/v1/inference'

logger.info(f"Loaded config - MINIMAX_VOICE_ID: {MINIMAX_VOICE_ID}, COSYVOICE_VOICE_ID: {COSYVOICE_VOICE_ID}")

tts_bp = Blueprint("tts", __name__)

@tts_bp.route('/synthesize', methods=['POST'])
def synthesize():
    d = request.get_json()
    t = d.get('text', '')
    m = d.get('model', 'minimax')
    if m == 'minimax': return synthesize_minimax(t)
    elif m == 'cosyvoice': return synthesize_cosyvoice(t)
    return jsonify({'error': '不支持的模型'}), 400

def synthesize_minimax(text):
    if not text:
        return jsonify({"success": False, "error": "文本不能为空"}), 400
    
    url = "https://api.minimaxi.com/v1/t2a_v2"
    headers = {
        "Authorization": f"Bearer {MINIMAX_API_KEY}", 
        "Content-Type": "application/json"
    }
    payload = {
        "model": "speech-2.8-hd",
        "text": text,
        "voice_setting": {"voice_id": MINIMAX_VOICE_ID},
        "audio_setting": {
            "sample_rate": 32000,
            "format": "mp3",
            "bitrate": 128000,
            "speed": 1.0,
            "volume": 1.0,
            "pitch": 0
        }
    }
    
    logger.info(f"Calling MiniMax API with voice_id: {MINIMAX_VOICE_ID}")
    
    try:
        resp = requests.post(url, headers=headers, json=payload, timeout=60)
        result = resp.json()
        logger.info(f"MiniMax response: {result}")
        
        if result.get("base_resp", {}).get("status_code") == 0:
            audio_hex = result.get("data", {}).get("audio", "")
            if audio_hex:
                filename = f"minimax_{uuid.uuid4().hex[:8]}.mp3"
                filepath = f"/workspace/military_tech_voice/audio_uploads/{filename}"
                with open(filepath, "wb") as f: f.write(bytes.fromhex(audio_hex))
                logger.info(f"Audio saved to: {filepath}")
                return jsonify({"success": True, "audio_url": f"/api/tts/audio/{filename}"})
        
        error_msg = result.get("base_resp", {}).get("status_msg", str(result))
        return jsonify({"success": False, "error": error_msg})
    except Exception as e:
        logger.error(f"MiniMax API error: {e}")
        return jsonify({"success": False, "error": str(e)}), 500

def synthesize_cosyvoice(text):
    if not text:
        return jsonify({"success": False, "error": "文本不能为空"}), 400
    
    logger.info(f"Calling CosyVoice WebSocket API with voice_id: {COSYVOICE_VOICE_ID}")
    
    try:
        # 使用DashScope SDK的WebSocket方式
        synthesizer = SpeechSynthesizer(
            model="cosyvoice-v3-flash",
            voice=COSYVOICE_VOICE_ID
        )
        
        # 调用WebSocket API进行语音合成
        audio = synthesizer.call(text)
        
        if audio:
            filename = f"cosyvoice_{uuid.uuid4().hex[:8]}.mp3"
            filepath = f"/workspace/military_tech_voice/audio_uploads/{filename}"
            with open(filepath, "wb") as f: 
                f.write(audio)
            logger.info(f"CosyVoice audio saved to: {filepath}")
            return jsonify({"success": True, "audio_url": f"/api/tts/audio/{filename}"})
        
        return jsonify({"success": False, "error": "未生成音频"})
    except Exception as e:
        logger.error(f"CosyVoice WebSocket API error: {e}")
        return jsonify({"success": False, "error": str(e)}), 500

@tts_bp.route('/audio/<filename>')
def get_audio(filename):
    filepath = f"/workspace/military_tech_voice/audio_uploads/{filename}"
    if os.path.exists(filepath): return send_file(filepath, mimetype="audio/mpeg")
    return jsonify({"error": "文件不存在"}), 404