import os
import sys
import argparse
import logging
import concurrent.futures
from pathlib import Path
from datetime import datetime
import cv2
import imageio
from PIL import Image
import subprocess
import msvcrt
# 修复Windows中文路径问题
sys.getfilesystemencoding = lambda: 'utf-8'
# 日志配置
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s [%(levelname)s] %(message)s',
handlers=[
logging.FileHandler('conversion.log', encoding='utf-8'),
logging.StreamHandler()
]
)
def win_path_convert(path):
"""Windows路径兼容性处理"""
return Path(path).resolve().as_posix()
def validate_environment():
"""环境检查与硬件加速支持检测"""
try:
result = subprocess.run(
['ffmpeg', '-hwaccels'],
capture_output=True,
text=True,
check=True
)
if 'dxva2' not in result.stdout.lower():
logging.warning(" 未检测到DXVA2硬件加速支持，性能将受影响")
except FileNotFoundError:
raise RuntimeError("请先安装FFmpeg并添加至PATH环境变量")
def generate_output_paths(src, output_root, platform):
"""生成标准化输出路径（含时间戳防重复）"""
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
src_stem = Path(src).stem[:30] # 限制文件名长度
base_name = f"{src_stem}_{timestamp}"
output_root = Path(output_root) / platform
output_root.mkdir(parents=True, exist_ok=True)
if platform == 'ios':
return (
output_root / f"{base_name}.jpeg",
output_root / f"{base_name}.mov"
)
else:
return (
output_root / f"{base_name}.jpg",
output_root / f"{base_name}.jpg"
)
def process_single_video(src, cover_path, video_path, platform):
"""单文件处理核心逻辑"""
try:
# 硬件加速解码配置
ffmpeg_base = [
'ffmpeg', '-hwaccel', 'dxva2',
'-hwaccel_output_format', 'dxva2_vld',
'-y', '-i', win_path_convert(src)
]
# 关键帧提取
cap = cv2.VideoCapture(win_path_convert(src))
cap.set(cv2.CAP_PROP_POS_MSEC, 2000) # 取2秒位置
ret, frame = cap.read()
if not ret:
raise ValueError("视频关键帧提取失败")
# 封面保存（带EXIF）
with Image.fromarray(cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)) as img:
exif_data = img.info.get('exif', b'')
img.save(
win_path_convert(cover_path),
quality=95,
exif=exif_data,
subsampling=0 # 关闭色度抽样保持最佳质量
)
# 视频转码
ffmpeg_cmd = ffmpeg_base + [
'-ss', '0', '-t', '15', # 截取前15秒
'-vcodec', 'hevc', '-tag:v', 'hvc1',
'-acodec', 'aac', '-b:a', '192k',
'-vf', 'scale=ceil(iw/2)*2:ceil(ih/2)*2', # 确保分辨率合规
'-movflags', '+faststart',
win_path_convert(video_path)
]
subprocess.run(
ffmpeg_cmd,
check=True,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL
)
# Android特殊封装
if platform == 'android':
with open(video_path, 'rb') as f:
video_data = f.read()
with Image.open(cover_path) as img:
img.save(
video_path,
'jpeg',
quality=95,
save_all=True,
append_images=[Image.new('RGB', (1, 1))],
append_images_data=[video_data]
)
logging.info(f" 成功处理：{Path(src).name}")
return True
except Exception as e:
logging.error(f" 处理失败：{Path(src).name} - {str(e)}")
return False
def batch_convert(input_path, output_root, platform='ios', max_workers=4):
"""批处理主程序"""
validate_environment()
# 构建任务列表
input_path = Path(input_path)
if input_path.is_file():
task_list = [input_path]
else:
task_list = list(input_path.rglob('*.mp4')) + list(input_path.rglob('*.mov'))
# 进度跟踪
total = len(task_list)
success = 0
# 多线程处理
with concurrent.futures.ThreadPoolExecutor(
max_workers=min(max_workers, os.cpu_count() * 2),
thread_name_prefix='ConvWorker'
) as executor:
futures = []
for src in task_list:
cover_path, video_path = generate_output_paths(src, output_root, platform)
futures.append(
executor.submit(
process_single_video,
str(src),
str(cover_path),
str(video_path),
platform
)
)
for future in concurrent.futures.as_completed(futures):
if future.result():
success += 1
# 生成报告
logging.info("\n=== 转换完成 ===")
logging.info(f" 总处理文件：{total} 个")
logging.info(f" 成功转换：{success} 个 ({success / total:.1%})")
logging.info(f" 输出目录：{Path(output_root).resolve()}")
'''
# 安装依赖（管理员权限运行）
pip install opencv-python imageio Pillow
scoop install ffmpeg
# 转换单个文件（iOS）
python mp4_to_live_photo.py "D:\My Videos\test.mp4" -o "C:\LivePhotos"
# 批量转换目录（Android平台/8线程）
python mp4_to_live_photo.py "E:\Camera" -p android -j 8
'''
if __name__ == "__main__":
# 命令行参数解析
parser = argparse.ArgumentParser(
description='MP4批量转LivePhoto工具（Windows优化版）',
formatter_class=argparse.ArgumentDefaultsHelpFormatter
)
parser.add_argument('input', help='输入文件/目录路径')
parser.add_argument('-o', '--output', default='./LiveOutput',
help='输出目录路径')
parser.add_argument('-p', '--platform', choices=['ios', 'android'],
default='ios', help='目标平台')
parser.add_argument('-j', '--jobs', type=int,
default=os.cpu_count(), help='并行任务数')
args = parser.parse_args()
try:
batch_convert(
win_path_convert(args.input),
win_path_convert(args.output),
args.platform,
args.jobs
)
except KeyboardInterrupt:
logging.warning(" 用户中断操作！")
except Exception as e:
logging.critical(f" 致命错误：{str(e)}")
