dsProject/dsLightRag/Routes/ttsRoute.py

import logging
import uuid
from typing import Optional

from fastapi import APIRouter, HTTPException, Query  # 添加Query导入
from pydantic import BaseModel

from Util.GengerateAudio import ByteDanceTTS

# 创建声音生成路由
router = APIRouter(prefix="/api/tts", tags=["声音生成"])

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

# 初始化TTS实例
tts_instance = ByteDanceTTS()

class TextToSpeechRequest(BaseModel):
    """文本转语音请求参数"""
    text: str
    voice_type: Optional[str] = None
    speed_ratio: Optional[float] = 1.0
    volume_ratio: Optional[float] = 1.0
    pitch_ratio: Optional[float] = 1.0
    encoding: Optional[str] = "mp3"


class TextToSpeechResponse(BaseModel):
    """文本转语音响应"""
    success: bool
    message: str
    audio_url: Optional[str] = None
    audio_size: Optional[float] = None
    audio_format: Optional[str] = None
    request_id: Optional[str] = None


@router.get("/voice-categories")
async def get_voice_categories():
    try:
        categories = tts_instance.get_all_categories()
        return {
            "success": True,
            "data": categories,  # 恢复为原始的 data 字段
            "message": "获取音色分类成功"
        }
    except Exception as e:
        logger.error(f"获取音色分类错误: {e}")
        raise HTTPException(
            status_code=500,
            detail=f"获取音色分类失败: {str(e)}"
        )

# 恢复原始的音色列表接口路由
@router.get("/voices")
async def get_voices_by_category(category: str = Query(...)):  # 现在Query已定义
    try:
        voices = tts_instance.get_voices_by_category(category)
        if not voices:
            return {
                "success": False,
                "message": f"未找到分类 '{category}' 下的音色"
            }
        
        return {
            "success": True,
            "data": voices,
            "message": f"获取分类 '{category}' 下的音色成功"
        }
    except Exception as e:
        logger.error(f"获取分类 '{category}' 下的音色错误: {e}")
        raise HTTPException(
            status_code=500,
            detail=f"获取分类 '{category}' 下的音色失败: {str(e)}"
        )


@router.get("/all")
async def get_all_voices():
    """
    获取所有音色分类和音色列表接口
    返回所有音色分类和每个分类下的音色列表
    """
    try:
        all_voices = tts_instance.get_all_voices()
        return {
            "success": True,
            "data": all_voices,
            "message": "获取所有音色分类和列表成功"
        }
    except Exception as e:
        logger.error(f"获取所有音色分类和列表错误: {e}")
        raise HTTPException(
            status_code=500,
            detail=f"获取所有音色分类和列表失败: {str(e)}"
        )


@router.post("/generate", response_model=TextToSpeechResponse)
async def generate_audio(request: TextToSpeechRequest):
    """
    文本转语音接口
    根据输入文本和语音参数生成音频文件
    """
    try:
        # 生成唯一文件名
        import os
        from datetime import datetime
        
        # 确保输出目录存在
        output_dir = "static/audio"
        os.makedirs(output_dir, exist_ok=True)
        
        # 生成唯一文件名
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        unique_id = str(uuid.uuid4())[:8]
        filename = f"tts_{timestamp}_{unique_id}.{request.encoding}"
        output_path = os.path.join(output_dir, filename)
        
        # 调用TTS工具生成音频
        audio_data = tts_instance.generate_audio(
            text=request.text,
            voice_type=request.voice_type,
            speed_ratio=request.speed_ratio,
            volume_ratio=request.volume_ratio,
            pitch_ratio=request.pitch_ratio,
            encoding=request.encoding,
            output_path=output_path
        )
        
        if audio_data:
            # 构建可访问的URL
            audio_url = f"/{output_path}"
            
            return TextToSpeechResponse(
                success=True,
                message="音频生成成功",
                audio_url=audio_url,
                audio_format=request.encoding,
                request_id=str(uuid.uuid4())
            )
        else:
            return TextToSpeechResponse(
                success=False,
                message="音频生成失败"
            )
    except Exception as e:
        logger.error(f"文本转语音接口错误: {e}")
        raise HTTPException(
            status_code=500,
            detail=f"音频生成失败: {str(e)}"
        )