#!/usr/bin/env python3 """Pipeline principal de processamento de vídeo.""" import json import os import re from pathlib import Path import anthropic from whisper_utils import transcrever_audio from pexels_utils import buscar_videos, baixar_video from ffmpeg_utils import cortar_segmentos as _cortar_segmentos, montar_video_final from music_utils import obter_musica_para_video ANTHROPIC_KEY = "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" PEXELS_KEY = os.getenv("PEXELS_API_KEY", "PEXELS_API_KEY_PLACEHOLDER") TEMP_DIR = Path(__file__).parent / "temp" OUTPUT_DIR = Path(__file__).parent / "output" MUSIC_DIR = Path(__file__).parent / "music" # Configuração de janelas de B-roll JANELA_SEG = 7.0 # cada janela tem ~7s de fala MAX_JANELAS = 8 # no máximo 8 B-rolls (evita estourar quota e ficar visualmente exagerado) def _construir_janelas(segmentos: list) -> list: """ Agrupa segmentos do Whisper em janelas de ~JANELA_SEG segundos. Retorna [{"inicio": float, "fim": float, "texto": str}, ...]. """ if not segmentos: return [] janelas = [] atual = {"inicio": segmentos[0]["inicio"], "fim": segmentos[0]["fim"], "texto": segmentos[0]["texto"]} for seg in segmentos[1:]: # Se ainda cabe na janela atual, agrega if (seg["fim"] - atual["inicio"]) <= JANELA_SEG: atual["fim"] = seg["fim"] atual["texto"] += " " + seg["texto"] else: janelas.append(atual) atual = {"inicio": seg["inicio"], "fim": seg["fim"], "texto": seg["texto"]} janelas.append(atual) # Limita a MAX_JANELAS, distribuídas uniformemente if len(janelas) > MAX_JANELAS: step = len(janelas) / MAX_JANELAS janelas = [janelas[int(i * step)] for i in range(MAX_JANELAS)] return janelas def analisar_para_broll(transcricao: dict) -> list: """ Para cada janela de ~7s da transcrição, gera uma query visual em inglês via Claude. Retorna lista alinhada: [{"inicio": float, "fim": float, "query": str}, ...]. """ segmentos = transcricao.get("segmentos", []) janelas = _construir_janelas(segmentos) if not janelas: return [] client = anthropic.Anthropic(api_key=ANTHROPIC_KEY) # Monta uma única chamada com todas as janelas pra economizar tokens bloco_janelas = "\n".join( f"{i+1}. [{j['inicio']:.1f}s - {j['fim']:.1f}s] {j['texto'].strip()}" for i, j in enumerate(janelas) ) prompt = f"""Você está ajudando a editar um Reel para Instagram. Para cada trecho de fala numerado abaixo, sugira UMA query de busca em inglês (3-5 palavras visuais concretas) para encontrar um vídeo no Pexels que ILUSTRE visualmente o conceito específico falado naquele trecho. Regras: - Sempre em inglês - 3 a 5 palavras concretas e visuais (ex: "person typing laptop", "city skyline sunset", "team meeting office") - NÃO repita a mesma query - Se o trecho for muito genérico/conectivo (sem conceito visual claro), retorne "skip" como query Trechos: {bloco_janelas} Retorne APENAS um JSON válido neste formato (sem texto antes ou depois): {{ "brolls": [ {{"n": 1, "query": "person typing laptop"}}, {{"n": 2, "query": "skip"}}, ... ] }}""" try: resp = client.messages.create( model="claude-haiku-4-5", max_tokens=1200, messages=[{"role": "user", "content": prompt}] ) text = resp.content[0].text match = re.search(r"\{.*\}", text, re.DOTALL) if not match: print(f"[claude] Resposta sem JSON: {text[:200]}") return [] data = json.loads(match.group()) items = data.get("brolls", []) # Mapeia query de volta pras janelas pelo n out = [] for it in items: try: idx = int(it.get("n", 0)) - 1 query = (it.get("query") or "").strip() if idx < 0 or idx >= len(janelas): continue if not query or query.lower() == "skip": continue j = janelas[idx] out.append({ "inicio": float(j["inicio"]), "fim": float(j["fim"]), "query": query }) except Exception: continue return out except Exception as e: print(f"[claude] Erro ao analisar para b-roll: {e}") return [] def escolher_musica() -> str: """Retorna o caminho de uma música aleatória do diretório music/, ou '' se vazio.""" if not MUSIC_DIR.exists(): return "" musicas = list(MUSIC_DIR.glob("*.mp3")) + list(MUSIC_DIR.glob("*.m4a")) + list(MUSIC_DIR.glob("*.aac")) if not musicas: return "" import random return str(random.choice(musicas)) def processar_video( job_id: str, video_path: str, modo: str, segmentos: list, usar_musica: bool, estilo_legenda: str, callback_status, tamanho_legenda: str = "medio", musica_path_direto: str = None, ): """ Pipeline completo: 1. Corta segmentos (modo live) 2. Transcrição Whisper 3. Análise Claude para b-roll (janelas alinhadas) 4. Download b-roll Pexels 5. Montagem final com FFmpeg """ try: # Fase 1 callback_status(job_id, "fila", 5, "Preparando vídeo...") if modo == "live" and segmentos: print(f"[proc] Cortando {len(segmentos)} segmentos...") video_path = _cortar_segmentos(video_path, segmentos, job_id) # Fase 2 callback_status(job_id, "transcrevendo", 15, "Transcrevendo áudio com IA...") print(f"[proc] Transcrevendo {video_path}...") transcricao = transcrever_audio(video_path) print(f"[proc] Transcrição OK. Segmentos: {len(transcricao['segmentos'])}") # Fase 3 callback_status(job_id, "analisando", 30, "Analisando conteúdo para buscar B-roll...") termos = analisar_para_broll(transcricao) print(f"[proc] B-rolls alinhados: {[(t['inicio'], t['fim'], t['query']) for t in termos]}") # Fase 4: baixa cada B-roll alinhado callback_status(job_id, "baixando_broll", 45, "Buscando vídeos de apoio...") clips_broll = [] for i, termo in enumerate(termos): query = termo.get("query", "") if not query: continue videos = buscar_videos(query, quantidade=2) baixado = False for v in videos[:2]: dest = str(TEMP_DIR / f"{job_id}_broll_{i:02d}.mp4") ok = baixar_video(v["url"], dest) if ok: clips_broll.append({ "path": dest, "inicio": termo["inicio"], "fim": termo["fim"] }) baixado = True break status_msg = f"B-roll {i+1}/{len(termos)}: {query}" + ("" if baixado else " (sem resultado)") callback_status(job_id, "baixando_broll", min(45 + i * 3, 60), status_msg) print(f"[proc] B-rolls baixados: {len(clips_broll)}") # Fase 5 callback_status(job_id, "montando", 65, "Montando vídeo final...") output_path = str(OUTPUT_DIR / f"{job_id}_final.mp4") # Define a trilha sonora (3 caminhos): # 1) musica_path_direto -> usuário escolheu via biblioteca # 2) ccMixter automático # 3) Fallback aleatório local (manifest) musica_path = "" if usar_musica: if musica_path_direto and Path(musica_path_direto).exists(): musica_path = musica_path_direto callback_status(job_id, "montando", 60, "Aplicando trilha selecionada...") print(f"[proc] Música escolhida pelo usuário: {musica_path}") else: callback_status(job_id, "montando", 60, "Buscando trilha sonora automática...") try: musica_path = obter_musica_para_video(transcricao.get("texto", ""), job_id) or "" except Exception as e: print(f"[proc] ccMixter falhou: {e}") musica_path = "" if musica_path: print(f"[proc] Música automática (ccMixter): {musica_path}") else: # Fallback: amostra aleatória da biblioteca local (varre subpastas) musicas_locais = [] if MUSIC_DIR.exists(): for sub in MUSIC_DIR.iterdir(): if sub.is_dir(): musicas_locais.extend(sub.glob("*.mp3")) musicas_locais.extend(MUSIC_DIR.glob("*.mp3")) if musicas_locais: import random musica_path = str(random.choice(musicas_locais)) print(f"[proc] Música local (fallback): {musica_path}") montar_video_final( video_principal=video_path, clips_broll=clips_broll, transcricao=transcricao, musica_path=musica_path, estilo_legenda=estilo_legenda, job_id=job_id, output_path=output_path, tamanho_legenda=tamanho_legenda, ) if not Path(output_path).exists(): raise RuntimeError("Arquivo de saída não foi gerado pelo FFmpeg.") callback_status(job_id, "concluido", 100, "Pronto! Seu Reel está pronto.", output_path) except Exception as e: print(f"[proc] ERRO no job {job_id}: {e}") import traceback traceback.print_exc() callback_status(job_id, "erro", 0, f"Erro no processamento: {str(e)}")