dsProject/dsLightRag/Routes/ttsRoute.py

import logging

from fastapi import APIRouter, HTTPException, BackgroundTasks, Query
from pydantic import BaseModel
from typing import Optional
import os
import uuid
import time
from Util.ObsUtil import ObsUploader
from Config.Config import OBS_BUCKET, OBS_SERVER
# 导入TTS生成类
from Util.GengerateAudio import ByteDanceTTS
from fastapi.responses import StreamingResponse
import requests

# 配置日志
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/api/tts", tags=["文生音频"])
# 初始化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"


@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")
async def generate_audio(request: TextToSpeechRequest, background_tasks: BackgroundTasks):
    try:
        # 1. 生成唯一文件名和路径
        timestamp = int(time.time())
        filename = f"tts_{uuid.uuid4().hex[:8]}_{timestamp}.mp3"
        output_dir = "static/audio"
        os.makedirs(output_dir, exist_ok=True)
        output_path = os.path.join(output_dir, filename)

        # 2. 使用ByteDanceTTS生成音频文件
        # ------------------- TTS生成逻辑开始 -------------------
        audio_data = tts_instance.generate_audio(
            text=request.text,
            output_path=output_path,
            voice_type=request.voice_type,
            speed_ratio=request.speed_ratio,
            volume_ratio=request.volume_ratio,
            pitch_ratio=request.pitch_ratio,
            encoding=request.encoding
        )

        if not audio_data:
            raise HTTPException(status_code=500, detail="TTS音频生成失败")
        # ------------------- TTS生成逻辑结束 -------------------

        # 3. 上传到OBS
        obs_uploader = ObsUploader()
        obs_object_key = f"HuangHai/tts/{filename}"
        upload_success, upload_result = obs_uploader.upload_file(obs_object_key, output_path)

        if not upload_success:
            # 上传失败，清理文件并抛出异常
            background_tasks.add_task(os.remove, output_path) if os.path.exists(output_path) else None
            raise HTTPException(status_code=500, detail=f"OBS上传失败: {upload_result.get('errorMessage', '未知错误')}")

        # 4. 构造OBS访问URL
        obs_url = f"https://{OBS_BUCKET}.{OBS_SERVER}/{obs_object_key}"

        # 5. 安排后台任务删除本地文件
        background_tasks.add_task(os.remove, output_path) if os.path.exists(output_path) else None

        # 6. 返回OBS URL
        return {
            "status": "success",
            "message": "音频生成并上传成功",
            "audio_url": obs_url,  # 确保此处没有添加任何引号或反引号
            "filename": filename,
            "audio_size": len(audio_data) / 1024  # 音频大小(KB)
        }

    except Exception as e:
        # 捕获所有异常并清理文件
        if 'output_path' in locals() and os.path.exists(output_path):
            background_tasks.add_task(os.remove, output_path)
        raise HTTPException(status_code=500, detail=f"处理失败: {str(e)}")


@router.get("/download/{filename}")
async def download_audio(filename: str):
    # 构建OBS文件URL
    obs_url = f"https://{OBS_BUCKET}.{OBS_SERVER}/HuangHai/tts/{filename}"
    
    try:
        # 流式获取OBS文件
        response = requests.get(obs_url, stream=True)
        if response.status_code != 200:
            raise HTTPException(status_code=404, detail="音频文件不存在")
        
        # 设置下载响应头
        headers = {
            "Content-Disposition": f"attachment; filename={filename}",
            "Content-Type": "audio/mpeg"
        }
        
        # 返回流式响应
        return StreamingResponse(response.iter_content(chunk_size=8192), headers=headers)
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"下载失败: {str(e)}")