xiaoxia-saas/apps/api/app/api/routes/tts.py

"""TTS 合成 API 路由。"""

from __future__ import annotations

import json
import logging
import subprocess
import tempfile
from pathlib import Path
from typing import Any, Optional

from app.auth import AuthenticatedUser, get_current_user
from app.core.celery_app import celery_app
from app.core.storage import get_storage_service
from app.dependencies import (
    get_asset_library_repository,
    get_asset_repository,
    get_audio_url_signer,
    get_cosyvoice_service,
    get_db_session,
    get_project_repository,
    get_voice_clone_profile_repository,
)
from app.schemas.tts import (
    ListTTSJobResponse,
    SaveToLibraryRequest,
    SaveToLibraryResponse,
    TTSJobResponse,
    TTSPreviewRequest,
    TTSPreviewResponse,
    TTSStatusResponse,
    TTSSynthesizeRequest,
    TTSSynthesizeResponse,
)
from fastapi import APIRouter, Depends, HTTPException, Query, Response, WebSocket, WebSocketDisconnect, status
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session

from packages.adapters.sqlalchemy_impl.tts_job_repository import (
    SQLAlchemyTTSJobRepository,
)
from packages.application.cosyvoice_service import CosyVoiceError, CosyVoiceService
from packages.application.tts_job.streaming_service import TTSStreamingService
from packages.application.tts_job.use_cases import (
    CreateTTSJobUseCase,
    DeleteTTSJobUseCase,
    GetTTSJobStatusUseCase,
    GetTTSJobUseCase,
    ListTTSJobsUseCase,
    TTSJobNotFoundError,
)
from packages.application.tts_job.workflow import TTSWorkflowService
from packages.domain import Asset, AssetLibrary, AssetLibraryKind, AssetStatus, ClassificationStatus
from packages.domain.voice_presets import list_voices
from packages.ports.asset_library_repository import AssetLibraryRepository
from packages.ports.asset_repository import AssetRepository
from packages.ports.project_repository import ProjectRepository
from packages.shared.storage import SharedStorageService

logger = logging.getLogger(__name__)

router = APIRouter()


@router.get("/presets", summary="获取预设音色列表")
def list_preset_voices(
    gender: Optional[str] = Query(None, description="按性别筛选: male/female/child"),
    style: Optional[str] = Query(None, description="按风格筛选: stable/lively/customer_service/narration/news/story"),
    keyword: Optional[str] = Query(None, description="按关键词搜索"),
    _user: AuthenticatedUser = Depends(get_current_user),
) -> list[dict]:
    """获取可用的预设音色列表。

    用于配音功能的音色选择。
    """
    voices = list_voices(gender=gender, style=style, keyword=keyword)
    return [
        {
            "voice_id": v.voice_id,
            "name": v.name,
            "gender": v.gender.value,
            "style": v.style.value,
            "description": v.description,
            "default_speed": v.default_speed,
            "default_pitch": v.default_pitch,
            "sample_rate": v.sample_rate,
            "language": v.language,
        }
        for v in voices
    ]


def _get_repository(session: Session = Depends(get_db_session)) -> SQLAlchemyTTSJobRepository:
    return SQLAlchemyTTSJobRepository(session)


def _to_response(job, sign_url=None) -> TTSJobResponse:
    output_url = job.output_audio_url
    if sign_url and output_url:
        output_url = sign_url(output_url)
    return TTSJobResponse(
        id=job.id,
        user_id=job.user_id,
        input_text=job.input_text,
        voice_id=job.voice_id,
        voice_model=job.voice_model,
        project_id=job.project_id,
        voice_clone_profile_id=job.voice_clone_profile_id,
        status=job.status,
        output_audio_url=output_url,
        output_audio_key=job.output_audio_key,
        duration=job.duration,
        file_size=job.file_size,
        sample_rate=job.sample_rate,
        format=job.format,
        error_message=job.error_message,
        retry_count=job.retry_count,
        max_retries=job.max_retries,
        metadata=job.metadata,
        started_at=job.started_at,
        completed_at=job.completed_at,
        created_at=job.created_at,
        updated_at=job.updated_at,
    )


