repo_description: AutoClip: AI-powered video clipping and highlight generation · 一款智能高光提取与剪辑的二创工具
repo_topics: ai-video-editing auto-highlight llm video video-editing
code: """
  视频处理工具 - 封装FFmpeg命令用于视频切割和拼接
  """
  import subprocess
  import json
  import logging
  import re
  from typing import List, Dict, Optional
  from pathlib import Path
  
  from ..config import CLIPS_DIR, COLLECTIONS_DIR
  
  logger = logging.getLogger(__name__)
  
  class VideoProcessor:
      """视频处理工具类"""
      
      def __init__(self, clips_dir: Optional[str] = None, collections_dir: Optional[str] = None):
          self.clips_dir = Path(clips_dir) if clips_dir else CLIPS_DIR
          self.collections_dir = Path(collections_dir) if collections_dir else COLLECTIONS_DIR
      
      @staticmethod
      def sanitize_filename(filename: str) -> str:
          """
          清理文件名，移除或替换不合法的字符
          
          Args:
              filename: 原始文件名
              
          Returns:
              清理后的文件名
          """
          # 移除或替换不合法的字符
          # Windows和Unix系统都不允许的字符: < > : " | ? * \ /
          # 替换为下划线
          sanitized = re.sub(r'[<>:"|?*\\/]', '_', filename)
          
          # 移除前后空格和点
          sanitized = sanitized.strip(' .')
          
          # 限制长度，避免文件名过长
          if len(sanitized) > 100:
              sanitized = sanitized[:100]
          
          # 确保文件名不为空
          if not sanitized:
              sanitized = "untitled"
              
          return sanitized
      
      @staticmethod
      def convert_srt_time_to_ffmpeg_time(srt_time: str) -> str:
          """
          将SRT时间格式转换为FFmpeg时间格式
          
          Args:
              srt_time: SRT时间格式 (如 "00:00:06,140")
              
          Returns:
              FFmpeg时间格式 (如 "00:00:06.140")
          """
          # 将逗号替换为点
          return srt_time.replace(',', '.')
      
      @staticmethod
      def extract_clip(input_video: Path, output_path: Path, 
                      start_time: str, end_time: str) -> bool:
          """
          从视频中提取指定时间段的片段
          
          Args:
              input_video: 输入视频路径
              output_path: 输出视频路径
              start_time: 开始时间 (格式: "00:01:25,140")
              end_time: 结束时间 (格式: "00:02:53,500")
              
          Returns:
              是否成功
          """
          try:
              # 确保输出目录存在
              output_path.parent.mkdir(parents=True, exist_ok=True)
              
              # 转换时间格式：从SRT格式转换为FFmpeg格式
              ffmpeg_start_time = VideoProcessor.convert_srt_time_to_ffmpeg_time(start_time)
              ffmpeg_end_time = VideoProcessor.convert_srt_time_to_ffmpeg_time(end_time)
              
              # 计算持续时间
              def time_to_seconds(time_str: str) -> float:
                  """将时间字符串转换为秒数"""
                  h, m, s = time_str.split(':')
                  s, ms = s.split('.')
                  return int(h) * 3600 + int(m) * 60 + int(s) + int(ms) / 1000
              
              start_seconds = time_to_seconds(ffmpeg_start_time)
              end_seconds = time_to_seconds(ffmpeg_end_time)
              duration = end_seconds - start_seconds
              
              # 构建优化的FFmpeg命令
              # 使用 -ss 在输入前进行精确定位，使用 -t 指定持续时间
              cmd = [
                  'ffmpeg',
                  '-ss', ffmpeg_start_time,  # 在输入前定位，更精确
                  '-i', str(input_video),
                  '-t', str(duration),  # 使用持续时间而不是绝对结束时间
                  '-c:v', 'copy',  # 复制视频流
                  '-c:a', 'copy',  # 复制音频流
                  '-avoid_negative_ts', 'make_zero',
                  '-y',  # 覆盖输出文件
                  str(output_path)
              ]
              
              # 执行命令
              result = subprocess.run(cmd, capture_output=True, text=True, encoding='utf-8', errors='ignore')
              
              if result.returncode == 0:
                  logger.info(f"成功提取视频片段: {output_path} ({ffmpeg_start_time} -> {ffmpeg_end_time}, 时长: {duration:.2f}秒)")
                  return True
              else:
                  logger.error(f"提取视频片段失败: {result.stderr}")
                  return False
                  
          except Exception as e:
              logger.error(f"视频处理异常: {str(e)}")
              return False
      
      @staticmethod
      def create_collection(clips_list: List[Path], output_path: Path) -> bool:
          """
          将多个视频片段拼接成合集
          
          Args:
              clips_list: 视频片段路径列表
              output_path: 输出合集路径
              
          Returns:
              是否成功
          """
          try:
              # 确保输出目录存在
              output_path.parent.mkdir(parents=True, exist_ok=True)
              
              # 创建concat文件
              concat_file = output_path.parent / "concat_list.txt"
              
              with open(concat_file, 'w', encoding='utf-8') as f:
                  for clip_path in clips_list:
                      f.write(f"file '{clip_path.absolute()}'\n")
              
              # 构建FFmpeg命令
              cmd = [
                  'ffmpeg',
                  '-f', 'concat',
                  '-safe', '0',
                  '-i', str(concat_file),
                  '-c', 'copy',
                  '-y',
                  str(output_path)
              ]
              
              # 执行命令
              result = subprocess.run(cmd, capture_output=True, text=True, encoding='utf-8', errors='ignore')
              
              # 清理临时文件
              concat_file.unlink(missing_ok=True)
              
              if result.returncode == 0:
                  logger.info(f"成功创建合集: {output_path}")
                  return True
              else:
                  logger.error(f"创建合集失败: {result.stderr}")
                  return False
                  
          except Exception as e:
              logger.error(f"视频拼接异常: {str(e)}")
              return False
      
      @staticmethod
      def get_video_info(video_path: Path) -> Dict:
          """
          获取视频信息
          
          Args:
              video_path: 视频文件路径
              
          Returns:
              视频信息字典
          """
          try:
              cmd = [
                  'ffprobe',
                  '-v', 'quiet',
                  '-print_format', 'json',
                  '-show_format',
                  '-show_streams',
                  str(video_path)
              ]
              
              result = subprocess.run(cmd, capture_output=True, text=True, encoding='utf-8', errors='ignore')
              
              if result.returncode == 0:
                  info = json.loads(result.stdout)
                  return {
                      'duration': float(info['format']['duration']),
                      'size': int(info['format']['size']),
                      'bitrate': int(info['format']['bit_rate']),
                      'streams': info['streams']
                  }
              else:
                  logger.error(f"获取视频信息失败: {result.stderr}")
                  return {}
                  
          except Exception as e:
              logger.error(f"获取视频信息异常: {str(e)}")
              return {}
      
      def batch_extract_clips(self, input_video: Path, clips_data: List[Dict]) -> List[Path]:
          """
          批量提取视频片段
          
          Args:
              input_video: 输入视频路径
              clips_data: 片段数据列表，每个元素包含id、title、start_time、end_time
              
          Returns:
              成功提取的片段路径列表
          """
          successful_clips = []
          
          for clip_data in clips_data:
              clip_id = clip_data['id']
              title = clip_data.get('title', f"片段_{clip_id}")
              start_time = clip_data['start_time']
              end_time = clip_data['end_time']
              
              # 使用标题作为文件名，并清理不合法的字符
              # 在文件名中包含clip_id，便于后续合集拼接时查找
              safe_title = VideoProcessor.sanitize_filename(title)
              output_path = self.clips_dir / f"{clip_id}_{safe_title}.mp4"
              
              if VideoProcessor.extract_clip(input_video, output_path, start_time, end_time):
                  successful_clips.append(output_path)
          
          return successful_clips
      
      def create_collections_from_metadata(self, collections_data: List[Dict]) -> List[Path]:
          """
          根据元数据创建合集
          
          Args:
              collections_data: 合集数据列表
              
          Returns:
              成功创建的合集路径列表
          """
          successful_collections = []
          
          for collection_data in collections_data:
              collection_id = collection_data['id']
              collection_title = collection_data.get('collection_title', f'合集_{collection_id}')
              clip_ids = collection_data['clip_ids']
              
              # 构建片段路径列表
              clips_list = []
              for clip_id in clip_ids:
                  # 查找对应的切片文件
                  # 新的文件名格式是: {clip_id}_{title}.mp4
                  clip_path = self.clips_dir / f"{clip_id}_*.mp4"
                  found_clips = list(self.clips_dir.glob(f"{clip_id}_*.mp4"))
                  
                  if found_clips:
                      found_clip = found_clips[0]  # 取第一个匹配的文件
                      clips_list.append(found_clip)
                      logger.info(f"找到合集 {collection_id} 的切片: {found_clip.name}")
                  else:
                      logger.warning(f"未找到合集 {collection_id} 的切片 {clip_id}")
              
              if clips_list:
                  # 使用collection_title作为文件名，并清理不合法的字符
                  safe_title = VideoProcessor.sanitize_filename(collection_title)
                  output_path = self.collections_dir / f"{safe_title}.mp4"
                  
                  if VideoProcessor.create_collection(clips_list, output_path):
                      successful_collections.append(output_path)
                      logger.info(f"成功创建合集 {collection_id}: {output_path}")
              else:
                  logger.warning(f"合集 {collection_id} 没有找到任何有效的切片文件")
          
          return successful_collections