feat(tts): P0 音频转存 OSS — 合成完成后下载临时音频上传 OSS 存储永久 URL

- TTSWorkflowService 新增 _transfer_audio_to_oss 方法
- start_synthesis 同步路径和 process_synthesis_result 均集成 OSS 转存
- OSS 转存失败时优雅降级，回退到 CosyVoice 临时 URL
- 支持 mp3/wav/pcm/opus 格式 content-type 映射
- 新增 9 个单元测试覆盖转存成功/失败/降级场景
- 向后兼容：storage_service 为可选参数，现有调用方无需修改

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

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:

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()