299 lines
11 KiB
Python
299 lines
11 KiB
Python
"""
|
|
视频处理工具
|
|
"""
|
|
import subprocess
|
|
import json
|
|
import logging
|
|
import re
|
|
from typing import List, Dict, Optional
|
|
from pathlib import Path
|
|
|
|
# 修复导入问题
|
|
try:
|
|
from config import CLIPS_DIR, COLLECTIONS_DIR
|
|
except ImportError:
|
|
# 如果相对导入失败，尝试绝对导入
|
|
import sys
|
|
from pathlib import Path
|
|
shared_path = Path(__file__).parent.parent
|
|
if str(shared_path) not in sys.path:
|
|
sys.path.insert(0, str(shared_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 |