灵应
e539105256
Deploy / Build Production Runtime Images (push) Has been skipped
Deploy / Deploy Production (push) Has been skipped
Deploy / Production Browser E2E (push) Has been skipped
CI/CD Pipeline / Frontend Lint (pull_request) Failing after 163h53m1s
CI/CD Pipeline / Validate Code Quality And Tests (pull_request) Failing after 163h53m6s
Deploy / Deploy Staging (push) Failing after 164h23m43s
CI/CD Pipeline / Frontend Lint (push) Failing after 164h23m43s
CI/CD Pipeline / Validate Code Quality And Tests (push) Failing after 164h23m50s
P1-1 (阻塞性): TTS 合成完整异步链路 - CosyVoiceService 新增 submit_synthesize_task() + poll_synthesize_task() - 新建 TTSWorkflowService 编排层 (packages/application/tts_job/workflow.py) - 新建 Celery 任务 process_tts_synthesis (apps/worker/worker_app/tasks/tts_synthesis.py) - 注册到 celery_app.conf.imports + tasks/__init__.py 懒加载 - TTS 路由 synthesize() 增加 CosyVoice 提交 + Celery 调度 P2-1: voice_clone.py 添加详细 Celery 重试策略注释 P2-2: 修复 voice_clone.py Session 泄漏 (session=None 安全模式) P2-3: ListVoiceLibraryUseCase 返回 (items, count) 元组，消除重复 count_by_user() P2-4: 新增 find_profile_ids_by_voice_ids() 批量方法，填充 voice_clone_profile_id 测试: 749 passed, 0 failed
56 lines
1.5 KiB
Python
56 lines
1.5 KiB
Python
"""VoiceCloneProfileRepository 端口接口 — Phase 3 CosyVoice 集成."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Dict, List, Protocol
|
|
|
|
from packages.domain.voice_clone_profile import VoiceCloneProfile, VoiceCloneStatus
|
|
|
|
|
|
class VoiceCloneProfileRepository(Protocol):
|
|
"""音色克隆档案仓储接口。"""
|
|
|
|
def create(self, profile: VoiceCloneProfile) -> VoiceCloneProfile:
|
|
"""持久化一个新音色克隆档案。"""
|
|
...
|
|
|
|
def get(self, profile_id: str) -> VoiceCloneProfile | None:
|
|
"""根据 ID 获取档案。"""
|
|
...
|
|
|
|
def update(self, profile: VoiceCloneProfile) -> VoiceCloneProfile:
|
|
"""更新档案状态。"""
|
|
...
|
|
|
|
def delete(self, profile_id: str) -> bool:
|
|
"""删除档案。"""
|
|
...
|
|
|
|
def list_by_user(
|
|
self,
|
|
user_id: str,
|
|
*,
|
|
status: VoiceCloneStatus | str | None = None,
|
|
limit: int = 50,
|
|
offset: int = 0,
|
|
) -> list[VoiceCloneProfile]:
|
|
"""按用户列出档案，支持状态过滤。"""
|
|
...
|
|
|
|
def count_by_user(
|
|
self,
|
|
user_id: str,
|
|
*,
|
|
status: VoiceCloneStatus | str | None = None,
|
|
) -> int:
|
|
"""按用户统计档案数量。"""
|
|
...
|
|
|
|
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 映射。"""
|
|
...
|