在需要对视频进行批量处理的时候，常见的就是批量去视频的片头和片尾：
具体代码示例：
import os import re import time import subprocess from decimal import Decimal from multiprocessing import Poolpath = r'E:\FYZ2025\deleteAD\原视频' # 原文件夹路径 new_path = r'E:\FYZ2025\deleteAD\新视频' # 新文件夹路径 if not os.path.exists(new_path):os.mkdir(new_path) else:pass# 获取视频的 duration 时长 长 宽 def get_video_length(file):process = subprocess.Popen(['ffmpeg', '-i', file],stdout=subprocess.PIPE,stderr=subprocess.STDOUT)stdout, stderr = process.communicate()# print(stdout)pattern_duration = re.compile(r"Duration:\s(\d+?):(\d+?):(\d+\.\d+?),")pattern_size = re.compile(r",\s(\d{3,4})x(\d{3,4})[\s or ,]")matches = re.search(pattern_duration, stdout.decode('utf-8'))size = re.search(pattern_size, stdout.decode('utf-8'))if size:size = size.groups()# print(size)if matches:matches = matches.groups()# print(matches)hours = Decimal(matches[0])minutes = Decimal(matches[1])seconds = Decimal(matches[2]) # 处理为十进制，避免小数点报错total = 0total += 60 * 60 * hourstotal += 60 * minutestotal += secondswidth = size[0]height = size[1]return {'total': total, 'width': width, 'height': height}def cutVideo(startPoint, file, endPoint, newFile):command = ['ffmpeg', '-ss', startPoint, '-i', file, '-acodec', 'copy', '-vcodec','copy', '-t', endPoint, newFile]subprocess.call(command)def millisecToAssFormat(t): # 取时间，单位秒s = t % 60m = t // 60if t < 3600:h = 00else:h = t // 3600m = t // 60 - h * 60return '%02d:%02d:%02d' % (h, m, s)def main(dict):file = dict['file'] # 文件名piantou = dict['piantou'] # 片头时长pianwei = dict['pianwei'] # 片尾时长videoInfo = get_video_length(file) # 视频信息# print(videoInfo)if videoInfo:duration = videoInfo.get('total') # 时长 秒startPoint = piantou # 剪掉片头时间,从原文件此处开始播放startPoint = millisecToAssFormat(startPoint)endPoint = duration - piantou - pianwei # 剪掉片头片尾时间和,结果等于新文件总时长endPoint = millisecToAssFormat(endPoint)new_File = new_path + file.replace(path, '') # 创建生成的文件路径+文件名# print(new_File, endPoint) cutVideo(startPoint, file, endPoint, new_File)if __name__ == '__main__':# main()timt0 = time.time()file = [os.path.join(path, file) for file in os.listdir(path)]a = input('输入片头时长：') # 片头时长b = input('输入片尾时长：') # 片尾时长piantou = [int(a)] * len(file) # 片头时长列表pianwei = [int(b)] * len(file) # 片尾时长列表dict_list = [] # main函数参数列表for x in range(len(file)):dict = {'file': file[x],'piantou': piantou[x],'pianwei': pianwei[x]} # main函数参数，文件路径，片头片尾时长 dict_list.append(dict)print('\n规则：去除片头%s秒片尾%s秒,开始。\n' % (a, b))pool = Pool()pool.map(main, dict_list)pool.close()pool.join()time1 = time.time() - timt0print('\n结束，处理%d个视频文件，共用时%.4f秒。\n' % (len(file), time1))input('回车键退出')
这个代码是基础的代码，片头和片尾的时长是秒。
但是我实际使用过程中，浮点的小数秒。
经过豆包AI的优化：
import os import re import time import subprocess from decimal import Decimal from multiprocessing import Pool from typing import List, Dict# 配置路径（建议用os.path.join统一处理路径分隔符） path = r'C:\2025电商美工\原视频' new_path = r'C:\2025电商美工\新视频' os.makedirs(new_path, exist_ok=True) # 简化文件夹创建逻辑# 支持的视频格式（可根据需求扩展） SUPPORTED_FORMATS = ('.mp4', '.avi', '.mov', '.mkv', '.flv', '.wmv')# 验证ffmpeg是否可用 def check_ffmpeg() -> bool:try:subprocess.run(['ffmpeg', '-version'], capture_output=True, check=True)return Trueexcept (subprocess.CalledProcessError, FileNotFoundError):print("错误：未找到ffmpeg！请先安装并配置环境变量（参考之前的指导）")return False# 获取视频信息（优化正则+错误处理） def get_video_length(file: str) -> Dict[str, Decimal | str] | None:try:process = subprocess.Popen(['ffmpeg', '-i', file],stdout=subprocess.PIPE,stderr=subprocess.STDOUT,encoding='utf-8', # 直接指定编码，避免手动decodeerrors='ignore' # 忽略编码错误 )stdout, _ = process.communicate(timeout=30) # 设置超时，避免卡死except Exception as e:print(f"获取视频信息失败：{file}，错误：{str(e)}")return None# 优化正则：放宽匹配条件，适应不同ffmpeg版本pattern_duration = re.compile(r"Duration:\s*(\d+):(\d+):(\d+\.\d+)", re.IGNORECASE)pattern_size = re.compile(r"(\d{3,4})x(\d{3,4})", re.IGNORECASE)duration_match = pattern_duration.search(stdout)size_match = pattern_size.search(stdout)if not (duration_match and size_match):print(f"无法提取视频信息：{file}（可能不是支持的视频格式）")return None# 解析时长hours = Decimal(duration_match.group(1))minutes = Decimal(duration_match.group(2))seconds = Decimal(duration_match.group(3))total_seconds = hours * 3600 + minutes * 60 + seconds# 解析分辨率width, height = size_match.groups()return {'total': total_seconds,'width': width,'height': height}# 裁剪视频（优化命令+错误处理） def cutVideo(startPoint: str, file: str, endPoint: str, newFile: str) -> None:# 确保输出目录存在（处理子文件夹场景）output_dir = os.path.dirname(newFile)os.makedirs(output_dir, exist_ok=True)command = ['ffmpeg', '-ss', startPoint, '-i', file,'-acodec', 'copy', '-vcodec', 'copy','-t', endPoint, '-y', # -y：覆盖已存在的输出文件 newFile]try:# 捕获ffmpeg输出，便于调试result = subprocess.run(command, capture_output=True, encoding='utf-8', errors='ignore', timeout=300)if result.returncode != 0:print(f"裁剪失败：{file}，错误信息：{result.stderr[:500]}") # 只打印前500字符else:print(f"裁剪成功：{newFile}")except Exception as e:print(f"裁剪异常：{file}，错误：{str(e)}")# 时间格式转换（支持小数秒，保留3位精度） def secToFFmpegFormat(t: float) -> str:if t < 0:return "00:00:00.000"hours = int(t // 3600)minutes = int((t % 3600) // 60)seconds = t % 60return f"{hours:02d}:{minutes:02d}:{seconds:06.3f}" # 格式：hh:mm:ss.sss# 输入验证函数 def get_valid_duration(prompt: str) -> float | None:while True:user_input = input(prompt).strip()try:duration = float(user_input)if duration >= 0:return durationelse:print("错误：时长不能为负数，请重新输入！")except ValueError:print("错误：请输入有效的数字（支持小数，如2.5）！")# 主处理函数 def main(task: Dict[str, str | float]) -> None:file = task['file']piantou = task['piantou']pianwei = task['pianwei']# 获取视频信息videoInfo = get_video_length(file)if not videoInfo:returnduration = videoInfo['total'] # Decimal类型total_cut = piantou + pianwei # float类型# 关键修改：统一数据类型为Decimal，避免运算错误total_cut_decimal = Decimal(str(total_cut)) # 先转字符串再转Decimal，保证精度# 验证裁剪时长是否合理if total_cut_decimal >= duration:print(f"跳过：{file}（片头+片尾时长{total_cut:.1f}秒 ≥ 视频总时长{duration:.1f}秒）")return# 计算裁剪参数startPoint = secToFFmpegFormat(piantou)endPoint_seconds = duration - total_cut_decimal # Decimal类型运算endPoint = secToFFmpegFormat(float(endPoint_seconds)) # 转为float适配时间格式函数# 生成输出路径（优化路径拼接逻辑）relative_path = os.path.relpath(file, path)new_File = os.path.join(new_path, relative_path)# 执行裁剪 cutVideo(startPoint, file, endPoint, new_File)# 递归获取所有支持的视频文件 def get_all_video_files(folder: str) -> List[str]:video_files = []for root, _, files in os.walk(folder):for file in files:if file.lower().endswith(SUPPORTED_FORMATS):video_files.append(os.path.join(root, file))return video_filesif __name__ == '__main__':# 检查ffmpeg可用性if not check_ffmpeg():input("按回车键退出...")exit(1)# 获取所有视频文件video_files = get_all_video_files(path)if not video_files:print(f"未在{path}找到支持的视频文件（支持格式：{SUPPORTED_FORMATS}）")input("按回车键退出...")exit(0)print(f"共找到{len(video_files)}个视频文件，准备处理...")# 获取用户输入的时长（带验证）piantou_duration = get_valid_duration("输入片头时长（秒，支持小数）：")pianwei_duration = get_valid_duration("输入片尾时长（秒，支持小数）：")if piantou_duration is None or pianwei_duration is None:exit(1)# 构建任务列表dict_list = [{'file': file,'piantou': piantou_duration,'pianwei': pianwei_duration}for file in video_files]# 多进程处理（限制进程数为CPU核心数的一半，避免资源占用过高）print(f"\n规则：去除片头{piantou_duration}秒、片尾{pianwei_duration}秒，开始处理...\n")timt0 = time.time()with Pool(processes=os.cpu_count() // 2) as pool: # 优雅的进程池管理 pool.map(main, dict_list)# 输出统计信息time1 = time.time() - timt0print(f"\n处理完成！共处理{len(video_files)}个视频文件，总耗时{time1:.4f}秒。")input("按回车键退出...")
简直了YYDS！
打完收工！
)