@router.post("/synthesize", response_model=TTSSynthesizeResponse, status_code=status.HTTP_201_CREATED)
def synthesize(
    request: TTSSynthesizeRequest,
    authenticated_user: AuthenticatedUser = Depends(get_current_user),
    repository: SQLAlchemyTTSJobRepository = Depends(_get_repository),
    cosyvoice_service: CosyVoiceService = Depends(get_cosyvoice_service),
    voice_clone_repo=Depends(get_voice_clone_profile_repository),
) -> TTSSynthesizeResponse:
    """发起 TTS 合成任务。

    创建 TTS 任务 → 提交 CosyVoice 合成 → 触发 Celery 异步轮询。
    与音色克隆接口保持一致：CosyVoice 失败时不抛 500，而是返回 201 + failed 状态任务记录。
    """
    user_id = authenticated_user.user.id

    # 解析 voice_id：前端可能传克隆音色 profile UUID（而非 CosyVoice voice_id），
    # 与 /tts/preview 保持一致：命中 profile → 校验归属 → 取 CosyVoice voice_id
    actual_voice_id = request.voice_id
    voice_clone_profile_id = request.voice_clone_profile_id
    resolved_profile = None
    if actual_voice_id:
        resolved_profile = voice_clone_repo.get(actual_voice_id)
        if resolved_profile is not None:
            voice_clone_profile_id = actual_voice_id

    # 显式传了 voice_clone_profile_id（且与 voice_id 不同）时再查一次归属
    if voice_clone_profile_id and (resolved_profile is None or resolved_profile.id != voice_clone_profile_id):
        resolved_profile = voice_clone_repo.get(voice_clone_profile_id)
        if resolved_profile is None:
            raise HTTPException(
                status_code=status.HTTP_404_NOT_FOUND,
                detail="Voice clone profile not found",
            )

    if resolved_profile is not None:
        if resolved_profile.user_id != user_id:
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail="无权访问该音色",
            )
        if not resolved_profile.voice_id:
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail="音色克隆尚未完成，请稍后再试",
            )
        # 命中克隆音色：无论 voice_id 直接传 profile UUID 还是显式传 voice_clone_profile_id，
        # job.voice_id 统一存解析后的 CosyVoice voice_id
        actual_voice_id = resolved_profile.voice_id

    use_case = CreateTTSJobUseCase(repository)
    job = use_case.execute(
        user_id=user_id,
        input_text=request.text,
        voice_id=actual_voice_id,
        voice_model=request.voice_model,
        voice_clone_profile_id=voice_clone_profile_id,
        metadata=request.metadata_,
    )

    # 提交 CosyVoice 合成任务
    workflow = TTSWorkflowService(
        repository=repository,
        cosyvoice_service=cosyvoice_service,
    )

    try:
        job = workflow.start_synthesis(job.id)
    except Exception as e:
        # 兜底：workflow 内部已捕获 CosyVoiceError / ValueError，
        # 但 DB 异常、网络异常等意外错误可能逃逸。
        # 与音色克隆接口保持一致：标记 failed，返回 201，不抛 500。
        logger.error(f"TTS 合成异常: job_id={job.id}, error={e}", exc_info=True)
        try:
            job = workflow.process_synthesis_failure(job.id, str(e))
        except Exception as inner_e:
            logger.error(f"标记 TTS job 失败时出错: job_id={job.id}, error={inner_e}")

    # 若任务处于 processing 状态（异步模式），触发 Celery 后台轮询
    if job.status.value == "processing":
        # 分段合成任务 vs 普通单段任务
        segment_task_ids = (job.metadata or {}).get("segment_task_ids", [])
        is_segment = len(segment_task_ids) > 0

        try:
            if is_segment:
                celery_app.send_task("worker.process_tts_segment_synthesis", args=[job.id])
            else:
                celery_app.send_task("worker.process_tts_synthesis", args=[job.id])
        except Exception as e:
            # Celery 调度失败，标记 job 为 failed
            try:
                workflow.process_synthesis_failure(job.id, f"Celery 任务调度失败: {e}")
            except Exception as inner_e:
                logger.error(f"Celery 调度后标记失败时出错: job_id={job.id}, error={inner_e}")

    return TTSSynthesizeResponse(
        job_id=job.id,
        status=job.status,
        message="合成任务已创建",
    )


