xiaoxia
a27ed596b4
- 对口型支持 TTS 直生模式：POST /lipsync/jobs 传 voice_id+script_text(+speed/emotion)， 后端内部调 CosyVoice 合成音频→转存OSS→提交 MediaKit；保留 audio_url 直接音频模式 - cosyvoice_service: 新增 emotion 参数 + normalize_emotion（中文 自然/兴奋/沉稳/亲切 映射 natural/excited/calm/friendly），空值不透传 - TTS 链路 speed/emotion 全链路透传：schema→use_case(metadata)→workflow(单段/分段/重合成) →submit_synthesize_task(rate/emotion)；preview 即时试听同步 - lipsync refresh: 中间状态(running/processing)同步DB；completed 输出视频转存自家OSS防过期 - 修复 lipsync/ai-avatar 路由 current_user.id → current_user.user.id（AuthenticatedUser 无 .id） - 新增 POST /ai-avatar/render/smart-cover 独立封面接口：复用 MediaKit extract_frames + cover_frame_scorer 评分选最佳帧（非 FFmpeg 首帧），渲染管线封面同样优先智能选帧 - 迁移 073: lipsync_jobs 增加 voice_id/script_text/speed/emotion 列，audio_url 改可空 - docs/ai-avatar-api-contract-1822.md: 前后端接口契约 + title_config 字段清单 - 新增 11 个单测（情绪归一化/payload透传/TTS直生/中间状态/智能封面），相关 82 测试全绿 Refs #1797 #1822
343 lines
13 KiB
Python
343 lines
13 KiB
Python
"""对口型 Service — #1796 MediaKit 对口型业务逻辑.
|
||
|
||
职责：
|
||
- 创建/查询/取消对口型任务
|
||
- 调用 MediaKit 客户端提交异步任务
|
||
- 轮询更新任务状态
|
||
- 用户隔离（每个用户只能操作自己的任务）
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
import uuid
|
||
from datetime import datetime, timezone
|
||
from typing import Optional
|
||
|
||
import io
|
||
|
||
from app.services.mediakit_client import (
|
||
STATUS_COMPLETED,
|
||
STATUS_FAILED,
|
||
STATUS_RUNNING,
|
||
MediaKitClient,
|
||
MediaKitError,
|
||
get_mediakit_client,
|
||
)
|
||
from sqlalchemy.orm import Session
|
||
|
||
from packages.adapters.sqlalchemy_impl.models import LipsyncJobModel
|
||
from packages.application.cosyvoice_service import CosyVoiceError, normalize_emotion
|
||
from packages.shared.storage import get_shared_storage_service
|
||
from packages.shared.url_security import ALLOWED_AUDIO_MIME_TYPES, safe_download_bytes
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class LipsyncService:
|
||
"""对口型任务 Service."""
|
||
|
||
def __init__(
|
||
self,
|
||
db: Session,
|
||
client: Optional[MediaKitClient] = None,
|
||
cosyvoice_service=None,
|
||
voice_clone_repo=None,
|
||
):
|
||
self.db = db
|
||
self.client = client or get_mediakit_client()
|
||
self._cosyvoice = cosyvoice_service
|
||
self._voice_clone_repo = voice_clone_repo
|
||
|
||
def _get_cosyvoice(self):
|
||
"""延迟获取 CosyVoiceService（与 tts 路由一致，含 OSS 预签名配置）."""
|
||
if self._cosyvoice is None:
|
||
from app.dependencies import get_cosyvoice_service
|
||
|
||
self._cosyvoice = get_cosyvoice_service()
|
||
return self._cosyvoice
|
||
|
||
def _resolve_voice_id(self, voice_id: str, user_id: str) -> str:
|
||
"""将克隆音色 profile UUID 解析为 CosyVoice voice_id。
|
||
|
||
与 /tts/synthesize 保持一致：命中 profile → 校验归属 → 返回其 voice_id；
|
||
未命中（预置音色 ID 或克隆 CosyVoice voice_id）原样返回。
|
||
"""
|
||
if not voice_id:
|
||
return ""
|
||
if self._voice_clone_repo is None:
|
||
try:
|
||
from app.dependencies import get_voice_clone_profile_repository
|
||
|
||
self._voice_clone_repo = get_voice_clone_profile_repository(self.db)
|
||
except Exception:
|
||
return voice_id
|
||
try:
|
||
profile = self._voice_clone_repo.get(voice_id)
|
||
except Exception:
|
||
return voice_id
|
||
if profile is None:
|
||
return voice_id
|
||
if getattr(profile, "user_id", "") != user_id:
|
||
raise MediaKitError("无权访问该音色", code="VoiceForbidden")
|
||
if not getattr(profile, "voice_id", ""):
|
||
raise MediaKitError("音色克隆尚未完成，请稍后再试", code="VoiceNotReady")
|
||
return profile.voice_id
|
||
|
||
def _synthesize_and_persist_audio(
|
||
self,
|
||
*,
|
||
user_id: str,
|
||
job_id: str,
|
||
voice_id: str,
|
||
script_text: str,
|
||
speed: float,
|
||
emotion: str,
|
||
) -> str:
|
||
"""TTS 直生：调 CosyVoice 合成音频并转存 OSS，返回可公网访问的音频 URL.
|
||
|
||
Raises:
|
||
MediaKitError: 合成失败
|
||
"""
|
||
actual_voice_id = self._resolve_voice_id(voice_id, user_id)
|
||
cosyvoice = self._get_cosyvoice()
|
||
try:
|
||
result = cosyvoice.submit_synthesize_task(
|
||
text=script_text,
|
||
voice_id=actual_voice_id,
|
||
speed=speed,
|
||
emotion=normalize_emotion(emotion),
|
||
)
|
||
except CosyVoiceError as exc:
|
||
raise MediaKitError(f"TTS 合成失败: {exc}", code="TTSSynthesisFailed") from exc
|
||
except ValueError as exc:
|
||
raise MediaKitError(f"TTS 参数错误: {exc}", code="TTSInvalidParam") from exc
|
||
|
||
temp_url = result.get("audio_url", "")
|
||
if not temp_url:
|
||
raise MediaKitError("TTS 未返回音频 URL", code="TTSNoAudio")
|
||
|
||
# 转存到自家 OSS，避免临时 URL 过期导致 MediaKit 拉取失败
|
||
try:
|
||
audio_data = safe_download_bytes(
|
||
temp_url,
|
||
purpose="lipsync_tts_audio",
|
||
allowed_mime_types=ALLOWED_AUDIO_MIME_TYPES,
|
||
timeout=60.0,
|
||
)
|
||
storage = get_shared_storage_service()
|
||
storage_key = f"lipsync-tts/{user_id}/{job_id}.mp3"
|
||
permanent_url = storage.upload_file(
|
||
io.BytesIO(audio_data), storage_key, content_type="audio/mpeg"
|
||
)
|
||
logger.info("对口型 TTS 音频已转存 OSS: job_id=%s key=%s", job_id, storage_key)
|
||
return permanent_url
|
||
except Exception as exc:
|
||
logger.warning("TTS 音频转存 OSS 失败，回退临时 URL: job_id=%s err=%s", job_id, exc)
|
||
return temp_url
|
||
|
||
# ── 创建任务 ──────────────────────────────────────────────────────────
|
||
|
||
def create_job(
|
||
self,
|
||
*,
|
||
user_id: str,
|
||
video_url: str,
|
||
audio_url: str = "",
|
||
voice_id: str = "",
|
||
script_text: str = "",
|
||
speed: float = 1.0,
|
||
emotion: str = "",
|
||
enable_video_loop: bool = False,
|
||
project_id: str = "",
|
||
) -> LipsyncJobModel:
|
||
"""创建对口型任务并提交到 MediaKit.
|
||
|
||
两种输入模式：
|
||
- TTS 直生：voice_id + script_text（audio_url 留空），后端先合成音频
|
||
- 直接音频：提供 audio_url
|
||
|
||
Raises:
|
||
MediaKitError: TTS 合成或 MediaKit 提交失败
|
||
"""
|
||
# 0. TTS 直生模式：先合成音频（在创建 DB 记录之前完成，失败直接抛出）
|
||
if not audio_url:
|
||
if not (voice_id and script_text):
|
||
raise MediaKitError(
|
||
"必须提供 audio_url 或 voice_id+script_text",
|
||
code="InvalidInput",
|
||
)
|
||
# 预合成：用临时 job_id 命名 OSS 对象
|
||
pre_job_id = str(uuid.uuid4())
|
||
audio_url = self._synthesize_and_persist_audio(
|
||
user_id=user_id,
|
||
job_id=pre_job_id,
|
||
voice_id=voice_id,
|
||
script_text=script_text,
|
||
speed=speed,
|
||
emotion=emotion,
|
||
)
|
||
|
||
# 1. 创建数据库记录
|
||
job_id = str(uuid.uuid4())
|
||
job = LipsyncJobModel(
|
||
id=job_id,
|
||
user_id=user_id,
|
||
project_id=project_id,
|
||
video_url=video_url,
|
||
audio_url=audio_url,
|
||
enable_video_loop=enable_video_loop,
|
||
voice_id=voice_id or "",
|
||
script_text=script_text or "",
|
||
speed=speed,
|
||
emotion=normalize_emotion(emotion),
|
||
status="pending",
|
||
)
|
||
self.db.add(job)
|
||
self.db.flush()
|
||
|
||
# 2. 提交到 MediaKit
|
||
try:
|
||
result = self.client.submit_lipsync(
|
||
video_url=video_url,
|
||
audio_url=audio_url,
|
||
enable_video_loop=enable_video_loop,
|
||
client_token=job_id, # 幂等控制
|
||
)
|
||
job.mediakit_task_id = result["task_id"]
|
||
job.status = "submitted"
|
||
job.submitted_at = datetime.now(timezone.utc)
|
||
except MediaKitError as exc:
|
||
job.status = "failed"
|
||
job.error_message = str(exc)
|
||
job.error_code = exc.code
|
||
logger.error("提交对口型任务失败: %s", exc)
|
||
raise
|
||
|
||
self.db.commit()
|
||
self.db.refresh(job)
|
||
return job
|
||
|
||
# ── 查询任务 ──────────────────────────────────────────────────────────
|
||
|
||
def get_job(self, job_id: str, user_id: str) -> Optional[LipsyncJobModel]:
|
||
"""获取任务详情（用户隔离）."""
|
||
return (
|
||
self.db.query(LipsyncJobModel)
|
||
.filter(LipsyncJobModel.id == job_id, LipsyncJobModel.user_id == user_id)
|
||
.first()
|
||
)
|
||
|
||
def list_jobs(
|
||
self,
|
||
*,
|
||
user_id: str,
|
||
project_id: str = "",
|
||
status: str = "",
|
||
offset: int = 0,
|
||
limit: int = 20,
|
||
) -> tuple[list[LipsyncJobModel], int]:
|
||
"""获取任务列表（分页 + 用户隔离）."""
|
||
query = self.db.query(LipsyncJobModel).filter(LipsyncJobModel.user_id == user_id)
|
||
if project_id:
|
||
query = query.filter(LipsyncJobModel.project_id == project_id)
|
||
if status:
|
||
query = query.filter(LipsyncJobModel.status == status)
|
||
|
||
total = query.count()
|
||
items = query.order_by(LipsyncJobModel.created_at.desc()).offset(offset).limit(limit).all()
|
||
return items, total
|
||
|
||
# ── 更新任务状态（轮询） ──────────────────────────────────────────────
|
||
|
||
def refresh_job_status(self, job_id: str, user_id: str) -> Optional[LipsyncJobModel]:
|
||
"""从 MediaKit 拉取最新状态并更新本地记录.
|
||
|
||
Returns:
|
||
更新后的 Job，或 None（任务不存在/不属于该用户）
|
||
"""
|
||
job = self.get_job(job_id, user_id)
|
||
if job is None:
|
||
return None
|
||
|
||
# 终态不需要再轮询
|
||
if job.status in (STATUS_COMPLETED, "failed"):
|
||
return job
|
||
|
||
# 未提交的任务不轮询
|
||
if not job.mediakit_task_id:
|
||
return job
|
||
|
||
try:
|
||
status_data = self.client.get_task_status(job.mediakit_task_id)
|
||
except MediaKitError as exc:
|
||
logger.error("轮询对口型任务状态失败 [%s]: %s", job_id, exc)
|
||
return job
|
||
|
||
mk_status = status_data.get("status", STATUS_RUNNING)
|
||
logger.info("MediaKit 对口型状态 [%s]: %s", job_id, mk_status)
|
||
|
||
if mk_status == STATUS_COMPLETED:
|
||
result = status_data.get("result", {})
|
||
job.status = STATUS_COMPLETED
|
||
output_url = result.get("video_url", "")
|
||
# MediaKit 输出为临时 URL，转存自家 OSS 防止过期（失败则回退临时 URL）
|
||
job.output_video_url = self._persist_output_video(output_url, job_id, user_id)
|
||
job.output_duration = result.get("duration", 0.0)
|
||
job.completed_at = datetime.now(timezone.utc)
|
||
elif mk_status == STATUS_FAILED:
|
||
error = status_data.get("error", {})
|
||
job.status = "failed"
|
||
job.error_message = error.get("message", "任务执行失败")
|
||
job.error_code = error.get("code", "TaskFailed")
|
||
job.completed_at = datetime.now(timezone.utc)
|
||
else:
|
||
# 中间状态（running/processing/queued 等）同步到 DB，避免前端永远卡在 submitted
|
||
if isinstance(mk_status, str) and mk_status:
|
||
job.status = mk_status
|
||
job.updated_at = datetime.now(timezone.utc)
|
||
self.db.commit()
|
||
self.db.refresh(job)
|
||
return job
|
||
|
||
def _persist_output_video(self, temp_url: str, job_id: str, user_id: str) -> str:
|
||
"""将 MediaKit 输出的临时视频 URL 转存到自家 OSS.
|
||
|
||
失败时回退返回原始临时 URL，不影响任务完成。
|
||
"""
|
||
if not temp_url:
|
||
return ""
|
||
try:
|
||
import httpx
|
||
|
||
with httpx.Client(timeout=180.0, follow_redirects=True) as client:
|
||
resp = client.get(temp_url)
|
||
resp.raise_for_status()
|
||
data = resp.content
|
||
storage = get_shared_storage_service()
|
||
storage_key = f"lipsync-outputs/{user_id}/{job_id}.mp4"
|
||
permanent_url = storage.upload_file(
|
||
io.BytesIO(data), storage_key, content_type="video/mp4"
|
||
)
|
||
logger.info("对口型输出视频已转存 OSS: job_id=%s key=%s", job_id, storage_key)
|
||
return permanent_url or temp_url
|
||
except Exception as exc:
|
||
logger.warning("对口型输出视频转存 OSS 失败，回退临时 URL: job_id=%s err=%s", job_id, exc)
|
||
return temp_url
|
||
|
||
# ── 取消任务 ──────────────────────────────────────────────────────────
|
||
|
||
def cancel_job(self, job_id: str, user_id: str) -> Optional[LipsyncJobModel]:
|
||
"""取消任务（仅 pending/submitted 状态可取消）."""
|
||
job = self.get_job(job_id, user_id)
|
||
if job is None:
|
||
return None
|
||
|
||
if job.status in ("pending", "submitted"):
|
||
job.status = "cancelled"
|
||
job.updated_at = datetime.now(timezone.utc)
|
||
self.db.commit()
|
||
self.db.refresh(job)
|
||
|
||
return job
|