基于FFmpeg与AI的自动化视频处理管道：深度解析数字人、转场、字幕、调色、配乐、特效的工程实现

原始素材（视频/图片） + 解说音频/文本
          ↓
[1] 音频预处理 → TTS（可选） + 节拍分析
          ↓
[2] 数字人生成 → Wav2Lip 合成说话人视频
          ↓
[3] 视频分镜 → 场景切分 + 转场插入
          ↓
[4] 字幕生成 → Whisper 识别 + ASS 渲染
          ↓
[5] 调色流水线 → 3D LUT 应用 + 色彩校正
          ↓
[6] 配乐混音 → 背景音乐 + 音量自动适配
          ↓
[7] 特效合成 → 粒子/文字动画（FFmpeg filter）
          ↓
      最终 MP4 输出

import subprocess
import os
import torch
from torch.nn import functional as F
from wav2lip.models.wav2lip import Wav2Lip
from wav2lip.audio import load_audio, melspectrogram
from wav2lip.face_detection import FaceDetection
import cv2
import numpy as np

class DigitalHumanGenerator:
    def __init__(self, model_path='checkpoints/wav2lip_gan.pth', face_path='static_face.jpg'):
        self.device = 'cuda' if torch.cuda.is_available() else 'cpu'
        self.model = Wav2Lip()
        self.model.load_state_dict(torch.load(model_path, map_location=self.device))
        self.model.to(self.device).eval()
        self.face_detector = FaceDetection()
        self.face_img = cv2.imread(face_path)
        self.face_roi, self.box = self._extract_face(self.face_img)
        
    def _extract_face(self, img):
        boxes = self.face_detector.detect(img)
        if len(boxes) == 0:
            raise RuntimeError("No face detected in static image")
        x1, y1, x2, y2 = boxes[0]
        roi = img[y1:y2, x1:x2]
        return roi, (x1, y1, x2, y2)
    
    def generate(self, audio_path, output_video='digital_human.mp4', fps=25):
        # 1. 提取音频 mel 谱
        wav = load_audio(audio_path, 16000)
        mel = melspectrogram(wav).T  # [T, 80]
        
        # 2. 逐帧推理
        frame_batch = []
        mel_chunks = torch.tensor(mel).unsqueeze(0).to(self.device)  # [1, T, 80]
        # 实际需要滑动窗口，每5帧取一次，此处简化演示
        with torch.no_grad():
            # 将静态脸 resize 到模型输入大小 (96x96)
            face_resized = cv2.resize(self.face_roi, (96, 96))
            face_tensor = torch.tensor(face_resized / 255.0).permute(2,0,1).unsqueeze(0).float().to(self.device)
            # Wav2Lip 需要 audio 特征和 face 序列，此处仅演示单帧循环
            # 真实场景需对每帧做仿射变换，这里简化生成全序列
            for i in range(mel_chunks.shape[1] - 5):  # 窗口长度5
                audio_window = mel_chunks[:, i:i+5, :]  # [1,5,80]
                # 复制 face 5次
                faces = face_tensor.repeat(1, 5, 1, 1)  # [1,5,3,96,96]
                pred = self.model(faces, audio_window)  # [1,5,3,96,96]
                # 取第一帧输出
                frame = pred[0,0].cpu().numpy().transpose(1,2,0) * 255
                frame = frame.astype(np.uint8)
                # 放回原图（简单替换）
                output_img = self.face_img.copy()
                x1,y1,x2,y2 = self.box
                resized_face = cv2.resize(frame, (x2-x1, y2-y1))
                output_img[y1:y2, x1:x2] = resized_face
                frame_batch.append(output_img)
        
        # 3. 写入视频
        h, w = self.face_img.shape[:2]
        out = cv2.VideoWriter(output_video, cv2.VideoWriter_fourcc(*'mp4v'), fps, (w, h))
        for f in frame_batch:
            out.write(f)
        out.release()
        # 4. 合并音频
        cmd = f"ffmpeg -y -i {output_video} -i {audio_path} -c:v libx264 -c:a aac -shortest {output_video.replace('.mp4','_with_audio.mp4')}"
        subprocess.run(cmd, shell=True)
        return output_video.replace('.mp4','_with_audio.mp4')

def add_transition(input1, input2, output, duration=0.5, transition='fade'):
    """
    将两个视频片段通过 transition 连接
    transition: fade, wipeleft, slideleft, circleopen, etc.
    """
    cmd = (
        f"ffmpeg -y -i {input1} -i {input2} "
        f"-filter_complex \"[0:v][1:v]xfade=transition={transition}:duration={duration}:offset={duration}[v]\" "
        f"-map \"[v]\" -c:v libx264 -preset fast {output}"
    )
    subprocess.run(cmd, shell=True, check=True)