@router.get("/jobs", response_model=ListTTSJobResponse)
def list_tts_jobs(
    page: int = Query(default=1, ge=1, description="页码"),
    page_size: int = Query(default=20, ge=1, le=100, description="每页数量"),
    status_filter: Optional[str] = Query(None, alias="status"),
    authenticated_user: AuthenticatedUser = Depends(get_current_user),
    repository: SQLAlchemyTTSJobRepository = Depends(_get_repository),
    sign_url=Depends(get_audio_url_signer),
) -> ListTTSJobResponse:
    """列出用户的 TTS 合成任务。"""
    user_id = authenticated_user.user.id
    use_case = ListTTSJobsUseCase(repository)
    skip = (page - 1) * page_size
    items, total = use_case.execute(user_id, status=status_filter, skip=skip, limit=page_size)
    return ListTTSJobResponse(
        items=[_to_response(j, sign_url) for j in items],
        total=total,
        page=page,
        page_size=page_size,
    )


@router.get("/jobs/{job_id}", response_model=TTSJobResponse)
def get_tts_job(
    job_id: str,
    authenticated_user: AuthenticatedUser = Depends(get_current_user),
    repository: SQLAlchemyTTSJobRepository = Depends(_get_repository),
    sign_url=Depends(get_audio_url_signer),
) -> TTSJobResponse:
    """获取 TTS 任务详情。"""
    user_id = authenticated_user.user.id
    use_case = GetTTSJobUseCase(repository)
    try:
        job = use_case.execute(job_id, user_id)
    except TTSJobNotFoundError as _e:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="TTS job not found") from _e
    return _to_response(job, sign_url)


@router.get("/jobs/{job_id}/status", response_model=TTSStatusResponse)
def get_tts_job_status(
    job_id: str,
    authenticated_user: AuthenticatedUser = Depends(get_current_user),
    repository: SQLAlchemyTTSJobRepository = Depends(_get_repository),
    sign_url=Depends(get_audio_url_signer),
) -> TTSStatusResponse:
    """查询 TTS 合成状态（用于前端轮询）。"""
    user_id = authenticated_user.user.id
    use_case = GetTTSJobStatusUseCase(repository)
    try:
        job = use_case.execute(job_id, user_id)
    except TTSJobNotFoundError as _e:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="TTS job not found") from _e
    output_url = job.output_audio_url
    if output_url:
        output_url = sign_url(output_url)
    return TTSStatusResponse(
        id=job.id,
        status=job.status,
        output_audio_url=output_url,
        error_message=job.error_message,
        duration=job.duration,
        retry_count=job.retry_count,
        created_at=job.created_at,
        updated_at=job.updated_at,
    )


@router.delete("/jobs/{job_id}", status_code=status.HTTP_204_NO_CONTENT, response_model=None, response_class=Response)
def delete_tts_job(
    job_id: str,
    authenticated_user: AuthenticatedUser = Depends(get_current_user),
    repository: SQLAlchemyTTSJobRepository = Depends(_get_repository),
) -> Response:
    """删除 TTS 合成任务。"""
    user_id = authenticated_user.user.id
    use_case = DeleteTTSJobUseCase(repository)
    deleted = use_case.execute(job_id, user_id)
    if not deleted:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="TTS job not found")
    return


