wehub-resource-sync
460 lines
17 KiB
Python
460 lines
17 KiB
Python
"""
|
|
视频处理工具
|
|
"""
|
|
import subprocess
|
|
import json
|
|
import logging
|
|
import re
|
|
from typing import List, Dict, Optional
|
|
from pathlib import Path
|
|
from .ffmpeg_utils import get_ffmpeg_path, get_ffprobe_path
|
|
|
|
# 修复导入问题
|
|
try:
|
|
from ..core.shared_config import CLIPS_DIR, COLLECTIONS_DIR
|
|
except ImportError:
|
|
# 如果相对导入失败，尝试绝对导入
|
|
import sys
|
|
from pathlib import Path
|
|
backend_path = Path(__file__).parent.parent
|
|
if str(backend_path) not in sys.path:
|
|
sys.path.insert(0, str(backend_path))
|
|
from ..core.shared_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):
|
|
# 强制使用传入的项目特定路径，不使用全局路径作为后备
|
|
if not clips_dir:
|
|
raise ValueError("clips_dir 参数是必需的，不能使用全局路径")
|
|
if not collections_dir:
|
|
raise ValueError("collections_dir 参数是必需的，不能使用全局路径")
|
|
|
|
self.clips_dir = Path(clips_dir)
|
|
self.collections_dir = Path(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" 或 "00:00:06.140")
|
|
|
|
Returns:
|
|
FFmpeg时间格式 (如 "00:00:06.140")
|
|
"""
|
|
# 将逗号替换为点
|
|
return srt_time.replace(',', '.')
|
|
|
|
@staticmethod
|
|
def convert_seconds_to_ffmpeg_time(seconds: float) -> str:
|
|
"""
|
|
将秒数转换为FFmpeg时间格式
|
|
|
|
Args:
|
|
seconds: 秒数
|
|
|
|
Returns:
|
|
FFmpeg时间格式 (如 "00:00:06.140")
|
|
"""
|
|
hours = int(seconds // 3600)
|
|
minutes = int((seconds % 3600) // 60)
|
|
secs = int(seconds % 60)
|
|
milliseconds = int((seconds % 1) * 1000)
|
|
|
|
return f"{hours:02d}:{minutes:02d}:{secs:02d}.{milliseconds:03d}"
|
|
|
|
@staticmethod
|
|
def convert_ffmpeg_time_to_seconds(time_str: str) -> float:
|
|
"""
|
|
将FFmpeg时间格式转换为秒数
|
|
|
|
Args:
|
|
time_str: FFmpeg时间格式 (如 "00:00:06.140")
|
|
|
|
Returns:
|
|
秒数
|
|
"""
|
|
try:
|
|
# 处理毫秒部分
|
|
if '.' in time_str:
|
|
time_part, ms_part = time_str.split('.')
|
|
milliseconds = int(ms_part)
|
|
else:
|
|
time_part = time_str
|
|
milliseconds = 0
|
|
|
|
# 解析时分秒
|
|
h, m, s = map(int, time_part.split(':'))
|
|
|
|
return h * 3600 + m * 60 + s + milliseconds / 1000
|
|
except Exception as e:
|
|
logger.error(f"时间格式转换失败: {time_str}, 错误: {e}")
|
|
return 0.0
|
|
|
|
@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)
|
|
|
|
# 计算持续时间
|
|
start_seconds = VideoProcessor.convert_ffmpeg_time_to_seconds(ffmpeg_start_time)
|
|
end_seconds = VideoProcessor.convert_ffmpeg_time_to_seconds(ffmpeg_end_time)
|
|
duration = end_seconds - start_seconds
|
|
|
|
# 构建优化的FFmpeg命令
|
|
# 使用 -ss 在输入前进行精确定位，使用 -t 指定持续时间
|
|
ffmpeg_bin = get_ffmpeg_path()
|
|
cmd = [
|
|
ffmpeg_bin,
|
|
'-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:
|
|
# 验证输入参数
|
|
if not clips_list:
|
|
logger.error("clips_list为空，无法创建合集")
|
|
return False
|
|
|
|
# 验证所有视频文件是否存在
|
|
valid_clips = []
|
|
for clip_path in clips_list:
|
|
if not clip_path.exists():
|
|
logger.warning(f"视频文件不存在，跳过: {clip_path}")
|
|
continue
|
|
valid_clips.append(clip_path)
|
|
|
|
if not valid_clips:
|
|
logger.error("没有有效的视频文件，无法创建合集")
|
|
return False
|
|
|
|
# 确保输出目录存在
|
|
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 valid_clips:
|
|
# 使用绝对路径并转义单引号
|
|
abs_path = clip_path.absolute()
|
|
escaped_path = str(abs_path).replace("'", "'\"'\"'")
|
|
f.write(f"file '{escaped_path}'\n")
|
|
|
|
# 验证concat文件内容
|
|
if concat_file.stat().st_size == 0:
|
|
logger.error("concat文件为空，无法创建合集")
|
|
concat_file.unlink(missing_ok=True)
|
|
return False
|
|
|
|
# 构建FFmpeg命令 - 使用H.264编码确保兼容性
|
|
ffmpeg_bin = get_ffmpeg_path()
|
|
cmd = [
|
|
ffmpeg_bin,
|
|
'-f', 'concat',
|
|
'-safe', '0',
|
|
'-i', str(concat_file),
|
|
'-c:v', 'libx264', # 使用H.264视频编码
|
|
'-preset', 'ultrafast', # 使用最快的编码预设
|
|
'-crf', '28', # 稍微降低质量以加快编码速度
|
|
'-c:a', 'aac', # 使用AAC音频编码
|
|
'-b:a', '128k', # 音频比特率
|
|
'-movflags', '+faststart', # 优化网络播放
|
|
'-y',
|
|
str(output_path)
|
|
]
|
|
|
|
logger.info(f"执行FFmpeg命令: {' '.join(cmd)}")
|
|
|
|
# 执行命令
|
|
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}")
|
|
logger.error(f"FFmpeg stdout: {result.stdout}")
|
|
return False
|
|
|
|
except Exception as e:
|
|
logger.error(f"视频拼接异常: {str(e)}")
|
|
return False
|
|
|
|
@staticmethod
|
|
def extract_thumbnail(video_path: Path, output_path: Path, time_offset: int = 5) -> bool:
|
|
"""
|
|
从视频中提取缩略图
|
|
|
|
Args:
|
|
video_path: 视频文件路径
|
|
output_path: 输出缩略图路径
|
|
time_offset: 提取时间点（秒）
|
|
|
|
Returns:
|
|
是否成功
|
|
"""
|
|
try:
|
|
# 确保输出目录存在
|
|
output_path.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
# 构建FFmpeg命令
|
|
cmd = [
|
|
'ffmpeg',
|
|
'-i', str(video_path),
|
|
'-ss', str(time_offset),
|
|
'-vframes', '1',
|
|
'-q:v', '2', # 高质量
|
|
'-y', # 覆盖输出文件
|
|
str(output_path)
|
|
]
|
|
|
|
# 执行命令
|
|
result = subprocess.run(cmd, capture_output=True, text=True, encoding='utf-8', errors='ignore')
|
|
|
|
if result.returncode == 0 and output_path.exists():
|
|
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:
|
|
ffprobe_bin = get_ffprobe_path()
|
|
cmd = [
|
|
ffprobe_bin,
|
|
'-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']
|
|
|
|
# 处理时间格式 - 如果是秒数，转换为SRT格式
|
|
if isinstance(start_time, (int, float)):
|
|
start_time = VideoProcessor.convert_seconds_to_ffmpeg_time(start_time)
|
|
if isinstance(end_time, (int, float)):
|
|
end_time = VideoProcessor.convert_seconds_to_ffmpeg_time(end_time)
|
|
|
|
# 使用标题作为文件名，并清理不合法的字符
|
|
# 在文件名中包含clip_id，便于后续合集拼接时查找
|
|
safe_title = VideoProcessor.sanitize_filename(title)
|
|
output_path = self.clips_dir / f"{clip_id}_{safe_title}.mp4"
|
|
|
|
logger.info(f"提取切片 {clip_id}: {start_time} -> {end_time}, 输出: {output_path}")
|
|
|
|
if VideoProcessor.extract_clip(input_video, output_path, start_time, end_time):
|
|
successful_clips.append(output_path)
|
|
logger.info(f"切片 {clip_id} 提取成功")
|
|
else:
|
|
logger.error(f"切片 {clip_id} 提取失败")
|
|
|
|
return successful_clips
|
|
|
|
def create_collections_from_metadata(self, collections_data: List[Dict]) -> List[Dict]:
|
|
"""
|
|
根据元数据创建合集
|
|
|
|
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):
|
|
# 生成合集缩略图
|
|
thumbnail_path = None
|
|
try:
|
|
thumbnail_filename = f"{collection_id}_{safe_title}_thumbnail.jpg"
|
|
thumbnail_path = self.collections_dir / thumbnail_filename
|
|
|
|
# 从视频中提取缩略图（第2秒的帧）
|
|
thumbnail_success = VideoProcessor.extract_thumbnail(output_path, thumbnail_path, time_offset=2)
|
|
if thumbnail_success:
|
|
logger.info(f"合集 {collection_id} 缩略图生成成功: {thumbnail_path}")
|
|
else:
|
|
logger.warning(f"合集 {collection_id} 缩略图生成失败")
|
|
thumbnail_path = None
|
|
except Exception as e:
|
|
logger.error(f"生成合集 {collection_id} 缩略图时出错: {e}")
|
|
thumbnail_path = None
|
|
|
|
# 返回包含视频路径和缩略图路径的信息
|
|
collection_info = {
|
|
'collection_id': collection_id,
|
|
'video_path': str(output_path),
|
|
'thumbnail_path': str(thumbnail_path) if thumbnail_path else None,
|
|
'title': collection_title
|
|
}
|
|
successful_collections.append(collection_info)
|
|
logger.info(f"成功创建合集 {collection_id}: {output_path}")
|
|
else:
|
|
logger.warning(f"合集 {collection_id} 没有找到任何有效的切片文件")
|
|
|
|
return successful_collections |