def custom_wipe_transition(vid1_path, vid2_path, output, duration=1.0, fps=30):
    cap1 = cv2.VideoCapture(vid1_path)
    cap2 = cv2.VideoCapture(vid2_path)
    total_frames = int(duration * fps)
    width = int(cap1.get(cv2.CAP_PROP_FRAME_WIDTH))
    height = int(cap1.get(cv2.CAP_PROP_FRAME_HEIGHT))
    out = cv2.VideoWriter(output, cv2.VideoWriter_fourcc(*'mp4v'), fps, (width, height))
    
    for i in range(total_frames):
        ret1, frame1 = cap1.read()
        ret2, frame2 = cap2.read()
        if not ret1 or not ret2:
            break
        ratio = i / total_frames
        # 对角线擦除：从左下到右上
        mask = np.zeros((height, width), dtype=np.float32)
        for y in range(height):
            for x in range(width):
                if (x / width + y / height) / 2 < ratio:
                    mask[y, x] = 1.0
        blended = frame1 * (1 - mask[:,:,np.newaxis]) + frame2 * mask[:,:,np.newaxis]
        out.write(blended.astype(np.uint8))
    out.release()
    # 音频处理需单独合成

import whisper
import json
from ass_generator import make_ass  # 自定义库，下文给出

def generate_subtitles(audio_path, model_size='base', lang='zh'):
    model = whisper.load_model(model_size)
    result = model.transcribe(audio_path, language=lang, word_timestamps=True)
    segments = result['segments']
    # 转换为 ASS
    ass_content = make_ass(segments, font='微软雅黑', font_size=48, primary_color='&H00FFFFFF', outline_color='&H00000000')
    with open('subtitles.ass', 'w', encoding='utf-8') as f:
        f.write(ass_content)
    return 'subtitles.ass'

