diff --git a/apps/api/app/api/routes/tts.py b/apps/api/app/api/routes/tts.py index 38ba5ccd1..0958763a2 100644 --- a/apps/api/app/api/routes/tts.py +++ b/apps/api/app/api/routes/tts.py @@ -130,18 +130,25 @@ def synthesize( # 若任务处于 processing 状态（异步模式），触发 Celery 后台轮询 if job.status.value == "processing": - task_id = (job.metadata or {}).get("cosyvoice_task_id", "") - if task_id: - try: + # 分段合成任务 vs 普通单段任务 + segment_task_ids = (job.metadata or {}).get("segment_task_ids", []) + is_segment = len(segment_task_ids) > 0 + + try: + if is_segment: + from worker_app.tasks import process_tts_segment_synthesis + + process_tts_segment_synthesis.delay(job.id) + else: from worker_app.tasks import process_tts_synthesis process_tts_synthesis.delay(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}") + 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, diff --git a/apps/worker/worker_app/tasks/__init__.py b/apps/worker/worker_app/tasks/__init__.py index cf7721f22..7c9c315a4 100755 --- a/apps/worker/worker_app/tasks/__init__.py +++ b/apps/worker/worker_app/tasks/__init__.py @@ -41,6 +41,10 @@ def __getattr__(name: str): from .tts_synthesis import process_tts_synthesis return process_tts_synthesis + elif name == "process_tts_segment_synthesis": + from .tts_synthesis import process_tts_segment_synthesis + + return process_tts_segment_synthesis elif name == "run_ai_recommend": from .ai_tasks import run_ai_recommend @@ -62,6 +66,7 @@ __all__ = [ "extract_background_task", "process_voice_clone", "process_tts_synthesis", + "process_tts_segment_synthesis", "run_ai_recommend", "run_generate_cover", ] diff --git a/apps/worker/worker_app/tasks/tts_synthesis.py b/apps/worker/worker_app/tasks/tts_synthesis.py index e8f5d5e1f..4fad944a4 100644 --- a/apps/worker/worker_app/tasks/tts_synthesis.py +++ b/apps/worker/worker_app/tasks/tts_synthesis.py @@ -104,3 +104,77 @@ def process_tts_synthesis(self: Task, job_id: str) -> dict: finally: if session is not None: session.close() + + +@celery_app.task(bind=True, max_retries=2, name="worker.process_tts_segment_synthesis") +def process_tts_segment_synthesis(self: Task, job_id: str) -> dict: + """分段合成轮询任务 — 轮询多个 CosyVoice 子任务并合并音频。 + + 与 process_tts_synthesis 类似，但超时更长（300s）， + 因为分段任务需要等待所有子任务完成。 + """ + 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=300) + session.commit() + + logger.info(f"TTS segment 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: + raise + + except CosyVoiceTimeoutError as e: + logger.warning(f"TTS segment synthesis timeout for {job_id}: {e}") + if session is not None: + session.rollback() + raise self.retry(exc=e, countdown=60) + + except CosyVoiceError as e: + logger.error(f"TTS segment synthesis failed for {job_id}: {e}") + if session is not None: + session.rollback() + 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 segment synthesis unexpected error for {job_id}: {e}") + if session is not None: + session.rollback() + 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/packages/application/tts_job/audio_merger.py b/packages/application/tts_job/audio_merger.py new file mode 100644 index 000000000..0d583f821 --- /dev/null +++ b/packages/application/tts_job/audio_merger.py @@ -0,0 +1,96 @@ +"""FFmpeg 音频合并器 — P1 长文本分段合成。 + +将多个分段音频文件合并为一个完整音频文件。 +""" + +from __future__ import annotations + +import logging +import os +import shutil +import subprocess +import tempfile + +logger = logging.getLogger(__name__) + + +class AudioMergeError(Exception): + """音频合并异常。""" + + pass + + +class AudioMerger: + """使用 FFmpeg 合并多个音频文件。""" + + def merge(self, audio_paths: list[str], output_format: str = "mp3") -> bytes: + """合并多个音频文件，返回合并后的音频数据。 + + 使用 FFmpeg concat demuxer 按顺序拼接音频。 + 所有输入文件必须为相同格式和采样率。 + + Args: + audio_paths: 音频文件路径列表（按合成顺序） + output_format: 输出格式（mp3/wav/pcm） + + Returns: + 合并后的音频文件字节数据 + + Raises: + AudioMergeError: 合并失败 + """ + if not audio_paths: + raise AudioMergeError("没有可合并的音频文件") + + if len(audio_paths) == 1: + with open(audio_paths[0], "rb") as f: + return f.read() + + temp_dir = tempfile.mkdtemp(prefix="tts_merge_") + try: + # 生成 concat demuxer 列表文件 + list_path = os.path.join(temp_dir, "concat_list.txt") + with open(list_path, "w") as f: + for path in audio_paths: + # FFmpeg concat 文件需要 file: 前缀，路径中的 ' 和 \n 需转义 + escaped = path.replace("'", "'\\''").replace("\n", "\\n") + f.write(f"file '{escaped}'\n") + + output_path = os.path.join(temp_dir, f"merged.{output_format}") + + cmd = [ + "ffmpeg", + "-y", + "-f", + "concat", + "-safe", + "0", + "-i", + list_path, + "-c", + "copy", + output_path, + ] + + result = subprocess.run( + cmd, + capture_output=True, + text=True, + timeout=120, + ) + + if result.returncode != 0: + logger.error(f"FFmpeg 合并失败: stderr={result.stderr}") + raise AudioMergeError(f"FFmpeg 合并失败: {result.stderr[:500]}") + + with open(output_path, "rb") as f: + return f.read() + + except subprocess.TimeoutExpired: + raise AudioMergeError("FFmpeg 合并超时（120 秒）") + except AudioMergeError: + raise + except Exception as e: + raise AudioMergeError(f"音频合并失败: {e}") + finally: + shutil.rmtree(temp_dir, ignore_errors=True) diff --git a/packages/application/tts_job/text_splitter.py b/packages/application/tts_job/text_splitter.py new file mode 100644 index 000000000..a34135bb1 --- /dev/null +++ b/packages/application/tts_job/text_splitter.py @@ -0,0 +1,70 @@ +"""长文本分段工具 — P1 长文本分段合成。 + +将超过阈值的文本按句子边界分段，供 CosyVoice 并发合成后合并。 +""" + +from __future__ import annotations + +# 中文句子结束符（含全角/半角） +_SENTENCE_ENDS = frozenset("。！？；\n.!?;") + + +def split_text(text: str, max_chars: int = 500) -> list[str]: + """将文本分段，每段不超过 max_chars 个字符。 + + 优先在句子边界（句号、问号、感叹号、换行符）处分段。 + 若单个句子超过 max_chars，则在逗号等次级标点处拆分。 + 若仍超长，则硬切。 + + Args: + text: 待分段文本 + max_chars: 每段最大字符数 + + Returns: + 分段列表，每段 ≤ max_chars。文本为空时返回空列表。 + """ + text = text.strip() + if not text: + return [] + if len(text) <= max_chars: + return [text] + + segments: list[str] = [] + current = "" + + for char in text: + current += char + if char in _SENTENCE_ENDS and len(current) >= 50: + # 句子边界且长度合理，切段 + segments.append(current.strip()) + current = "" + elif len(current) >= max_chars: + # 达到上限，强制切段 + segments.append(current.strip()) + current = "" + + if current.strip(): + segments.append(current.strip()) + + # 合并过短的段（< 50 字符且不是最后一段），减少 API 调用次数 + merged: list[str] = [] + buffer = "" + for seg in segments: + if buffer: + combined = buffer + seg + if len(combined) <= max_chars: + buffer = combined + continue + merged.append(buffer) + buffer = "" + if len(seg) < 50: + buffer = seg + else: + merged.append(seg) + if buffer: + if merged and len(merged[-1]) + len(buffer) <= max_chars: + merged[-1] = merged[-1] + buffer + else: + merged.append(buffer) + + return [s for s in merged if s] diff --git a/packages/application/tts_job/workflow.py b/packages/application/tts_job/workflow.py index 4bdcab63d..1b6e9c36e 100644 --- a/packages/application/tts_job/workflow.py +++ b/packages/application/tts_job/workflow.py @@ -11,6 +11,11 @@ from __future__ import annotations import io import logging +import os +import shutil +import tempfile +import time +from concurrent.futures import ThreadPoolExecutor, as_completed from typing import Optional import httpx @@ -20,12 +25,19 @@ from packages.application.cosyvoice_service import ( CosyVoiceError, CosyVoiceService, ) +from packages.application.tts_job.audio_merger import AudioMergeError, AudioMerger +from packages.application.tts_job.text_splitter import split_text from packages.domain.tts_job import TTSJob, TTSJobStatus from packages.ports.tts_job_repository import TTSJobRepository from packages.shared.storage import SharedStorageService, get_shared_storage_service logger = logging.getLogger(__name__) +# 长文本分段阈值：超过此字符数自动分段合成 +_SEGMENT_THRESHOLD = 500 +# 分段并发上限 +_MAX_SEGMENT_WORKERS = 5 + class TTSWorkflowError(Exception): """TTS 合成工作流异常。""" @@ -130,6 +142,10 @@ class TTSWorkflowService: job.mark_processing() job = self.repository.update(job) + # 长文本自动分段合成 + if len(job.input_text) > _SEGMENT_THRESHOLD: + return self._start_segment_synthesis(job) + try: submit_result = self.cosyvoice_service.submit_synthesize_task( text=job.input_text, @@ -185,6 +201,11 @@ class TTSWorkflowService: if job is None: raise TTSJobNotFoundError(f"TTS job {job_id} not found") + # 检查是否为分段合成任务 + segment_task_ids = (job.metadata or {}).get("segment_task_ids", []) + if segment_task_ids: + return self._poll_segment_tasks(job) + 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") @@ -257,3 +278,224 @@ class TTSWorkflowService: job = self.repository.update(job) logger.error(f"TTS 合成失败: job_id={job_id}, error={error_message}") return job + + # ── P1: 长文本分段合成 ───────────────────────────────────── + + def _upload_merged_to_oss( + self, merged_data: bytes, user_id: str, job_id: str, audio_format: str + ) -> tuple[str, str]: + """上传合并后的音频数据到 OSS。 + + Returns: + (permanent_url, storage_key) 元组。 + 上传失败时返回 ("", "")。 + """ + storage_key = f"tts-outputs/{user_id}/{job_id}.{audio_format}" + content_type_map = { + "mp3": "audio/mpeg", + "wav": "audio/wav", + "pcm": "audio/pcm", + "opus": "audio/opus", + } + content_type = content_type_map.get(audio_format, "application/octet-stream") + try: + file_obj = io.BytesIO(merged_data) + permanent_url = self._storage.upload_file(file_obj, storage_key, content_type=content_type) + return permanent_url, storage_key + except Exception as e: + logger.warning(f"分段合并音频转存 OSS 失败: job_id={job_id}, error={e}") + return "", "" + + def _start_segment_synthesis(self, job: TTSJob) -> TTSJob: + """长文本分段合成入口。 + + 将文本分段后并发提交到 CosyVoice，根据同步/异步结果走不同路径。 + """ + segments = split_text(job.input_text, max_chars=_SEGMENT_THRESHOLD) + logger.info(f"长文本分段合成: job_id={job.id}, " f"原文={len(job.input_text)}字, 段数={len(segments)}") + + # 记录分段信息到 metadata + job_metadata = dict(job.metadata) + job_metadata["segment_count"] = len(segments) + + # 并发提交所有分段 + results = self._submit_segments_concurrent(segments, job) + if results is None: + # 提交阶段已失败，_submit_segments_concurrent 内部已标记 failed + return self.repository.get(job.id) + + # 判断同步还是异步 + has_audio_urls = any(r.get("audio_url", "") for r in results) + has_task_ids = any(r.get("task_id", "") for r in results) + + if has_audio_urls and not has_task_ids: + # 所有分段同步返回音频，直接合并 + return self._process_segments_sync(job, results) + + # 异步路径：保存各分段的 task_id 供后续轮询 + segment_task_ids = [r.get("task_id", "") for r in results] + segment_audio_urls = [r.get("audio_url", "") for r in results] + job_metadata["segment_task_ids"] = segment_task_ids + job_metadata["segment_audio_urls"] = segment_audio_urls + job_metadata["segment_format"] = job.format + + job.metadata = job_metadata + job = self.repository.update(job) + logger.info(f"分段合成任务已提交（异步）: job_id={job.id}, " f"段数={len(segments)}") + return job + + def _submit_segments_concurrent(self, segments: list[str], job: TTSJob) -> list[dict] | None: + """并发提交分段合成任务。 + + Returns: + 各分段的结果列表（保持顺序），提交失败时返回 None。 + """ + max_workers = min(len(segments), _MAX_SEGMENT_WORKERS) + results: list[dict | None] = [None] * len(segments) + + with ThreadPoolExecutor(max_workers=max_workers) as executor: + future_to_idx = {} + for idx, segment_text in enumerate(segments): + future = executor.submit( + self.cosyvoice_service.submit_synthesize_task, + text=segment_text, + voice_id=job.voice_id, + sample_rate=job.sample_rate, + format=job.format, + ) + future_to_idx[future] = idx + + for future in as_completed(future_to_idx): + idx = future_to_idx[future] + try: + results[idx] = future.result() + except Exception as e: + logger.error(f"分段合成提交失败: job_id={job.id}, " f"segment={idx}, error={e}") + self._handle_segment_failure(job, f"分段 {idx + 1} 合成提交失败: {e}") + return None + + return results # type: ignore[return-value] + + def _process_segments_sync(self, job: TTSJob, results: list[dict]) -> TTSJob: + """同步路径：所有分段已返回 audio_url，下载合并后转存 OSS。""" + merged_data, total_duration = self._download_and_merge_segments(results, job) + + # 直接上传合并后的音频 bytes 到 OSS + permanent_url, storage_key = self._upload_merged_to_oss(merged_data, job.user_id, job.id, job.format) + + job.mark_completed( + output_audio_url=permanent_url, + output_audio_key=storage_key, + duration=total_duration, + file_size=len(merged_data), + ) + job = self.repository.update(job) + logger.info(f"分段合成完成: job_id={job.id}, " f"merged_size={len(merged_data)}, duration={total_duration:.1f}") + return job + + def _download_and_merge_segments(self, results: list[dict], job: TTSJob) -> tuple[bytes, float]: + """下载各分段音频并合并。 + + Returns: + (merged_audio_bytes, total_duration) + """ + temp_dir = tempfile.mkdtemp(prefix="tts_segments_") + try: + audio_paths: list[str] = [] + total_duration = 0.0 + + for idx, result in enumerate(results): + audio_url = result.get("audio_url", "") + if not audio_url: + raise TTSWorkflowError(f"分段 {idx + 1} 没有返回 audio_url") + + total_duration += result.get("duration", 0.0) + + # 下载分段音频到临时文件 + resp = httpx.get(audio_url, timeout=60.0, follow_redirects=True) + resp.raise_for_status() + + seg_path = os.path.join(temp_dir, f"seg_{idx:03d}.{job.format}") + with open(seg_path, "wb") as f: + f.write(resp.content) + audio_paths.append(seg_path) + + # 合并 + merger = AudioMerger() + merged_data = merger.merge(audio_paths, output_format=job.format) + return merged_data, total_duration + + finally: + shutil.rmtree(temp_dir, ignore_errors=True) + + def _poll_segment_tasks(self, job: TTSJob) -> TTSJob: + """轮询所有分段异步任务，全部完成后合并音频。""" + segment_task_ids: list[str] = (job.metadata or {}).get("segment_task_ids", []) + segment_audio_urls: list[str] = (job.metadata or {}).get("segment_audio_urls", []) + segment_count = len(segment_task_ids) + + poll_start = time.monotonic() + poll_timeout = 300.0 # 分段任务超时更长 + poll_interval = 2.0 + + while time.monotonic() - poll_start < poll_timeout: + all_done = True + results: list[dict | None] = [None] * segment_count + + for idx, task_id in enumerate(segment_task_ids): + # 已经有音频的分段跳过轮询 + if idx < len(segment_audio_urls) and segment_audio_urls[idx]: + results[idx] = { + "audio_url": segment_audio_urls[idx], + "duration": 0.0, + "file_size": 0, + } + continue + + try: + result = self.cosyvoice_service.poll_synthesize_task(task_id, timeout=poll_timeout) + results[idx] = result + except Exception as e: + logger.error(f"分段任务轮询失败: job_id={job.id}, " f"segment={idx}, error={e}") + self._handle_segment_failure(job, f"分段 {idx + 1} 轮询失败: {e}") + return self.repository.get(job.id) + + if results[idx] is None: + all_done = False + + if all_done and all(r is not None for r in results): + # 所有分段完成，下载合并 + try: + merged_data, total_duration = self._download_and_merge_segments(results, job) + + # 转存 OSS + permanent_url, storage_key = self._upload_merged_to_oss( + merged_data, job.user_id, job.id, job.format + ) + + job.mark_completed( + output_audio_url=permanent_url, + output_audio_key=storage_key, + duration=total_duration, + file_size=len(merged_data), + ) + job = self.repository.update(job) + logger.info(f"分段合成轮询完成: job_id={job.id}, " f"merged_size={len(merged_data)}") + return job + + except Exception as e: + self._handle_segment_failure(job, f"分段合并失败: {e}") + return self.repository.get(job.id) + + # 等待后重试 + time.sleep(poll_interval) + + # 超时 + self._handle_segment_failure(job, "分段合成轮询超时（300 秒）") + return self.repository.get(job.id) + + def _handle_segment_failure(self, job: TTSJob, error_message: str) -> None: + """分段合成失败处理。""" + job.mark_failed(error_message) + self.repository.update(job) + logger.error(f"分段合成失败: job_id={job.id}, error={error_message}") diff --git a/tests/unit/test_tts_segment_synthesis.py b/tests/unit/test_tts_segment_synthesis.py new file mode 100644 index 000000000..f903b5d51 --- /dev/null +++ b/tests/unit/test_tts_segment_synthesis.py @@ -0,0 +1,485 @@ +"""P1 长文本分段合成单元测试。 + +覆盖： +- text_splitter.split_text 分段逻辑 +- audio_merger.AudioMerger 合并逻辑 +- workflow 分段合成路径（同步 / 异步 / 失败） +""" + +from __future__ import annotations + +import os +import tempfile +from datetime import datetime, timezone +from unittest.mock import MagicMock, call, patch + +import pytest + +from packages.application.cosyvoice_service import CosyVoiceError, CosyVoiceService +from packages.application.tts_job.audio_merger import AudioMergeError, AudioMerger +from packages.application.tts_job.text_splitter import split_text +from packages.application.tts_job.workflow import TTSWorkflowService +from packages.domain.tts_job import TTSJob, TTSJobStatus + +# ── text_splitter ───────────────────────────────────────────── + + +class TestSplitText: + """测试文本分段工具。""" + + def test_short_text_no_split(self) -> None: + """短文本不拆分。""" + assert split_text("你好世界", max_chars=500) == ["你好世界"] + + def test_empty_text(self) -> None: + """空文本返回空列表。""" + assert split_text("") == [] + assert split_text(" ") == [] + + def test_exact_threshold(self) -> None: + """恰好等于阈值不拆分。""" + text = "a" * 500 + assert split_text(text, max_chars=500) == [text] + + def test_split_at_sentence_boundary(self) -> None: + """在句子边界处分段。""" + text = "第一句话。" * 60 # 300 chars + text += "第二句话。" * 60 # 300 chars → total 600 + segments = split_text(text, max_chars=500) + assert len(segments) >= 2 + for seg in segments: + assert len(seg) <= 500 + + def test_split_at_newline(self) -> None: + """在换行符处分段。""" + text = "段落一\n" * 100 # 300 chars + text += "段落二\n" * 100 # 300 chars + segments = split_text(text, max_chars=500) + assert len(segments) >= 2 + + def test_long_sentence_hard_split(self) -> None: + """超长句子硬切。""" + text = "a" * 1200 + segments = split_text(text, max_chars=500) + assert len(segments) >= 3 + for seg in segments: + assert len(seg) <= 500 + + def test_merge_short_segments(self) -> None: + """短段合并减少 API 调用。""" + # 多个短句子应该被合并 + text = "你好。" * 120 # 360 chars, each sentence 3 chars + segments = split_text(text, max_chars=500) + # 短段应该被合并，段数应该比较少 + assert len(segments) < 120 + + def test_preserves_order(self) -> None: + """分段保持原始顺序。""" + text = "第一段。第二段。第三段。" + "x" * 490 + segments = split_text(text, max_chars=500) + # 第一个段应该以 "第一段" 开头 + assert segments[0].startswith("第一段") + + +# ── audio_merger ───────────────────────────────────────────── + + +class TestAudioMerger: + """测试 FFmpeg 音频合并器。""" + + def test_empty_list_raises(self) -> None: + """空列表抛出 AudioMergeError。""" + merger = AudioMerger() + with pytest.raises(AudioMergeError, match="没有可合并"): + merger.merge([]) + + def test_single_file_returns_bytes(self) -> None: + """单文件直接返回内容。""" + with tempfile.NamedTemporaryFile(suffix=".mp3", delete=False) as f: + f.write(b"fake audio content") + f.flush() + path = f.name + + try: + merger = AudioMerger() + data = merger.merge([path]) + assert data == b"fake audio content" + finally: + os.unlink(path) + + @patch("packages.application.tts_job.audio_merger.subprocess.run") + def test_ffmpeg_called_correctly(self, mock_run: MagicMock) -> None: + """多文件调用 FFmpeg concat。""" + mock_run.return_value = MagicMock(returncode=0) + + # 创建临时文件 + paths = [] + for i in range(3): + with tempfile.NamedTemporaryFile(suffix=".mp3", delete=False) as f: + f.write(b"audio") + paths.append(f.name) + + try: + merger = AudioMerger() + # Mock open for reading the merged output + with patch("builtins.open", create=True) as mock_open: + mock_open.return_value.__enter__ = lambda s: s + mock_open.return_value.read = lambda: b"merged audio" + try: + merger.merge(paths, output_format="mp3") + except (FileNotFoundError, OSError): + pass # Expected since we're mocking + + # 验证 FFmpeg 被调用 + mock_run.assert_called_once() + cmd = mock_run.call_args[0][0] + assert cmd[0] == "ffmpeg" + assert "-f" in cmd + assert "concat" in cmd + finally: + for p in paths: + os.unlink(p) + + @patch("packages.application.tts_job.audio_merger.subprocess.run") + def test_ffmpeg_failure_raises(self, mock_run: MagicMock) -> None: + """FFmpeg 失败抛出 AudioMergeError。""" + mock_run.return_value = MagicMock(returncode=1, stderr="error details") + + paths = [] + for i in range(2): + with tempfile.NamedTemporaryFile(suffix=".mp3", delete=False) as f: + f.write(b"audio") + paths.append(f.name) + + try: + merger = AudioMerger() + with pytest.raises(AudioMergeError, match="FFmpeg 合并失败"): + merger.merge(paths) + finally: + for p in paths: + os.unlink(p) + + +# ── workflow segment methods ───────────────────────────────── + + +def _make_job(**kwargs) -> TTSJob: + defaults = { + "id": "test_job_seg", + "user_id": "user_001", + "input_text": "x" * 600, # > 500 threshold + "voice_id": "voice_001", + "voice_model": "", + "project_id": "", + "voice_clone_profile_id": "", + "status": TTSJobStatus.PENDING, + "output_audio_url": "", + "output_audio_key": "", + "duration": 0.0, + "file_size": 0, + "sample_rate": 22050, + "format": "mp3", + "error_message": "", + "retry_count": 0, + "max_retries": 3, + "metadata": {}, + "started_at": None, + "completed_at": None, + "created_at": datetime.now(timezone.utc), + "updated_at": datetime.now(timezone.utc), + } + defaults.update(kwargs) + return TTSJob(**defaults) + + +def _make_workflow( + cosyvoice_service: MagicMock | None = None, + repo: MagicMock | None = None, + storage: MagicMock | None = None, +) -> TTSWorkflowService: + if cosyvoice_service is None: + cosyvoice_service = MagicMock(spec=CosyVoiceService) + if repo is None: + repo = MagicMock() + repo.get.return_value = _make_job() + repo.update.side_effect = lambda j: j + if storage is None: + storage = MagicMock() + storage.upload_file.return_value = "https://oss.example.com/merged.mp3" + return TTSWorkflowService( + repository=repo, + cosyvoice_service=cosyvoice_service, + storage_service=storage, + ) + + +class TestStartSegmentSynthesis: + """测试 _start_segment_synthesis 分段合成入口。""" + + def test_short_text_no_segment(self) -> None: + """短文本不触发分段。""" + service = MagicMock(spec=CosyVoiceService) + service.submit_synthesize_task.return_value = { + "task_id": "task_1", + "audio_url": "", + } + repo = MagicMock() + repo.get.return_value = _make_job(input_text="短文本") + repo.update.side_effect = lambda j: j + + workflow = _make_workflow(cosyvoice_service=service, repo=repo) + job = workflow.start_synthesis("test_job_seg") + + # 短文本走普通路径，不调用分段 + assert job.status == TTSJobStatus.PROCESSING + + @patch("packages.application.tts_job.workflow.httpx") + def test_long_text_sync_segments(self, mock_httpx: MagicMock) -> None: + """长文本同步分段：所有段立即返回 audio_url，直接合并。""" + # Mock 分段音频下载 + mock_resp = MagicMock() + mock_resp.content = b"segment audio" + mock_resp.raise_for_status.return_value = None + mock_httpx.get.return_value = mock_resp + + service = MagicMock(spec=CosyVoiceService) + # 每个分段都同步返回 audio_url + service.submit_synthesize_task.side_effect = [ + {"task_id": "", "audio_url": "https://temp.com/seg1.mp3", "duration": 2.0, "file_size": 1000}, + {"task_id": "", "audio_url": "https://temp.com/seg2.mp3", "duration": 3.0, "file_size": 1500}, + ] + + storage = MagicMock() + storage.upload_file.return_value = "https://oss.example.com/merged.mp3" + + repo = MagicMock() + repo.get.return_value = _make_job() + repo.update.side_effect = lambda j: j + + workflow = _make_workflow(cosyvoice_service=service, repo=repo, storage=storage) + + # Mock AudioMerger 避免真实 FFmpeg 调用 + with patch("packages.application.tts_job.workflow.AudioMerger") as MockMerger: + mock_merger = MagicMock() + mock_merger.merge.return_value = b"merged audio data" + MockMerger.return_value = mock_merger + + job = workflow.start_synthesis("test_job_seg") + + assert job.status == TTSJobStatus.COMPLETED + assert job.output_audio_url == "https://oss.example.com/merged.mp3" + assert job.duration == 5.0 # 2.0 + 3.0 + + def test_long_text_async_segments(self) -> None: + """长文本异步分段：返回 task_id，存入 metadata。""" + service = MagicMock(spec=CosyVoiceService) + # 每个分段返回 task_id（异步） + service.submit_synthesize_task.side_effect = [ + {"task_id": "seg_task_1", "audio_url": "", "duration": 0.0, "file_size": 0}, + {"task_id": "seg_task_2", "audio_url": "", "duration": 0.0, "file_size": 0}, + ] + + repo = MagicMock() + repo.get.return_value = _make_job() + repo.update.side_effect = lambda j: j + + workflow = _make_workflow(cosyvoice_service=service, repo=repo) + job = workflow.start_synthesis("test_job_seg") + + assert job.status == TTSJobStatus.PROCESSING + assert "segment_task_ids" in job.metadata + assert job.metadata["segment_task_ids"] == ["seg_task_1", "seg_task_2"] + + def test_segment_submit_failure_marks_failed(self) -> None: + """分段提交失败时标记 job 为 failed。""" + service = MagicMock(spec=CosyVoiceService) + service.submit_synthesize_task.side_effect = CosyVoiceError("API error") + + repo = MagicMock() + repo.get.return_value = _make_job() + repo.update.side_effect = lambda j: j + + workflow = _make_workflow(cosyvoice_service=service, repo=repo) + job = workflow.start_synthesis("test_job_seg") + + assert job.status == TTSJobStatus.FAILED + + +class TestUploadMergedToOSS: + """测试 _upload_merged_to_oss 辅助方法。""" + + def test_success(self) -> None: + """成功上传返回 URL 和 key。""" + storage = MagicMock() + storage.upload_file.return_value = "https://oss.example.com/merged.mp3" + + workflow = _make_workflow(storage=storage) + url, key = workflow._upload_merged_to_oss(b"audio data", "user_001", "job_001", "mp3") + + assert url == "https://oss.example.com/merged.mp3" + assert key == "tts-outputs/user_001/job_001.mp3" + storage.upload_file.assert_called_once() + call_args = storage.upload_file.call_args + assert call_args[1]["content_type"] == "audio/mpeg" + + def test_failure_returns_empty(self) -> None: + """上传失败返回空字符串。""" + storage = MagicMock() + storage.upload_file.side_effect = Exception("OSS error") + + workflow = _make_workflow(storage=storage) + url, key = workflow._upload_merged_to_oss(b"audio data", "user_001", "job_001", "mp3") + + assert url == "" + assert key == "" + + +class TestHandleSegmentFailure: + """测试 _handle_segment_failure。""" + + def test_marks_job_failed(self) -> None: + """标记 job 为 failed 并更新。""" + repo = MagicMock() + job = _make_job(status=TTSJobStatus.PROCESSING) + repo.get.return_value = job + repo.update.side_effect = lambda j: j + + workflow = _make_workflow(repo=repo) + workflow._handle_segment_failure(job, "分段 1 合成失败") + + assert job.status == TTSJobStatus.FAILED + assert "分段 1 合成失败" in job.error_message + repo.update.assert_called_once() + + +class TestPollSegmentTasks: + """测试 _poll_segment_tasks 异步轮询。""" + + @patch("packages.application.tts_job.workflow.time") + @patch("packages.application.tts_job.workflow.httpx") + def test_all_segments_done(self, mock_httpx: MagicMock, mock_time: MagicMock) -> None: + """所有分段完成后合并并标记完成。""" + # Mock time.monotonic 让循环只执行一次 + mock_time.monotonic.side_effect = [0.0, 1.0, 2.0] + mock_time.sleep = MagicMock() + + # Mock 下载分段音频 + mock_resp = MagicMock() + mock_resp.content = b"seg audio" + mock_resp.raise_for_status.return_value = None + mock_httpx.get.return_value = mock_resp + + service = MagicMock(spec=CosyVoiceService) + service.poll_synthesize_task.side_effect = [ + {"audio_url": "https://temp.com/seg1.mp3", "duration": 2.0, "file_size": 100}, + {"audio_url": "https://temp.com/seg2.mp3", "duration": 3.0, "file_size": 200}, + ] + + storage = MagicMock() + storage.upload_file.return_value = "https://oss.example.com/merged.mp3" + + repo = MagicMock() + job = _make_job( + status=TTSJobStatus.PROCESSING, + metadata={ + "segment_task_ids": ["task_1", "task_2"], + "segment_audio_urls": ["", ""], + "segment_format": "mp3", + }, + ) + repo.get.return_value = job + repo.update.side_effect = lambda j: j + + workflow = _make_workflow(cosyvoice_service=service, repo=repo, storage=storage) + + with patch("packages.application.tts_job.workflow.AudioMerger") as MockMerger: + mock_merger = MagicMock() + mock_merger.merge.return_value = b"merged data" + MockMerger.return_value = mock_merger + + result = workflow._poll_segment_tasks(job) + + assert result.status == TTSJobStatus.COMPLETED + + @patch("packages.application.tts_job.workflow.time") + def test_segment_poll_failure(self, mock_time: MagicMock) -> None: + """分段轮询失败时标记 job failed。""" + mock_time.monotonic.side_effect = [0.0, 1.0] + mock_time.sleep = MagicMock() + + service = MagicMock(spec=CosyVoiceService) + service.poll_synthesize_task.side_effect = CosyVoiceError("Poll failed") + + repo = MagicMock() + job = _make_job( + status=TTSJobStatus.PROCESSING, + metadata={ + "segment_task_ids": ["task_1"], + "segment_audio_urls": [""], + }, + ) + repo.get.return_value = job + repo.update.side_effect = lambda j: j + + workflow = _make_workflow(cosyvoice_service=service, repo=repo) + result = workflow._poll_segment_tasks(job) + + assert result.status == TTSJobStatus.FAILED + + +class TestPollAndProcessSynthesisSegmentDetection: + """测试 poll_and_process_synthesis 正确识别分段任务。""" + + def test_detects_segment_task(self) -> None: + """metadata 中有 segment_task_ids 时走分段轮询路径。""" + service = MagicMock(spec=CosyVoiceService) + + repo = MagicMock() + job = _make_job( + status=TTSJobStatus.PROCESSING, + metadata={ + "segment_task_ids": ["task_1", "task_2"], + "segment_audio_urls": ["", ""], + }, + ) + repo.get.return_value = job + repo.update.side_effect = lambda j: j + + workflow = _make_workflow(cosyvoice_service=service, repo=repo) + + with patch.object(workflow, "_poll_segment_tasks") as mock_poll: + mock_poll.return_value = job + workflow.poll_and_process_synthesis("test_job_seg") + mock_poll.assert_called_once() + + def test_normal_task_no_segment(self) -> None: + """普通任务不走分段路径。""" + service = MagicMock(spec=CosyVoiceService) + service.poll_synthesize_task.return_value = { + "audio_url": "https://temp.com/audio.mp3", + "duration": 5.0, + "file_size": 5000, + } + + repo = MagicMock() + job = _make_job( + status=TTSJobStatus.PROCESSING, + metadata={"cosyvoice_task_id": "task_normal"}, + ) + repo.get.return_value = job + repo.update.side_effect = lambda j: j + + storage = MagicMock() + storage.upload_file.return_value = "https://oss.example.com/audio.mp3" + + workflow = _make_workflow(cosyvoice_service=service, repo=repo, storage=storage) + + with patch("packages.application.tts_job.workflow.httpx") as mock_httpx: + mock_resp = MagicMock() + mock_resp.content = b"audio" + mock_resp.raise_for_status.return_value = None + mock_httpx.get.return_value = mock_resp + + result = workflow.poll_and_process_synthesis("test_job_seg") + + assert result.status == TTSJobStatus.COMPLETED