def _find_or_create_voice_library(
    *,
    user_id: str,
    project_repository: ProjectRepository,
    asset_library_repository: Any,  # port Protocol 声明为 async，SQLAlchemy 实现为同步，与 upload/asset_libraries 路由惯例一致用 Any
) -> AssetLibrary:
    """在用户可访问的项目中找到（或自动创建）voice 素材库。

    与前端配音素材页逻辑一致：素材库挂在项目下，配音素材读取
    getAssetsByKind("voice") → 用户所有可访问项目中的 voice 库。
    优先使用已有 voice 库；没有则在第一个可访问项目中自动创建。
    """
    projects = project_repository.find_accessible_projects(user_id)
    if not projects:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="没有可用的项目，请先创建项目后再保存配音素材",
        )

    for project in projects:
        for lib in asset_library_repository.find_by_project(project.id):
            kind = lib.kind.value if hasattr(lib.kind, "value") else lib.kind
            if kind == AssetLibraryKind.VOICE.value:
                return lib

    # 所有项目都没有 voice 库 → 在第一个可访问项目中自动创建默认配音素材库。
    # asset_libraries 有 (project_id, kind) 唯一索引兜底并发：若两个请求同时创建，
    # 落败方捕获 IntegrityError 回滚后重新查询，返回抢先创建成功的库。
    project = projects[0]
    library = AssetLibrary.create(
        project_id=project.id,
        name="配音素材库",
        kind=AssetLibraryKind.VOICE,
    )
    try:
        return asset_library_repository.create(library)
    except IntegrityError:
        # 并发下另一个请求已抢先创建：回滚当前事务（立即 commit 模式下 session 已
        # 自动回滚，rollback 为幂等 no-op；UoW/flush 模式下必须显式回滚才能继续查询），
        # 再重查返回抢先创建成功的库。
        session = getattr(asset_library_repository, "session", None)
        if session is not None:
            try:
                session.rollback()
            except Exception:
                logger.warning("IntegrityError 后回滚 session 失败（可能已关闭）", exc_info=True)
        for lib in asset_library_repository.find_by_project(project.id):
            kind = lib.kind.value if hasattr(lib.kind, "value") else lib.kind
            if kind == AssetLibraryKind.VOICE.value:
                return lib
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail="配音素材库创建失败，请重试",
        ) from None  # IntegrityError 已处理，不保留异常链


