diff --git a/apps/api/app/api/routes/tts.py b/apps/api/app/api/routes/tts.py index 00d6a886b..2f6a8da30 100644 --- a/apps/api/app/api/routes/tts.py +++ b/apps/api/app/api/routes/tts.py @@ -5,7 +5,7 @@ from __future__ import annotations from typing import Optional from app.auth import AuthenticatedUser, get_current_user -from app.dependencies import get_db_session +from app.dependencies import get_cosyvoice_service, get_db_session from app.schemas.tts import ( ListTTSJobResponse, TTSSynthesizeRequest, @@ -19,6 +19,7 @@ from sqlalchemy.orm import Session from packages.adapters.sqlalchemy_impl.tts_job_repository import ( SQLAlchemyTTSJobRepository, ) +from packages.application.cosyvoice_service import CosyVoiceService from packages.application.tts_job.use_cases import ( CreateTTSJobUseCase, DeleteTTSJobUseCase, @@ -27,6 +28,7 @@ from packages.application.tts_job.use_cases import ( ListTTSJobsUseCase, TTSJobNotFoundError, ) +from packages.application.tts_job.workflow import TTSWorkflowService router = APIRouter() @@ -67,10 +69,11 @@ def synthesize( request: TTSSynthesizeRequest, authenticated_user: AuthenticatedUser = Depends(get_current_user), repository: SQLAlchemyTTSJobRepository = Depends(_get_repository), + cosyvoice_service: CosyVoiceService = Depends(get_cosyvoice_service), ) -> TTSSynthesizeResponse: """发起 TTS 合成任务。 - 创建 TTS 任务，状态为 pending，等待后续 CosyVoice API 调用。 + 创建 TTS 任务 → 提交 CosyVoice 合成 → 触发 Celery 异步轮询。 """ user_id = authenticated_user.user.id use_case = CreateTTSJobUseCase(repository) @@ -82,6 +85,26 @@ def synthesize( voice_clone_profile_id=request.voice_clone_profile_id, metadata=request.metadata_, ) + + # 提交 CosyVoice 合成任务 + workflow = TTSWorkflowService( + repository=repository, cosyvoice_service=cosyvoice_service, + ) + job = workflow.start_synthesis(job.id) + + # 若任务处于 processing 状态（异步模式），触发 Celery 后台轮询 + if job.status.value == "processing": + task_id = (job.metadata or {}).get("cosyvoice_task_id", "") + if task_id: + try: + from worker_app.tasks import process_tts_synthesis + process_tts_synthesis.delay(job.id) + except Exception as e: + # Celery 调度失败，标记 job 为 failed + workflow.process_synthesis_failure( + job.id, f"Celery 任务调度失败: {e}" + ) + return TTSSynthesizeResponse( job_id=job.id, status=job.status, diff --git a/apps/api/app/api/routes/voices.py b/apps/api/app/api/routes/voices.py index d7cf61252..a91b4b4da 100644 --- a/apps/api/app/api/routes/voices.py +++ b/apps/api/app/api/routes/voices.py @@ -24,6 +24,7 @@ from app.schemas.voice_library import ( from fastapi import APIRouter, Depends, HTTPException, Query, Response, status from sqlalchemy.orm import Session +from packages.adapters.sqlalchemy_impl.voice_clone_profile_repository import SQLAlchemyVoiceCloneProfileRepository from packages.adapters.sqlalchemy_impl.voice_library_repository import SQLAlchemyVoiceLibraryRepository from packages.application.voice_library.commands import CreateVoiceLibraryCommand, UpdateVoiceLibraryCommand from packages.application.voice_library.use_cases import ( @@ -45,6 +46,10 @@ def _get_voice_repository(session: Session = Depends(get_db_session)) -> SQLAlch return SQLAlchemyVoiceLibraryRepository(session) +def _get_clone_profile_repository(session: Session = Depends(get_db_session)) -> SQLAlchemyVoiceCloneProfileRepository: + return SQLAlchemyVoiceCloneProfileRepository(session) + + def _to_response(item) -> VoiceLibraryItemResponse: return VoiceLibraryItemResponse( id=item.id, @@ -65,8 +70,16 @@ def _to_response(item) -> VoiceLibraryItemResponse: ) -def _to_unified_response(item) -> UnifiedVoiceItemResponse: - """将数据库音色转换为统一响应格式。""" +def _to_unified_response(item, profile_id_map: dict | None = None) -> UnifiedVoiceItemResponse: + """将数据库音色转换为统一响应格式。 + + Args: + item: VoiceLibraryItem + profile_id_map: voice_id → profile_id 映射，用于填充 voice_clone_profile_id + """ + profile_id = None + if profile_id_map and item.voice_id: + profile_id = profile_id_map.get(item.voice_id) return UnifiedVoiceItemResponse( id=item.id, type="clone", @@ -83,6 +96,7 @@ def _to_unified_response(item) -> UnifiedVoiceItemResponse: tags=item.tags, user_id=item.user_id, project_id=item.project_id, + voice_clone_profile_id=profile_id, created_at=item.created_at, updated_at=item.updated_at, ) @@ -125,6 +139,7 @@ def list_voices_unified( limit: int = Query(50, ge=1, le=200), authenticated_user: AuthenticatedUser = Depends(get_current_user), voice_repository: SQLAlchemyVoiceLibraryRepository = Depends(_get_voice_repository), + clone_profile_repository: SQLAlchemyVoiceCloneProfileRepository = Depends(_get_clone_profile_repository), ) -> UnifiedVoiceListResponse: """获取配音列表（预置音色 + 用户克隆音色）。 @@ -148,9 +163,11 @@ def list_voices_unified( # 获取克隆音色 if has_clone: use_case = ListVoiceLibraryUseCase(voice_repository) - clone_items_raw = use_case.execute(user_id, status=status_filter, skip=skip, limit=limit) - clone_items = [_to_unified_response(i) for i in clone_items_raw] - clone_count = voice_repository.count_by_user(user_id, status=status_filter) if status_filter else voice_repository.count_by_user(user_id) + clone_items_raw, clone_count = use_case.execute(user_id, status=status_filter, skip=skip, limit=limit) + # 批量查询 voice_id → profile_id 映射，填充 voice_clone_profile_id + voice_ids = [i.voice_id for i in clone_items_raw if i.voice_id] + profile_id_map = clone_profile_repository.find_profile_ids_by_voice_ids(voice_ids) if voice_ids else {} + clone_items = [_to_unified_response(i, profile_id_map) for i in clone_items_raw] # 组装结果 if type == "preset": @@ -214,8 +231,7 @@ def list_voices_legacy( """ user_id = authenticated_user.user.id use_case = ListVoiceLibraryUseCase(voice_repository) - items = use_case.execute(user_id, status=status_filter, skip=skip, limit=limit) - total = voice_repository.count_by_user(user_id) + items, total = use_case.execute(user_id, status=status_filter, skip=skip, limit=limit) return ListVoiceLibraryResponse( items=[_to_response(i) for i in items], total=total, diff --git a/apps/worker/worker_app/celery_app.py b/apps/worker/worker_app/celery_app.py index 4d401eb44..352775112 100755 --- a/apps/worker/worker_app/celery_app.py +++ b/apps/worker/worker_app/celery_app.py @@ -13,6 +13,7 @@ celery_app.conf.imports = ( "worker_app.tasks.generation", "worker_app.tasks.voice_extraction", "worker_app.tasks.voice_clone", + "worker_app.tasks.tts_synthesis", "worker_app.tasks.edit_plan_generation", "worker_app.tasks.compose_video", "apps.worker.video_processing.dedup", diff --git a/apps/worker/worker_app/tasks/__init__.py b/apps/worker/worker_app/tasks/__init__.py index 535c8886a..ec088c631 100755 --- a/apps/worker/worker_app/tasks/__init__.py +++ b/apps/worker/worker_app/tasks/__init__.py @@ -37,6 +37,10 @@ def __getattr__(name: str): from .voice_clone import process_voice_clone return process_voice_clone + elif name == "process_tts_synthesis": + from .tts_synthesis import process_tts_synthesis + + return process_tts_synthesis raise AttributeError(f"module {__name__!r} has no attribute {name!r}") @@ -49,4 +53,5 @@ __all__ = [ "extract_voice_task", "extract_background_task", "process_voice_clone", + "process_tts_synthesis", ] diff --git a/apps/worker/worker_app/tasks/tts_synthesis.py b/apps/worker/worker_app/tasks/tts_synthesis.py new file mode 100644 index 000000000..e8ce6c816 --- /dev/null +++ b/apps/worker/worker_app/tasks/tts_synthesis.py @@ -0,0 +1,108 @@ +"""TTS synthesis tasks - process TTS synthesis requests via CosyVoice API.""" + +import logging + +from celery import Task +from celery.exceptions import Retry +from worker_app.celery_app import celery_app +from worker_app.db import SessionLocal + +from packages.adapters.sqlalchemy_impl.tts_job_repository import ( + SQLAlchemyTTSJobRepository, +) +from packages.application.cosyvoice_service import ( + CosyVoiceError, + CosyVoiceService, + CosyVoiceTimeoutError, +) +from packages.application.tts_job.workflow import TTSWorkflowService + +logger = logging.getLogger(__name__) + + +@celery_app.task(bind=True, max_retries=3, name="worker.process_tts_synthesis") +def process_tts_synthesis(self: Task, job_id: str) -> dict: + """处理 TTS 合成任务。 + + 通过 TTSWorkflowService.poll_and_process_synthesis() 轮询 CosyVoice + 合成任务状态，更新 TTSJob。 + 超时自动重试（最多 3 次），其他错误标记 job 为 failed。 + + Args: + job_id: TTSJob ID + + Returns: + dict: {"ok": True, "job_id": str, "audio_url": str} 或 + {"ok": False, "job_id": str, "error": str} + """ + session = None + try: + session = SessionLocal() + repo = SQLAlchemyTTSJobRepository(session) + workflow = TTSWorkflowService( + repository=repo, cosyvoice_service=CosyVoiceService(), + ) + + updated_job = workflow.poll_and_process_synthesis(job_id, timeout=120) + session.commit() + + logger.info( + f"TTS synthesis completed: job_id={job_id}, " + f"audio_url={updated_job.output_audio_url}" + ) + return { + "ok": True, + "job_id": job_id, + "audio_url": updated_job.output_audio_url, + } + + except Retry: + # Celery Retry 异常必须向上传播，不能被后续 except 捕获 + raise + + except CosyVoiceTimeoutError as e: + logger.warning(f"TTS synthesis timeout for {job_id}: {e}") + if session is not None: + session.rollback() + # 超时重试，指数退避 + raise self.retry(exc=e, countdown=30) + + except CosyVoiceError as e: + logger.error(f"TTS synthesis failed for {job_id}: {e}") + if session is not None: + session.rollback() + # 标记 job 为 failed + try: + if session is not None: + job = repo.get(job_id) + if job is not None: + job.mark_failed(str(e)) + repo.update(job) + session.commit() + except Exception as inner_e: + logger.error(f"Failed to mark job as failed: {inner_e}") + if session is not None: + session.rollback() + return {"ok": False, "job_id": job_id, "error": str(e)} + + except Exception as e: + logger.error(f"TTS synthesis unexpected error for {job_id}: {e}") + if session is not None: + session.rollback() + # 标记 job 为 failed + try: + if session is not None: + job = repo.get(job_id) + if job is not None: + job.mark_failed(str(e)) + repo.update(job) + session.commit() + except Exception as inner_e: + logger.error(f"Failed to mark job as failed: {inner_e}") + if session is not None: + session.rollback() + return {"ok": False, "job_id": job_id, "error": str(e)} + + finally: + if session is not None: + session.close() diff --git a/apps/worker/worker_app/tasks/voice_clone.py b/apps/worker/worker_app/tasks/voice_clone.py index 0924f9b74..8375054cf 100644 --- a/apps/worker/worker_app/tasks/voice_clone.py +++ b/apps/worker/worker_app/tasks/voice_clone.py @@ -26,7 +26,13 @@ def process_voice_clone(self: Task, profile_id: str) -> dict: 通过 VoiceCloneWorkflowService.poll_and_process_clone() 轮询 CosyVoice 克隆任务状态，更新 VoiceCloneProfile。 - 失败时自动重试（最多 2 次）。 + + 重试策略（Celery 5.x bind=True 模式）： + - max_retries=2：最多重试 2 次（共执行 3 次），超过后抛出 MaxRetriesExceededError + - CosyVoiceTimeoutError：网络超时属于临时性故障，使用 countdown=30 延迟 30 秒后重试 + - CosyVoiceError：API 业务错误（如任务失败），属于永久性故障，不重试直接标记 failed + - Exception：未知错误，不重试直接标记 failed，避免无限重试掩盖 bug + - Retry 异常：Celery 内部重试信号，必须向上传播不能被捕获 Args: profile_id: VoiceCloneProfile ID @@ -34,8 +40,11 @@ def process_voice_clone(self: Task, profile_id: str) -> dict: Returns: dict: {"ok": True, "profile_id": str, "voice_id": str} """ - session = SessionLocal() + # P2-2 修复：session 初始化为 None，避免 SessionLocal() 抛异常时 + # finally 块中 session.close() 触发 UnboundLocalError + session = None try: + session = SessionLocal() repo = SQLAlchemyVoiceCloneProfileRepository(session) workflow = VoiceCloneWorkflowService( repository=repo, cosyvoice_service=CosyVoiceService(), @@ -60,39 +69,47 @@ def process_voice_clone(self: Task, profile_id: str) -> dict: except CosyVoiceTimeoutError as e: logger.error(f"Voice clone timeout for {profile_id}: {e}") - session.rollback() - # 超时重试 + if session is not None: + session.rollback() + # 超时属于临时性故障，延迟 30 秒后重试 raise self.retry(exc=e, countdown=30) except CosyVoiceError as e: logger.error(f"Voice clone failed for {profile_id}: {e}") - session.rollback() - # 标记 profile 为 failed + if session is not None: + session.rollback() + # API 业务错误属于永久性故障，不重试，标记 profile 为 failed try: - profile = repo.get(profile_id) - if profile is not None: - profile.mark_failed(str(e)) - repo.update(profile) - session.commit() + if session is not None: + profile = repo.get(profile_id) + if profile is not None: + profile.mark_failed(str(e)) + repo.update(profile) + session.commit() except Exception as inner_e: logger.error(f"Failed to mark profile as failed: {inner_e}") - session.rollback() + if session is not None: + session.rollback() return {"ok": False, "profile_id": profile_id, "error": str(e)} except Exception as e: logger.error(f"Voice clone unexpected error for {profile_id}: {e}") - session.rollback() - # 标记 profile 为 failed + if session is not None: + session.rollback() + # 未知错误不重试，标记 profile 为 failed，避免无限重试掩盖 bug try: - profile = repo.get(profile_id) - if profile is not None: - profile.mark_failed(str(e)) - repo.update(profile) - session.commit() + if session is not None: + profile = repo.get(profile_id) + if profile is not None: + profile.mark_failed(str(e)) + repo.update(profile) + session.commit() except Exception as inner_e: logger.error(f"Failed to mark profile as failed: {inner_e}") - session.rollback() + if session is not None: + session.rollback() return {"ok": False, "profile_id": profile_id, "error": str(e)} finally: - session.close() + if session is not None: + session.close() diff --git a/packages/adapters/sqlalchemy_impl/voice_clone_profile_repository.py b/packages/adapters/sqlalchemy_impl/voice_clone_profile_repository.py index fa1a37f16..821cb125c 100644 --- a/packages/adapters/sqlalchemy_impl/voice_clone_profile_repository.py +++ b/packages/adapters/sqlalchemy_impl/voice_clone_profile_repository.py @@ -2,7 +2,7 @@ from __future__ import annotations -from typing import List, Optional +from typing import Dict, List, Optional from sqlalchemy.orm import Session @@ -130,6 +130,23 @@ class SQLAlchemyVoiceCloneProfileRepository: return None return self._model_to_entity(model) + def find_profile_ids_by_voice_ids(self, voice_ids: List[str]) -> Dict[str, str]: + """批量查询 voice_id → profile_id 映射。用于填充统一列表的 voice_clone_profile_id。""" + if not voice_ids: + return {} + rows = ( + self.session.query( + VoiceCloneProfileModel.voice_id, + VoiceCloneProfileModel.id, + ) + .filter( + VoiceCloneProfileModel.voice_id.in_(voice_ids), + VoiceCloneProfileModel.status != "deleted", + ) + .all() + ) + return {voice_id: profile_id for voice_id, profile_id in rows} + @staticmethod def _model_to_entity(model: VoiceCloneProfileModel) -> VoiceCloneProfile: return VoiceCloneProfile( diff --git a/packages/application/cosyvoice_service.py b/packages/application/cosyvoice_service.py index 15ce4dce7..79c72a6f4 100644 --- a/packages/application/cosyvoice_service.py +++ b/packages/application/cosyvoice_service.py @@ -391,6 +391,100 @@ class CosyVoiceService: # ── 语音合成 ───────────────────────────────────────── + def submit_synthesize_task( + self, + text: str, + voice_id: str = "", + sample_rate: int = 0, + format: str = "", + speed: float = 1.0, + ) -> dict: + """提交语音合成任务（非阻塞）。 + + 只提交任务到 CosyVoice API，不轮询结果。 + 返回的 dict 包含 task_id（异步）或 audio_url（同步）。 + + Args: + text: 要合成的文本 + voice_id: 音色 ID（预置音色或克隆音色） + sample_rate: 采样率（Hz），0 表示使用配置默认值 + format: 输出格式（mp3/wav/pcm），空表示使用配置默认值 + speed: 语速（0.5-2.0），1.0 为正常速度 + + Returns: + dict: {"task_id": str, "audio_url": str, "request_id": str} + task_id 和 audio_url 至少有一个非空 + + Raises: + CosyVoiceError: API 调用失败 + CosyVoiceAuthError: 认证失败 + ValueError: 参数无效 + """ + if not text: + raise ValueError("text 不能为空") + if not voice_id: + raise ValueError("voice_id 不能为空") + if not self._api_key: + raise CosyVoiceAuthError("CosyVoice API Key 未配置") + + settings = get_shared_settings() + + payload = { + "model": self._model, + "input": { + "text": text, + }, + "parameters": { + "voice": voice_id, + "sample_rate": sample_rate or settings.cosyvoice_sample_rate, + "format": format or settings.cosyvoice_format, + "rate": speed, + }, + } + + response = self._call_api( + method="POST", + path="/services/aigc/text2audio/generation", + json=payload, + timeout=60.0, + ) + + output = response.get("output", {}) + task_id = output.get("task_id", "") + audio_url = output.get("audio_url", "") + request_id = response.get("request_id", "") + + if not task_id and not audio_url: + raise CosyVoiceError( + f"CosyVoice API 未返回 task_id 或 audio_url: {response}" + ) + + return { + "task_id": task_id, + "audio_url": audio_url, + "duration": output.get("duration", 0.0), + "file_size": output.get("file_size", 0), + "request_id": request_id, + } + + def poll_synthesize_task(self, task_id: str, timeout: float = 120.0) -> dict: + """轮询语音合成任务状态（公开方法）。 + + 供 Celery 后台任务调用，轮询直到完成或超时。 + + Args: + task_id: CosyVoice 任务 ID + timeout: 超时时间（秒），默认 120 + + Returns: + dict: {"audio_url": str, "duration": float, "file_size": int} + + Raises: + CosyVoiceError: 任务失败 + CosyVoiceTimeoutError: 超时 + """ + return self._poll_synthesize_task(task_id, timeout=timeout) + def synthesize_speech( self, text: str, diff --git a/packages/application/tts_job/workflow.py b/packages/application/tts_job/workflow.py new file mode 100644 index 000000000..682b99e2e --- /dev/null +++ b/packages/application/tts_job/workflow.py @@ -0,0 +1,214 @@ +"""TTS Job workflow orchestration — Phase 3. + +编排 TTS 合成的完整流程： +1. 创建 TTSJob（pending） +2. 提交 CosyVoice 合成任务 +3. 轮询处理合成结果（成功/失败） +4. 重试失败的合成 +""" + +from __future__ import annotations + +import logging +from typing import Optional + +from packages.application.cosyvoice_service import ( + CosyVoiceAuthError, + CosyVoiceError, + CosyVoiceService, +) +from packages.domain.tts_job import TTSJob, TTSJobStatus +from packages.ports.tts_job_repository import TTSJobRepository + +logger = logging.getLogger(__name__) + + +class TTSWorkflowError(Exception): + """TTS 合成工作流异常。""" + + pass + + +class TTSJobNotFoundError(Exception): + """TTS 任务未找到。""" + + pass + + +class TTSWorkflowService: + """TTS 合成工作流编排服务。 + + 协调 TTSJobRepository + CosyVoiceService， + 实现完整的 TTS 合成生命周期管理。 + """ + + def __init__( + self, + repository: TTSJobRepository, + cosyvoice_service: CosyVoiceService, + ) -> None: + self.repository = repository + self.cosyvoice_service = cosyvoice_service + + def start_synthesis( + self, + job_id: str, + ) -> TTSJob: + """启动 TTS 合成流程。 + + 1. 获取 pending 状态的 TTSJob + 2. 标记为 processing + 3. 提交 CosyVoice 合成任务 + 4. 保存 task_id 到 metadata + 5. 返回 job（Celery task 由调用方触发） + + Args: + job_id: TTSJob ID + + Returns: + TTSJob: 更新后的 job + + Raises: + TTSJobNotFoundError: job 不存在 + TTSWorkflowError: CosyVoice 提交失败 + """ + job = self.repository.get(job_id) + if job is None: + raise TTSJobNotFoundError(f"TTS job {job_id} not found") + + # 标记为 processing + job.mark_processing() + job = self.repository.update(job) + + try: + submit_result = self.cosyvoice_service.submit_synthesize_task( + text=job.input_text, + voice_id=job.voice_id, + sample_rate=job.sample_rate, + format=job.format, + ) + + # 保存 task_id / request_id 到 metadata + job_metadata = dict(job.metadata) + job_metadata["cosyvoice_task_id"] = submit_result.get("task_id", "") + job_metadata["cosyvoice_request_id"] = submit_result.get("request_id", "") + + # 如果 CosyVoice 同步返回了 audio_url，直接标记完成 + audio_url = submit_result.get("audio_url", "") + if audio_url: + job.mark_completed( + output_audio_url=audio_url, + duration=submit_result.get("duration", 0.0), + file_size=submit_result.get("file_size", 0), + ) + job.metadata = job_metadata + job = self.repository.update(job) + logger.info( + f"TTS 合成同步完成: job_id={job.id}, audio_url={audio_url}" + ) + return job + + job.metadata = job_metadata + job = self.repository.update(job) + logger.info( + f"TTS 合成任务已提交: job_id={job.id}, " + f"task_id={submit_result.get('task_id')}" + ) + + except (CosyVoiceError, CosyVoiceAuthError) as e: + job.mark_failed(str(e)) + job = self.repository.update(job) + logger.error(f"TTS 合成提交失败: job_id={job.id}, error={e}") + except ValueError as e: + job.mark_failed(str(e)) + job = self.repository.update(job) + logger.error(f"TTS 合成参数错误: job_id={job.id}, error={e}") + + return job + + def poll_and_process_synthesis( + self, job_id: str, timeout: float = 120.0 + ) -> TTSJob: + """轮询 CosyVoice 合成任务并处理结果。 + + 从 job.metadata 获取 task_id，调用 CosyVoiceService.poll_synthesize_task() + 轮询状态，然后通过 process_synthesis_result / process_synthesis_failure 更新 job。 + + 供 Celery 后台任务调用。 + """ + job = self.repository.get(job_id) + if job is None: + raise TTSJobNotFoundError(f"TTS job {job_id} not found") + + task_id = (job.metadata or {}).get("cosyvoice_task_id", "") + if not task_id: + raise ValueError( + f"TTSJob {job_id} has no cosyvoice_task_id in metadata" + ) + + result = self.cosyvoice_service.poll_synthesize_task(task_id, timeout=timeout) + return self.process_synthesis_result( + job_id, + audio_url=result["audio_url"], + duration=result.get("duration", 0.0), + file_size=result.get("file_size", 0), + ) + + def process_synthesis_result( + self, + job_id: str, + audio_url: str, + *, + duration: float = 0.0, + file_size: int = 0, + ) -> TTSJob: + """处理合成成功结果。 + + Args: + job_id: TTSJob ID + audio_url: 输出音频 URL + duration: 音频时长 + file_size: 文件大小 + + Returns: + TTSJob: 更新后的 job + + Raises: + TTSJobNotFoundError: job 不存在 + """ + job = self.repository.get(job_id) + if job is None: + raise TTSJobNotFoundError(f"TTS job {job_id} not found") + + job.mark_completed( + output_audio_url=audio_url, + duration=duration, + file_size=file_size, + ) + job = self.repository.update(job) + logger.info(f"TTS 合成成功: job_id={job_id}, audio_url={audio_url}") + return job + + def process_synthesis_failure( + self, job_id: str, error_message: str + ) -> TTSJob: + """处理合成失败结果。 + + Args: + job_id: TTSJob ID + error_message: 错误信息 + + Returns: + TTSJob: 更新后的 job + + Raises: + TTSJobNotFoundError: job 不存在 + """ + job = self.repository.get(job_id) + if job is None: + raise TTSJobNotFoundError(f"TTS job {job_id} not found") + + job.mark_failed(error_message) + job = self.repository.update(job) + logger.error(f"TTS 合成失败: job_id={job_id}, error={error_message}") + return job diff --git a/packages/application/voice_library/use_cases.py b/packages/application/voice_library/use_cases.py index b02a0c76b..c81ee51a6 100644 --- a/packages/application/voice_library/use_cases.py +++ b/packages/application/voice_library/use_cases.py @@ -25,8 +25,11 @@ class ListVoiceLibraryUseCase: status: Optional[str] = None, skip: int = 0, limit: int = 50, - ) -> List[VoiceLibraryItem]: - return self.repository.list_by_user(user_id, status=status, skip=skip, limit=limit) + ) -> tuple[List[VoiceLibraryItem], int]: + """返回 (items, total_count)，避免调用方再单独查一次 count。""" + items = self.repository.list_by_user(user_id, status=status, skip=skip, limit=limit) + total = self.repository.count_by_user(user_id, status=status) if status else self.repository.count_by_user(user_id) + return items, total class GetVoiceLibraryUseCase: diff --git a/packages/ports/voice_clone_profile_repository.py b/packages/ports/voice_clone_profile_repository.py index c8c46e96d..81f1b8637 100644 --- a/packages/ports/voice_clone_profile_repository.py +++ b/packages/ports/voice_clone_profile_repository.py @@ -2,7 +2,7 @@ from __future__ import annotations -from typing import Protocol +from typing import Dict, List, Protocol from packages.domain.voice_clone_profile import VoiceCloneProfile, VoiceCloneStatus @@ -49,3 +49,7 @@ class VoiceCloneProfileRepository(Protocol): def find_by_voice_id(self, voice_id: str) -> VoiceCloneProfile | None: """根据 CosyVoice 返回的音色 ID 查找档案。""" ... + + def find_profile_ids_by_voice_ids(self, voice_ids: List[str]) -> Dict[str, str]: + """批量查询 voice_id → profile_id 映射。""" + ... diff --git a/tests/unit/test_voice_library_use_cases.py b/tests/unit/test_voice_library_use_cases.py index ca21bdc39..3fa913525 100644 --- a/tests/unit/test_voice_library_use_cases.py +++ b/tests/unit/test_voice_library_use_cases.py @@ -586,11 +586,13 @@ class TestListVoiceLibraryUseCase: VoiceLibraryItem(id="v2", user_id="user-001", name="B"), ] mock_repo.list_by_user.return_value = items + mock_repo.count_by_user.return_value = 2 use_case = ListVoiceLibraryUseCase(repository=mock_repo) - result = use_case.execute("user-001") + result_items, total = use_case.execute("user-001") - assert len(result) == 2 + assert len(result_items) == 2 + assert total == 2 mock_repo.list_by_user.assert_called_once_with("user-001", status=None, skip=0, limit=50) def test_list_with_status_filter(self, mock_repo): @@ -605,11 +607,13 @@ class TestListVoiceLibraryUseCase: def test_list_empty(self, mock_repo): """测试空列表""" mock_repo.list_by_user.return_value = [] + mock_repo.count_by_user.return_value = 0 use_case = ListVoiceLibraryUseCase(repository=mock_repo) - result = use_case.execute("user-001") + items, total = use_case.execute("user-001") - assert result == [] + assert items == [] + assert total == 0 # ===========================================================================