diff --git a/packages/application/tts_job/workflow.py b/packages/application/tts_job/workflow.py index b49e60eef..4bdcab63d 100644 --- a/packages/application/tts_job/workflow.py +++ b/packages/application/tts_job/workflow.py @@ -9,9 +9,12 @@ from __future__ import annotations +import io import logging from typing import Optional +import httpx + from packages.application.cosyvoice_service import ( CosyVoiceAuthError, CosyVoiceError, @@ -19,6 +22,7 @@ from packages.application.cosyvoice_service import ( ) 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__) @@ -46,9 +50,55 @@ class TTSWorkflowService: self, repository: TTSJobRepository, cosyvoice_service: CosyVoiceService, + storage_service: Optional[SharedStorageService] = None, ) -> None: self.repository = repository self.cosyvoice_service = cosyvoice_service + self._storage_service = storage_service + + @property + def _storage(self) -> SharedStorageService: + if self._storage_service is None: + self._storage_service = get_shared_storage_service() + return self._storage_service + + def _transfer_audio_to_oss( + self, + temp_url: str, + user_id: str, + job_id: str, + audio_format: str = "mp3", + ) -> tuple[str, str]: + """下载 CosyVoice 临时音频并转存到 OSS。 + + Returns: + (permanent_url, storage_key) 元组。 + 转存失败时回退到原始临时 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: + # 下载临时音频 + resp = httpx.get(temp_url, timeout=60.0, follow_redirects=True) + resp.raise_for_status() + audio_data = resp.content + + # 上传到 OSS + file_obj = io.BytesIO(audio_data) + permanent_url = self._storage.upload_file(file_obj, storage_key, content_type=content_type) + logger.info(f"音频转存 OSS 成功: job_id={job_id}, " f"storage_key={storage_key}, size={len(audio_data)}") + return permanent_url, storage_key + + except Exception as e: + logger.warning(f"音频转存 OSS 失败，使用临时 URL: " f"job_id={job_id}, error={e}") + return temp_url, "" def start_synthesis( self, @@ -93,17 +143,19 @@ class TTSWorkflowService: job_metadata["cosyvoice_task_id"] = submit_result.get("task_id", "") job_metadata["cosyvoice_request_id"] = submit_result.get("request_id", "") - # 如果 CosyVoice 同步返回了 audio_url，直接标记完成 + # 如果 CosyVoice 同步返回了 audio_url，转存 OSS 后标记完成 audio_url = submit_result.get("audio_url", "") if audio_url: + permanent_url, storage_key = self._transfer_audio_to_oss(audio_url, job.user_id, job.id, job.format) job.mark_completed( - output_audio_url=audio_url, + output_audio_url=permanent_url, + output_audio_key=storage_key, 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}") + logger.info(f"TTS 合成同步完成: job_id={job.id}, audio_url={permanent_url}") return job job.metadata = job_metadata @@ -171,13 +223,17 @@ class TTSWorkflowService: if job is None: raise TTSJobNotFoundError(f"TTS job {job_id} not found") + # 转存音频到 OSS，获取永久 URL + permanent_url, storage_key = self._transfer_audio_to_oss(audio_url, job.user_id, job.id, job.format) + job.mark_completed( - output_audio_url=audio_url, + output_audio_url=permanent_url, + output_audio_key=storage_key, duration=duration, file_size=file_size, ) job = self.repository.update(job) - logger.info(f"TTS 合成成功: job_id={job_id}, audio_url={audio_url}") + logger.info(f"TTS 合成成功: job_id={job_id}, audio_url={permanent_url}") return job def process_synthesis_failure(self, job_id: str, error_message: str) -> TTSJob: diff --git a/tests/unit/test_tts_oss_transfer.py b/tests/unit/test_tts_oss_transfer.py new file mode 100644 index 000000000..0fa0ed7ac --- /dev/null +++ b/tests/unit/test_tts_oss_transfer.py @@ -0,0 +1,299 @@ +"""TTS 音频转存 OSS 单元测试。 + +验证 TTSWorkflowService 在合成完成后将 CosyVoice 临时音频转存到 OSS， +存储永久 URL 到 TTSJob.output_audio_url，OSS key 到 output_audio_key。 +""" + +from __future__ import annotations + +from datetime import datetime, timezone +from unittest.mock import MagicMock, patch + +import pytest + +from packages.application.cosyvoice_service import CosyVoiceService +from packages.application.tts_job.workflow import TTSWorkflowService +from packages.domain.tts_job import TTSJob, TTSJobStatus + + +def _make_job(**kwargs) -> TTSJob: + defaults = { + "id": "test_job_001", + "user_id": "user_001", + "input_text": "测试文本", + "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/tts-outputs/user_001/test_job_001.mp3" + return TTSWorkflowService( + repository=repo, + cosyvoice_service=cosyvoice_service, + storage_service=storage, + ) + + +class TestTransferAudioToOSS: + """测试 _transfer_audio_to_oss 方法。""" + + @patch("packages.application.tts_job.workflow.httpx") + def test_success_download_and_upload(self, mock_httpx: MagicMock) -> None: + """成功下载音频并上传到 OSS，返回永久 URL 和 storage_key。""" + mock_resp = MagicMock() + mock_resp.content = b"fake audio data" + mock_resp.raise_for_status.return_value = None + mock_httpx.get.return_value = mock_resp + + storage = MagicMock() + storage.upload_file.return_value = "https://oss.example.com/tts-outputs/user_001/job_123.mp3" + + workflow = _make_workflow(storage=storage) + url, key = workflow._transfer_audio_to_oss( + "https://cosyvoice-temp.com/audio.mp3", + "user_001", + "job_123", + "mp3", + ) + + assert url == "https://oss.example.com/tts-outputs/user_001/job_123.mp3" + assert key == "tts-outputs/user_001/job_123.mp3" + + mock_httpx.get.assert_called_once_with( + "https://cosyvoice-temp.com/audio.mp3", + timeout=60.0, + follow_redirects=True, + ) + storage.upload_file.assert_called_once() + call_args = storage.upload_file.call_args + assert call_args[0][1] == "tts-outputs/user_001/job_123.mp3" + assert call_args[1]["content_type"] == "audio/mpeg" + + @patch("packages.application.tts_job.workflow.httpx") + def test_download_failure_fallback(self, mock_httpx: MagicMock) -> None: + """下载失败时回退到原始临时 URL，storage_key 为空。""" + mock_httpx.get.side_effect = Exception("Network error") + + workflow = _make_workflow() + url, key = workflow._transfer_audio_to_oss( + "https://cosyvoice-temp.com/audio.mp3", + "user_001", + "job_123", + ) + + assert url == "https://cosyvoice-temp.com/audio.mp3" + assert key == "" + + @patch("packages.application.tts_job.workflow.httpx") + def test_upload_failure_fallback(self, mock_httpx: MagicMock) -> None: + """上传 OSS 失败时回退到原始临时 URL。""" + mock_resp = MagicMock() + mock_resp.content = b"fake audio data" + mock_resp.raise_for_status.return_value = None + mock_httpx.get.return_value = mock_resp + + storage = MagicMock() + storage.upload_file.side_effect = Exception("OSS bucket error") + + workflow = _make_workflow(storage=storage) + url, key = workflow._transfer_audio_to_oss( + "https://cosyvoice-temp.com/audio.mp3", + "user_001", + "job_123", + ) + + assert url == "https://cosyvoice-temp.com/audio.mp3" + assert key == "" + + @patch("packages.application.tts_job.workflow.httpx") + def test_wav_content_type(self, mock_httpx: MagicMock) -> None: + """wav 格式使用正确的 content_type。""" + mock_resp = MagicMock() + mock_resp.content = b"fake wav data" + mock_resp.raise_for_status.return_value = None + mock_httpx.get.return_value = mock_resp + + storage = MagicMock() + storage.upload_file.return_value = "https://oss.example.com/audio.wav" + + workflow = _make_workflow(storage=storage) + workflow._transfer_audio_to_oss( + "https://cosyvoice-temp.com/audio.wav", + "user_001", + "job_456", + "wav", + ) + + call_args = storage.upload_file.call_args + assert call_args[1]["content_type"] == "audio/wav" + assert call_args[0][1] == "tts-outputs/user_001/job_456.wav" + + +class TestProcessSynthesisResultWithOSS: + """测试 process_synthesis_result 集成 OSS 转存。""" + + @patch("packages.application.tts_job.workflow.httpx") + def test_stores_permanent_url_and_key(self, mock_httpx: MagicMock) -> None: + """合成结果存 OSS 永久 URL 和 storage_key。""" + mock_resp = MagicMock() + mock_resp.content = b"audio bytes" + mock_resp.raise_for_status.return_value = None + mock_httpx.get.return_value = mock_resp + + storage = MagicMock() + storage.upload_file.return_value = "https://oss.example.com/tts-outputs/user_001/test_job_001.mp3" + + 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, storage=storage) + result = workflow.process_synthesis_result( + "test_job_001", + audio_url="https://cosyvoice-temp.com/expiring.mp3", + duration=5.0, + file_size=50000, + ) + + assert result.status == TTSJobStatus.COMPLETED + assert result.output_audio_url == "https://oss.example.com/tts-outputs/user_001/test_job_001.mp3" + assert result.output_audio_key == "tts-outputs/user_001/test_job_001.mp3" + assert result.duration == 5.0 + assert result.file_size == 50000 + + @patch("packages.application.tts_job.workflow.httpx") + def test_fallback_to_temp_url_on_oss_failure(self, mock_httpx: MagicMock) -> None: + """OSS 转存失败时，使用 CosyVoice 临时 URL（不阻塞合成流程）。""" + mock_httpx.get.side_effect = Exception("Download 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) + result = workflow.process_synthesis_result( + "test_job_001", + audio_url="https://cosyvoice-temp.com/expiring.mp3", + ) + + assert result.status == TTSJobStatus.COMPLETED + assert result.output_audio_url == "https://cosyvoice-temp.com/expiring.mp3" + assert result.output_audio_key == "" + + +class TestStartSynthesisSyncWithOSS: + """测试 start_synthesis 同步路径的 OSS 转存。""" + + @patch("packages.application.tts_job.workflow.httpx") + def test_sync_path_transfers_to_oss(self, mock_httpx: MagicMock) -> None: + """CosyVoice 同步返回 audio_url 时，也走 OSS 转存。""" + mock_resp = MagicMock() + mock_resp.content = b"sync audio bytes" + mock_resp.raise_for_status.return_value = None + mock_httpx.get.return_value = mock_resp + + storage = MagicMock() + storage.upload_file.return_value = "https://oss.example.com/tts-outputs/user_001/test_job_001.mp3" + + service = MagicMock(spec=CosyVoiceService) + service.submit_synthesize_task.return_value = { + "task_id": "", + "audio_url": "https://cosyvoice-temp.com/sync.mp3", + "duration": 2.0, + "file_size": 20000, + "request_id": "req_sync", + } + + 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) + job = workflow.start_synthesis("test_job_001") + + assert job.status == TTSJobStatus.COMPLETED + assert job.output_audio_url == "https://oss.example.com/tts-outputs/user_001/test_job_001.mp3" + assert job.output_audio_key == "tts-outputs/user_001/test_job_001.mp3" + assert job.duration == 2.0 + + @patch("packages.application.tts_job.workflow.httpx") + def test_sync_path_oss_failure_stores_temp_url(self, mock_httpx: MagicMock) -> None: + """同步路径 OSS 失败时，降级存储临时 URL。""" + mock_httpx.get.side_effect = Exception("Network error") + + service = MagicMock(spec=CosyVoiceService) + service.submit_synthesize_task.return_value = { + "task_id": "", + "audio_url": "https://cosyvoice-temp.com/sync.mp3", + "duration": 2.0, + "file_size": 20000, + "request_id": "req_sync", + } + + 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_001") + + assert job.status == TTSJobStatus.COMPLETED + assert job.output_audio_url == "https://cosyvoice-temp.com/sync.mp3" + assert job.output_audio_key == "" + + def test_async_path_no_oss_transfer(self) -> None: + """异步路径（返回 task_id，无 audio_url）不触发 OSS 转存。""" + service = MagicMock(spec=CosyVoiceService) + service.submit_synthesize_task.return_value = { + "task_id": "cosy_task_async", + "audio_url": "", + "duration": 0.0, + "file_size": 0, + "request_id": "req_async", + } + + repo = MagicMock() + repo.get.return_value = _make_job() + repo.update.side_effect = lambda j: j + + storage = MagicMock() + workflow = _make_workflow(cosyvoice_service=service, repo=repo, storage=storage) + job = workflow.start_synthesis("test_job_001") + + assert job.status == TTSJobStatus.PROCESSING + # 异步路径不应调用 OSS 上传 + storage.upload_file.assert_not_called()