Maas-Model-platform/backend/app/services/video_service.py

"""
视频合成服务

提供文生视频和图生视频功能
"""
import logging
import threading
import httpx
from datetime import datetime
from decimal import Decimal
from typing import Optional
from fastapi import HTTPException

from sqlalchemy.orm import Session

from app.models.ai_video import AIVideo
from app.schemas.video_schema import (
    VideoGenerateRequest, VideoTaskResponse, VideoTaskResult,
    VideoHistoryItem, VideoHistoryResponse
)
from app.services.oss_service import get_oss_service

logger = logging.getLogger(__name__)

DASHSCOPE_BASE_URL = "https://dashscope.aliyuncs.com/api/v1"

# 分辨率映射
RESOLUTION_MAP = {
    "720P": {"16:9": "1280*720", "9:16": "720*1280", "1:1": "960*960"},
    "1080P": {"16:9": "1920*1080", "9:16": "1080*1920", "1:1": "1440*1440"}
}


class VideoService:
    """视频合成服务"""

    def __init__(self, db: Session, user_id: int, api_key: str):
        self.db = db
        self.user_id = user_id
        self.api_key = api_key
        self.oss_service = get_oss_service()

    async def generate(self, request: VideoGenerateRequest) -> VideoTaskResponse:
        """统一视频生成入口"""
        # 统一策略：只要有首帧（不管是否包含尾帧）都使用图生视频接口
        if request.first_frame_url:
            return await self._image_to_video(request)
        else:
            return await self._text_to_video(request)

    async def _text_to_video(self, request: VideoGenerateRequest) -> VideoTaskResponse:
        """文生视频（HTTP接口）"""
        # 使用前端传入的模型，默认 wan2.6-t2v
        model = request.model or "wan2.6-t2v"

        url = f"{DASHSCOPE_BASE_URL}/services/aigc/video-generation/video-synthesis"
        headers = {
            "Content-Type": "application/json",
            "Authorization": f"Bearer {self.api_key}",
            "X-DashScope-Async": "enable"
        }

        # 获取分辨率尺寸
        size = RESOLUTION_MAP.get(request.resolution, RESOLUTION_MAP["720P"])["16:9"]

        body = {
            "model": model,
            "input": {
                "prompt": request.prompt
            },
            "parameters": {
                "size": size,
                "duration": request.duration,
                "prompt_extend": request.prompt_extend,
                "shot_type": request.shot_type,
                "watermark": request.watermark
            }
        }

        if request.negative_prompt:
            body["input"]["negative_prompt"] = request.negative_prompt
        if request.audio_url:
            body["input"]["audio_url"] = request.audio_url
        if request.seed:
            body["parameters"]["seed"] = request.seed

        async with httpx.AsyncClient(timeout=30.0) as client:
            response = await client.post(url, json=body, headers=headers)
            result = response.json()

        if "code" in result:
            raise Exception(f"创建任务失败: {result.get('message')}")

        task_id = result["output"]["task_id"]
        task_status = result["output"]["task_status"]

        # 保存记录
        record = AIVideo(
            user_id=self.user_id,
            task_id=task_id,
            model_name=model,
            video_type="t2v",
            input_params=body,
            prompt=request.prompt,
            audio_url=request.audio_url,
            resolution=request.resolution,
            status=task_status,
            submit_time=datetime.now()
        )
        self.db.add(record)
        self.db.commit()

        return VideoTaskResponse(task_id=task_id, task_status=task_status)

    async def _first_last_frame_to_video(self, request: VideoGenerateRequest) -> VideoTaskResponse:
        """首尾帧生视频 - 调用wan2.2-kf2v-flash（HTTP接口）"""
        url = f"{DASHSCOPE_BASE_URL}/services/aigc/image2video/video-synthesis"
        headers = {
            "Content-Type": "application/json",
            "Authorization": f"Bearer {self.api_key}",
            "X-DashScope-Async": "enable"
        }

        body = {
            "model": "wan2.2-kf2v-flash",
            "input": {
                "first_frame_url": request.first_frame_url,
                "last_frame_url": request.last_frame_url,
                "prompt": request.prompt
            },
            "parameters": {
                "resolution": request.resolution,
                "prompt_extend": request.prompt_extend,
                "watermark": request.watermark
            }
        }

        if request.negative_prompt:
            body["input"]["negative_prompt"] = request.negative_prompt
        if request.seed:
            body["parameters"]["seed"] = request.seed

        async with httpx.AsyncClient(timeout=30.0) as client:
            response = await client.post(url, json=body, headers=headers)
            result = response.json()

        if "code" in result:
            raise Exception(f"创建任务失败: {result.get('message')}")

        task_id = result["output"]["task_id"]
        task_status = result["output"]["task_status"]

        # 保存记录
        record = AIVideo(
            user_id=self.user_id,
            task_id=task_id,
            model_name="wan2.2-kf2v-flash",
            video_type="kf2v",
            input_params=body,
            prompt=request.prompt,
            first_frame_url=request.first_frame_url,
            last_frame_url=request.last_frame_url,
            resolution=request.resolution,
            status=task_status,
            submit_time=datetime.now()
        )
        self.db.add(record)
        self.db.commit()

        return VideoTaskResponse(task_id=task_id, task_status=task_status)

    async def _image_to_video(self, request: VideoGenerateRequest) -> VideoTaskResponse:
        """图生视频（HTTP接口）"""
        # 使用前端传入的模型，默认 wan2.6-i2v
        model = request.model or "wan2.6-i2v"

        url = f"{DASHSCOPE_BASE_URL}/services/aigc/video-generation/video-synthesis"
        headers = {
            "Content-Type": "application/json",
            "Authorization": f"Bearer {self.api_key}",
            "X-DashScope-Async": "enable"
        }

        # 获取分辨率尺寸
        size = RESOLUTION_MAP.get(request.resolution, RESOLUTION_MAP["720P"])["16:9"]

        body = {
            "model": model,
            "input": {
                "prompt": request.prompt,
                "img_url": request.first_frame_url
            },
            "parameters": {
                "size": size,
                "duration": request.duration,
                "prompt_extend": request.prompt_extend,
                "watermark": request.watermark
            }
        }

        if request.negative_prompt:
            body["input"]["negative_prompt"] = request.negative_prompt
        if request.last_frame_url:
            body["input"]["last_img_url"] = request.last_frame_url
        if request.audio_url:
            body["input"]["audio_url"] = request.audio_url
        if request.audio is not None:
            body["parameters"]["audio"] = request.audio
        if request.seed:
            body["parameters"]["seed"] = request.seed

        async with httpx.AsyncClient(timeout=30.0) as client:
            response = await client.post(url, json=body, headers=headers)
            result = response.json()

        if "code" in result:
            raise Exception(f"创建任务失败: {result.get('message')}")

        task_id = result["output"]["task_id"]
        task_status = result["output"]["task_status"]

        # 保存记录
        record = AIVideo(
            user_id=self.user_id,
            task_id=task_id,
            model_name=model,
            video_type="i2v",
            input_params={
                "prompt": request.prompt,
                "first_frame_url": request.first_frame_url,
                "last_frame_url": request.last_frame_url,
                "resolution": request.resolution,
                "duration": request.duration
            },
            prompt=request.prompt,
            first_frame_url=request.first_frame_url,
            last_frame_url=request.last_frame_url,
            audio_url=request.audio_url,
            resolution=request.resolution,
            status=task_status,
            submit_time=datetime.now()
        )
        self.db.add(record)
        self.db.commit()

        return VideoTaskResponse(task_id=task_id, task_status=task_status)

    async def get_task_status(self, task_id: str) -> VideoTaskResult:
        """查询任务状态"""
        # 验证任务归属
        record = self.db.query(AIVideo).filter(
            AIVideo.task_id == task_id,
            AIVideo.user_id == self.user_id
        ).first()
        if not record:
            raise Exception("任务不存在")

        # 如果已完成，直接返回
        if record.status in ["SUCCEEDED", "FAILED"]:
            return VideoTaskResult(
                task_id=task_id,
                task_status=record.status,
                video_url=record.video_url,
                video_duration=float(record.video_duration) if record.video_duration else None,
                actual_prompt=record.actual_prompt,
                error_message=record.error_message
            )

        # 查询百炼API
        url = f"{DASHSCOPE_BASE_URL}/tasks/{task_id}"
        headers = {"Authorization": f"Bearer {self.api_key}"}

        async with httpx.AsyncClient(timeout=30.0) as client:
            response = await client.get(url, headers=headers)
            result = response.json()

        output = result.get("output", {})
        task_status = output.get("task_status", "UNKNOWN")

        # 更新数据库
        record.status = task_status
        if task_status == "RUNNING" and not record.scheduled_time:
            record.scheduled_time = datetime.now()

        if task_status == "SUCCEEDED":
            record.end_time = datetime.now()

            # 保存actual_prompt
            actual_prompt = output.get("actual_prompt")
            if actual_prompt:
                record.actual_prompt = actual_prompt

            # 费用（API调用免费）
            record.bill = Decimal("0")

            # 先落库（状态 SUCCEEDED），让前端立刻拿到结果
            # OSS 上传放到后台线程，不阻塞当前请求
            dashscope_video_url = output.get("video_url")
            self.db.commit()

            if dashscope_video_url:
                threading.Thread(
                    target=self._upload_video_to_oss_sync,
                    args=(record.id, dashscope_video_url),
                    daemon=True
                ).start()

            return VideoTaskResult(
                task_id=task_id,
                task_status=record.status,
                # OSS 上传未完成前先返回 DashScope 原始 URL，前端可立即播放
                video_url=dashscope_video_url,
                video_duration=float(record.video_duration) if record.video_duration else None,
                actual_prompt=record.actual_prompt,
                error_message=record.error_message
            )

        elif task_status == "FAILED":
            record.end_time = datetime.now()
            record.error_message = output.get("message", "任务失败")

        self.db.commit()

        # 返回时以数据库中最终的 record.status 为准，确保扣费失败后前端不会误判任务为 SUCCEEDED
        return VideoTaskResult(
            task_id=task_id,
            task_status=record.status if record.status else task_status,
            video_url=record.video_url,
            video_duration=float(record.video_duration) if record.video_duration else None,
            actual_prompt=record.actual_prompt,
            error_message=record.error_message
        )

    def _upload_video_to_oss_sync(self, record_id: int, video_url: str) -> None:
        """后台线程：将视频从 DashScope URL 上传到 OSS，完成后更新数据库。

        使用独立的数据库会话，避免与请求会话冲突。
        """
        from app.database import SessionLocal

        try:
            oss_url = self.oss_service.upload_from_url_sync(video_url, "ai-videos")
        except Exception as e:
            logger.error(f"OSS 上传失败: record_id={record_id}, error={e}")
            return

        db = SessionLocal()
        try:
            rec = db.query(AIVideo).filter(AIVideo.id == record_id).first()
            if rec:
                rec.video_url = oss_url
                db.commit()
                logger.info(f"OSS 上传完成: record_id={record_id}, oss_url={oss_url}")
        except Exception as e:
            logger.error(f"更新 OSS URL 失败: record_id={record_id}, error={e}")
            db.rollback()
        finally:
            db.close()

    def get_history(self, page: int = 1, page_size: int = 20) -> VideoHistoryResponse:
        """获取用户历史记录"""
        query = self.db.query(AIVideo).filter(
            AIVideo.user_id == self.user_id,
            AIVideo.review_status != 'rejected'  # 排除被拒绝的内容
        ).order_by(AIVideo.created_at.desc())

        total = query.count()
        offset = (page - 1) * page_size
        records = query.offset(offset).limit(page_size).all()

        items = [
            VideoHistoryItem(
                id=r.id,
                task_id=r.task_id,
                model_name=r.model_name,
                video_type=r.video_type,
                prompt=r.prompt,
                custom_name=r.custom_name,
                video_url=r.video_url,
                video_duration=float(r.video_duration) if r.video_duration else None,
                resolution=r.resolution,
                status=r.status,
                bill=float(r.bill) if r.bill else 0,
                created_at=r.created_at.isoformat() if r.created_at else ""
            )
            for r in records
        ]

        return VideoHistoryResponse(
            items=items,
            total=total,
            page=page,
            page_size=page_size
        )