def make_ass(segments, **style):
    ass_template = f"""
[Script Info]
ScriptType: v4.00+
WrapStyle: 0
[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,{style.get('font','Arial')},{style.get('font_size',36)},{style.get('primary_color','&H00FFFFFF')},&H000000FF,{style.get('outline_color','&H00000000')},&H00000000,0,0,0,0,100,100,0,0,1,{style.get('outline',2)},{style.get('shadow',1)},2,20,20,20,1
[Events]
Format: Layer, Start, End, Style, Name, MarginL, MarginR, MarginV, Effect, Text
"""
    events = []
    for seg in segments:
        start = seg['start']
        end = seg['end']
        text = seg['text'].strip()
        # 转换为 ASS 时间格式 H:MM:SS.cc
        def to_ass(t):
            h = int(t // 3600)
            m = int((t % 3600) // 60)
            s = int(t % 60)
            cs = int((t - int(t)) * 100)
            return f"{h:01d}:{m:02d}:{s:02d}.{cs:02d}"
        events.append(f"Dialogue: 0,{to_ass(start)},{to_ass(end)},Default,,0,0,0,,{text}")
    return ass_template + "\n".join(events)

ffmpeg -i video.mp4 -vf "ass=subtitles.ass" -c:a copy output.mp4

def apply_lut(input_video, lut_cube='film_look.cube', output='graded.mp4'):
    cmd = f"ffmpeg -y -i {input_video} -vf \"lut3d={lut_cube}\" -c:a copy {output}"
    subprocess.run(cmd, shell=True)

def color_correct(input_video, brightness=0.05, contrast=1.1, saturation=1.2, output='corrected.mp4'):
    # FFmpeg eq 滤镜: brightness (-1~1), contrast (0~2), saturation (0~3)
    cmd = f"ffmpeg -y -i {input_video} -vf \"eq=brightness={brightness}:contrast={contrast}:saturation={saturation}\" -c:a copy {output}"
    subprocess.run(cmd, shell=True)

import numpy as np
def generate_lut_from_reference(ref_img, source_img, cube_size=33):
    # 将 RGB 颜色空间离散化为 cube_size^3 网格
    # 使用均值-方差颜色迁移
    ref = cv2.cvtColor(cv2.imread(ref_img), cv2.COLOR_BGR2RGB).reshape(-1, 3)
    src = cv2.cvtColor(cv2.imread(source_img), cv2.COLOR_BGR2RGB).reshape(-1, 3)
    mean_ref, std_ref = ref.mean(axis=0), ref.std(axis=0)
    mean_src, std_src = src.mean(axis=0), src.std(axis=0)
    # 生成 LUT 映射表
    lut = np.zeros((cube_size**3, 3), dtype=np.float32)
    for r in range(cube_size):
        for g in range(cube_size):
            for b in range(cube_size):
                idx = (r*cube_size + g)*cube_size + b
                # 归一化值 0~1
                rgb = np.array([r/(cube_size-1), g/(cube_size-1), b/(cube_size-1)])
                # 颜色迁移公式
                mapped = (rgb - mean_src/255) * (std_ref / (std_src + 1e-6)) + mean_ref/255
                lut[idx] = np.clip(mapped, 0, 1)
    # 写入 .cube 文件
    with open('auto_lut.cube', 'w') as f:
        f.write(f"TITLE Generated LUT\nLUT_3D_SIZE {cube_size}\n")
        for rgb in lut:
            f.write(f"{rgb[0]:.6f} {rgb[1]:.6f} {rgb[2]:.6f}\n")
    return 'auto_lut.cube'

import librosa
import numpy as np

def detect_beats(audio_path, sr=22050):
    y, sr = librosa.load(audio_path, sr=sr)
    tempo, beat_frames = librosa.beat.beat_track(y=y, sr=sr)
    beat_times = librosa.frames_to_time(beat_frames, sr=sr)
    return tempo, beat_times

def duck_audio(voice_path, music_path, output, duck_db=-12, threshold=0.3):
    """
    当语音存在时降低音乐音量，实现侧链压缩效果
    使用 ffmpeg 的 sidechaincompress 或 volume + aresample
    """
    cmd = (
        f"ffmpeg -y -i {music_path} -i {voice_path} "
        f"-filter_complex \"[1:a]aformat=channel_layouts=stereo,compand=attacks=0.1:decays=0.1:points=-80/-80|-45/-15|-27/-9|0/-7|20/-7:gain=5[comp]; "
        f"[0:a][comp]sidechaincompress=threshold={threshold}:ratio=4:release=50:makeup=1[out]\" "
        f"-map \"[out]\" -c:a aac -b:a 192k {output}"
    )
    subprocess.run(cmd, shell=True, check=True)

ffmpeg -i input.mp4 -vf "drawtext=text='Hello':fontsize=60:fontcolor=white:x=(w-text_w)/2:y=(h-text_h)/2:enable='between(t,2,4)', \
scale=iw*1.1:ih*1.1:zoompan='z=if(lte(on,1),1,min(z+0.1,1.5))':d=1:fps=30" \
-c:a copy output.mp4

def add_snow_effect(input_video, output, intensity=50):
    # geq 生成随机噪点作为雪花
    cmd = (
        f"ffmpeg -y -i {input_video} -vf \"geq=r='r(X,Y)+random(0,{intensity})':g='g(X,Y)+random(0,{intensity})':b='b(X,Y)+random(0,{intensity})'\" "
        f"-c:a copy {output}"
    )
    subprocess.run(cmd, shell=True)

import argparse

def main(args):
    # 1. 若没有解说音频，先 TTS（调用 edge-tts 或 tacotron2）
    if args.tts_text:
        audio_path = tts_generate(args.tts_text, 'output.wav')
    else:
        audio_path = args.audio
    
    # 2. 生成数字人视频（或跳过）
    if args.digital_human:
        dh_video = DigitalHumanGenerator().generate(audio_path)
        main_video = dh_video
    else:
        main_video = args.video
    
    # 3. 分镜与转场：按场景列表或随机裁剪
    segments = split_scenes(main_video, num_segments=5)
    transitioned = []
    for i in range(len(segments)-1):
        out_seg = f'seg_{i}.mp4'
        add_transition(segments[i], segments[i+1], out_seg, duration=0.5, transition='slideleft')
        transitioned.append(out_seg)
    # 合并所有转场片段
    concat_list = 'concat:' + '+'.join(transitioned)
    final_video = 'with_transitions.mp4'
    subprocess.run(f"ffmpeg -y -i concat_list -c copy {final_video}", shell=True)
    
    # 4. 字幕
    if args.subtitle:
        ass_file = generate_subtitles(audio_path)
        final_video = overlay_subtitles(final_video, ass_file)
    
    # 5. 调色
    if args.lut:
        final_video = apply_lut(final_video, args.lut)
    
    # 6. 配乐混音
    if args.music:
        duck_audio(audio_path, args.music, 'mixed_audio.m4a')
        # 替换视频音轨
        final_video = replace_audio(final_video, 'mixed_audio.m4a')
    
    # 7. 特效
    if args.effect == 'snow':
        final_video = add_snow_effect(final_video, 'final.mp4')
    else:
        shutil.copy(final_video, 'final.mp4')
    
    print(f"✅ 成片已生成：final.mp4")

if __name__ == '__main__':
    parser = argparse.ArgumentParser()
    parser.add_argument('--video', help='原始视频素材')
    parser.add_argument('--audio', help='解说音频')
    parser.add_argument('--tts_text', help='TTS 文字')
    parser.add_argument('--digital_human', action='store_true')
    parser.add_argument('--subtitle', action='store_true')
    parser.add_argument('--lut', help='LUT 文件')
    parser.add_argument('--music', help='背景音乐')
    parser.add_argument('--effect', choices=['snow', 'none'])
    args = parser.parse_args()
    main(args)