"""Transcodifica vídeos de origem para um proxy leve usado na leitura visual e de cenas. O gargalo de ler frames direto dos arquivos de origem (ex.: MOV com vários GB) está em buscar com ``-ss`` em arquivos pesados e não indexados. Este módulo produz, uma única vez por arquivo e por resolução, uma cópia leve (proxy) em H.264 com keyframes frequentes e ``+faststart``, sem áudio. Toda a análise visual e detecção de cena passa a ler desse proxy, que é pequeno, tem seek rápido e leitura sequencial barata — enquanto o arquivo original fica intocado para áudio/transcrição. """ from __future__ import annotations import hashlib from pathlib import Path import subprocess from ...dominio.entidades.modelos import Clipe, IntervaloDeTempo from ..visual.contratos import ExtratorDeQuadros # Resolução padrão do proxy: Full HD (1080) — mantém a qualidade de leitura # visual (faces/OCR/pose) num arquivo bem mais leve e rápido de processar do # que o original em 4K. O formato H.264/MP4 +faststart com keyframes frequentes # é o que torna o seek barato. RESOLUCAO_PADRAO_DO_PROXY = 1080 # Prazo máximo para transcodificar um único arquivo (protetor contra falhas). LIMITE_DE_TEMPO_DA_TRANSCODIFICACAO = 1800 # 30 min class CompactadorDeProxy: """Transcodifica arquivos de vídeo para um proxy leve e o reaproveita.""" def __init__(self, ffmpeg: str = "ffmpeg", ffprobe: str = "ffprobe", diretorio_de_cache: str | Path | None = None, largura_maxima: int = RESOLUCAO_PADRAO_DO_PROXY) -> None: self.ffmpeg = ffmpeg self.ffprobe = ffprobe self.largura_maxima = int(largura_maxima) self.diretorio_de_cache = ( Path(diretorio_de_cache) if diretorio_de_cache else Path.home() / ".cache" / "premiere-mcp" / "proxies" ) self.diretorio_de_cache.mkdir(parents=True, exist_ok=True) def obter(self, arquivo: str | Path, largura_maxima: int | None = None) -> Path: """Devolve o caminho do proxy do arquivo, gerando-o quando necessário. O proxy é idempotente: se ele já existir e o arquivo de origem não tiver mudado desde a última geração, devolve o caminho sem re-transcodificar. """ largura = int(largura_maxima or self.largura_maxima) origem = Path(arquivo).expanduser().resolve(strict=False) if not origem.is_file(): raise FileNotFoundError(f"Arquivo de origem não encontrado: {origem}") proxy = self._caminho_do_proxy(origem, largura) if proxy.is_file() and self._proxies_sao_validos(origem, proxy): return proxy self._transcodificar(origem, proxy, largura) return proxy def obter_por_clipe(self, clipe, largura_maxima: int | None = None) -> Path | None: """Devolve o proxy do arquivo do clipe, ou None quando não houver arquivo. Se o clipe for offline ou não tiver arquivo, devolve None (sem proxy). """ if not getattr(clipe, "arquivo", None) or getattr(clipe, "offline", False): return None try: return self.obter(clipe.arquivo, largura_maxima=largura_maxima) except (FileNotFoundError, RuntimeError): return None # ------------------------------------------------------------------ internos def _caminho_do_proxy(self, origem: Path, largura_maxima: int) -> Path: """Monta o nome estável do proxy a partir do caminho e da largura.""" resumo = hashlib.sha1(str(origem).encode("utf-8")).hexdigest()[:12] return self.diretorio_de_cache / f"{origem.stem}-{resumo}-{largura_maxima}p.mp4" @staticmethod def _alteracao_do_arquivo(caminho: Path) -> float: """Última alteração do arquivo, útil para detectar que a origem mudou.""" try: return caminho.stat().st_mtime_ns except OSError: return 0.0 def _proxies_sao_validos(self, origem: Path, proxy: Path) -> bool: """O proxy é válido só se nasceu depois do arquivo de origem mudar.""" return proxy.stat().st_mtime_ns >= self._alteracao_do_arquivo(origem) def _transcodificar(self, origem: Path, proxy: Path, largura_maxima: int) -> None: """Transcodifica o arquivo de origem para o proxy H.264 de leitura visual.""" comando = [ self.ffmpeg, "-y", "-hide_banner", "-loglevel", "error", "-i", str(origem), "-map", "0:v:0", "-vf", ( f"scale=w={largura_maxima}:h={largura_maxima}:" "force_original_aspect_ratio=decrease" ), "-c:v", "libx264", "-preset", "veryfast", "-crf", "28", "-g", "30", "-keyint_min", "30", "-sc_threshold", "0", "-pix_fmt", "yuv420p", "-an", "-movflags", "+faststart", str(proxy), ] try: processo = subprocess.run( comando, capture_output=True, text=True, timeout=LIMITE_DE_TEMPO_DA_TRANSCODIFICACAO, ) except subprocess.TimeoutExpired as exc: raise RuntimeError( f"Transcodificação do proxy excedeu o tempo: {origem}" ) from exc if processo.returncode != 0 or not proxy.is_file(): detalhe = (processo.stderr or processo.stdout or "").strip() raise RuntimeError( f"Não foi possível gerar o proxy de {origem}: {detalhe}" ) def _clipe_com_arquivo(clipe: Clipe, caminho: Path) -> Clipe: """Devolve uma cópia do clipe com o ``arquivo`` substituído pelo caminho dado. Mantém a identidade, os intervalos e os metadados; só troca o arquivo lido. """ origem = clipe.intervalo_na_origem return Clipe( identificador=clipe.identificador, nome=clipe.nome, intervalo_na_timeline=clipe.intervalo_na_timeline, intervalo_na_origem=IntervaloDeTempo(origem.inicio, origem.fim) if origem else None, arquivo=str(caminho), identificador_da_faixa=clipe.identificador_da_faixa, offline=clipe.offline, metadados=clipe.metadados, ) class ExtratorDeQuadrosDoProxy: """Extrai frames de um clipe a partir do seu proxy, sem duplicar lógica. Delega todo o trabalho ao ``ExtratorDeQuadrosFFmpeg`` informado, apontando-o para o proxy. Os timestamps continuam relativos ao tempo do arquivo original, pois o proxy preserva a mesma cronologia. """ def __init__(self, extrator: ExtratorDeQuadros, compactador: CompactadorDeProxy, largura_maxima: int | None = None) -> None: self.extrator = extrator self.compactador = compactador self.largura_maxima = largura_maxima def extrair(self, clipe: Clipe) -> list: """Extrai frames usando o proxy; usa o arquivo original quando não há proxy.""" proxy = ( self.compactador.obter_por_clipe(clipe) if self.largura_maxima is None else self.compactador.obter_por_clipe(clipe, largura_maxima=self.largura_maxima) ) alvo = clipe if proxy is None else _clipe_com_arquivo(clipe, proxy) return self.extrator.extrair(alvo)