xiaoxia-saas/apps/api/app/services/lipsync_service.py

"""对口型 Service — #1796 MediaKit 对口型业务逻辑.

职责：
- 创建/查询/取消对口型任务
- 调用 MediaKit 客户端提交异步任务
- 轮询更新任务状态
- 用户隔离（每个用户只能操作自己的任务）
"""

from __future__ import annotations

import logging
import uuid
from datetime import datetime, timezone
from typing import Optional

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

logger = logging.getLogger(__name__)


class LipsyncService:
    """对口型任务 Service."""

    def __init__(self, db: Session, client: Optional[MediaKitClient] = None):
        self.db = db
        self.client = client or get_mediakit_client()

    # ── 创建任务 ──────────────────────────────────────────────────────────

    def create_job(
        self,
        *,
        user_id: str,
        video_url: str,
        audio_url: str,
        enable_video_loop: bool = False,
        project_id: str = "",
    ) -> LipsyncJobModel:
        """创建对口型任务并提交到 MediaKit.

        Raises:
            MediaKitError: API 调用失败
        """
        # 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,
            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)

        if mk_status == STATUS_COMPLETED:
            result = status_data.get("result", {})
            job.status = STATUS_COMPLETED
            job.output_video_url = result.get("video_url", "")
            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)
        # running 状态只更新时间戳
        job.updated_at = datetime.now(timezone.utc)
        self.db.commit()
        self.db.refresh(job)
        return job

    # ── 取消任务 ──────────────────────────────────────────────────────────

    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