Automatizando o Pipeline com FFmpeg

after renderiza ProRes
       |
Script detecta novo arquivo
       |
FFmpeg converte para múltiplos formatos
       |
Arquivos organizados por plataforma
       |
Notificação enviada (email, Slack)

import subprocess, os, glob, json
from pathlib import Path

class RenderPipeline:
    def __init__(self, project_name, output_dir):
        self.project = project_name
        self.output_dir = Path(output_dir)
        self.output_dir.mkdir(parents=True, exist_ok=True)

    def sequence_to_prores(self, pattern, fps=24):
        output = self.output_dir / f"{self.project}_master.mov"
        subprocess.run([
            'ffmpeg', '-framerate', str(fps), '-i', pattern,
            '-c:v', 'prores_ks', '-profile:v', '3',
            '-c:a', 'pcm_s16le', str(output)
        ], check=True)
        return output

    def export_youtube(self, master):
        output = self.output_dir / f"{self.project}_youtube.mp4"
        subprocess.run([
            'ffmpeg', '-i', str(master),
            '-c:v', 'libx264', '-crf', '18', '-preset', 'slow',
            '-pix_fmt', 'yuv420p', '-c:a', 'aac', '-b:a', '192k',
            '-movflags', '+faststart', str(output)
        ], check=True)
        return output

    def export_instagram(self, master):
        output = self.output_dir / f"{self.project}_instagram.mp4"
        subprocess.run([
            'ffmpeg', '-i', str(master),
            '-vf', 'crop=1080:1920:420:0',
            '-c:v', 'libx264', '-crf', '18', '-pix_fmt', 'yuv420p',
            '-c:a', 'aac', '-b:a', '192k', str(output)
        ], check=True)
        return output

    def run(self, frame_pattern, fps=24):
        print(f"Pipeline: {self.project}")

        print("1/3 Convertendo sequência para ProRes...")
        master = self.sequence_to_prores(frame_pattern, fps)

        print("2/3 Exportando YouTube...")
        yt = self.export_youtube(master)

        print("3/3 Exportando Instagram...")
        ig = self.export_instagram(master)

        for label, path in [("YouTube", yt), ("Instagram", ig)]:
            size = os.path.getsize(path) / (1024*1024)
            print(f"  {label}: {path.name} ({size:.1f}MB)")

# Uso
pipeline = RenderPipeline("hero_animation_v03", "/delivery/hero_animation")
pipeline.run("/renders/hero_animation/frame_%04d.png", fps=24)

import time, os, glob, subprocess

def watch_and_process(watch_dir, output_dir, interval=10):
    processed = set()
    print(f"Monitorando: {watch_dir}")

    while True:
        files = set(glob.glob(os.path.join(watch_dir, '*.mov')))
        new_files = files - processed

        for filepath in new_files:
            size1 = os.path.getsize(filepath)
            time.sleep(2)
            size2 = os.path.getsize(filepath)

            if size1 == size2:  # arquivo estável (não está sendo gravado)
                print(f"Novo: {os.path.basename(filepath)}")
                output = os.path.join(output_dir,
                    os.path.basename(filepath).replace('.mov', '.mp4'))
                subprocess.run([
                    'ffmpeg', '-i', filepath,
                    '-c:v', 'libx264', '-crf', '18', '-pix_fmt', 'yuv420p',
                    '-movflags', '+faststart', output
                ])
                processed.add(filepath)

        time.sleep(interval)

watch_and_process('/renders/watch_folder', '/delivery/auto')

import subprocess, os, glob, time, requests, json

WEBHOOK_SLACK = 'https://hooks.slack.com/services/XXX/YYY/ZZZ'
PASTA_RENDERS = '/renders/prores'
PASTA_ENTREGA = '/entregas'

def converter(entrada, saida, preset):
    os.makedirs(os.path.dirname(saida), exist_ok=True)
    subprocess.run(['ffmpeg', '-i', entrada, '-y'] + preset + [saida], check=True)

def processar_arquivo(arquivo):
    nome = os.path.splitext(os.path.basename(arquivo))[0]

    print(f'Processando: {nome}')

    converter(arquivo,
        f'{PASTA_ENTREGA}/youtube/{nome}_youtube.mp4',
        ['-c:v', 'libx264', '-crf', '18', '-pix_fmt', 'yuv420p', '-c:a', 'aac', '-b:a', '192k']
    )

    converter(arquivo,
        f'{PASTA_ENTREGA}/instagram/{nome}_square.mp4',
        ['-vf', 'scale=1080:1080:force_original_aspect_ratio=decrease,pad=1080:1080:(ow-iw)/2:(oh-ih)/2:black',
         '-c:v', 'libx264', '-crf', '20', '-pix_fmt', 'yuv420p']
    )

    notificar(f'Entrega concluída: {nome}')

def notificar(mensagem):
    payload = {'text': mensagem}
    requests.post(WEBHOOK_SLACK, data=json.dumps(payload),
                  headers={'Content-Type': 'application/json'})

processados = set()

print(f'Monitorando {PASTA_RENDERS}...')
while True:
    arquivos = set(glob.glob(os.path.join(PASTA_RENDERS, '*.mov')))
    novos = arquivos - processados

    for arquivo in novos:
        processar_arquivo(arquivo)
        processados.add(arquivo)

    time.sleep(30)

import subprocess, json

def verificar_arquivo(arquivo):
    cmd = ['ffprobe', '-v', 'quiet', '-print_format', 'json',
           '-show_streams', '-show_format', arquivo]
    result = subprocess.run(cmd, capture_output=True, text=True)
    data = json.loads(result.stdout)

    video = next(s for s in data['streams'] if s['codec_type'] == 'video')

    checks = {
        'resolucao': f"{video['width']}x{video['height']}",
        'fps': eval(video['r_frame_rate']),
        'codec': video['codec_name'],
        'duracao': float(data['format']['duration'])
    }

    return checks

info = verificar_arquivo('/entregas/youtube/projeto_youtube.mp4')
print(f"Codec: {info['codec']}")
print(f"Resolução: {info['resolucao']}")
print(f"FPS: {info['fps']:.2f}")
print(f"Duração: {info['duracao']:.1f}s")