AI视频编辑与后期制作完全教程
从零掌握AI视频编辑技术栈，构建智能视频处理工具链
一、前言
视频内容已成为互联网信息传播的主流形式。传统视频编辑依赖专业软件和人工操作，耗时且门槛高。随着AI技术的突破——从自动字幕生成到风格迁移、从智能裁剪到物体移除——视频编辑正在经历一场效率革命。
本教程系统讲解AI视频编辑与后期制作的核心技术，涵盖商业工具对比、开源方案实现、自动化流水线搭建等内容。无论你是短视频创作者、视频平台开发者，还是企业内容团队的技术负责人，都能从中找到可落地的解决方案。
二、AI视频编辑工具全景
2.1 商业工具对比
| 工具 | 核心AI能力 | 适用场景 | 价格 | API |
|---|---|---|---|---|
| Runway Gen-3 | 文生视频、图生视频、视频编辑 | 创意视频、特效制作 | $12/月起 | 有 |
| Premiere Pro AI | 自动剪辑、场景检测、音频清理 | 专业剪辑、影视制作 | $23/月 | 有限 |
| DaVinci Resolve AI | 人脸追踪、速度扭曲、调色 | 调色、后期合成 | 免费+Studio $295 | 无 |
| CapCut/剪映 | 自动字幕、智能抠图、AI特效 | 短视频、社交媒体 | 免费+Pro | 有 |
| Descript | 文本驱动编辑、AI配音、填充词删除 | 播客、访谈、教程 | $24/月 | 有 |
| Synthesia | AI数字人、文本转视频 | 企业培训、产品介绍 | $22/月 | 有 |
2.2 工具选择决策树
需要AI生成全新视频？
├─ 是 → Runway Gen-3 / Pika / Sora
└─ 否 → 需要编辑已有视频？
├─ 短视频/社交媒体 → CapCut/剪映
├─ 专业影视后期 → DaVinci Resolve + Premiere
├─ 播客/访谈 → Descript
├─ 企业批量生产 → Synthesia + 自建Pipeline
└─ 开发者集成 → FFmpeg + 开源AI模型
三、自动字幕生成与翻译
3.1 基于Whisper的语音转文字
OpenAI的Whisper是目前最强大的开源语音识别模型，支持99种语言，中文识别准确率极高。
import whisper
import json
import srt
from datetime import timedelta
class SubtitleGenerator:
"""基于Whisper的自动字幕生成器"""
def __init__(self, model_size: str = "medium"):
"""
初始化字幕生成器
Args:
model_size: 模型大小 (tiny/base/small/medium/large)
中文推荐 medium 或 large
"""
print(f"加载Whisper模型: {model_size}")
self.model = whisper.load_model(model_size)
print("模型加载完成")
def transcribe(self, audio_path: str, language: str = "zh",
task: str = "transcribe") -> dict:
"""
转录音频/视频
Args:
audio_path: 音频或视频文件路径
language: 语言代码 (zh/en/ja等)
task: transcribe(转录) 或 translate(翻译成英文)
Returns:
包含时间戳的转录结果
"""
result = self.model.transcribe(
audio_path,
language=language,
task=task,
verbose=False,
word_timestamps=True # 启用词级时间戳
)
return result
def generate_srt(self, result: dict, output_path: str,
max_chars_per_line: int = 30):
"""生成SRT字幕文件"""
subtitles = []
for i, segment in enumerate(result['segments'], 1):
text = segment['text'].strip()
# 长文本分行
lines = self._split_text(text, max_chars_per_line)
start = timedelta(seconds=segment['start'])
end = timedelta(seconds=segment['end'])
subtitles.append(srt.Subtitle(
index=i,
start=start,
end=end,
content='\n'.join(lines)
))
with open(output_path, 'w', encoding='utf-8') as f:
f.write(srt.compose(subtitles))
print(f"SRT字幕已生成: {output_path}")
return output_path
def generate_ass(self, result: dict, output_path: str,
font_size: int = 24, font_name: str = "Microsoft YaHei",
position: str = "bottom"):
"""生成ASS字幕文件（支持更丰富的样式）"""
# ASS文件头
ass_content = f"""[Script Info]
Title: AI Generated Subtitles
ScriptType: v4.00+
PlayResX: 1920
PlayResY: 1080
[V4+ Styles]
Format: Name,Fontname,Fontsize,PrimaryColour,SecondaryColour,OutlineColour,BackColour,Bold,Italic,Underline,StrikeOut,ScaleX,ScaleY,Spacing,Angle,BorderStyle,Outline,Shadow,Alignment,MarginL,MarginR,MarginV,Encoding
Style: Default,{font_name},{font_size},&H00FFFFFF,&H000000FF,&H00000000,&H80000000,-1,0,0,0,100,100,0,0,1,2,1,{2 if position == 'bottom' else 8},20,20,30,1
[Events]
Format: Layer,Start,End,Style,Name,MarginL,MarginR,MarginV,Effect,Text
"""
for segment in result['segments']:
start = self._seconds_to_ass_time(segment['start'])
end = self._seconds_to_ass_time(segment['end'])
text = segment['text'].strip().replace('\n', '\\N')
ass_content += f"Dialogue: 0,{start},{end},Default,,0,0,0,,{text}\n"
with open(output_path, 'w', encoding='utf-8') as f:
f.write(ass_content)
print(f"ASS字幕已生成: {output_path}")
return output_path
def _split_text(self, text: str, max_chars: int) -> list:
"""智能分行"""
if len(text) <= max_chars:
return [text]
# 优先在标点处断行
punctuations = '，。、；！？,;!?'
lines = []
current = ""
for char in text:
current += char
if char in punctuations and len(current) >= max_chars // 2:
lines.append(current)
current = ""
if current:
if lines and len(current) < max_chars // 3:
lines[-1] += current
else:
lines.append(current)
return lines
def _seconds_to_ass_time(self, seconds: float) -> str:
"""秒数转ASS时间格式"""
h = int(seconds // 3600)
m = int((seconds % 3600) // 60)
s = seconds % 60
return f"{h}:{m:02d}:{s:05.2f}"
# 使用示例
# generator = SubtitleGenerator(model_size="medium")
# result = generator.transcribe("input_video.mp4", language="zh")
# generator.generate_srt(result, "output.srt")
# generator.generate_ass(result, "output.ass", font_size=28)
3.2 字幕翻译
from openai import OpenAI
class SubtitleTranslator:
"""AI字幕翻译器"""
def __init__(self, api_key: str, base_url: str = None,
model: str = "gpt-4"):
self.client = OpenAI(api_key=api_key, base_url=base_url)
self.model = model
def translate_segments(self, segments: list,
source_lang: str = "zh",
target_lang: str = "en",
batch_size: int = 20) -> list:
"""批量翻译字幕片段"""
translated = []
for i in range(0, len(segments), batch_size):
batch = segments[i:i+batch_size]
# 构建批量翻译请求
texts = [seg['text'].strip() for seg in batch]
prompt = f"""请将以下{source_lang}字幕翻译为{target_lang}。
要求：
1. 保持口语化风格
2. 保持原文的情感和语气
3. 专有名词保留原文
4. 输出JSON数组，每个元素包含原文和译文
原文：
{json.dumps(texts, ensure_ascii=False)}"""
response = self.client.chat.completions.create(
model=self.model,
messages=[{"role": "user", "content": prompt}],
response_format={"type": "json_object"},
temperature=0.3
)
result = json.loads(response.choices[0].message.content)
translations = result.get('translations', result.get('translated', []))
for j, seg in enumerate(batch):
new_seg = seg.copy()
if j < len(translations):
if isinstance(translations[j], dict):
new_seg['translated_text'] = translations[j].get('translation',
translations[j].get('translated', ''))
else:
new_seg['translated_text'] = translations[j]
new_seg['original_text'] = seg['text']
translated.append(new_seg)
print(f" 翻译进度: {min(i+batch_size, len(segments))}/{len(segments)}")
return translated
def generate_bilingual_srt(self, translated_segments: list,
output_path: str,
primary: str = "translated"):
"""生成双语字幕"""
import srt
from datetime import timedelta
subtitles = []
for i, seg in enumerate(translated_segments, 1):
if primary == "translated":
text = f"{seg.get('translated_text', '')}\n{seg['text']}"
else:
text = f"{seg['text']}\n{seg.get('translated_text', '')}"
subtitles.append(srt.Subtitle(
index=i,
start=timedelta(seconds=seg['start']),
end=timedelta(seconds=seg['end']),
text.strip()
))
with open(output_path, 'w', encoding='utf-8') as f:
f.write(srt.compose(subtitles))
print(f"双语字幕已生成: {output_path}")
# 使用示例
# translator = SubtitleTranslator(api_key="your-key")
# translated = translator.translate_segments(result['segments'], "zh", "en")
# translator.generate_bilingual_srt(translated, "bilingual.srt")
3.3 字幕烧录（硬字幕）
import subprocess
import os
class SubtitleBurner:
"""字幕烧录工具 - 将字幕嵌入视频"""
@staticmethod
def burn_srt(video_path: str, srt_path: str, output_path: str,
font_size: int = 24, font_color: str = "white",
outline_color: str = "black", outline_width: int = 2):
"""烧录SRT字幕到视频"""
# 字幕样式滤镜
style = (
f"FontSize={font_size},"
f"PrimaryColour=&H00FFFFFF," # 白色
f"OutlineColour=&H00000000," # 黑色描边
f"Outline={outline_width},"
f"Shadow=1,"
f"MarginV=30"
)
cmd = [
'ffmpeg', '-i', video_path,
'-vf', f"subtitles={srt_path}:force_style='{style}'",
'-c:a', 'copy',
'-y', output_path
]
result = subprocess.run(cmd, capture_output=True, text=True)
if result.returncode != 0:
raise RuntimeError(f"字幕烧录失败: {result.stderr}")
print(f"硬字幕视频已生成: {output_path}")
return output_path
@staticmethod
def burn_ass(video_path: str, ass_path: str, output_path: str):
"""烧录ASS字幕到视频（支持更丰富的样式）"""
cmd = [
'ffmpeg', '-i', video_path,
'-vf', f"ass={ass_path}",
'-c:a', 'copy',
'-y', output_path
]
result = subprocess.run(cmd, capture_output=True, text=True)
if result.returncode != 0:
raise RuntimeError(f"ASS字幕烧录失败: {result.stderr}")
print(f"ASS字幕视频已生成: {output_path}")
return output_path
@staticmethod
def soft_sub(video_path: str, srt_path: str, output_path: str,
language: str = "chi"):
"""封装软字幕（可开关）"""
cmd = [
'ffmpeg', '-i', video_path,
'-i', srt_path,
'-c', 'copy',
'-c:s', 'mov_text',
'-metadata:s:s:0', f'language={language}',
'-y', output_path
]
result = subprocess.run(cmd, capture_output=True, text=True)
if result.returncode != 0:
raise RuntimeError(f"软字幕封装失败: {result.stderr}")
print(f"软字幕视频已生成: {output_path}")
return output_path
四、AI配音与语音合成
4.1 基于开源TTS的配音生成
import torch
import numpy as np
import soundfile as sf
from dataclasses import dataclass
@dataclass
class VoiceConfig:
"""语音配置"""
speaker: str = "default"
speed: float = 1.0
pitch: float = 1.0
emotion: str = "neutral" # neutral/happy/sad/angry
language: str = "zh"
class AIVoiceGenerator:
"""AI语音合成器"""
def __init__(self, engine: str = "edge-tts"):
"""
Args:
engine: TTS引擎 (edge-tts/coqui/bark)
"""
self.engine = engine
if engine == "edge-tts":
self._init_edge_tts()
elif engine == "coqui":
self._init_coqui()
def _init_edge_tts(self):
"""初始化Edge TTS（微软免费TTS）"""
try:
import edge_tts
self.edge_tts = edge_tts
except ImportError:
raise ImportError("请安装edge-tts: pip install edge-tts")
def _init_coqui(self):
"""初始化Coqui TTS"""
try:
from TTS.api import TTS
self.tts = TTS(model_name="tts_models/zh-CN/baker/tacotron2-DDC-GST",
progress_bar=False)
except ImportError:
raise ImportError("请安装TTS: pip install TTS")
async def generate_edge_tts(self, text: str, output_path: str,
voice: str = "zh-CN-XiaoxiaoNeural",
rate: str = "+0%", pitch: str = "+0Hz"):
"""使用Edge TTS生成语音"""
import edge_tts
communicate = edge_tts.Communicate(text, voice, rate=rate, pitch=pitch)
await communicate.save(output_path)
print(f"语音已生成: {output_path}")
return output_path
def generate_coqui_tts(self, text: str, output_path: str,
speaker_wav: str = None):
"""使用Coqui TTS生成语音（支持声音克隆）"""
if speaker_wav:
self.tts.tts_to_file(text=text, speaker_wav=speaker_wav,
file_path=output_path)
else:
self.tts.tts_to_file(text=text, file_path=output_path)
print(f"语音已生成: {output_path}")
return output_path
def list_edge_voices(self, language: str = "zh"):
"""列出可用的Edge TTS语音"""
import asyncio
async def _list():
voices = await self.edge_tts.list_voices()
return [v for v in voices if v['Locale'].startswith(language)]
return asyncio.run(_list())
class VoiceOverGenerator:
"""视频配音生成器 - 自动为视频生成配音"""
def __init__(self, tts_engine: AIVoiceGenerator):
self.tts = tts_engine
def generate_voiceover_from_script(self, script: list,
output_dir: str,
voice: str = "zh-CN-YunxiNeural"):
"""
从脚本生成配音
Args:
script: [{"text": "旁白内容", "start": 0.0, "end": 5.0}, ...]
output_dir: 输出目录
voice: 语音角色
"""
import asyncio
import os
os.makedirs(output_dir, exist_ok=True)
audio_files = []
for i, segment in enumerate(script):
output_path = os.path.join(output_dir, f"voiceover_{i:03d}.mp3")
# 根据时长调整语速
duration = segment['end'] - segment['start']
char_count = len(segment['text'])
# 估算语速调整（中文约4字/秒为正常语速）
normal_duration = char_count / 4.0
rate = int((normal_duration / duration - 1) * 100)
rate_str = f"+{rate}%" if rate >= 0 else f"{rate}%"
asyncio.run(self.tts.generate_edge_tts(
segment['text'], output_path,
voice=voice, rate=rate_str
))
audio_files.append({
'path': output_path,
'start': segment['start'],
'end': segment['end']
})
return audio_files
def merge_voiceover_with_video(self, video_path: str,
voiceover_files: list,
output_path: str,
bgm_path: str = None,
bgm_volume: float = 0.15):
"""将配音合并到视频"""
import subprocess
import tempfile
# 先生成配音时间线音频
timeline_audio = self._create_audio_timeline(
voiceover_files, video_path
)
if bgm_path:
# 混合背景音乐
mixed_audio = tempfile.mktemp(suffix='.wav')
cmd = [
'ffmpeg',
'-i', timeline_audio,
'-i', bgm_path,
'-filter_complex',
f'[1:a]volume={bgm_volume}[bgm];[0:a][bgm]amix=inputs=2:duration=first',
'-y', mixed_audio
]
subprocess.run(cmd, capture_output=True)
timeline_audio = mixed_audio
# 合并到视频
cmd = [
'ffmpeg', '-i', video_path,
'-i', timeline_audio,
'-c:v', 'copy',
'-map', '0:v:0',
'-map', '1:a:0',
'-shortest',
'-y', output_path
]
result = subprocess.run(cmd, capture_output=True, text=True)
if result.returncode != 0:
raise RuntimeError(f"合并失败: {result.stderr}")
print(f"配音视频已生成: {output_path}")
return output_path
def _create_audio_timeline(self, voiceover_files: list,
video_path: str) -> str:
"""创建音频时间线"""
import subprocess
import tempfile
# 获取视频时长
probe = subprocess.run(
['ffprobe', '-v', 'error', '-show_entries',
'format=duration', '-of', 'csv=p=0', video_path],
capture_output=True, text=True
)
duration = float(probe.stdout.strip())
# 使用FFmpeg的amerge/amix创建时间线
inputs = []
filter_parts = []
for i, vf in enumerate(voiceover_files):
inputs.extend(['-i', vf['path']])
delay_ms = int(vf['start'] * 1000)
filter_parts.append(
f"[{i}:a]adelay={delay_ms}|{delay_ms}[a{i}]"
)
if not filter_parts:
raise ValueError("没有配音文件")
mix_inputs = ''.join(f'[a{i}]' for i in range(len(voiceover_files)))
filter_parts.append(
f"{mix_inputs}amix=inputs={len(voiceover_files)}:duration=longest[out]"
)
output_path = tempfile.mktemp(suffix='.wav')
cmd = ['ffmpeg'] + inputs + [
'-filter_complex', ';'.join(filter_parts),
'-map', '[out]',
'-t', str(duration),
'-y', output_path
]
subprocess.run(cmd, capture_output=True)
return output_path
五、视频风格迁移与AI滤镜
5.1 基于神经网络的风格迁移
import torch
import torch.nn as nn
import torchvision.transforms as transforms
import torchvision.models as models
from PIL import Image
import cv2
import numpy as np
class VideoStyleTransfer:
"""视频风格迁移处理器"""
def __init__(self, style_image_path: str, model: str = "vgg19"):
self.device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
# 加载预训练模型
if model == "vgg19":
self.model = models.vgg19(pretrained=True).features.to(self.device).eval()
# 加载风格图片
self.style_image = self._load_image(style_image_path)
self.style_features = self._extract_features(self.style_image)
# 图像变换
self.transform = transforms.Compose([
transforms.ToTensor(),
transforms.Normalize(mean=[0.485, 0.456, 0.406],
std=[0.229, 0.224, 0.225])
])
def _load_image(self, path: str, max_size: int = 512) -> torch.Tensor:
"""加载并预处理图片"""
img = Image.open(path).convert('RGB')
# 保持比例缩放
ratio = max_size / max(img.size)
if ratio < 1:
new_size = tuple(int(s * ratio) for s in img.size)
img = img.resize(new_size, Image.LANCZOS)
tensor = self.transform(img).unsqueeze(0)
return tensor.to(self.device)
def _extract_features(self, x: torch.Tensor) -> list:
"""提取特征层"""
features = []
layer_names = ['relu1_1', 'relu2_1', 'relu3_1', 'relu4_1', 'relu5_1']
layer_indices = [1, 6, 11, 20, 29] # VGG19中的层索引
for i, layer in enumerate(self.model):
x = layer(x)
if i in layer_indices:
features.append(x)
return features
def _gram_matrix(self, tensor: torch.Tensor) -> torch.Tensor:
"""计算Gram矩阵（风格表示）"""
b, c, h, w = tensor.size()
features = tensor.view(b, c, h * w)
gram = torch.bmm(features, features.transpose(1, 2))
return gram / (c * h * w)
def apply_to_frame(self, frame: np.ndarray,
num_steps: int = 100,
style_weight: float = 1e6,
content_weight: int = 1) -> np.ndarray:
"""对单帧应用风格迁移"""
# 转换帧格式
frame_rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
frame_pil = Image.fromarray(frame_rgb)
# 保持原始尺寸
original_size = frame_pil.size
# 缩放到处理尺寸
max_dim = 512
ratio = max_dim / max(original_size)
if ratio < 1:
process_size = tuple(int(s * ratio) for s in original_size)
frame_pil = frame_pil.resize(process_size, Image.LANCZOS)
content_tensor = self.transform(frame_pil).unsqueeze(0).to(self.device)
target = content_tensor.clone().requires_grad_(True)
content_features = self._extract_features(content_tensor)
# 优化器
optimizer = torch.optim.Adam([target], lr=0.01)
for step in range(num_steps):
target_features = self._extract_features(target)
content_loss = 0
style_loss = 0
for cf, tf, sf in zip(content_features, target_features,
self.style_features):
content_loss += torch.mean((tf - cf) ** 2)
target_gram = self._gram_matrix(tf)
style_gram = self._gram_matrix(sf)
style_loss += torch.mean((target_gram - style_gram) ** 2)
total_loss = content_weight * content_loss + style_weight * style_loss
optimizer.zero_grad()
total_loss.backward()
optimizer.step()
# 转换回图片
result = target.detach().squeeze(0).cpu()
result = result * torch.tensor([0.229, 0.224, 0.225]).view(3, 1, 1)
result = result + torch.tensor([0.485, 0.456, 0.406]).view(3, 1, 1)
result = result.clamp(0, 1)
result_np = (result.permute(1, 2, 0).numpy() * 255).astype(np.uint8)
result_pil = Image.fromarray(result_np)
# 恢复原始尺寸
result_pil = result_pil.resize(original_size, Image.LANCZOS)
return cv2.cvtColor(np.array(result_pil), cv2.COLOR_RGB2BGR)
def process_video(self, input_path: str, output_path: str,
num_steps: int = 50, fps: float = None):
"""处理整个视频"""
cap = cv2.VideoCapture(input_path)
if not cap.isOpened():
raise RuntimeError(f"无法打开视频: {input_path}")
# 获取视频参数
width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
original_fps = cap.get(cv2.CAP_PROP_FPS)
total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
if fps is None:
fps = original_fps
# 创建视频写入器
fourcc = cv2.VideoWriter_fourcc(*'mp4v')
writer = cv2.VideoWriter(output_path, fourcc, fps, (width, height))
frame_count = 0
while True:
ret, frame = cap.read()
if not ret:
break
# 应用风格迁移
styled_frame = self.apply_to_frame(frame, num_steps=num_steps)
writer.write(styled_frame)
frame_count += 1
if frame_count % 10 == 0:
print(f"处理进度: {frame_count}/{total_frames} "
f"({frame_count/total_frames*100:.1f}%)")
cap.release()
writer.release()
print(f"风格迁移视频已生成: {output_path}")
# 使用示例
# transfer = VideoStyleTransfer("starry_night.jpg")
# transfer.process_video("input.mp4", "styled_output.mp4", num_steps=30)
5.2 快速滤镜方案（基于预训练模型）
import cv2
import numpy as np
class QuickFilters:
"""快速AI风格滤镜（无需训练）"""
@staticmethod
def apply_pencil_sketch(frame: np.ndarray) -> np.ndarray:
"""素描风格"""
gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
inv = 255 - gray
blur = cv2.GaussianBlur(inv, (21, 21), 0)
sketch = cv2.divide(gray, 255 - blur, scale=256)
return cv2.cvtColor(sketch, cv2.COLOR_GRAY2BGR)
@staticmethod
def apply_oil_painting(frame: np.ndarray,
size: int = 5, dynRatio: int = 1) -> np.ndarray:
"""油画风格"""
return cv2.xphoto.oilPainting(frame, size, dynRatio)
@staticmethod
def apply_cartoon(frame: np.ndarray) -> np.ndarray:
"""卡通风格"""
# 边缘检测
gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
gray = cv2.medianBlur(gray, 5)
edges = cv2.adaptiveThreshold(
gray, 255, cv2.ADAPTIVE_THRESH_MEAN_C,
cv2.THRESH_BINARY, 9, 9
)
# 颜色量化
color = cv2.bilateralFilter(frame, 9, 300, 300)
# 合并
cartoon = cv2.bitwise_and(color, color, mask=edges)
return cartoon
@staticmethod
def apply_cinematic(frame: np.ndarray) -> np.ndarray:
"""电影色调"""
# 调整对比度和饱和度
lab = cv2.cvtColor(frame, cv2.COLOR_BGR2LAB)
l, a, b = cv2.split(lab)
# CLAHE增强对比度
clahe = cv2.createCLAHE(clipLimit=2.0, tileGridSize=(8, 8))
l = clahe.apply(l)
# 调整色温（偏暖）
b = cv2.add(b, 10)
merged = cv2.merge([l, a, b])
result = cv2.cvtColor(merged, cv2.COLOR_LAB2BGR)
# 添加暗角
rows, cols = result.shape[:2]
kernel_x = cv2.getGaussianKernel(cols, cols * 0.6)
kernel_y = cv2.getGaussianKernel(rows, rows * 0.6)
kernel = kernel_y * kernel_x.T
mask = kernel / kernel.max()
vignette = np.copy(result)
for i in range(3):
vignette[:, :, i] = vignette[:, :, i] * mask
return vignette
@staticmethod
def apply_color_grading(frame: np.ndarray,
shadows: tuple = (30, 20, 10),
highlights: tuple = (240, 235, 250)) -> np.ndarray:
"""专业调色"""
result = frame.astype(np.float32)
# 分离暗部和亮部
for c in range(3):
channel = result[:, :, c]
shadow_mask = channel < 128
highlight_mask = channel >= 128
# 暗部偏色
result[:, :, c] = np.where(
shadow_mask,
channel * (shadows[c] / 128.0),
channel
)
# 亮部偏色
result[:, :, c] = np.where(
highlight_mask,
channel + (highlights[c] - 255) * (channel - 128) / 127.0,
result[:, :, c]
)
return np.clip(result, 0, 255).astype(np.uint8)
class VideoFilterProcessor:
"""视频滤镜批量处理器"""
def __init__(self):
self.filters = QuickFilters()
def apply_filter_to_video(self, input_path: str, output_path: str,
filter_name: str = "cinematic",
fps: float = None):
"""对视频应用滤镜"""
filter_func = getattr(self.filters, f"apply_{filter_name}", None)
if filter_func is None:
raise ValueError(f"未知滤镜: {filter_name}")
cap = cv2.VideoCapture(input_path)
width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
original_fps = cap.get(cv2.CAP_PROP_FPS)
total = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
if fps is None:
fps = original_fps
fourcc = cv2.VideoWriter_fourcc(*'mp4v')
writer = cv2.VideoWriter(output_path, fourcc, fps, (width, height))
count = 0
while True:
ret, frame = cap.read()
if not ret:
break
filtered = filter_func(frame)
writer.write(filtered)
count += 1
if count % 30 == 0:
print(f"进度: {count}/{total}")
cap.release()
writer.release()
print(f"滤镜视频已保存: {output_path}")
六、智能裁剪与构图优化
6.1 基于人脸检测的智能裁剪
import cv2
import numpy as np
from dataclasses import dataclass
from typing import Optional
@dataclass
class CropRegion:
"""裁剪区域"""
x: int
y: int
width: int
height: int
confidence: float = 1.0
class SmartCropper:
"""智能裁剪器 - 基于内容感知的自动裁剪"""
def __init__(self):
# 加载人脸检测器
self.face_cascade = cv2.CascadeClassifier(
cv2.data.haarcascades + 'haarcascade_frontalface_default.xml'
)
# 尝试加载DNN人脸检测器（更准确）
try:
self.face_net = cv2.dnn.readNetFromCaffe(
"deploy.prototxt",
"res10_300x300_ssd_iter_140000.caffemodel"
)
self.use_dnn = True
except:
self.use_dnn = False
print("DNN模型不可用，使用Haar级联检测器")
def detect_faces(self, frame: np.ndarray) -> list:
"""检测人脸位置"""
if self.use_dnn:
return self._detect_faces_dnn(frame)
return self._detect_faces_haar(frame)
def _detect_faces_haar(self, frame: np.ndarray) -> list:
"""Haar级联人脸检测"""
gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
faces = self.face_cascade.detectMultiScale(
gray, scaleFactor=1.1, minNeighbors=5, minSize=(30, 30)
)
return [(x, y, w, h) for (x, y, w, h) in faces]
def _detect_faces_dnn(self, frame: np.ndarray) -> list:
"""DNN人脸检测（更准确）"""
h, w = frame.shape[:2]
blob = cv2.dnn.blobFromImage(
cv2.resize(frame, (300, 300)), 1.0,
(300, 300), (104.0, 177.0, 123.0)
)
self.face_net.setInput(blob)
detections = self.face_net.forward()
faces = []
for i in range(detections.shape[2]):
confidence = detections[0, 0, i, 2]
if confidence > 0.5:
box = detections[0, 0, i, 3:7] * np.array([w, h, w, h])
x1, y1, x2, y2 = box.astype(int)
faces.append((x1, y1, x2 - x1, y2 - y1))
return faces
def auto_crop_for_portrait(self, frame: np.ndarray,
target_ratio: float = 9/16) -> np.ndarray:
"""人像智能裁剪（适配竖屏）"""
h, w = frame.shape[:2]
faces = self.detect_faces(frame)
if not faces:
# 无人脸时居中裁剪
return self._center_crop(frame, target_ratio)
# 找到最大的人脸
largest_face = max(faces, key=lambda f: f[2] * f[3])
fx, fy, fw, fh = largest_face
# 计算人脸中心
face_cx = fx + fw // 2
face_cy = fy + fh // 2
# 计算裁剪区域（确保人脸在上方1/3处）
target_w = int(h * target_ratio)
if target_w > w:
target_w = w
target_h = int(target_w / target_ratio)
# 人脸应在垂直方向的1/3处
crop_y = face_cy - target_h // 3
crop_x = face_cx - target_w // 2
# 边界修正
crop_x = max(0, min(crop_x, w - target_w))
crop_y = max(0, min(crop_y, h - target_h))
return frame[crop_y:crop_y+target_h, crop_x:crop_x+target_w]
def auto_crop_for_landscape(self, frame: np.ndarray,
target_ratio: float = 16/9) -> np.ndarray:
"""风景智能裁剪（适配横屏）"""
h, w = frame.shape[:2]
# 使用显著性检测找到重要区域
saliency = cv2.saliency.StaticSaliencySpectralResidual_create()
success, saliency_map = saliency.computeSaliency(frame)
if success:
saliency_map = (saliency_map * 255).astype(np.uint8)
# 找到显著性中心
moments = cv2.moments(saliency_map)
if moments['m00'] > 0:
cx = int(moments['m10'] / moments['m00'])
cy = int(moments['m01'] / moments['m00'])
else:
cx, cy = w // 2, h // 2
else:
cx, cy = w // 2, h // 2
# 计算裁剪区域
target_h = h
target_w = int(target_h * target_ratio)
if target_w > w:
target_w = w
target_h = int(target_w / target_ratio)
crop_x = cx - target_w // 2
crop_y = cy - target_h // 2
crop_x = max(0, min(crop_x, w - target_w))
crop_y = max(0, min(crop_y, h - target_h))
return frame[crop_y:crop_y+target_h, crop_x:crop_x+target_w]
def _center_crop(self, frame: np.ndarray,
target_ratio: float) -> np.ndarray:
"""居中裁剪"""
h, w = frame.shape[:2]
current_ratio = w / h
if current_ratio > target_ratio:
new_w = int(h * target_ratio)
x = (w - new_w) // 2
return frame[:, x:x+new_w]
else:
new_h = int(w / target_ratio)
y = (h - new_h) // 2
return frame[y:y+new_h, :]
def smart_crop_video(self, input_path: str, output_path: str,
target_ratio: float = 9/16,
content_type: str = "portrait"):
"""智能裁剪视频"""
cap = cv2.VideoCapture(input_path)
# 先分析第一帧确定裁剪参数
ret, first_frame = cap.read()
if not ret:
raise RuntimeError("无法读取视频")
if content_type == "portrait":
cropped = self.auto_crop_for_portrait(first_frame, target_ratio)
else:
cropped = self.auto_crop_for_landscape(first_frame, target_ratio)
crop_h, crop_w = cropped.shape[:2]
# 重置视频
cap.set(cv2.CAP_PROP_POS_FRAMES, 0)
fps = cap.get(cv2.CAP_PROP_FPS)
fourcc = cv2.VideoWriter_fourcc(*'mp4v')
writer = cv2.VideoWriter(output_path, fourcc, fps, (crop_w, crop_h))
count = 0
while True:
ret, frame = cap.read()
if not ret:
break
if content_type == "portrait":
cropped = self.auto_crop_for_portrait(frame, target_ratio)
else:
cropped = self.auto_crop_for_landscape(frame, target_ratio)
# 确保尺寸一致
cropped = cv2.resize(cropped, (crop_w, crop_h))
writer.write(cropped)
count += 1
if count % 30 == 0:
print(f"裁剪进度: {count} 帧")
cap.release()
writer.release()
print(f"智能裁剪视频已生成: {output_path}")
七、物体移除与替换
7.1 基于图像修复的物体移除
import cv2
import numpy as np
import torch
from PIL import Image
class ObjectRemover:
"""AI物体移除器"""
def __init__(self, method: str = "opencv"):
"""
Args:
method: 修复方法 (opencv/lama/sd)
"""
self.method = method
if method == "lama":
self._init_lama()
elif method == "sd":
self._init_sd_inpaint()
def _init_lama(self):
"""初始化LaMa修复模型"""
try:
# 需要下载LaMa模型
print("LaMa模型需要单独下载，请参考: https://github.com/advimman/lama")
except Exception as e:
print(f"LaMa初始化失败: {e}")
def _init_sd_inpaint(self):
"""初始化Stable Diffusion修复"""
try:
from diffusers import StableDiffusionInpaintPipeline
self.sd_pipe = StableDiffusionInpaintPipeline.from_pretrained(
"runwayml/stable-diffusion-inpainting",
torch_dtype=torch.float16
).to("cuda" if torch.cuda.is_available() else "cpu")
except ImportError:
raise ImportError("请安装diffusers: pip install diffusers")
def remove_with_mask(self, image: np.ndarray,
mask: np.ndarray) -> np.ndarray:
"""使用掩码移除物体"""
if self.method == "opencv":
return self._opencv_inpaint(image, mask)
elif self.method == "lama":
return self._lama_inpaint(image, mask)
elif self.method == "sd":
return self._sd_inpaint(image, mask, "")
return image
def _opencv_inpaint(self, image: np.ndarray,
mask: np.ndarray) -> np.ndarray:
"""OpenCV传统修复（快速但效果一般）"""
# 使用NS方法
result_ns = cv2.inpaint(image, mask, 3, cv2.INPAINT_NS)
# 使用Telea方法
result_telea = cv2.inpaint(image, mask, 3, cv2.INPAINT_TELEA)
# 混合两种结果
alpha = 0.5
result = cv2.addWeighted(result_ns, alpha, result_telea, 1-alpha, 0)
return result
def _sd_inpaint(self, image: np.ndarray, mask: np.ndarray,
prompt: str = "clean background") -> np.ndarray:
"""Stable Diffusion修复（效果最好但较慢）"""
image_pil = Image.fromarray(cv2.cvtColor(image, cv2.COLOR_BGR2RGB))
mask_pil = Image.fromarray(mask)
result = self.sd_pipe(
prompt=prompt,
image=image_pil,
mask_image=mask_pil,
num_inference_steps=20
).images[0]
return cv2.cvtColor(np.array(result), cv2.COLOR_RGB2BGR)
def auto_detect_and_remove(self, image: np.ndarray,
object_class: str = "person") -> np.ndarray:
"""自动检测并移除指定类型的物体"""
# 使用YOLO检测物体
try:
from ultralytics import YOLO
model = YOLO('yolov8n.pt')
results = model(image)
except ImportError:
print("请安装ultralytics: pip install ultralytics")
return image
mask = np.zeros(image.shape[:2], dtype=np.uint8)
for result in results:
for box in result.boxes:
cls = result.names[int(box.cls)]
if cls == object_class:
x1, y1, x2, y2 = box.xyxy[0].int().tolist()
# 扩展掩码区域
margin = 10
cv2.rectangle(mask,
(x1-margin, y1-margin),
(x2+margin, y2+margin),
255, -1)
if mask.sum() == 0:
print(f"未检测到 {object_class}")
return image
return self.remove_with_mask(image, mask)
def process_video(self, input_path: str, output_path: str,
mask_video_path: str = None,
object_class: str = None):
"""处理视频中的物体移除"""
cap = cv2.VideoCapture(input_path)
width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
fps = cap.get(cv2.CAP_PROP_FPS)
if mask_video_path:
mask_cap = cv2.VideoCapture(mask_video_path)
fourcc = cv2.VideoWriter_fourcc(*'mp4v')
writer = cv2.VideoWriter(output_path, fourcc, fps, (width, height))
count = 0
while True:
ret, frame = cap.read()
if not ret:
break
if mask_video_path:
ret_m, mask_frame = mask_cap.read()
if ret_m:
mask = cv2.cvtColor(mask_frame, cv2.COLOR_BGR2GRAY)
result = self.remove_with_mask(frame, mask)
else:
result = frame
elif object_class:
result = self.auto_detect_and_remove(frame, object_class)
else:
result = frame
writer.write(result)
count += 1
if count % 30 == 0:
print(f"物体移除进度: {count} 帧")
cap.release()
if mask_video_path:
mask_cap.release()
writer.release()
print(f"物体移除视频已生成: {output_path}")
八、AI转场与特效生成
8.1 智能转场效果
import cv2
import numpy as np
class AITransitions:
"""AI转场效果库"""
@staticmethod
def crossfade(frame1: np.ndarray, frame2: np.ndarray,
progress: float) -> np.ndarray:
"""交叉淡入淡出"""
return cv2.addWeighted(frame1, 1 - progress, frame2, progress, 0)
@staticmethod
def morph_transition(frame1: np.ndarray, frame2: np.ndarray,
progress: float) -> np.ndarray:
"""基于光流的形变转场"""
# 计算光流
gray1 = cv2.cvtColor(frame1, cv2.COLOR_BGR2GRAY)
gray2 = cv2.cvtColor(frame2, cv2.COLOR_BGR2GRAY)
flow = cv2.calcOpticalFlowFarneback(
gray1, gray2, None, 0.5, 3, 15, 3, 5, 1.2, 0
)
h, w = frame1.shape[:2]
flow_map = np.column_stack([
np.repeat(np.arange(w), h),
np.tile(np.arange(h), w)
]).reshape(h, w, 2).astype(np.float32)
# 根据进度插值光流
flow_map += flow * progress
# 应用形变
warped1 = cv2.remap(frame1, flow_map[:, :, 0], flow_map[:, :, 1],
cv2.INTER_LINEAR)
warped2 = cv2.remap(frame2,
flow_map[:, :, 0] - flow[:, :, 0] * (1 - progress),
flow_map[:, :, 1] - flow[:, :, 1] * (1 - progress),
cv2.INTER_LINEAR)
return cv2.addWeighted(warped1, 1 - progress, warped2, progress, 0)
@staticmethod
def zoom_transition(frame1: np.ndarray, frame2: np.ndarray,
progress: float) -> np.ndarray:
"""缩放转场"""
h, w = frame1.shape[:2]
# frame1缩小
scale1 = 1 + progress * 0.5
center = (w // 2, h // 2)
M1 = cv2.getRotationMatrix2D(center, 0, 1/scale1)
zoomed1 = cv2.warpAffine(frame1, M1, (w, h))
# frame2从小变大
scale2 = 0.5 + progress * 0.5
M2 = cv2.getRotationMatrix2D(center, 0, scale2)
zoomed2 = cv2.warpAffine(frame2, M2, (w, h))
# 混合
alpha = min(1.0, progress * 2)
return cv2.addWeighted(zoomed1, 1 - alpha, zoomed2, alpha, 0)
@staticmethod
def slide_transition(frame1: np.ndarray, frame2: np.ndarray,
progress: float, direction: str = "left") -> np.ndarray:
"""滑动转场"""
h, w = frame1.shape[:2]
result = np.zeros_like(frame1)
if direction == "left":
offset = int(w * progress)
if offset < w:
result[:, :w-offset] = frame1[:, offset:]
if offset > 0:
result[:, w-offset:] = frame2[:, :offset]
elif direction == "right":
offset = int(w * (1 - progress))
if offset > 0:
result[:, :offset] = frame2[:, w-offset:]
if offset < w:
result[:, offset:] = frame1[:, :w-offset]
elif direction == "up":
offset = int(h * progress)
if offset < h:
result[:h-offset, :] = frame1[offset:, :]
if offset > 0:
result[h-offset:, :] = frame2[:offset, :]
return result
@staticmethod
def glitch_transition(frame1: np.ndarray, frame2: np.ndarray,
progress: float) -> np.ndarray:
"""故障风格转场"""
h, w = frame1.shape[:2]
if progress < 0.5:
# 前半段：frame1 + 故障效果
alpha = progress * 2
result = frame1.copy()
# RGB通道偏移
offset = int(alpha * 20)
result[:, offset:, 0] = frame1[:, :w-offset, 0] # R偏移
result[:, :w-offset, 2] = frame1[:, offset:, 2] # B偏移
# 随机水平条纹
num_stripes = int(alpha * 10)
for _ in range(num_stripes):
y = np.random.randint(0, h)
stripe_h = np.random.randint(2, 10)
shift = np.random.randint(-30, 30)
result[y:y+stripe_h] = np.roll(result[y:y+stripe_h], shift, axis=1)
else:
# 后半段：过渡到frame2
alpha = (progress - 0.5) * 2
result = cv2.addWeighted(frame1, 1 - alpha, frame2, alpha, 0)
# 逐渐减少故障效果
glitch_alpha = 1 - alpha
offset = int(glitch_alpha * 15)
if offset > 0:
result[:, offset:, 0] = result[:, :w-offset, 0]
return result
class TransitionApplicator:
"""转场应用器 - 为视频片段添加转场"""
def __init__(self):
self.transitions = AITransitions()
def apply_transitions(self, video_clips: list, output_path: str,
transition_type: str = "crossfade",
transition_duration: float = 1.0):
"""
为多个视频片段添加转场
Args:
video_clips: 视频文件路径列表
output_path: 输出路径
transition_type: 转场类型
transition_duration: 转场时长（秒）
"""
if not video_clips:
raise ValueError("没有视频片段")
# 读取所有视频的参数
caps = [cv2.VideoCapture(p) for p in video_clips]
width = int(caps[0].get(cv2.CAP_PROP_FRAME_WIDTH))
height = int(caps[0].get(cv2.CAP_PROP_FRAME_HEIGHT))
fps = caps[0].get(cv2.CAP_PROP_FPS)
fourcc = cv2.VideoWriter_fourcc(*'mp4v')
writer = cv2.VideoWriter(output_path, fourcc, fps, (width, height))
transition_frames = int(transition_duration * fps)
transition_func = getattr(self.transitions, transition_type,
self.transitions.crossfade)
for clip_idx, cap in enumerate(caps):
# 读取当前片段的所有帧
frames = []
while True:
ret, frame = cap.read()
if not ret:
break
frame = cv2.resize(frame, (width, height))
frames.append(frame)
if clip_idx == 0:
# 第一个片段：写入除最后transition_frames帧外的所有帧
for frame in frames[:-transition_frames]:
writer.write(frame)
if clip_idx < len(caps) - 1:
# 读取下一个片段的前transition_frames帧
next_cap = caps[clip_idx + 1]
next_frames = []
for _ in range(transition_frames):
ret, frame = next_cap.read()
if ret:
frame = cv2.resize(frame, (width, height))
next_frames.append(frame)
# 生成转场帧
last_frames = frames[-transition_frames:]
for i in range(transition_frames):
progress = i / transition_frames
f1 = last_frames[min(i, len(last_frames)-1)]
f2 = next_frames[min(i, len(next_frames)-1)]
blended = transition_func(f1, f2, progress)
writer.write(blended)
# 将next_frames中未使用的帧留给下一个循环处理
# （这里简化处理，实际需要更复杂的缓冲管理）
else:
# 最后一个片段：写入所有剩余帧
for frame in frames:
writer.write(frame)
cap.release()
writer.release()
print(f"转场视频已生成: {output_path}")
九、视频摘要与精彩片段提取
9.1 基于AI的视频摘要
import cv2
import numpy as np
from dataclasses import dataclass
from typing import List
@dataclass
class VideoSegment:
"""视频片段"""
start_time: float
end_time: float
score: float
description: str = ""
has_face: bool = False
motion_level: float = 0.0
audio_energy: float = 0.0
class VideoSummarizer:
"""AI视频摘要生成器"""
def __init__(self):
self.face_cascade = cv2.CascadeClassifier(
cv2.data.haarcascades + 'haarcascade_frontalface_default.xml'
)
def analyze_video(self, video_path: str,
sample_interval: float = 1.0) -> List[VideoSegment]:
"""分析视频，提取关键特征"""
cap = cv2.VideoCapture(video_path)
fps = cap.get(cv2.CAP_PROP_FPS)
total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
duration = total_frames / fps
segments = []
prev_frame = None
sample_frames = int(fps * sample_interval)
frame_idx = 0
while True:
ret, frame = cap.read()
if not ret:
break
if frame_idx % sample_frames == 0:
current_time = frame_idx / fps
# 计算运动量
motion = 0.0
if prev_frame is not None:
diff = cv2.absdiff(
cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY),
cv2.cvtColor(prev_frame, cv2.COLOR_BGR2GRAY)
)
motion = np.mean(diff) / 255.0
# 检测人脸
gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
faces = self.face_cascade.detectMultiScale(
gray, 1.3, 5, minSize=(30, 30)
)
has_face = len(faces) > 0
# 计算视觉显著性
saliency_score = self._compute_saliency(frame)
# 综合评分
score = (
motion * 0.3 +
(1.0 if has_face else 0.0) * 0.3 +
saliency_score * 0.4
)
segments.append(VideoSegment(
start_time=current_time,
end_time=current_time + sample_interval,
score=score,
has_face=has_face,
motion_level=motion
))
prev_frame = frame.copy()
frame_idx += 1
cap.release()
return segments
def _compute_saliency(self, frame: np.ndarray) -> float:
"""计算视觉显著性"""
try:
saliency = cv2.saliency.StaticSaliencySpectralResidual_create()
success, saliency_map = saliency.computeSaliency(frame)
if success:
return float(np.mean(saliency_map))
except:
pass
# 备用方案：基于边缘密度
gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
edges = cv2.Canny(gray, 50, 150)
return np.mean(edges) / 255.0
def extract_highlights(self, segments: List[VideoSegment],
target_duration: float = 60.0,
min_gap: float = 5.0) -> List[VideoSegment]:
"""提取精彩片段"""
# 按分数排序
sorted_segs = sorted(segments, key=lambda s: s.score, reverse=True)
highlights = []
total_duration = 0.0
used_times = set()
for seg in sorted_segs:
if total_duration >= target_duration:
break
# 检查是否与已选片段重叠
start = seg.start_time
if any(abs(start - t) < min_gap for t in used_times):
continue
highlights.append(seg)
used_times.add(start)
total_duration += seg.end_time - seg.start_time
# 按时间排序
highlights.sort(key=lambda s: s.start_time)
return highlights
def generate_summary_video(self, input_path: str, output_path: str,
highlight_segments: List[VideoSegment],
transition_frames: int = 15):
"""生成摘要视频"""
cap = cv2.VideoCapture(input_path)
fps = cap.get(cv2.CAP_PROP_FPS)
width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
fourcc = cv2.VideoWriter_fourcc(*'mp4v')
writer = cv2.VideoWriter(output_path, fourcc, fps, (width, height))
for seg in highlight_segments:
start_frame = int(seg.start_time * fps)
end_frame = int(seg.end_time * fps)
cap.set(cv2.CAP_PROP_POS_FRAMES, start_frame)
for f_idx in range(start_frame, end_frame):
ret, frame = cap.read()
if not ret:
break
# 添加淡入效果
if f_idx - start_frame < transition_frames:
alpha = (f_idx - start_frame) / transition_frames
frame = (frame * alpha).astype(np.uint8)
# 添加淡出效果
if end_frame - f_idx < transition_frames:
alpha = (end_frame - f_idx) / transition_frames
frame = (frame * alpha).astype(np.uint8)
writer.write(frame)
cap.release()
writer.release()
print(f"摘要视频已生成: {output_path}")
def add_timestamp_overlay(self, frame: np.ndarray,
time_seconds: float) -> np.ndarray:
"""添加时间戳水印"""
minutes = int(time_seconds // 60)
seconds = int(time_seconds % 60)
time_str = f"{minutes:02d}:{seconds:02d}"
cv2.putText(frame, time_str, (20, 40),
cv2.FONT_HERSHEY_SIMPLEX, 1.0,
(255, 255, 255), 2, cv2.LINE_AA)
return frame
十、批量处理与自动化工作流
10.1 视频批量处理框架
import os
import json
from concurrent.futures import ThreadPoolExecutor, as_completed
from dataclasses import dataclass, field
from typing import Callable, Optional
@dataclass
class ProcessingTask:
"""处理任务"""
input_path: str
output_path: str
operations: list = field(default_factory=list)
status: str = "pending"
error: str = ""
@dataclass
class ProcessingPipeline:
"""处理流水线配置"""
name: str
steps: list = field(default_factory=list)
parallel: bool = False
max_workers: int = 4
class VideoBatchProcessor:
"""视频批量处理器"""
def __init__(self, max_workers: int = 4):
self.max_workers = max_workers
self.operations = {}
self._register_default_operations()
def _register_default_operations(self):
"""注册默认操作"""
self.operations['subtitle'] = self._op_add_subtitle
self.operations['filter'] = self._op_apply_filter
self.operations['crop'] = self._op_smart_crop
self.operations['resize'] = self._op_resize
self.operations['watermark'] = self._op_add_watermark
self.operations['trim'] = self._op_trim
self.operations['speed'] = self._op_change_speed
self.operations['extract_audio'] = self._op_extract_audio
def register_operation(self, name: str, func: Callable):
"""注册自定义操作"""
self.operations[name] = func
def process_batch(self, tasks: list, pipeline: ProcessingPipeline) -> dict:
"""批量处理视频"""
results = {
'total': len(tasks),
'success': 0,
'failed': 0,
'details': []
}
if pipeline.parallel:
with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
futures = {}
for task in tasks:
future = executor.submit(
self._process_single, task, pipeline
)
futures[future] = task
for future in as_completed(futures):
task = futures[future]
try:
result = future.result()
results['success'] += 1
results['details'].append({
'input': task.input_path,
'status': 'success',
'output': task.output_path
})
except Exception as e:
results['failed'] += 1
results['details'].append({
'input': task.input_path,
'status': 'failed',
'error': str(e)
})
print(f" [{'✓' if results['details'][-1]['status'] == 'success' else '✗'}] "
f"{os.path.basename(task.input_path)}")
else:
for task in tasks:
try:
self._process_single(task, pipeline)
results['success'] += 1
results['details'].append({
'input': task.input_path,
'status': 'success',
'output': task.output_path
})
print(f" [✓] {os.path.basename(task.input_path)}")
except Exception as e:
results['failed'] += 1
results['details'].append({
'input': task.input_path,
'status': 'failed',
'error': str(e)
})
print(f" [✗] {os.path.basename(task.input_path)}: {e}")
return results
def _process_single(self, task: ProcessingTask,
pipeline: ProcessingPipeline):
"""处理单个视频"""
current_input = task.input_path
for step in pipeline.steps:
op_name = step.get('operation')
op_params = step.get('params', {})
if op_name not in self.operations:
raise ValueError(f"未知操作: {op_name}")
# 中间步骤使用临时文件
if step != pipeline.steps[-1]:
temp_output = current_input + f".temp_{op_name}.mp4"
else:
temp_output = task.output_path
self.operations[op_name](current_input, temp_output, **op_params)
# 清理临时文件
if current_input != task.input_path and os.path.exists(current_input):
os.remove(current_input)
current_input = temp_output
task.status = "completed"
# ===== 内置操作实现 =====
def _op_add_subtitle(self, input_path: str, output_path: str,
srt_path: str = "", **kwargs):
"""添加字幕操作"""
import subprocess
cmd = [
'ffmpeg', '-i', input_path,
'-vf', f"subtitles={srt_path}",
'-c:a', 'copy', '-y', output_path
]
subprocess.run(cmd, capture_output=True, check=True)
def _op_apply_filter(self, input_path: str, output_path: str,
filter_name: str = "cinematic", **kwargs):
"""应用滤镜操作"""
processor = VideoFilterProcessor()
processor.apply_filter_to_video(input_path, output_path, filter_name)
def _op_smart_crop(self, input_path: str, output_path: str,
ratio: float = 9/16, **kwargs):
"""智能裁剪操作"""
cropper = SmartCropper()
cropper.smart_crop_video(input_path, output_path, ratio)
def _op_resize(self, input_path: str, output_path: str,
width: int = 1920, height: int = 1080, **kwargs):
"""调整尺寸操作"""
import subprocess
cmd = [
'ffmpeg', '-i', input_path,
'-vf', f'scale={width}:{height}:force_original_aspect_ratio=decrease,'
f'pad={width}:{height}:(ow-iw)/2:(oh-ih)/2',
'-c:a', 'copy', '-y', output_path
]
subprocess.run(cmd, capture_output=True, check=True)
def _op_add_watermark(self, input_path: str, output_path: str,
text: str = "© 2024", position: str = "bottom-right",
**kwargs):
"""添加水印操作"""
import subprocess
positions = {
'top-left': 'x=20:y=20',
'top-right': 'x=w-tw-20:y=20',
'bottom-left': 'x=20:y=h-th-20',
'bottom-right': 'x=w-tw-20:y=h-th-20',
'center': 'x=(w-tw)/2:y=(h-th)/2'
}
pos = positions.get(position, positions['bottom-right'])
cmd = [
'ffmpeg', '-i', input_path,
'-vf', f"drawtext=text='{text}':fontsize=24:fontcolor=white@"
f"0.5:{pos}:shadowcolor=black@0.5:shadowx=2:shadowy=2",
'-c:a', 'copy', '-y', output_path
]
subprocess.run(cmd, capture_output=True, check=True)
def _op_trim(self, input_path: str, output_path: str,
start: float = 0, end: float = None, **kwargs):
"""裁剪时长操作"""
import subprocess
cmd = ['ffmpeg', '-ss', str(start), '-i', input_path]
if end:
cmd.extend(['-to', str(end)])
cmd.extend(['-c', 'copy', '-y', output_path])
subprocess.run(cmd, capture_output=True, check=True)
def _op_change_speed(self, input_path: str, output_path: str,
speed: float = 1.0, **kwargs):
"""变速操作"""
import subprocess
video_filter = f"setpts={1/speed}*PTS"
audio_filter = f"atempo={min(2.0, max(0.5, speed))}"
# 对于极端速度，链式使用atempo
if speed > 2.0:
factors = []
remaining = speed
while remaining > 2.0:
factors.append(2.0)
remaining /= 2.0
factors.append(remaining)
audio_filter = ','.join(f'atempo={f}' for f in factors)
elif speed < 0.5:
factors = []
remaining = speed
while remaining < 0.5:
factors.append(0.5)
remaining /= 0.5
factors.append(remaining)
audio_filter = ','.join(f'atempo={f}' for f in factors)
cmd = [
'ffmpeg', '-i', input_path,
'-filter_complex', f'[0:v]{video_filter}[v];[0:a]{audio_filter}[a]',
'-map', '[v]', '-map', '[a]',
'-y', output_path
]
subprocess.run(cmd, capture_output=True, check=True)
def _op_extract_audio(self, input_path: str, output_path: str,
format: str = "mp3", **kwargs):
"""提取音频操作"""
import subprocess
cmd = [
'ffmpeg', '-i', input_path,
'-vn', '-acodec', 'libmp3lame' if format == 'mp3' else 'copy',
'-y', output_path
]
subprocess.run(cmd, capture_output=True, check=True)
# 使用示例
def demo_batch_processing():
"""批量处理示例"""
processor = VideoBatchProcessor(max_workers=4)
# 定义处理流水线
pipeline = ProcessingPipeline(
name="social_media_pipeline",
steps=[
{'operation': 'crop', 'params': {'ratio': 9/16}},
{'operation': 'filter', 'params': {'filter_name': 'cinematic'}},
{'operation': 'resize', 'params': {'width': 1080, 'height': 1920}},
{'operation': 'watermark', 'params': {'text': '@MyChannel', 'position': 'top-right'}},
],
parallel=True,
max_workers=4
)
# 创建任务列表
video_dir = "./raw_videos"
output_dir = "./processed_videos"
os.makedirs(output_dir, exist_ok=True)
tasks = []
for filename in os.listdir(video_dir):
if filename.endswith(('.mp4', '.avi', '.mov')):
tasks.append(ProcessingTask(
input_path=os.path.join(video_dir, filename),
output_path=os.path.join(output_dir, f"processed_{filename}"),
))
# 执行批量处理
print(f"开始批量处理 {len(tasks)} 个视频...")
results = processor.process_batch(tasks, pipeline)
print(f"\n处理完成:")
print(f" 成功: {results['success']}")
print(f" 失败: {results['failed']}")
print(f" 总计: {results['total']}")
# demo_batch_processing()
十一、FFmpeg + AI Pipeline 实战组合
11.1 完整的视频处理流水线
"""
FFmpeg + AI 完整视频处理流水线
将多个AI能力串联为自动化工作流
"""
import os
import subprocess
import json
from datetime import datetime
class VideoProductionPipeline:
"""视频生产流水线"""
def __init__(self, config: dict):
self.config = config
self.temp_dir = config.get('temp_dir', './temp')
os.makedirs(self.temp_dir, exist_ok=True)
def produce(self, input_video: str, script: str = None,
output_path: str = None) -> dict:
"""
完整视频制作流水线
Args:
input_video: 原始视频路径
script: 配音脚本（可选）
output_path: 最终输出路径
Returns:
各阶段产出文件路径
"""
results = {'input': input_video}
# Step 1: 智能裁剪
print("[1/7] 智能裁剪...")
cropped = self._step_smart_crop(input_video)
results['cropped'] = cropped
# Step 2: 风格滤镜
print("[2/7] 应用风格滤镜...")
filtered = self._step_apply_filter(cropped)
results['filtered'] = filtered
# Step 3: 自动字幕
print("[3/7] 生成自动字幕...")
srt_path = self._step_generate_subtitles(filtered)
results['subtitles'] = srt_path
# Step 4: 字幕烧录
print("[4/7] 烧录字幕...")
subtitled = self._step_burn_subtitles(filtered, srt_path)
results['subtitled'] = subtitled
# Step 5: AI配音（如果有脚本）
if script:
print("[5/7] 生成AI配音...")
voiced = self._step_add_voiceover(subtitled, script)
results['voiced'] = voiced
else:
voiced = subtitled
print("[5/7] 跳过配音（无脚本）")
# Step 6: 提取精彩片段
print("[6/7] 提取精彩片段...")
highlights = self._step_extract_highlights(voiced)
results['highlights'] = highlights
# Step 7: 最终输出
print("[7/7] 生成最终输出...")
final = self._step_final_output(voiced, output_path)
results['final'] = final
print("\n✓ 视频制作完成!")
for key, path in results.items():
if path and os.path.exists(str(path)):
size = os.path.getsize(path) / 1024 / 1024
print(f" {key}: {path} ({size:.1f}MB)")
return results
def _step_smart_crop(self, input_path: str) -> str:
"""智能裁剪步骤"""
output = os.path.join(self.temp_dir, "01_cropped.mp4")
target = self.config.get('target_ratio', 16/9)
cropper = SmartCropper()
cropper.smart_crop_video(input_path, output, target,
self.config.get('content_type', 'landscape'))
return output
def _step_apply_filter(self, input_path: str) -> str:
"""滤镜步骤"""
output = os.path.join(self.temp_dir, "02_filtered.mp4")
filter_name = self.config.get('filter', 'cinematic')
processor = VideoFilterProcessor()
processor.apply_filter_to_video(input_path, output, filter_name)
return output
def _step_generate_subtitles(self, input_path: str) -> str:
"""字幕生成步骤"""
srt_path = os.path.join(self.temp_dir, "03_subtitles.srt")
generator = SubtitleGenerator(
model_size=self.config.get('whisper_model', 'medium')
)
result = generator.transcribe(input_path, language='zh')
generator.generate_srt(result, srt_path)
return srt_path
def _step_burn_subtitles(self, input_path: str, srt_path: str) -> str:
"""字幕烧录步骤"""
output = os.path.join(self.temp_dir, "04_subtitled.mp4")
SubtitleBurner.burn_srt(
input_path, srt_path, output,
font_size=self.config.get('subtitle_font_size', 24)
)
return output
def _step_add_voiceover(self, input_path: str, script: str) -> str:
"""配音步骤"""
output = os.path.join(self.temp_dir, "05_voiceover.mp4")
tts = AIVoiceGenerator(engine='edge-tts')
vo_gen = VoiceOverGenerator(tts)
# 解析脚本
script_data = json.loads(script) if isinstance(script, str) else script
voiceover_files = vo_gen.generate_voiceover_from_script(
script_data,
os.path.join(self.temp_dir, 'voiceovers'),
voice=self.config.get('voice', 'zh-CN-YunxiNeural')
)
vo_gen.merge_voiceover_with_video(
input_path, voiceover_files, output,
bgm_path=self.config.get('bgm_path'),
bgm_volume=self.config.get('bgm_volume', 0.15)
)
return output
def _step_extract_highlights(self, input_path: str) -> str:
"""精彩片段提取步骤"""
output = os.path.join(self.temp_dir, "06_highlights.mp4")
summarizer = VideoSummarizer()
segments = summarizer.analyze_video(input_path, sample_interval=2.0)
highlights = summarizer.extract_highlights(
segments,
target_duration=self.config.get('highlight_duration', 60)
)
summarizer.generate_summary_video(input_path, output, highlights)
return output
def _step_final_output(self, input_path: str, output_path: str = None) -> str:
"""最终输出步骤"""
if output_path is None:
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
output_path = f"./output/final_{timestamp}.mp4"
os.makedirs(os.path.dirname(output_path), exist_ok=True)
# 最终编码优化
cmd = [
'ffmpeg', '-i', input_path,
'-c:v', 'libx264',
'-preset', 'medium',
'-crf', '23',
'-c:a', 'aac',
'-b:a', '128k',
'-movflags', '+faststart', # 优化网络播放
'-y', output_path
]
subprocess.run(cmd, capture_output=True, check=True)
return output_path
# 使用示例
def demo_production_pipeline():
"""演示完整制作流水线"""
config = {
'target_ratio': 16/9,
'content_type': 'landscape',
'filter': 'cinematic',
'whisper_model': 'medium',
'subtitle_font_size': 24,
'voice': 'zh-CN-YunxiNeural',
'highlight_duration': 120,
'temp_dir': './pipeline_temp',
}
# 配音脚本（可选）
script = [
{"text": "欢迎观看本期视频", "start": 0.0, "end": 3.0},
{"text": "今天我们来聊聊AI技术的最新进展", "start": 3.0, "end": 7.0},
{"text": "首先让我们看看大语言模型的发展", "start": 7.0, "end": 11.0},
]
pipeline = VideoProductionPipeline(config)
results = pipeline.produce(
input_video="raw_footage.mp4",
script=script,
output_path="./output/final_video.mp4"
)
# demo_production_pipeline()
十二、最佳实践与进阶建议
12.1 性能优化
- GPU加速：使用CUDA加速视频编解码
# FFmpeg NVIDIA GPU加速
cmd = ['ffmpeg', '-hwaccel', 'cuda', '-i', input_path, ...]
# PyTorch模型GPU推理
device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
model = model.to(device)
- 批量帧处理：避免逐帧I/O，使用批量推理
# 批量处理而非逐帧
batch_size = 16
for i in range(0, len(frames), batch_size):
batch = frames[i:i+batch_size]
results = model(batch) # 批量推理
- 流水线并行：使用多线程/多进程并行处理不同阶段
import multiprocessing as mp
def parallel_pipeline(input_files, num_workers=4):
with mp.Pool(num_workers) as pool:
results = pool.map(process_single_video, input_files)
return results
- 内存管理：大视频分段处理，及时释放资源
import gc
def process_large_video(video_path, chunk_duration=60):
"""分段处理大视频"""
# ... 按chunk_duration秒分段处理
gc.collect() # 每段处理后释放内存
12.2 常见问题
Q: Whisper中文识别准确率不高？
A: 使用medium或large模型，并指定language="zh"：
model = whisper.load_model("large")
result = model.transcribe("audio.mp3", language="zh")
Q: FFmpeg字幕烧录中文乱码？ A: 确保字幕文件编码为UTF-8，并指定字体：
ffmpeg -i input.mp4 -vf "subtitles=sub.srt:force_style='FontName=Microsoft YaHei'" output.mp4
Q: 视频风格迁移速度太慢？
A: 降低num_steps参数，或使用快速滤镜替代：
# 原始：100步优化
transfer.apply_to_frame(frame, num_steps=100)
# 快速：30步（质量略降但速度快3倍）
transfer.apply_to_frame(frame, num_steps=30)
Q: 批量处理时内存溢出？ A: 减少并行数，或使用流式处理：
# 降低并行度
pipeline = ProcessingPipeline(max_workers=2)
# 或改用串行处理
pipeline = ProcessingPipeline(parallel=False)
十三、总结
AI视频编辑与后期制作技术正在快速成熟，从"辅助工具"进化为"自动化流水线"。核心要点：
- 字幕是基础：Whisper + SRT/ASS是最成熟的AI视频应用，投入产出比最高
- 滤镜要快：实时滤镜用OpenCV，高质量风格迁移用神经网络（但慢）
- 智能裁剪：人脸检测 + 显著性检测是跨平台适配的关键
- 物体移除：OpenCV修复最快，Stable Diffusion效果最好，LaMa是平衡方案
- 批量为王：构建可复用的处理流水线，一次配置，批量执行
- FFmpeg为骨：所有视频操作最终都通过FFmpeg执行，它是不可替代的基础设施
掌握这套技术栈，你可以为团队构建一个"输入原始素材，输出成品视频"的全自动视频工厂，将传统需要数小时的后期制作压缩到分钟级别。
本教程配套完整代码已开源，欢迎Star和PR。
