""" Quick Mode 任务服务 处理一键训练任务的业务逻辑 """ import re import uuid from datetime import datetime from pathlib import Path from typing import AsyncGenerator, Dict, Optional, Any, Tuple from project_config import settings from ..core.adapters import get_database_adapter, get_task_queue_adapter, get_storage_adapter from ..models.domain import Task, TaskStatus from ..models.schemas.task import ( QuickModeRequest, TaskResponse, TaskListResponse, InferenceOutputItem, InferenceOutputsResponse, ) # 质量预设配置 QUALITY_PRESETS = { "fast": { "sovits_epochs": 4, "gpt_epochs": 8, "description": "快速训练，约10分钟", }, "standard": { "sovits_epochs": 8, "gpt_epochs": 15, "description": "标准训练，约20分钟", }, "high": { "sovits_epochs": 16, "gpt_epochs": 30, "description": "高质量训练，约40分钟", }, } # 各语言的默认推理测试文本 DEFAULT_TARGET_TEXTS = { "zh": "这是一段测试语音合成的文本。请你用自然、清晰、不过度夸张的语气朗读，并在逗号和句号处做适当停顿：先慢一点，再稍微快一点，最后恢复正常语速。", "en": "This is a test text for speech synthesis. Please read it naturally and clearly, without exaggeration, pausing appropriately at commas and periods: start slowly, then speed up a bit, and finally return to normal pace.", "ja": "これは音声合成のテストテキストです。自然で明瞭に、大げさにならないように朗読してください。読点と句点で適切に間を置いて：最初はゆっくり、少し速く、最後は普通の速さに戻してください。", "ko": "이것은 음성 합성을 위한 테스트 텍스트입니다. 자연스럽고 명확하게, 과장하지 않고 읽어주세요. 쉼표와 마침표에서 적절히 멈추며: 먼저 천천히, 그 다음 조금 빠르게, 마지막으로 보통 속도로 돌아오세요.", "yue": "呢段係測試語音合成嘅文字。請你用自然、清楚、唔好太誇張嘅語氣讀出嚟，喺逗號同句號嗰度要適當噉停一停：開頭慢啲，跟住快少少，最後返返正常語速。", } class TaskService: """ Quick Mode 任务服务 提供一键训练任务的完整生命周期管理： - 创建任务 - 查询任务状态 - 取消任务 - 订阅进度更新 Example: >>> service = TaskService() >>> task = await service.create_quick_task(request) >>> status = await service.get_task(task.id) >>> await service.cancel_task(task.id) """ def __init__(self): """初始化服务""" self._db = None self._queue = None self._storage = None @property def db(self): """延迟获取数据库适配器""" if self._db is None: self._db = get_database_adapter() return self._db @property def queue(self): """延迟获取任务队列适配器""" if self._queue is None: self._queue = get_task_queue_adapter() return self._queue @property def storage(self): """延迟获取存储适配器""" if self._storage is None: self._storage = get_storage_adapter() return self._storage async def check_exp_name_exists(self, exp_name: str) -> bool: """ 检查实验名称是否已存在 Args: exp_name: 实验名称 Returns: 如果存在返回 True，否则返回 False """ existing_task = await self.db.get_task_by_exp_name(exp_name) return existing_task is not None async def validate_audio_file(self, audio_file_id: str) -> tuple[bool, str]: """ 验证音频文件是否存在 Args: audio_file_id: 音频文件 ID 或路径 Returns: (是否存在, 实际文件路径) """ import os # 尝试获取文件元数据 file_metadata = await self.storage.get_file_metadata(audio_file_id) if file_metadata: # 文件存储在 storage.base_path / file_id audio_file_path = str(self.storage.base_path / audio_file_id) exists = os.path.exists(audio_file_path) return exists, audio_file_path else: # 如果找不到元数据，将 audio_file_id 当作路径 exists = os.path.exists(audio_file_id) return exists, audio_file_id async def create_quick_task(self, request: QuickModeRequest) -> TaskResponse: """ 创建一键训练任务 根据请求参数和质量预设，自动配置训练参数并创建任务。 Args: request: 快速模式请求 Returns: TaskResponse: 任务响应 """ # 生成任务ID task_id = f"task-{uuid.uuid4().hex[:12]}" # 获取质量预设 quality = request.options.quality preset = QUALITY_PRESETS.get(quality, QUALITY_PRESETS["standard"]) # 验证并解析音频文件路径 audio_file_id = request.audio_file_id _, audio_file_path = await self.validate_audio_file(audio_file_id) # 构建阶段列表 stages = [ "audio_slice", "asr", "text_feature", "hubert_feature", "semantic_token", "sovits_train", "gpt_train", ] # 获取推理配置 inference_opts = request.options.inference inference_enabled = inference_opts is None or inference_opts.enabled # 如果启用推理，添加推理阶段 if inference_enabled: stages.append("inference") # 构建任务配置 config = { "exp_name": request.exp_name, "audio_file_id": audio_file_id, "input_path": audio_file_path, # 音频文件的实际路径 "version": request.options.version, "language": request.options.language, "quality": quality, # 训练参数 "total_epoch": preset["sovits_epochs"], # SoVITS epoch "sovits_epochs": preset["sovits_epochs"], "gpt_epochs": preset["gpt_epochs"], # 预训练模型路径 "bert_pretrained_dir": str(settings.BERT_PRETRAINED_DIR), "ssl_pretrained_dir": str(settings.SSL_PRETRAINED_DIR), "pretrained_s2G": str(settings.PRETRAINED_S2G), "pretrained_s2D": str(settings.PRETRAINED_S2D), "pretrained_s1": str(settings.PRETRAINED_S1), # 执行阶段 "stages": stages, } # 添加推理配置（如果启用） if inference_enabled: if inference_opts: config["ref_text"] = inference_opts.ref_text or "" config["ref_audio_path"] = inference_opts.ref_audio_path or "" config["target_text"] = inference_opts.target_text else: # 使用默认值 - 空字符串表示由推理阶段从 asr_opt/slicer_opt.list 文件解析 config["ref_text"] = "" config["ref_audio_path"] = "" # 根据语言选择默认的测试文本 language = request.options.language config["target_text"] = DEFAULT_TARGET_TEXTS.get(language, DEFAULT_TARGET_TEXTS["zh"]) # 创建 Task 领域模型 task = Task( id=task_id, exp_name=request.exp_name, config=config, status=TaskStatus.QUEUED, created_at=datetime.utcnow(), ) # 保存到数据库 await self.db.create_task(task) # 入队执行 job_id = await self.queue.enqueue(task_id, config) # 更新 job_id await self.db.update_task(task_id, {"job_id": job_id}) task.job_id = job_id return self._task_to_response(task) async def get_task(self, task_id: str) -> Optional[TaskResponse]: """ 获取任务详情 Args: task_id: 任务ID Returns: TaskResponse 或 None（不存在时） """ task = await self.db.get_task(task_id) if not task: return None return self._task_to_response(task) async def list_tasks( self, status: Optional[str] = None, limit: int = 50, offset: int = 0 ) -> TaskListResponse: """ 获取任务列表 Args: status: 按状态筛选 limit: 每页数量 offset: 偏移量 Returns: TaskListResponse """ tasks = await self.db.list_tasks(status=status, limit=limit, offset=offset) total = await self.db.count_tasks(status=status) return TaskListResponse( items=[self._task_to_response(t) for t in tasks], total=total, limit=limit, offset=offset, ) async def cancel_task(self, task_id: str) -> bool: """ 取消任务 Args: task_id: 任务ID Returns: 是否成功取消 """ # 获取任务 task = await self.db.get_task(task_id) if not task: return False # 只有排队中或运行中的任务可以取消 if task.status not in (TaskStatus.QUEUED, TaskStatus.RUNNING): return False # 如果有 job_id，尝试取消队列任务 if task.job_id: await self.queue.cancel(task.job_id) # 更新状态 await self.db.update_task(task_id, { "status": TaskStatus.CANCELLED, "completed_at": datetime.utcnow(), "message": "任务已取消", }) return True async def subscribe_progress( self, task_id: str ) -> AsyncGenerator[Dict[str, Any], None]: """ 订阅任务进度（SSE 流） Args: task_id: 任务ID Yields: 进度信息字典 """ # 检查任务是否存在 task = await self.db.get_task(task_id) if not task: yield {"type": "error", "message": "任务不存在"} return # 如果任务已结束，直接返回最终状态 if task.status in (TaskStatus.COMPLETED, TaskStatus.FAILED, TaskStatus.CANCELLED): yield { "type": "final", "status": task.status.value, "message": task.message or task.error_message, "progress": task.progress, } return # 订阅进度更新 async for progress in self.queue.subscribe_progress(task_id): yield progress # 检查是否为终态 if progress.get("status") in ("completed", "failed", "cancelled"): break async def get_inference_outputs(self, task_id: str) -> Optional[InferenceOutputsResponse]: """ 获取任务的推理输出列表 扫描 logs/{exp_name}/inference/ 目录，返回所有推理生成的音频文件元信息。 Args: task_id: 任务 ID Returns: InferenceOutputsResponse 或 None（任务不存在时） """ # 获取任务 task = await self.db.get_task(task_id) if not task: return None exp_name = task.exp_name inference_dir = Path(settings.EXP_ROOT) / exp_name / "inference" # 从 task.config 获取推理配置 ref_text = task.config.get("ref_text", "") ref_audio_path = task.config.get("ref_audio_path", "") target_text = task.config.get("target_text", "") # 如果为空，从 .list 文件解析 if not ref_text or not ref_audio_path: parsed_audio, parsed_text = self._parse_list_file(exp_name) ref_audio_path = parsed_audio ref_text = parsed_text # 获取模型版本以确定权重目录 version = task.config.get("version", "v2") gpt_weight_dir = self._get_gpt_weight_dir(version) sovits_weight_dir = self._get_sovits_weight_dir(version) outputs = [] if inference_dir.exists() and inference_dir.is_dir(): # 扫描目录中的所有 .wav 文件 for file_path in inference_dir.glob("*.wav"): filename = file_path.name # 解析文件名获取模型信息 # 格式: {exp_name}_gpt-{gpt_name}_sovits-{sovits_name}.wav gpt_model, sovits_model = self._parse_inference_filename(filename, exp_name) # 构建模型完整路径 gpt_path = str(Path(settings.EXP_ROOT) / exp_name / gpt_weight_dir / f"{gpt_model}.ckpt") sovits_path = str(Path(settings.EXP_ROOT) / exp_name / sovits_weight_dir / f"{sovits_model}.pth") # 获取文件信息 stat = file_path.stat() outputs.append(InferenceOutputItem( filename=filename, gpt_model=gpt_model, sovits_model=sovits_model, gpt_path=gpt_path, sovits_path=sovits_path, file_path=str(file_path.relative_to(settings.PROJECT_ROOT)), size_bytes=stat.st_size, created_at=datetime.fromtimestamp(stat.st_ctime), )) return InferenceOutputsResponse( task_id=task_id, exp_name=exp_name, ref_text=ref_text, ref_audio_path=ref_audio_path, target_text=target_text, outputs=outputs, total=len(outputs), ) async def download_inference_output( self, task_id: str, filename: str ) -> Optional[Tuple[bytes, str, str]]: """ 下载指定的推理输出文件 Args: task_id: 任务 ID filename: 文件名 Returns: (文件内容, 文件名, content_type) 或 None（不存在时） """ # 获取任务 task = await self.db.get_task(task_id) if not task: return None exp_name = task.exp_name file_path = Path(settings.EXP_ROOT) / exp_name / "inference" / filename # 安全检查：确保文件在预期目录内 try: file_path = file_path.resolve() expected_parent = (Path(settings.EXP_ROOT) / exp_name / "inference").resolve() if not str(file_path).startswith(str(expected_parent)): return None except (ValueError, OSError): return None if not file_path.exists() or not file_path.is_file(): return None # 读取文件内容 with open(file_path, "rb") as f: file_data = f.read() return file_data, filename, "audio/wav" async def download_file( self, task_id: str, file_type: str, filename: str ) -> Optional[Tuple[bytes, str, str]]: """ 下载指定类型的文件 Args: task_id: 任务 ID file_type: 文件类型 (output/ref_audio/gpt_model/sovits_model) filename: 文件名 Returns: (文件内容, 文件名, content_type) 或 None（不存在时） """ # 获取任务 task = await self.db.get_task(task_id) if not task: return None exp_name = task.exp_name version = task.config.get("version", "v2") # 根据文件类型确定路径和 content_type if file_type == "output": file_path = Path(settings.EXP_ROOT) / exp_name / "inference" / filename content_type = "audio/wav" elif file_type == "ref_audio": # ref_audio 使用完整路径（filename 参数实际上是完整路径） file_path = Path(filename) content_type = "audio/wav" elif file_type == "gpt_model": gpt_dir = self._get_gpt_weight_dir(version) file_path = Path(settings.EXP_ROOT) / exp_name / gpt_dir / filename content_type = "application/octet-stream" elif file_type == "sovits_model": sovits_dir = self._get_sovits_weight_dir(version) file_path = Path(settings.EXP_ROOT) / exp_name / sovits_dir / filename content_type = "application/octet-stream" else: return None # 安全检查：确保文件路径有效 try: file_path = file_path.resolve() except (ValueError, OSError): return None if not file_path.exists() or not file_path.is_file(): return None # 读取文件内容 with open(file_path, "rb") as f: file_data = f.read() # 使用文件名（不含路径）作为下载文件名 download_filename = file_path.name return file_data, download_filename, content_type def _parse_inference_filename(self, filename: str, exp_name: str) -> Tuple[str, str]: """ 解析推理输出文件名，提取 GPT 和 SoVITS 模型名称 文件名格式: {exp_name}_gpt-{gpt_name}_sovits-{sovits_name}.wav Args: filename: 文件名 exp_name: 实验名称 Returns: (gpt_model, sovits_model) """ # 移除扩展名 name = filename.rsplit(".", 1)[0] if "." in filename else filename # 尝试解析模型名称 # 格式: {exp_name}_gpt-{gpt_name}_sovits-{sovits_name} pattern = rf"^{re.escape(exp_name)}_gpt-(.+)_sovits-(.+)$" match = re.match(pattern, name) if match: return match.group(1), match.group(2) # 备用解析：尝试匹配 gpt- 和 sovits- 部分 gpt_match = re.search(r"gpt-([^_]+)", name) sovits_match = re.search(r"sovits-([^_]+)", name) gpt_model = gpt_match.group(1) if gpt_match else "unknown" sovits_model = sovits_match.group(1) if sovits_match else "unknown" return gpt_model, sovits_model def _parse_list_file(self, exp_name: str) -> Tuple[str, str]: """ 从 asr_opt/slicer_opt.list 解析第一行获取 ref_audio_path 和 ref_text Args: exp_name: 实验名称 Returns: (ref_audio_path, ref_text) 元组，解析失败返回空字符串 """ list_path = Path(settings.EXP_ROOT) / exp_name / 'asr_opt' / 'slicer_opt.list' if not list_path.exists(): return "", "" with open(list_path, 'r', encoding='utf-8') as f: first_line = f.readline().strip() if not first_line: return "", "" # 格式: {音频路径}|{文件夹名}|{语言}|{识别文本} parts = first_line.split('|') if len(parts) >= 4: return parts[0], parts[3] return "", "" def _get_gpt_weight_dir(self, version: str) -> str: """根据模型版本获取 GPT 权重目录名""" version_to_dir = { "v1": "GPT_weights", "v2": "GPT_weights_v2", "v3": "GPT_weights_v3", "v4": "GPT_weights_v4", "v2Pro": "GPT_weights_v2Pro", "v2ProPlus": "GPT_weights_v2ProPlus", } return version_to_dir.get(version, "GPT_weights_v2") def _get_sovits_weight_dir(self, version: str) -> str: """根据模型版本获取 SoVITS 权重目录名""" version_to_dir = { "v1": "SoVITS_weights", "v2": "SoVITS_weights_v2", "v3": "SoVITS_weights_v3", "v4": "SoVITS_weights_v4", "v2Pro": "SoVITS_weights_v2Pro", "v2ProPlus": "SoVITS_weights_v2ProPlus", } return version_to_dir.get(version, "SoVITS_weights_v2") def _task_to_response(self, task: Task) -> TaskResponse: """将 Task 领域模型转换为 TaskResponse""" return TaskResponse( id=task.id, exp_name=task.exp_name, status=task.status.value if isinstance(task.status, TaskStatus) else task.status, current_stage=task.current_stage, progress=task.stage_progress, overall_progress=task.progress, message=task.message, error_message=task.error_message, created_at=task.created_at, started_at=task.started_at, completed_at=task.completed_at, )