@router.post(
    "/jobs/{job_id}/save-to-library",
    response_model=SaveToLibraryResponse,
    status_code=status.HTTP_201_CREATED,
)
def save_tts_job_to_library(
    job_id: str,
    request: SaveToLibraryRequest = SaveToLibraryRequest(),
    authenticated_user: AuthenticatedUser = Depends(get_current_user),
    tts_repository: SQLAlchemyTTSJobRepository = Depends(_get_repository),
    asset_repository: AssetRepository = Depends(get_asset_repository),
    asset_library_repository: AssetLibraryRepository = Depends(get_asset_library_repository),
    project_repository: ProjectRepository = Depends(get_project_repository),
    storage_service: SharedStorageService = Depends(get_storage_service),
    sign_url=Depends(get_audio_url_signer),
) -> SaveToLibraryResponse:
    """将已完成的 TTS 合成结果保存到配音素材库（assets 表新素材体系）。

    流程：把 TTS 输出音频转存到用户素材 OSS 路径 → 创建 file_type=audio、
    status=ready 的 asset（挂用户 voice 素材库）→ 返回前端可用结构。
    配额策略与素材上传一致（上传/ingest 链路无额外配额拦截）。
    """
    user_id = authenticated_user.user.id

    # 获取 TTS job
    get_use_case = GetTTSJobUseCase(tts_repository)
    try:
        job = get_use_case.execute(job_id, user_id)
    except TTSJobNotFoundError as _e:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="TTS job not found") from _e

    # 校验已完成
    if not job.is_completed:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="TTS job is not completed yet",
        )

    if not job.output_audio_url and not job.output_audio_key:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="TTS job 缺少输出音频，无法保存",
        )

    # 素材名称
    name = request.name or f"TTS-{job.id[:8]}"

    # 找到（或自动创建）用户 voice 素材库
    library = _find_or_create_voice_library(
        user_id=user_id,
        project_repository=project_repository,
        asset_library_repository=asset_library_repository,
    )

    # 转存音频到素材 OSS 路径（tts-outputs/ 下的产物归 TTS 任务所有，
    # 素材独立持有副本，删除 TTS 任务不影响配音库素材）
    audio_format = (job.format or "mp3").strip() or "mp3"
    content_type_map = {
        "mp3": "audio/mpeg",
        "wav": "audio/wav",
        "pcm": "audio/pcm",
        "opus": "audio/opus",
    }
    content_type = content_type_map.get(audio_format, "audio/mpeg")
    storage_key = f"uploads/voice/tts/{job.id}.{audio_format}"

    tmp_path: Path | None = None
    audio_duration: float | None = None
    file_size = 0
    try:
        with tempfile.NamedTemporaryFile(suffix=f".{audio_format}", delete=False) as tmp:
            tmp_path = Path(tmp.name)
        # 优先用 OSS storage_key（走 oss2 SDK，私有 bucket 也可下载）；
        # 兜底用 output_audio_url（旧任务可能没有 key）。
        # download_asset 自动识别输入：http(s):// 开头走 HTTP 下载，否则按 OSS key 走 SDK。
        download_source = job.output_audio_key or job.output_audio_url
        downloaded = storage_service.download_asset(download_source, tmp_path)
        if not downloaded or not tmp_path.exists() or tmp_path.stat().st_size == 0:
            raise HTTPException(
                status_code=status.HTTP_502_BAD_GATEWAY,
                detail="TTS 音频下载失败，无法保存到配音库",
            )
        file_size = tmp_path.stat().st_size
        storage_service.upload_file(tmp_path, storage_key, content_type=content_type)

        # 从音频文件提取时长（ffprobe），作为 job.duration 的兜底
        try:
            proc = subprocess.run(
                [
                    "ffprobe", "-v", "quiet", "-print_format", "json",
                    "-show_format", str(tmp_path),
                ],
                capture_output=True, text=True, timeout=10,
            )
            if proc.returncode == 0:
                fmt = json.loads(proc.stdout).get("format", {})
                dur = float(fmt.get("duration", 0))
                if dur > 0:
                    audio_duration = dur
        except Exception:
            logger.warning("ffprobe 提取时长失败: job_id=%s", job.id, exc_info=True)
    except HTTPException:
        raise
    except Exception as e:
        logger.error("TTS 音频转存素材失败: job_id=%s, error=%s", job.id, e, exc_info=True)
        raise HTTPException(
            status_code=status.HTTP_502_BAD_GATEWAY,
            detail="TTS 音频转存失败，无法保存到配音库",
        ) from e
    finally:
        if tmp_path and tmp_path.exists():
            try:
                tmp_path.unlink()
            except OSError:
                pass

    # 构建素材元信息
    metadata_: dict[str, object] = {
        "source": "tts_job",
        "tts_job_id": job.id,
        "format": job.format,
        "sample_rate": job.sample_rate,
        "voice_id": job.voice_id,
        "voice_name": job.voice_model or "",
    }
    if job.metadata:
        for key in ("speed", "language"):
            if key in job.metadata:
                metadata_[key] = job.metadata[key]

    asset = Asset.create(
        project_id=library.project_id,
        library_id=library.id,
        name=name,
        storage_key=storage_key,
        mime_type=content_type,
        metadata=metadata_,
        file_size=file_size,
        duration=job.duration or audio_duration or None,
        status=AssetStatus.READY,
        classification_status=ClassificationStatus.PENDING,  # 音频不参与内容分类，保持 pending 与 ingest 链路一致
        uploaded_by_user_id=user_id,
    )
    try:
        asset = asset_repository.create(asset)
    except Exception as e:
        # DB 写入失败：清理已上传到 OSS 的素材文件，避免产生无法索引的孤儿文件
        logger.error("素材记录创建失败，清理 OSS 文件: %s, error=%s", storage_key, e, exc_info=True)
        try:
            storage_service.delete_file(storage_key)
        except Exception:
            logger.warning("清理孤儿 OSS 文件失败: %s", storage_key, exc_info=True)
        raise HTTPException(
            status_code=status.HTTP_502_BAD_GATEWAY,
            detail="素材保存失败，请重试",
        ) from e

    return SaveToLibraryResponse(
        id=asset.id,
        name=asset.name,
        audio_url=sign_url(storage_key),
        duration=asset.duration or 0.0,
        voice_id=job.voice_id,
        voice_name=job.voice_model or "",
        status="completed",
    )


