From f7c8b441d6b9441813c144888a4d47104c95c7fe Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=81=B5=E5=BA=94?= Date: Thu, 2 Jul 2026 07:31:26 +0800 Subject: [PATCH] =?UTF-8?q?feat(phase3):=20CosyVoice=20=E9=85=8D=E7=BD=AE?= =?UTF-8?q?=20+=20VoiceCloneProfile/TTSJob=20=E9=A2=86=E5=9F=9F=E6=A8=A1?= =?UTF-8?q?=E5=9E=8B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Task 3.01: CosyVoice 配置 - 在 SharedSettings 中添加 CosyVoice 配置字段 - 更新 .env.example 和 .env.production.example Task 3.02: VoiceCloneProfile 领域模型 - 创建 VoiceCloneProfile 实体（音色克隆档案） - 状态机: pending → processing → ready/failed → disabled - 支持重试机制（retry_count/max_retries） - 创建 VoiceCloneProfileRepository 端口接口 Task 3.03: TTSJob 领域模型 - 创建 TTSJob 实体（TTS 任务） - 关联 VoiceCloneProfile（voice_clone_profile_id） - 状态机: pending → processing → completed/failed → cancelled - 支持重试机制 - 创建 TTSJobRepository 端口接口 单元测试: - VoiceCloneProfile: 31 个测试用例 - TTSJob: 34 个测试用例 - 全部 65 个测试通过 遵循六边形架构: Domain → Port → Adapter --- .env.example | 8 + .env.production.example | 8 + packages/domain/tts_job.py | 299 +++++++++++++++ packages/domain/voice_clone_profile.py | 261 +++++++++++++ packages/ports/tts_job_repository.py | 58 +++ .../ports/voice_clone_profile_repository.py | 51 +++ packages/shared/config.py | 8 + tests/unit/test_tts_job.py | 344 ++++++++++++++++++ tests/unit/test_voice_clone_profile.py | 306 ++++++++++++++++ 9 files changed, 1343 insertions(+) create mode 100644 packages/domain/tts_job.py create mode 100644 packages/domain/voice_clone_profile.py create mode 100644 packages/ports/tts_job_repository.py create mode 100644 packages/ports/voice_clone_profile_repository.py create mode 100644 tests/unit/test_tts_job.py create mode 100644 tests/unit/test_voice_clone_profile.py diff --git a/.env.example b/.env.example index dc55a01ae..3bcd412c9 100755 --- a/.env.example +++ b/.env.example @@ -42,3 +42,11 @@ OSS_ENDPOINT=oss-cn-hangzhou.aliyuncs.com OSS_ACCESS_KEY_ID=your-access-key-id OSS_ACCESS_KEY_SECRET=your-access-key-secret OSS_BUCKET_NAME=xiaoxia-autocut + +# ==================== CosyVoice 语音合成配置 ==================== +COSYVOICE_API_KEY=your-cosyvoice-api-key +COSYVOICE_BASE_URL=https://dashscope.aliyuncs.com/api/v1/services/aigc/text2audio +COSYVOICE_MODEL=cosyvoice-v1 +COSYVOICE_VOICE=longxiaochun +COSYVOICE_SAMPLE_RATE=22050 +COSYVOICE_FORMAT=mp3 diff --git a/.env.production.example b/.env.production.example index 833bbbdc8..6082c3926 100644 --- a/.env.production.example +++ b/.env.production.example @@ -41,6 +41,14 @@ OSS_BUCKET_NAME=xiaoxia-autocut OSS_DIRECT_UPLOAD_MAX_MB=2000 OSS_DIRECT_UPLOAD_EXPIRE_SECONDS=900 +# ==================== CosyVoice 语音合成（必须配置）==================== +COSYVOICE_API_KEY=CHANGE_ME_COSYVOICE_API_KEY +COSYVOICE_BASE_URL=https://dashscope.aliyuncs.com/api/v1/services/aigc/text2audio +COSYVOICE_MODEL=cosyvoice-v1 +COSYVOICE_VOICE=longxiaochun +COSYVOICE_SAMPLE_RATE=22050 +COSYVOICE_FORMAT=mp3 + # ==================== 生成文件 ==================== GENERATED_FILES_DIR=/app/generated GENERATED_FILES_URL_PREFIX=/generated-files diff --git a/packages/domain/tts_job.py b/packages/domain/tts_job.py new file mode 100644 index 000000000..288889b86 --- /dev/null +++ b/packages/domain/tts_job.py @@ -0,0 +1,299 @@ +"""TTSJob 领域模型 — Phase 3 CosyVoice 集成. + +TTS 任务，用于管理文本转语音的合成请求。 + +状态机： + pending → processing → completed + ↘ failed → pending (重试) + ↘ cancelled +""" + +from __future__ import annotations + +import sys +from dataclasses import dataclass, field +from datetime import datetime, timezone + +if sys.version_info >= (3, 11): + from enum import StrEnum +else: + from enum import Enum + + class StrEnum(str, Enum): + pass + + +from uuid import uuid4 + + +class TTSJobStatus(StrEnum): + """TTS 任务状态枚举。""" + + PENDING = "pending" + """待处理（任务已创建，等待执行）""" + + PROCESSING = "processing" + """处理中（正在调用 CosyVoice API 合成）""" + + COMPLETED = "completed" + """已完成（音频合成成功）""" + + FAILED = "failed" + """失败（合成失败）""" + + CANCELLED = "cancelled" + """已取消（用户取消或系统取消）""" + + +# 终态集合 +TERMINAL_STATUSES = frozenset({TTSJobStatus.COMPLETED, TTSJobStatus.FAILED, TTSJobStatus.CANCELLED}) + +# 合法状态转换 +_VALID_TRANSITIONS: dict[TTSJobStatus, set[TTSJobStatus]] = { + TTSJobStatus.PENDING: {TTSJobStatus.PROCESSING, TTSJobStatus.FAILED, TTSJobStatus.CANCELLED}, + TTSJobStatus.PROCESSING: {TTSJobStatus.COMPLETED, TTSJobStatus.FAILED, TTSJobStatus.CANCELLED}, + TTSJobStatus.FAILED: {TTSJobStatus.PENDING}, # 重试回到 pending +} + + +@dataclass(slots=True) +class TTSJob: + """TTS 任务实体。 + + Attributes: + id: 任务唯一标识 + user_id: 所属用户 + project_id: 所属项目（可选） + voice_clone_profile_id: 关联的音色克隆档案 ID（可选，使用自定义音色时必填） + status: 当前状态 + input_text: 输入文本 + voice_id: 使用的音色 ID（CosyVoice 内置音色或克隆音色 ID） + voice_model: 使用的模型名称 + output_audio_url: 输出音频文件 URL + output_audio_key: 输出音频 OSS key + duration: 音频时长（秒） + file_size: 文件大小（字节） + sample_rate: 采样率 + format: 输出格式（mp3/wav/pcm） + error_message: 错误信息 + retry_count: 已重试次数 + max_retries: 最大重试次数 + metadata: 扩展元数据（JSON） + started_at: 开始处理时间 + completed_at: 完成时间 + created_at: 创建时间 + updated_at: 最后更新时间 + """ + + id: str + user_id: str + input_text: str + voice_id: str = "" + voice_model: str = "" + project_id: str = "" + voice_clone_profile_id: str = "" + status: TTSJobStatus = TTSJobStatus.PENDING + output_audio_url: str = "" + output_audio_key: str = "" + duration: float = 0.0 + file_size: int = 0 + sample_rate: int = 22050 + format: str = "mp3" + error_message: str = "" + retry_count: int = 0 + max_retries: int = 3 + metadata: dict = field(default_factory=dict) + started_at: datetime | None = None + completed_at: datetime | None = None + created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc)) + updated_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc)) + + @classmethod + def create( + cls, + user_id: str, + input_text: str, + *, + voice_id: str = "", + voice_model: str = "", + project_id: str = "", + voice_clone_profile_id: str = "", + sample_rate: int = 22050, + format: str = "mp3", + max_retries: int = 3, + metadata: dict | None = None, + ) -> TTSJob: + """创建 TTS 任务。 + + Args: + user_id: 用户 ID + input_text: 输入文本 + voice_id: 使用的音色 ID + voice_model: 使用的模型名称 + project_id: 项目 ID + voice_clone_profile_id: 关联的音色克隆档案 ID + sample_rate: 采样率 + format: 输出格式 + max_retries: 最大重试次数 + metadata: 扩展元数据 + + Returns: + 新建的 TTSJob 实例 + + Raises: + ValueError: 参数校验失败 + """ + if not user_id.strip(): + raise ValueError("user_id 不能为空") + if not input_text.strip(): + raise ValueError("input_text 不能为空") + if len(input_text.strip()) > 10000: + raise ValueError("input_text 长度不能超过 10000 字符") + if format not in ("mp3", "wav", "pcm"): + raise ValueError(f"不支持的输出格式: {format}，支持: mp3/wav/pcm") + + return cls( + id=uuid4().hex, + user_id=user_id.strip(), + input_text=input_text.strip(), + voice_id=voice_id.strip(), + voice_model=voice_model.strip(), + project_id=project_id.strip(), + voice_clone_profile_id=voice_clone_profile_id.strip(), + sample_rate=sample_rate, + format=format.strip(), + max_retries=max_retries, + metadata=metadata or {}, + ) + + @property + def is_terminal(self) -> bool: + """是否处于终态。""" + return self.status in TERMINAL_STATUSES + + @property + def is_retryable(self) -> bool: + """是否可重试（失败且未超过重试上限）。""" + return self.status == TTSJobStatus.FAILED and self.retry_count < self.max_retries + + @property + def is_completed(self) -> bool: + """是否已完成。""" + return self.status == TTSJobStatus.COMPLETED and bool(self.output_audio_url) + + def transition_to(self, new_status: TTSJobStatus | str) -> None: + """执行状态转换。 + + Args: + new_status: 目标状态 + + Raises: + ValueError: 非法状态转换 + """ + if isinstance(new_status, str): + try: + new_status = TTSJobStatus(new_status) + except ValueError: + raise ValueError(f"无效状态: {new_status}") + + allowed = _VALID_TRANSITIONS.get(self.status, set()) + if new_status not in allowed: + raise ValueError( + f"非法状态转换: {self.status.value} → {new_status.value}，" + f"允许: {{{', '.join(s.value for s in allowed)}}}" + ) + + now = datetime.now(timezone.utc) + self.status = new_status + self.updated_at = now + + def mark_processing(self) -> None: + """标记为处理中。""" + self.transition_to(TTSJobStatus.PROCESSING) + self.started_at = datetime.now(timezone.utc) + self.error_message = "" + + def mark_completed( + self, + output_audio_url: str, + *, + output_audio_key: str = "", + duration: float = 0.0, + file_size: int = 0, + ) -> None: + """标记为已完成。 + + Args: + output_audio_url: 输出音频 URL + output_audio_key: 输出音频 OSS key + duration: 音频时长 + file_size: 文件大小 + """ + if not output_audio_url.strip(): + raise ValueError("output_audio_url 不能为空") + self.transition_to(TTSJobStatus.COMPLETED) + self.output_audio_url = output_audio_url.strip() + self.output_audio_key = output_audio_key.strip() + self.duration = duration + self.file_size = file_size + self.completed_at = datetime.now(timezone.utc) + self.error_message = "" + + def mark_failed(self, error_message: str) -> None: + """标记为失败。 + + Args: + error_message: 错误信息 + """ + self.transition_to(TTSJobStatus.FAILED) + self.error_message = error_message + + def mark_cancelled(self) -> None: + """标记为已取消。""" + self.transition_to(TTSJobStatus.CANCELLED) + + def prepare_retry(self) -> None: + """准备重试：重置状态为 pending。 + + Raises: + ValueError: 不可重试 + """ + if not self.is_retryable: + raise ValueError( + f"TTS 任务不可重试: status={self.status.value}, " + f"retry_count={self.retry_count}, max_retries={self.max_retries}" + ) + self.retry_count += 1 + self.transition_to(TTSJobStatus.PENDING) + self.error_message = "" + self.started_at = None + self.completed_at = None + + def to_dict(self) -> dict: + """序列化为字典。""" + return { + "id": self.id, + "user_id": self.user_id, + "project_id": self.project_id, + "voice_clone_profile_id": self.voice_clone_profile_id, + "status": self.status.value, + "input_text": self.input_text, + "voice_id": self.voice_id, + "voice_model": self.voice_model, + "output_audio_url": self.output_audio_url, + "output_audio_key": self.output_audio_key, + "duration": self.duration, + "file_size": self.file_size, + "sample_rate": self.sample_rate, + "format": self.format, + "error_message": self.error_message, + "retry_count": self.retry_count, + "max_retries": self.max_retries, + "is_retryable": self.is_retryable, + "is_completed": self.is_completed, + "metadata": self.metadata, + "started_at": self.started_at.isoformat() if self.started_at else None, + "completed_at": self.completed_at.isoformat() if self.completed_at else None, + "created_at": self.created_at.isoformat() if self.created_at else None, + "updated_at": self.updated_at.isoformat() if self.updated_at else None, + } diff --git a/packages/domain/voice_clone_profile.py b/packages/domain/voice_clone_profile.py new file mode 100644 index 000000000..f9e70785a --- /dev/null +++ b/packages/domain/voice_clone_profile.py @@ -0,0 +1,261 @@ +"""VoiceCloneProfile 领域模型 — Phase 3 CosyVoice 集成. + +音色克隆档案，用于管理用户的自定义音色。 + +状态机： + pending → processing → ready + ↘ failed → pending (重试) + ↘ disabled +""" + +from __future__ import annotations + +import sys +from dataclasses import dataclass, field +from datetime import datetime, timezone + +if sys.version_info >= (3, 11): + from enum import StrEnum +else: + from enum import Enum + + class StrEnum(str, Enum): + pass + + +from uuid import uuid4 + + +class VoiceCloneStatus(StrEnum): + """音色克隆状态枚举。""" + + PENDING = "pending" + """待处理（音频已上传，等待克隆）""" + + PROCESSING = "processing" + """处理中（正在调用 CosyVoice API 克隆）""" + + READY = "ready" + """就绪（克隆完成，可用于 TTS）""" + + FAILED = "failed" + """失败（克隆失败）""" + + DISABLED = "disabled" + """已禁用（用户手动禁用或违规）""" + + +# 终态集合 +TERMINAL_STATUSES = frozenset({VoiceCloneStatus.READY, VoiceCloneStatus.FAILED, VoiceCloneStatus.DISABLED}) + +# 合法状态转换 +_VALID_TRANSITIONS: dict[VoiceCloneStatus, set[VoiceCloneStatus]] = { + VoiceCloneStatus.PENDING: {VoiceCloneStatus.PROCESSING, VoiceCloneStatus.FAILED, VoiceCloneStatus.DISABLED}, + VoiceCloneStatus.PROCESSING: {VoiceCloneStatus.READY, VoiceCloneStatus.FAILED, VoiceCloneStatus.DISABLED}, + VoiceCloneStatus.FAILED: {VoiceCloneStatus.PENDING}, # 重试回到 pending + VoiceCloneStatus.READY: {VoiceCloneStatus.DISABLED}, # 可以禁用已就绪的音色 +} + + +@dataclass(slots=True) +class VoiceCloneProfile: + """音色克隆档案实体。 + + Attributes: + id: 档案唯一标识 + user_id: 所属用户 + name: 音色名称（用户可见） + description: 音色描述 + status: 当前状态 + source_audio_url: 源音频文件 URL（用户上传的用于克隆的音频） + voice_id: CosyVoice 返回的音色 ID（克隆成功后填充） + voice_model: 使用的模型名称 + language: 语言代码（如 zh-CN, en-US） + gender: 性别（male/female/unknown） + error_message: 错误信息 + retry_count: 已重试次数 + max_retries: 最大重试次数 + metadata: 扩展元数据（JSON） + created_at: 创建时间 + updated_at: 最后更新时间 + """ + + id: str + user_id: str + name: str + description: str = "" + status: VoiceCloneStatus = VoiceCloneStatus.PENDING + source_audio_url: str = "" + voice_id: str = "" + voice_model: str = "" + language: str = "zh-CN" + gender: str = "unknown" + error_message: str = "" + retry_count: int = 0 + max_retries: int = 3 + metadata: dict = field(default_factory=dict) + created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc)) + updated_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc)) + + @classmethod + def create( + cls, + user_id: str, + name: str, + *, + description: str = "", + source_audio_url: str = "", + voice_model: str = "", + language: str = "zh-CN", + gender: str = "unknown", + max_retries: int = 3, + metadata: dict | None = None, + ) -> VoiceCloneProfile: + """创建音色克隆档案。 + + Args: + user_id: 用户 ID + name: 音色名称 + description: 音色描述 + source_audio_url: 源音频文件 URL + voice_model: 使用的模型名称 + language: 语言代码 + gender: 性别 + max_retries: 最大重试次数 + metadata: 扩展元数据 + + Returns: + 新建的 VoiceCloneProfile 实例 + + Raises: + ValueError: 参数校验失败 + """ + if not user_id.strip(): + raise ValueError("user_id 不能为空") + if not name.strip(): + raise ValueError("name 不能为空") + if len(name.strip()) > 100: + raise ValueError("name 长度不能超过 100 字符") + + return cls( + id=uuid4().hex, + user_id=user_id.strip(), + name=name.strip(), + description=description.strip(), + source_audio_url=source_audio_url.strip(), + voice_model=voice_model.strip(), + language=language.strip(), + gender=gender.strip().lower(), + max_retries=max_retries, + metadata=metadata or {}, + ) + + @property + def is_terminal(self) -> bool: + """是否处于终态。""" + return self.status in TERMINAL_STATUSES + + @property + def is_retryable(self) -> bool: + """是否可重试（失败且未超过重试上限）。""" + return self.status == VoiceCloneStatus.FAILED and self.retry_count < self.max_retries + + @property + def is_ready(self) -> bool: + """是否就绪（可用于 TTS）。""" + return self.status == VoiceCloneStatus.READY and bool(self.voice_id) + + def transition_to(self, new_status: VoiceCloneStatus | str) -> None: + """执行状态转换。 + + Args: + new_status: 目标状态 + + Raises: + ValueError: 非法状态转换 + """ + if isinstance(new_status, str): + try: + new_status = VoiceCloneStatus(new_status) + except ValueError: + raise ValueError(f"无效状态: {new_status}") + + allowed = _VALID_TRANSITIONS.get(self.status, set()) + if new_status not in allowed: + raise ValueError( + f"非法状态转换: {self.status.value} → {new_status.value}，" + f"允许: {{{', '.join(s.value for s in allowed)}}}" + ) + + now = datetime.now(timezone.utc) + self.status = new_status + self.updated_at = now + + def mark_processing(self) -> None: + """标记为处理中。""" + self.transition_to(VoiceCloneStatus.PROCESSING) + self.error_message = "" + + def mark_ready(self, voice_id: str) -> None: + """标记为就绪。 + + Args: + voice_id: CosyVoice 返回的音色 ID + """ + if not voice_id.strip(): + raise ValueError("voice_id 不能为空") + self.transition_to(VoiceCloneStatus.READY) + self.voice_id = voice_id.strip() + self.error_message = "" + + def mark_failed(self, error_message: str) -> None: + """标记为失败。 + + Args: + error_message: 错误信息 + """ + self.transition_to(VoiceCloneStatus.FAILED) + self.error_message = error_message + + def mark_disabled(self) -> None: + """标记为禁用。""" + self.transition_to(VoiceCloneStatus.DISABLED) + + def prepare_retry(self) -> None: + """准备重试：重置状态为 pending。 + + Raises: + ValueError: 不可重试 + """ + if not self.is_retryable: + raise ValueError( + f"音色克隆不可重试: status={self.status.value}, " + f"retry_count={self.retry_count}, max_retries={self.max_retries}" + ) + self.retry_count += 1 + self.transition_to(VoiceCloneStatus.PENDING) + self.error_message = "" + self.voice_id = "" + + def to_dict(self) -> dict: + """序列化为字典。""" + return { + "id": self.id, + "user_id": self.user_id, + "name": self.name, + "description": self.description, + "status": self.status.value, + "source_audio_url": self.source_audio_url, + "voice_id": self.voice_id, + "voice_model": self.voice_model, + "language": self.language, + "gender": self.gender, + "error_message": self.error_message, + "retry_count": self.retry_count, + "max_retries": self.max_retries, + "is_retryable": self.is_retryable, + "is_ready": self.is_ready, + "metadata": self.metadata, + "created_at": self.created_at.isoformat() if self.created_at else None, + "updated_at": self.updated_at.isoformat() if self.updated_at else None, + } diff --git a/packages/ports/tts_job_repository.py b/packages/ports/tts_job_repository.py new file mode 100644 index 000000000..32a58b779 --- /dev/null +++ b/packages/ports/tts_job_repository.py @@ -0,0 +1,58 @@ +"""TTSJobRepository 端口接口 — Phase 3 CosyVoice 集成.""" + +from __future__ import annotations + +from typing import Protocol + +from packages.domain.tts_job import TTSJob, TTSJobStatus + + +class TTSJobRepository(Protocol): + """TTS 任务仓储接口。""" + + def create(self, job: TTSJob) -> TTSJob: + """持久化一个新 TTS 任务。""" + ... + + def get(self, job_id: str) -> TTSJob | None: + """根据 ID 获取任务。""" + ... + + def update(self, job: TTSJob) -> TTSJob: + """更新任务状态。""" + ... + + def delete(self, job_id: str) -> bool: + """删除任务。""" + ... + + def list_by_user( + self, + user_id: str, + *, + status: TTSJobStatus | str | None = None, + limit: int = 50, + offset: int = 0, + ) -> list[TTSJob]: + """按用户列出任务，支持状态过滤。""" + ... + + def count_by_user( + self, + user_id: str, + *, + status: TTSJobStatus | str | None = None, + ) -> int: + """按用户统计任务数量。""" + ... + + def list_by_profile( + self, + voice_clone_profile_id: str, + *, + status: TTSJobStatus | str | None = None, + limit: int = 50, + offset: int = 0, + ) -> list[TTSJob]: + """按音色克隆档案列出任务。""" + ... diff --git a/packages/ports/voice_clone_profile_repository.py b/packages/ports/voice_clone_profile_repository.py new file mode 100644 index 000000000..c8c46e96d --- /dev/null +++ b/packages/ports/voice_clone_profile_repository.py @@ -0,0 +1,51 @@ +"""VoiceCloneProfileRepository 端口接口 — Phase 3 CosyVoice 集成.""" + +from __future__ import annotations + +from typing import 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 查找档案。""" + ... diff --git a/packages/shared/config.py b/packages/shared/config.py index a612a043d..87ca3d213 100644 --- a/packages/shared/config.py +++ b/packages/shared/config.py @@ -29,6 +29,14 @@ class SharedSettings(BaseSettings): oss_access_key_secret: str = "" oss_bucket_name: str = "xiaoxia-autocut" + # CosyVoice (阿里云语音合成) + cosyvoice_api_key: str = "" + cosyvoice_base_url: str = "https://dashscope.aliyuncs.com/api/v1/services/aigc/text2audio" + cosyvoice_model: str = "cosyvoice-v1" + cosyvoice_voice: str = "longxiaochun" # 默认音色 + cosyvoice_sample_rate: int = 22050 + cosyvoice_format: str = "mp3" # 输出格式：mp3/wav/pcm + # Environment environment: str = "development" auto_create_schema: bool = False diff --git a/tests/unit/test_tts_job.py b/tests/unit/test_tts_job.py new file mode 100644 index 000000000..f00294adc --- /dev/null +++ b/tests/unit/test_tts_job.py @@ -0,0 +1,344 @@ +"""TTSJob 领域模型单元测试 — Phase 3 CosyVoice 集成.""" + +from __future__ import annotations + +import pytest + +from packages.domain.tts_job import TTSJob, TTSJobStatus + + +class TestTTSJobCreate: + """测试 TTSJob.create() 工厂方法。""" + + def test_create_success(self) -> None: + """正常创建 TTS 任务。""" + job = TTSJob.create( + user_id="user_001", + input_text="这是一段测试文本", + voice_id="longxiaochun", + voice_model="cosyvoice-v1", + project_id="project_001", + voice_clone_profile_id="profile_001", + sample_rate=22050, + format="mp3", + ) + + assert job.id + assert job.user_id == "user_001" + assert job.input_text == "这是一段测试文本" + assert job.voice_id == "longxiaochun" + assert job.voice_model == "cosyvoice-v1" + assert job.project_id == "project_001" + assert job.voice_clone_profile_id == "profile_001" + assert job.status == TTSJobStatus.PENDING + assert job.sample_rate == 22050 + assert job.format == "mp3" + assert job.retry_count == 0 + assert job.max_retries == 3 + + def test_create_minimal(self) -> None: + """使用最小参数创建。""" + job = TTSJob.create(user_id="user_001", input_text="测试文本") + + assert job.user_id == "user_001" + assert job.input_text == "测试文本" + assert job.status == TTSJobStatus.PENDING + assert job.voice_id == "" + assert job.project_id == "" + assert job.voice_clone_profile_id == "" + assert job.format == "mp3" + + def test_create_empty_user_id_raises(self) -> None: + """空 user_id 应抛出 ValueError。""" + with pytest.raises(ValueError, match="user_id 不能为空"): + TTSJob.create(user_id="", input_text="测试") + + def test_create_empty_input_text_raises(self) -> None: + """空 input_text 应抛出 ValueError。""" + with pytest.raises(ValueError, match="input_text 不能为空"): + TTSJob.create(user_id="user_001", input_text="") + + def test_create_whitespace_input_text_raises(self) -> None: + """空白 input_text 应抛出 ValueError。""" + with pytest.raises(ValueError, match="input_text 不能为空"): + TTSJob.create(user_id="user_001", input_text=" ") + + def test_create_input_text_too_long_raises(self) -> None: + """input_text 超过 10000 字符应抛出 ValueError。""" + with pytest.raises(ValueError, match="input_text 长度不能超过 10000 字符"): + TTSJob.create(user_id="user_001", input_text="a" * 10001) + + def test_create_invalid_format_raises(self) -> None: + """不支持的输出格式应抛出 ValueError。""" + with pytest.raises(ValueError, match="不支持的输出格式"): + TTSJob.create(user_id="user_001", input_text="测试", format="aac") + + def test_create_valid_formats(self) -> None: + """所有支持的格式都应正常创建。""" + for fmt in ("mp3", "wav", "pcm"): + job = TTSJob.create(user_id="user_001", input_text="测试", format=fmt) + assert job.format == fmt + + def test_create_strips_whitespace(self) -> None: + """应去除首尾空白。""" + job = TTSJob.create( + user_id=" user_001 ", + input_text=" 测试文本 ", + voice_id=" voice_001 ", + ) + + assert job.user_id == "user_001" + assert job.input_text == "测试文本" + assert job.voice_id == "voice_001" + + +class TestTTSJobStatus: + """测试状态相关属性和方法。""" + + def test_initial_status_is_pending(self) -> None: + """初始状态应为 PENDING。""" + job = TTSJob.create(user_id="user_001", input_text="测试") + assert job.status == TTSJobStatus.PENDING + + def test_is_terminal_pending(self) -> None: + """PENDING 不是终态。""" + job = TTSJob.create(user_id="user_001", input_text="测试") + assert not job.is_terminal + + def test_is_terminal_completed(self) -> None: + """COMPLETED 是终态。""" + job = TTSJob.create(user_id="user_001", input_text="测试") + job.mark_processing() + job.mark_completed(output_audio_url="https://example.com/audio.mp3") + assert job.is_terminal + + def test_is_terminal_failed(self) -> None: + """FAILED 是终态。""" + job = TTSJob.create(user_id="user_001", input_text="测试") + job.mark_processing() + job.mark_failed("合成失败") + assert job.is_terminal + + def test_is_terminal_cancelled(self) -> None: + """CANCELLED 是终态。""" + job = TTSJob.create(user_id="user_001", input_text="测试") + job.mark_cancelled() + assert job.is_terminal + + def test_is_retryable_not_failed(self) -> None: + """非 FAILED 状态不可重试。""" + job = TTSJob.create(user_id="user_001", input_text="测试") + assert not job.is_retryable + + def test_is_retryable_failed_under_limit(self) -> None: + """FAILED 且未超过重试上限时可重试。""" + job = TTSJob.create(user_id="user_001", input_text="测试") + job.mark_processing() + job.mark_failed("合成失败") + assert job.is_retryable + + def test_is_retryable_failed_over_limit(self) -> None: + """超过重试上限时不可重试。""" + job = TTSJob.create(user_id="user_001", input_text="测试", max_retries=1) + job.mark_processing() + job.mark_failed("第一次失败") + job.prepare_retry() + job.mark_processing() + job.mark_failed("第二次失败") + assert not job.is_retryable + + def test_is_completed_with_url(self) -> None: + """COMPLETED 且有 output_audio_url 时应返回 True。""" + job = TTSJob.create(user_id="user_001", input_text="测试") + job.mark_processing() + job.mark_completed(output_audio_url="https://example.com/audio.mp3") + assert job.is_completed + + def test_is_completed_without_url(self) -> None: + """COMPLETED 但无 output_audio_url 时应返回 False。""" + job = TTSJob.create(user_id="user_001", input_text="测试") + job.mark_processing() + job.status = TTSJobStatus.COMPLETED + job.output_audio_url = "" + assert not job.is_completed + + +class TestTTSJobTransitions: + """测试状态转换。""" + + def test_mark_processing(self) -> None: + """PENDING → PROCESSING 转换。""" + job = TTSJob.create(user_id="user_001", input_text="测试") + job.mark_processing() + assert job.status == TTSJobStatus.PROCESSING + assert job.error_message == "" + assert job.started_at is not None + + def test_mark_completed(self) -> None: + """PROCESSING → COMPLETED 转换。""" + job = TTSJob.create(user_id="user_001", input_text="测试") + job.mark_processing() + job.mark_completed( + output_audio_url="https://example.com/audio.mp3", + output_audio_key="tts/audio.mp3", + duration=5.5, + file_size=102400, + ) + assert job.status == TTSJobStatus.COMPLETED + assert job.output_audio_url == "https://example.com/audio.mp3" + assert job.output_audio_key == "tts/audio.mp3" + assert job.duration == 5.5 + assert job.file_size == 102400 + assert job.completed_at is not None + assert job.error_message == "" + + def test_mark_completed_empty_url_raises(self) -> None: + """mark_completed 空 output_audio_url 应抛出 ValueError。""" + job = TTSJob.create(user_id="user_001", input_text="测试") + job.mark_processing() + with pytest.raises(ValueError, match="output_audio_url 不能为空"): + job.mark_completed(output_audio_url="") + + def test_mark_completed_minimal(self) -> None: + """mark_completed 最小参数。""" + job = TTSJob.create(user_id="user_001", input_text="测试") + job.mark_processing() + job.mark_completed(output_audio_url="https://example.com/audio.mp3") + assert job.status == TTSJobStatus.COMPLETED + assert job.duration == 0.0 + assert job.file_size == 0 + + def test_mark_failed(self) -> None: + """PROCESSING → FAILED 转换。""" + job = TTSJob.create(user_id="user_001", input_text="测试") + job.mark_processing() + job.mark_failed("API 调用失败") + assert job.status == TTSJobStatus.FAILED + assert job.error_message == "API 调用失败" + + def test_mark_cancelled_from_pending(self) -> None: + """PENDING → CANCELLED 转换。""" + job = TTSJob.create(user_id="user_001", input_text="测试") + job.mark_cancelled() + assert job.status == TTSJobStatus.CANCELLED + + def test_mark_cancelled_from_processing(self) -> None: + """PROCESSING → CANCELLED 转换。""" + job = TTSJob.create(user_id="user_001", input_text="测试") + job.mark_processing() + job.mark_cancelled() + assert job.status == TTSJobStatus.CANCELLED + + def test_invalid_transition_raises(self) -> None: + """非法状态转换应抛出 ValueError。""" + job = TTSJob.create(user_id="user_001", input_text="测试") + with pytest.raises(ValueError, match="非法状态转换"): + job.mark_completed(output_audio_url="https://example.com/audio.mp3") + + def test_invalid_status_string_raises(self) -> None: + """无效状态字符串应抛出 ValueError。""" + job = TTSJob.create(user_id="user_001", input_text="测试") + with pytest.raises(ValueError, match="无效状态"): + job.transition_to("invalid_status") + + def test_transition_to_with_string(self) -> None: + """支持字符串形式的状态转换。""" + job = TTSJob.create(user_id="user_001", input_text="测试") + job.transition_to("processing") + assert job.status == TTSJobStatus.PROCESSING + + +class TestTTSJobRetry: + """测试重试逻辑。""" + + def test_prepare_retry_success(self) -> None: + """成功重试。""" + job = TTSJob.create(user_id="user_001", input_text="测试") + job.mark_processing() + job.mark_failed("失败") + job.prepare_retry() + + assert job.status == TTSJobStatus.PENDING + assert job.retry_count == 1 + assert job.error_message == "" + assert job.started_at is None + assert job.completed_at is None + + def test_prepare_retry_not_failed_raises(self) -> None: + """非 FAILED 状态重试应抛出 ValueError。""" + job = TTSJob.create(user_id="user_001", input_text="测试") + with pytest.raises(ValueError, match="不可重试"): + job.prepare_retry() + + def test_prepare_retry_over_limit_raises(self) -> None: + """超过重试上限重试应抛出 ValueError。""" + job = TTSJob.create(user_id="user_001", input_text="测试", max_retries=1) + job.mark_processing() + job.mark_failed("第一次失败") + job.prepare_retry() + job.mark_processing() + job.mark_failed("第二次失败") + + with pytest.raises(ValueError, match="不可重试"): + job.prepare_retry() + + +class TestTTSJobToDict: + """测试序列化。""" + + def test_to_dict_contains_all_fields(self) -> None: + """to_dict 应包含所有字段。""" + job = TTSJob.create( + user_id="user_001", + input_text="测试文本", + voice_id="longxiaochun", + voice_model="cosyvoice-v1", + project_id="project_001", + voice_clone_profile_id="profile_001", + sample_rate=22050, + format="mp3", + max_retries=5, + metadata={"key": "value"}, + ) + + result = job.to_dict() + + assert result["id"] == job.id + assert result["user_id"] == "user_001" + assert result["input_text"] == "测试文本" + assert result["voice_id"] == "longxiaochun" + assert result["voice_model"] == "cosyvoice-v1" + assert result["project_id"] == "project_001" + assert result["voice_clone_profile_id"] == "profile_001" + assert result["status"] == "pending" + assert result["sample_rate"] == 22050 + assert result["format"] == "mp3" + assert result["retry_count"] == 0 + assert result["max_retries"] == 5 + assert result["is_retryable"] is False + assert result["is_completed"] is False + assert result["metadata"] == {"key": "value"} + assert result["started_at"] is None + assert result["completed_at"] is None + assert result["created_at"] is not None + assert result["updated_at"] is not None + + def test_to_dict_after_completion(self) -> None: + """任务完成后 to_dict 应反映最新状态。""" + job = TTSJob.create(user_id="user_001", input_text="测试") + job.mark_processing() + job.mark_completed( + output_audio_url="https://example.com/audio.mp3", + duration=10.5, + file_size=204800, + ) + + result = job.to_dict() + + assert result["status"] == "completed" + assert result["output_audio_url"] == "https://example.com/audio.mp3" + assert result["duration"] == 10.5 + assert result["file_size"] == 204800 + assert result["is_completed"] is True + assert result["started_at"] is not None + assert result["completed_at"] is not None diff --git a/tests/unit/test_voice_clone_profile.py b/tests/unit/test_voice_clone_profile.py new file mode 100644 index 000000000..8f1840c2b --- /dev/null +++ b/tests/unit/test_voice_clone_profile.py @@ -0,0 +1,306 @@ +"""VoiceCloneProfile 领域模型单元测试 — Phase 3 CosyVoice 集成.""" + +from __future__ import annotations + +import pytest + +from packages.domain.voice_clone_profile import VoiceCloneProfile, VoiceCloneStatus + + +class TestVoiceCloneProfileCreate: + """测试 VoiceCloneProfile.create() 工厂方法。""" + + def test_create_success(self) -> None: + """正常创建音色克隆档案。""" + profile = VoiceCloneProfile.create( + user_id="user_001", + name="我的音色", + description="用于配音的自定义音色", + source_audio_url="https://example.com/audio.wav", + voice_model="cosyvoice-v1", + language="zh-CN", + gender="female", + ) + + assert profile.id + assert profile.user_id == "user_001" + assert profile.name == "我的音色" + assert profile.description == "用于配音的自定义音色" + assert profile.status == VoiceCloneStatus.PENDING + assert profile.source_audio_url == "https://example.com/audio.wav" + assert profile.voice_model == "cosyvoice-v1" + assert profile.language == "zh-CN" + assert profile.gender == "female" + assert profile.retry_count == 0 + assert profile.max_retries == 3 + assert profile.created_at + assert profile.updated_at + + def test_create_minimal(self) -> None: + """使用最小参数创建。""" + profile = VoiceCloneProfile.create(user_id="user_001", name="测试音色") + + assert profile.user_id == "user_001" + assert profile.name == "测试音色" + assert profile.status == VoiceCloneStatus.PENDING + assert profile.description == "" + assert profile.source_audio_url == "" + assert profile.language == "zh-CN" + assert profile.gender == "unknown" + + def test_create_empty_user_id_raises(self) -> None: + """空 user_id 应抛出 ValueError。""" + with pytest.raises(ValueError, match="user_id 不能为空"): + VoiceCloneProfile.create(user_id="", name="测试") + + def test_create_whitespace_user_id_raises(self) -> None: + """空白 user_id 应抛出 ValueError。""" + with pytest.raises(ValueError, match="user_id 不能为空"): + VoiceCloneProfile.create(user_id=" ", name="测试") + + def test_create_empty_name_raises(self) -> None: + """空 name 应抛出 ValueError。""" + with pytest.raises(ValueError, match="name 不能为空"): + VoiceCloneProfile.create(user_id="user_001", name="") + + def test_create_name_too_long_raises(self) -> None: + """name 超过 100 字符应抛出 ValueError。""" + with pytest.raises(ValueError, match="name 长度不能超过 100 字符"): + VoiceCloneProfile.create(user_id="user_001", name="a" * 101) + + def test_create_strips_whitespace(self) -> None: + """应去除首尾空白。""" + profile = VoiceCloneProfile.create( + user_id=" user_001 ", + name=" 测试音色 ", + description=" 描述 ", + ) + + assert profile.user_id == "user_001" + assert profile.name == "测试音色" + assert profile.description == "描述" + + def test_create_gender_normalized(self) -> None: + """gender 应转换为小写。""" + profile = VoiceCloneProfile.create( + user_id="user_001", + name="测试", + gender="FEMALE", + ) + + assert profile.gender == "female" + + +class TestVoiceCloneProfileStatus: + """测试状态相关属性和方法。""" + + def test_initial_status_is_pending(self) -> None: + """初始状态应为 PENDING。""" + profile = VoiceCloneProfile.create(user_id="user_001", name="测试") + assert profile.status == VoiceCloneStatus.PENDING + + def test_is_terminal_pending(self) -> None: + """PENDING 不是终态。""" + profile = VoiceCloneProfile.create(user_id="user_001", name="测试") + assert not profile.is_terminal + + def test_is_terminal_ready(self) -> None: + """READY 是终态。""" + profile = VoiceCloneProfile.create(user_id="user_001", name="测试") + profile.mark_processing() + profile.mark_ready(voice_id="voice_001") + assert profile.is_terminal + + def test_is_terminal_failed(self) -> None: + """FAILED 是终态。""" + profile = VoiceCloneProfile.create(user_id="user_001", name="测试") + profile.mark_processing() + profile.mark_failed("克隆失败") + assert profile.is_terminal + + def test_is_retryable_not_failed(self) -> None: + """非 FAILED 状态不可重试。""" + profile = VoiceCloneProfile.create(user_id="user_001", name="测试") + assert not profile.is_retryable + + def test_is_retryable_failed_under_limit(self) -> None: + """FAILED 且未超过重试上限时可重试。""" + profile = VoiceCloneProfile.create(user_id="user_001", name="测试") + profile.mark_processing() + profile.mark_failed("克隆失败") + assert profile.is_retryable + + def test_is_retryable_failed_over_limit(self) -> None: + """超过重试上限时不可重试。""" + profile = VoiceCloneProfile.create(user_id="user_001", name="测试", max_retries=1) + profile.mark_processing() + profile.mark_failed("第一次失败") + profile.prepare_retry() + profile.mark_processing() + profile.mark_failed("第二次失败") + assert not profile.is_retryable + + def test_is_ready_with_voice_id(self) -> None: + """READY 且有 voice_id 时应返回 True。""" + profile = VoiceCloneProfile.create(user_id="user_001", name="测试") + profile.mark_processing() + profile.mark_ready(voice_id="voice_001") + assert profile.is_ready + + def test_is_ready_without_voice_id(self) -> None: + """READY 但无 voice_id 时应返回 False。""" + profile = VoiceCloneProfile.create(user_id="user_001", name="测试") + profile.mark_processing() + profile.status = VoiceCloneStatus.READY + profile.voice_id = "" + assert not profile.is_ready + + +class TestVoiceCloneProfileTransitions: + """测试状态转换。""" + + def test_mark_processing(self) -> None: + """PENDING → PROCESSING 转换。""" + profile = VoiceCloneProfile.create(user_id="user_001", name="测试") + profile.mark_processing() + assert profile.status == VoiceCloneStatus.PROCESSING + assert profile.error_message == "" + + def test_mark_ready(self) -> None: + """PROCESSING → READY 转换。""" + profile = VoiceCloneProfile.create(user_id="user_001", name="测试") + profile.mark_processing() + profile.mark_ready(voice_id="voice_001") + assert profile.status == VoiceCloneStatus.READY + assert profile.voice_id == "voice_001" + assert profile.error_message == "" + + def test_mark_ready_empty_voice_id_raises(self) -> None: + """mark_ready 空 voice_id 应抛出 ValueError。""" + profile = VoiceCloneProfile.create(user_id="user_001", name="测试") + profile.mark_processing() + with pytest.raises(ValueError, match="voice_id 不能为空"): + profile.mark_ready(voice_id="") + + def test_mark_failed(self) -> None: + """PROCESSING → FAILED 转换。""" + profile = VoiceCloneProfile.create(user_id="user_001", name="测试") + profile.mark_processing() + profile.mark_failed("API 调用失败") + assert profile.status == VoiceCloneStatus.FAILED + assert profile.error_message == "API 调用失败" + + def test_mark_disabled_from_pending(self) -> None: + """PENDING → DISABLED 转换。""" + profile = VoiceCloneProfile.create(user_id="user_001", name="测试") + profile.mark_disabled() + assert profile.status == VoiceCloneStatus.DISABLED + + def test_mark_disabled_from_ready(self) -> None: + """READY → DISABLED 转换。""" + profile = VoiceCloneProfile.create(user_id="user_001", name="测试") + profile.mark_processing() + profile.mark_ready(voice_id="voice_001") + profile.mark_disabled() + assert profile.status == VoiceCloneStatus.DISABLED + + def test_invalid_transition_raises(self) -> None: + """非法状态转换应抛出 ValueError。""" + profile = VoiceCloneProfile.create(user_id="user_001", name="测试") + with pytest.raises(ValueError, match="非法状态转换"): + profile.mark_ready(voice_id="voice_001") # PENDING → READY 非法 + + def test_invalid_status_string_raises(self) -> None: + """无效状态字符串应抛出 ValueError。""" + profile = VoiceCloneProfile.create(user_id="user_001", name="测试") + with pytest.raises(ValueError, match="无效状态"): + profile.transition_to("invalid_status") + + def test_transition_to_with_string(self) -> None: + """支持字符串形式的状态转换。""" + profile = VoiceCloneProfile.create(user_id="user_001", name="测试") + profile.transition_to("processing") + assert profile.status == VoiceCloneStatus.PROCESSING + + +class TestVoiceCloneProfileRetry: + """测试重试逻辑。""" + + def test_prepare_retry_success(self) -> None: + """成功重试。""" + profile = VoiceCloneProfile.create(user_id="user_001", name="测试") + profile.mark_processing() + profile.mark_failed("失败") + profile.prepare_retry() + + assert profile.status == VoiceCloneStatus.PENDING + assert profile.retry_count == 1 + assert profile.error_message == "" + assert profile.voice_id == "" + + def test_prepare_retry_not_failed_raises(self) -> None: + """非 FAILED 状态重试应抛出 ValueError。""" + profile = VoiceCloneProfile.create(user_id="user_001", name="测试") + with pytest.raises(ValueError, match="不可重试"): + profile.prepare_retry() + + def test_prepare_retry_over_limit_raises(self) -> None: + """超过重试上限重试应抛出 ValueError。""" + profile = VoiceCloneProfile.create(user_id="user_001", name="测试", max_retries=1) + profile.mark_processing() + profile.mark_failed("第一次失败") + profile.prepare_retry() + profile.mark_processing() + profile.mark_failed("第二次失败") + + with pytest.raises(ValueError, match="不可重试"): + profile.prepare_retry() + + +class TestVoiceCloneProfileToDict: + """测试序列化。""" + + def test_to_dict_contains_all_fields(self) -> None: + """to_dict 应包含所有字段。""" + profile = VoiceCloneProfile.create( + user_id="user_001", + name="测试音色", + description="描述", + source_audio_url="https://example.com/audio.wav", + voice_model="cosyvoice-v1", + language="zh-CN", + gender="female", + max_retries=5, + metadata={"key": "value"}, + ) + + result = profile.to_dict() + + assert result["id"] == profile.id + assert result["user_id"] == "user_001" + assert result["name"] == "测试音色" + assert result["description"] == "描述" + assert result["status"] == "pending" + assert result["source_audio_url"] == "https://example.com/audio.wav" + assert result["voice_model"] == "cosyvoice-v1" + assert result["language"] == "zh-CN" + assert result["gender"] == "female" + assert result["retry_count"] == 0 + assert result["max_retries"] == 5 + assert result["is_retryable"] is False + assert result["is_ready"] is False + assert result["metadata"] == {"key": "value"} + assert result["created_at"] is not None + assert result["updated_at"] is not None + + def test_to_dict_after_state_change(self) -> None: + """状态变更后 to_dict 应反映最新状态。""" + profile = VoiceCloneProfile.create(user_id="user_001", name="测试") + profile.mark_processing() + profile.mark_ready(voice_id="voice_001") + + result = profile.to_dict() + + assert result["status"] == "ready" + assert result["voice_id"] == "voice_001" + assert result["is_ready"] is True