diff --git a/apps/api/app/api/routes/lipsync.py b/apps/api/app/api/routes/lipsync.py index 9e478bd43..47a40b019 100644 --- a/apps/api/app/api/routes/lipsync.py +++ b/apps/api/app/api/routes/lipsync.py @@ -1,4 +1,4 @@ -"""对口型 API 路由 — #1796 MediaKit 对口型. +"""对口型 API 路由 — #1796 MediaKit 对口型, #1809 参数调整. 接口： POST /api/v1/lipsync/jobs 提交对口型任务 @@ -13,20 +13,44 @@ from __future__ import annotations import logging from app.auth import AuthenticatedUser, get_current_user -from app.dependencies import get_db_session +from app.dependencies import get_cosyvoice_service, get_db_session, get_voice_clone_profile_repository from app.schemas.lipsync import CreateLipsyncJobRequest, LipsyncJobResponse from app.services.lipsync_service import LipsyncService from app.services.mediakit_client import MediaKitError from fastapi import APIRouter, Depends, HTTPException, Query from sqlalchemy.orm import Session +from packages.application.cosyvoice_service import CosyVoiceError, CosyVoiceService + logger = logging.getLogger(__name__) router = APIRouter() -def _get_service(db: Session = Depends(get_db_session)) -> LipsyncService: - return LipsyncService(db) +def _get_service( + db: Session = Depends(get_db_session), + cosyvoice_service: CosyVoiceService = Depends(get_cosyvoice_service), +) -> LipsyncService: + return LipsyncService(db, cosyvoice_service=cosyvoice_service) + + +def _resolve_voice_id( + raw_voice_id: str, + user_id: str, + voice_clone_repo, +) -> str: + """解析 voice_id：支持预设音色 ID 或克隆音色 profile UUID. + + 与 TTS 路由保持一致：命中 profile → 校验归属 → 取 CosyVoice voice_id。 + """ + profile = voice_clone_repo.get(raw_voice_id) + if profile is not None: + if profile.user_id != user_id: + raise HTTPException(status_code=403, detail="无权访问该音色") + if not profile.voice_id: + raise HTTPException(status_code=400, detail="音色克隆尚未完成，请稍后再试") + return profile.voice_id + return raw_voice_id # ── POST /jobs — 提交对口型任务 ─────────────────────────────────────────── @@ -37,21 +61,31 @@ def create_lipsync_job( body: CreateLipsyncJobRequest, current_user: AuthenticatedUser = Depends(get_current_user), svc: LipsyncService = Depends(_get_service), + voice_clone_repo=Depends(get_voice_clone_profile_repository), ): """提交对口型任务. - 输入人物视频 + 驱动音频，异步生成口型对齐视频。 + #1809: 前端传 {voice_id, script_text, video_url}， + 后端内部调 TTS 合成音频，再提交 MediaKit。 """ + # 解析 voice_id（支持克隆音色 profile UUID） + actual_voice_id = _resolve_voice_id(body.voice_id, current_user.id, voice_clone_repo) + try: job = svc.create_job( user_id=current_user.id, video_url=body.video_url, - audio_url=body.audio_url, + voice_id=actual_voice_id, + script_text=body.script_text, enable_video_loop=body.enable_video_loop, project_id=body.project_id, ) + except CosyVoiceError as exc: + raise HTTPException( + status_code=502, + detail={"code": "TTSSynthesisFailed", "message": str(exc)}, + ) from exc except MediaKitError as exc: - # 创建失败（job 已记录 error），返回 502 raise HTTPException( status_code=502, detail={ diff --git a/apps/api/app/schemas/lipsync.py b/apps/api/app/schemas/lipsync.py index 2429b6977..ac216b343 100644 --- a/apps/api/app/schemas/lipsync.py +++ b/apps/api/app/schemas/lipsync.py @@ -1,4 +1,4 @@ -"""对口型 API Schema 定义 — #1796.""" +"""对口型 API Schema 定义 — #1796, #1809 参数调整.""" from __future__ import annotations @@ -33,10 +33,15 @@ class LipsyncJobResponse(BaseModel): class CreateLipsyncJobRequest(BaseModel): - """创建对口型任务请求.""" + """创建对口型任务请求 — #1809. + + 前端传 {voice_id, script_text, video_url}， + 后端内部调 TTS 生成 audio_url 再提交 MediaKit。 + """ video_url: str = Field(..., description="人物视频 URL（MP4，≤30min，单人真人）") - audio_url: str = Field(..., description="驱动音频 URL（mp3/aac/wav/m4a/flac）") + voice_id: str = Field(..., description="音色 ID（预设音色或克隆音色 profile ID）") + script_text: str = Field(..., description="要合成的脚本文本") enable_video_loop: bool = Field(False, description="音频长于视频时是否循环画面") project_id: str = Field("", description="项目 ID（可选）") @@ -48,23 +53,25 @@ class CreateLipsyncJobRequest(BaseModel): raise ValueError("video_url 不能为空") if not v.startswith(("http://", "https://")): raise ValueError("video_url 必须是 HTTP/HTTPS URL") - # 仅支持 MP4 lower = v.lower().split("?")[0] if not lower.endswith(".mp4"): raise ValueError("video_url 仅支持 MP4 格式") return v - @field_validator("audio_url") + @field_validator("voice_id") @classmethod - def validate_audio_url(cls, v: str) -> str: + def validate_voice_id(cls, v: str) -> str: v = v.strip() if not v: - raise ValueError("audio_url 不能为空") - if not v.startswith(("http://", "https://")): - raise ValueError("audio_url 必须是 HTTP/HTTPS URL") - # 支持的音频格式 - lower = v.lower().split("?")[0] - allowed_exts = (".mp3", ".aac", ".wav", ".m4a", ".flac") - if not any(lower.endswith(ext) for ext in allowed_exts): - raise ValueError(f"audio_url 格式不支持，仅支持: {', '.join(allowed_exts)}") + raise ValueError("voice_id 不能为空") + return v + + @field_validator("script_text") + @classmethod + def validate_script_text(cls, v: str) -> str: + v = v.strip() + if not v: + raise ValueError("script_text 不能为空") + if len(v) > 5000: + raise ValueError("script_text 最长 5000 字符") return v diff --git a/apps/api/app/services/lipsync_service.py b/apps/api/app/services/lipsync_service.py index 3f7dd142e..5d349441e 100644 --- a/apps/api/app/services/lipsync_service.py +++ b/apps/api/app/services/lipsync_service.py @@ -1,7 +1,8 @@ -"""对口型 Service — #1796 MediaKit 对口型业务逻辑. +"""对口型 Service — #1796 MediaKit 对口型业务逻辑, #1809 参数调整. 职责： - 创建/查询/取消对口型任务 +- 调用 TTS 合成音频（#1809：前端不再传 audio_url） - 调用 MediaKit 客户端提交异步任务 - 轮询更新任务状态 - 用户隔离（每个用户只能操作自己的任务） @@ -25,6 +26,7 @@ from app.services.mediakit_client import ( from sqlalchemy.orm import Session from packages.adapters.sqlalchemy_impl.models import LipsyncJobModel +from packages.application.cosyvoice_service import CosyVoiceError, CosyVoiceService logger = logging.getLogger(__name__) @@ -32,9 +34,23 @@ logger = logging.getLogger(__name__) class LipsyncService: """对口型任务 Service.""" - def __init__(self, db: Session, client: Optional[MediaKitClient] = None): + def __init__( + self, + db: Session, + client: Optional[MediaKitClient] = None, + cosyvoice_service: Optional[CosyVoiceService] = None, + ): self.db = db self.client = client or get_mediakit_client() + self._cosyvoice_service = cosyvoice_service + + @property + def cosyvoice_service(self) -> CosyVoiceService: + if self._cosyvoice_service is None: + from app.dependencies import get_cosyvoice_service + + self._cosyvoice_service = get_cosyvoice_service() + return self._cosyvoice_service # ── 创建任务 ────────────────────────────────────────────────────────── @@ -43,16 +59,47 @@ class LipsyncService: *, user_id: str, video_url: str, - audio_url: str, + voice_id: str, + script_text: str, enable_video_loop: bool = False, project_id: str = "", ) -> LipsyncJobModel: """创建对口型任务并提交到 MediaKit. + #1809: 内部调 TTS 合成音频，不再由前端传 audio_url。 + Raises: + CosyVoiceError: TTS 合成失败 MediaKitError: API 调用失败 """ - # 1. 创建数据库记录 + # 1. 调 TTS 合成音频 + try: + tts_result = self.cosyvoice_service.synthesize_speech( + text=script_text, + voice_id=voice_id, + ) + audio_url = tts_result.audio_url + except CosyVoiceError as exc: + logger.error("TTS 合成失败: voice_id=%s, error=%s", voice_id, exc) + # 创建失败记录 + job_id = str(uuid.uuid4()) + job = LipsyncJobModel( + id=job_id, + user_id=user_id, + project_id=project_id, + video_url=video_url, + audio_url="", + enable_video_loop=enable_video_loop, + status="failed", + error_message=f"TTS 合成失败: {exc}", + error_code="TTSSynthesisFailed", + ) + self.db.add(job) + self.db.commit() + self.db.refresh(job) + raise + + # 2. 创建数据库记录 job_id = str(uuid.uuid4()) job = LipsyncJobModel( id=job_id, @@ -66,7 +113,7 @@ class LipsyncService: self.db.add(job) self.db.flush() - # 2. 提交到 MediaKit + # 3. 提交到 MediaKit try: result = self.client.submit_lipsync( video_url=video_url, diff --git a/tests/unit/test_lipsync_routes.py b/tests/unit/test_lipsync_routes.py index b489bc028..b4788171a 100644 --- a/tests/unit/test_lipsync_routes.py +++ b/tests/unit/test_lipsync_routes.py @@ -1,4 +1,4 @@ -"""对口型 API 路由 + Service 单元测试 — #1796. +"""对口型 API 路由 + Service 单元测试 — #1796, #1809 参数调整. CI 增量映射: lipsync.py (route) + lipsync_service.py → test_lipsync_routes.py """ @@ -33,6 +33,19 @@ def mock_mediakit(): return client +@pytest.fixture +def mock_cosyvoice(): + """Mock CosyVoice 服务.""" + service = MagicMock() + service.synthesize_speech.return_value = MagicMock( + audio_url="https://oss.example.com/tts-output.mp3", + duration=15.0, + file_size=12345, + request_id="tts-req-789", + ) + return service + + def _make_mock_job( job_id="job-1", user_id="user-1", @@ -48,7 +61,7 @@ def _make_mock_job( m.user_id = user_id m.project_id = "" m.video_url = "https://example.com/video.mp4" - m.audio_url = "https://example.com/audio.mp3" + m.audio_url = "https://oss.example.com/tts-output.mp3" m.enable_video_loop = False m.mediakit_task_id = mediakit_task_id m.status = status @@ -64,16 +77,19 @@ def _make_mock_job( class TestSchemaValidation: - """Schema 验证测试.""" + """Schema 验证测试 — #1809 新参数结构.""" - def test_valid_video_url(self): + def test_valid_request(self): from app.schemas.lipsync import CreateLipsyncJobRequest req = CreateLipsyncJobRequest( video_url="https://example.com/video.mp4", - audio_url="https://example.com/audio.mp3", + voice_id="longxiaochun_v3", + script_text="大家好，欢迎来到直播间", ) assert req.video_url == "https://example.com/video.mp4" + assert req.voice_id == "longxiaochun_v3" + assert req.script_text == "大家好，欢迎来到直播间" def test_invalid_video_url_not_mp4(self): from app.schemas.lipsync import CreateLipsyncJobRequest @@ -81,7 +97,8 @@ class TestSchemaValidation: with pytest.raises(ValueError, match="MP4"): CreateLipsyncJobRequest( video_url="https://example.com/video.mov", - audio_url="https://example.com/audio.mp3", + voice_id="longxiaochun_v3", + script_text="测试文本", ) def test_invalid_video_url_empty(self): @@ -90,7 +107,8 @@ class TestSchemaValidation: with pytest.raises(ValueError, match="不能为空"): CreateLipsyncJobRequest( video_url=" ", - audio_url="https://example.com/audio.mp3", + voice_id="longxiaochun_v3", + script_text="测试文本", ) def test_invalid_video_url_not_http(self): @@ -99,26 +117,38 @@ class TestSchemaValidation: with pytest.raises(ValueError, match="HTTP"): CreateLipsyncJobRequest( video_url="ftp://example.com/video.mp4", - audio_url="https://example.com/audio.mp3", + voice_id="longxiaochun_v3", + script_text="测试文本", ) - def test_valid_audio_formats(self): + def test_empty_voice_id_rejected(self): from app.schemas.lipsync import CreateLipsyncJobRequest - for ext in [".mp3", ".aac", ".wav", ".m4a", ".flac"]: - req = CreateLipsyncJobRequest( - video_url="https://example.com/video.mp4", - audio_url=f"https://example.com/audio{ext}", - ) - assert req.audio_url.endswith(ext) - - def test_invalid_audio_format(self): - from app.schemas.lipsync import CreateLipsyncJobRequest - - with pytest.raises(ValueError, match="格式不支持"): + with pytest.raises(ValueError, match="voice_id"): CreateLipsyncJobRequest( video_url="https://example.com/video.mp4", - audio_url="https://example.com/audio.ogg", + voice_id=" ", + script_text="测试文本", + ) + + def test_empty_script_text_rejected(self): + from app.schemas.lipsync import CreateLipsyncJobRequest + + with pytest.raises(ValueError, match="script_text"): + CreateLipsyncJobRequest( + video_url="https://example.com/video.mp4", + voice_id="longxiaochun_v3", + script_text="", + ) + + def test_script_text_too_long(self): + from app.schemas.lipsync import CreateLipsyncJobRequest + + with pytest.raises(ValueError, match="5000"): + CreateLipsyncJobRequest( + video_url="https://example.com/video.mp4", + voice_id="longxiaochun_v3", + script_text="x" * 5001, ) def test_enable_video_loop_default(self): @@ -126,7 +156,8 @@ class TestSchemaValidation: req = CreateLipsyncJobRequest( video_url="https://example.com/video.mp4", - audio_url="https://example.com/audio.mp3", + voice_id="longxiaochun_v3", + script_text="测试文本", ) assert req.enable_video_loop is False @@ -136,53 +167,111 @@ class TestSchemaValidation: req = CreateLipsyncJobRequest( video_url="https://example.com/video.mp4?token=abc", - audio_url="https://example.com/audio.mp3?sign=xyz", + voice_id="longxiaochun_v3", + script_text="测试文本", ) assert "?token=" in req.video_url + def test_no_audio_url_in_request(self): + """#1809: 请求体不应包含 audio_url 字段.""" + from app.schemas.lipsync import CreateLipsyncJobRequest + + req = CreateLipsyncJobRequest( + video_url="https://example.com/video.mp4", + voice_id="longxiaochun_v3", + script_text="测试文本", + ) + assert not hasattr(req, "audio_url") + fields = req.model_fields.keys() + assert "audio_url" not in fields + assert "voice_id" in fields + assert "script_text" in fields + class TestLipsyncServiceUnit: - """Service 层单元测试（纯 mock，不依赖数据库）.""" + """Service 层单元测试（纯 mock，不依赖数据库）— #1809 更新.""" - def test_create_job_success(self, mock_mediakit): + def test_create_job_success(self, mock_mediakit, mock_cosyvoice): from app.services.lipsync_service import LipsyncService mock_db = MagicMock() - svc = LipsyncService(mock_db, client=mock_mediakit) - - # 模拟 db.add + db.flush 不报错 mock_db.add = MagicMock() mock_db.flush = MagicMock() mock_db.commit = MagicMock() mock_db.refresh = MagicMock() + svc = LipsyncService(mock_db, client=mock_mediakit, cosyvoice_service=mock_cosyvoice) + job = svc.create_job( user_id="user-1", video_url="https://example.com/video.mp4", - audio_url="https://example.com/audio.mp3", + voice_id="longxiaochun_v3", + script_text="大家好，欢迎来到直播间", ) assert job.status == "submitted" assert job.mediakit_task_id == "mk-task-123" + # TTS 应该被调用 + mock_cosyvoice.synthesize_speech.assert_called_once_with( + text="大家好，欢迎来到直播间", + voice_id="longxiaochun_v3", + ) + # MediaKit 应该用 TTS 生成的 audio_url mock_mediakit.submit_lipsync.assert_called_once() + call_kwargs = mock_mediakit.submit_lipsync.call_args + assert call_kwargs.kwargs["audio_url"] == "https://oss.example.com/tts-output.mp3" - def test_create_job_api_failure(self, mock_mediakit): + def test_create_job_tts_failure(self, mock_mediakit): + """TTS 合成失败时，应创建 failed 记录并抛出 CosyVoiceError.""" + from app.services.lipsync_service import LipsyncService + + from packages.application.cosyvoice_service import CosyVoiceError + + mock_cosyvoice = MagicMock() + mock_cosyvoice.synthesize_speech.side_effect = CosyVoiceError("TTS 服务不可用") + + mock_db = MagicMock() + mock_db.add = MagicMock() + mock_db.flush = MagicMock() + mock_db.commit = MagicMock() + mock_db.refresh = MagicMock() + + svc = LipsyncService(mock_db, client=mock_mediakit, cosyvoice_service=mock_cosyvoice) + + with pytest.raises(CosyVoiceError, match="TTS 服务不可用"): + svc.create_job( + user_id="user-1", + video_url="https://example.com/video.mp4", + voice_id="longxiaochun_v3", + script_text="测试文本", + ) + + # 不应提交到 MediaKit + mock_mediakit.submit_lipsync.assert_not_called() + # 应该记录了失败状态 + added_job = mock_db.add.call_args[0][0] + assert added_job.status == "failed" + assert "TTS" in added_job.error_message + + def test_create_job_api_failure(self, mock_mediakit, mock_cosyvoice): + """MediaKit 提交失败.""" from app.services.lipsync_service import LipsyncService from app.services.mediakit_client import MediaKitError mock_mediakit.submit_lipsync.side_effect = MediaKitError("API 调用失败", code="SubmitFailed") mock_db = MagicMock() - svc = LipsyncService(mock_db, client=mock_mediakit) + svc = LipsyncService(mock_db, client=mock_mediakit, cosyvoice_service=mock_cosyvoice) with pytest.raises(MediaKitError, match="API 调用失败"): svc.create_job( user_id="user-1", video_url="https://example.com/video.mp4", - audio_url="https://example.com/audio.mp3", + voice_id="longxiaochun_v3", + script_text="测试文本", ) - def test_get_job_delegates_to_db(self, mock_mediakit): + def test_get_job_delegates_to_db(self, mock_mediakit, mock_cosyvoice): from app.services.lipsync_service import LipsyncService mock_job = _make_mock_job() @@ -193,13 +282,13 @@ class TestLipsyncServiceUnit: mock_query.filter.return_value = mock_filter mock_db.query.return_value = mock_query - svc = LipsyncService(mock_db, client=mock_mediakit) + svc = LipsyncService(mock_db, client=mock_mediakit, cosyvoice_service=mock_cosyvoice) result = svc.get_job("job-1", "user-1") assert result is mock_job mock_db.query.assert_called_once() - def test_get_job_not_found(self, mock_mediakit): + def test_get_job_not_found(self, mock_mediakit, mock_cosyvoice): from app.services.lipsync_service import LipsyncService mock_db = MagicMock() @@ -209,11 +298,11 @@ class TestLipsyncServiceUnit: mock_query.filter.return_value = mock_filter mock_db.query.return_value = mock_query - svc = LipsyncService(mock_db, client=mock_mediakit) + svc = LipsyncService(mock_db, client=mock_mediakit, cosyvoice_service=mock_cosyvoice) result = svc.get_job("nonexistent", "user-1") assert result is None - def test_refresh_job_completed(self, mock_mediakit): + def test_refresh_job_completed(self, mock_mediakit, mock_cosyvoice): from app.services.lipsync_service import LipsyncService mock_job = _make_mock_job(status="submitted") @@ -224,14 +313,14 @@ class TestLipsyncServiceUnit: mock_query.filter.return_value = mock_filter mock_db.query.return_value = mock_query - svc = LipsyncService(mock_db, client=mock_mediakit) + svc = LipsyncService(mock_db, client=mock_mediakit, cosyvoice_service=mock_cosyvoice) result = svc.refresh_job_status("job-1", "user-1") assert result.status == "completed" assert result.output_video_url == "https://output.mp4" assert result.output_duration == 30.0 - def test_refresh_job_failed(self, mock_mediakit): + def test_refresh_job_failed(self, mock_mediakit, mock_cosyvoice): from app.services.lipsync_service import LipsyncService mock_mediakit.get_task_status.return_value = { @@ -251,13 +340,13 @@ class TestLipsyncServiceUnit: mock_query.filter.return_value = mock_filter mock_db.query.return_value = mock_query - svc = LipsyncService(mock_db, client=mock_mediakit) + svc = LipsyncService(mock_db, client=mock_mediakit, cosyvoice_service=mock_cosyvoice) result = svc.refresh_job_status("job-1", "user-1") assert result.status == "failed" assert result.error_code == "DownloadFailed" - def test_refresh_job_already_completed(self, mock_mediakit): + def test_refresh_job_already_completed(self, mock_mediakit, mock_cosyvoice): """已完成的任务不轮询.""" from app.services.lipsync_service import LipsyncService @@ -269,14 +358,14 @@ class TestLipsyncServiceUnit: mock_query.filter.return_value = mock_filter mock_db.query.return_value = mock_query - svc = LipsyncService(mock_db, client=mock_mediakit) + svc = LipsyncService(mock_db, client=mock_mediakit, cosyvoice_service=mock_cosyvoice) result = svc.refresh_job_status("job-1", "user-1") # 不应调用 MediaKit mock_mediakit.get_task_status.assert_not_called() assert result.status == "completed" - def test_cancel_job_pending(self, mock_mediakit): + def test_cancel_job_pending(self, mock_mediakit, mock_cosyvoice): from app.services.lipsync_service import LipsyncService mock_job = _make_mock_job(status="pending") @@ -287,12 +376,12 @@ class TestLipsyncServiceUnit: mock_query.filter.return_value = mock_filter mock_db.query.return_value = mock_query - svc = LipsyncService(mock_db, client=mock_mediakit) + svc = LipsyncService(mock_db, client=mock_mediakit, cosyvoice_service=mock_cosyvoice) result = svc.cancel_job("job-1", "user-1") assert result.status == "cancelled" - def test_cancel_job_completed_not_allowed(self, mock_mediakit): + def test_cancel_job_completed_not_allowed(self, mock_mediakit, mock_cosyvoice): from app.services.lipsync_service import LipsyncService mock_job = _make_mock_job(status="completed") @@ -303,8 +392,30 @@ class TestLipsyncServiceUnit: mock_query.filter.return_value = mock_filter mock_db.query.return_value = mock_query - svc = LipsyncService(mock_db, client=mock_mediakit) + svc = LipsyncService(mock_db, client=mock_mediakit, cosyvoice_service=mock_cosyvoice) result = svc.cancel_job("job-1", "user-1") # 已完成不可取消 assert result.status == "completed" + + def test_create_job_stores_tts_audio_url(self, mock_mediakit, mock_cosyvoice): + """#1809: 验证 job 的 audio_url 来自 TTS 合成结果.""" + from app.services.lipsync_service import LipsyncService + + mock_db = MagicMock() + mock_db.add = MagicMock() + mock_db.flush = MagicMock() + mock_db.commit = MagicMock() + mock_db.refresh = MagicMock() + + svc = LipsyncService(mock_db, client=mock_mediakit, cosyvoice_service=mock_cosyvoice) + + job = svc.create_job( + user_id="user-1", + video_url="https://example.com/video.mp4", + voice_id="my-clone-voice", + script_text="这是一段测试文本", + ) + + # job.audio_url 应该是 TTS 返回的 URL + assert job.audio_url == "https://oss.example.com/tts-output.mp3"