@router.post("/preview", response_model=TTSPreviewResponse)
def preview_tts(
    request: TTSPreviewRequest,
    authenticated_user: AuthenticatedUser = Depends(get_current_user),
    cosyvoice_service: CosyVoiceService = Depends(get_cosyvoice_service),
    voice_clone_repo=Depends(get_voice_clone_profile_repository),
) -> TTSPreviewResponse:
    """TTS 预览（试听）——同步合成，立即返回音频 URL。

    用于前端预览配音效果，限制文本长度 200 字以内。
    支持预设音色和克隆音色：克隆音色传的是 profile UUID，需解析为 CosyVoice voice_id。
    """
    # 解析 voice_id：前端可能传 VoiceCloneProfile UUID 或预设音色 ID
    actual_voice_id = request.voice_id
    profile = voice_clone_repo.get(request.voice_id)
    if profile is not None:
        # 命中克隆音色 profile — 校验归属权限
        if profile.user_id != authenticated_user.user.id:
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail="无权访问该音色",
            )
        if not profile.voice_id:
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail="音色克隆尚未完成，请稍后再试",
            )
        actual_voice_id = profile.voice_id

    try:
        result = cosyvoice_service.synthesize_speech(
            text=request.text,
            voice_id=actual_voice_id,
            speed=request.speed,
        )
    except CosyVoiceError as e:
        raise HTTPException(
            status_code=status.HTTP_502_BAD_GATEWAY,
            detail=f"TTS 合成失败: {e}",
        ) from e
    except ValueError as e:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail=str(e),
        ) from e

    return TTSPreviewResponse(
        audio_url=result.audio_url,
        duration=result.duration if result.duration and result.duration > 0 else None,
    )


@router.websocket("/ws/tts/stream")
async def tts_websocket_stream(
    websocket: WebSocket,
    cosyvoice_service: CosyVoiceService = Depends(get_cosyvoice_service),
) -> None:
    """WebSocket 流式 TTS 合成。

    协议：
    1. 客户端发送 JSON 文本帧: {"text": "...", "voice_id": "...", ...}
    2. 服务端发送 JSON 状态帧 + 二进制音频帧
    3. 完成时发送 JSON 结束帧
    """
    await websocket.accept()
    try:
        message = await websocket.receive_json()
        params = {
            "text": message.get("text", ""),
            "voice_id": message.get("voice_id", ""),
            "sample_rate": message.get("sample_rate", 0),
            "format": message.get("format", "mp3"),
            "speed": message.get("speed", 1.0),
        }
        streaming_service = TTSStreamingService(cosyvoice_service)
        await streaming_service.synthesize_and_stream(websocket, params)
    except WebSocketDisconnect:
        logger.info("WebSocket 客户端主动断开连接")
    except Exception as e:
        logger.error(f"WebSocket 流式合成异常: {e}", exc_info=True)
        try:
            await websocket.send_json({"type": "error", "message": f"服务异常: {e}"})
        except Exception as send_err:
            logger.warning("WebSocket 错误消息发送失败（连接可能已断开）: %s", send_err)