NarratoAI/app/services/video_service.py

import os
from uuid import uuid4
from loguru import logger
from typing import Dict, List, Optional, Tuple

from app.services import material


class VideoService:
    @staticmethod
    async def crop_video(
        video_path: str,
        video_script: List[dict]
    ) -> Tuple[str, Dict[str, str]]:
        """
        裁剪视频服务
        
        Args:
            video_path: 视频文件路径
            video_script: 视频脚本列表
            
        Returns:
            Tuple[str, Dict[str, str]]: (task_id, 裁剪后的视频片段字典)
            视频片段字典格式: {timestamp: video_path}
        """
        try:
            task_id = str(uuid4())
            
            # 从脚本中提取时间戳列表
            time_list = [scene['timestamp'] for scene in video_script]
            
            # 调用裁剪服务
            subclip_videos = material.clip_videos(
                task_id=task_id,
                timestamp_terms=time_list,
                origin_video=video_path
            )
            
            if subclip_videos is None:
                raise ValueError("裁剪视频失败")
                
            # 更新脚本中的视频路径
            for scene in video_script:
                try:
                    scene['path'] = subclip_videos[scene['timestamp']]
                except KeyError as err:
                    logger.error(f"更新视频路径失败: {err}")
                    
            logger.debug(f"裁剪视频成功，共生成 {len(time_list)} 个视频片段")
            logger.debug(f"视频片段路径: {subclip_videos}")
            
            return task_id, subclip_videos
            
        except Exception as e:
            logger.exception("裁剪视频失败")
            raise