api/ltx_server.py

# ltx_server.py — VideoService (beta 1.1)
# Sempre output_type="latent"; no final: VAE (bloco inteiro) → pixels → MP4.
# Ignora UserWarning/FutureWarning e injeta VAE no manager com dtype/device corretos.
# --- 0. WARNINGS E AMBIENTE ---

import warnings
warnings.filterwarnings("ignore", category=UserWarning)
warnings.filterwarnings("ignore", category=FutureWarning)
warnings.filterwarnings("ignore", message=".*")
from huggingface_hub import logging
logging.set_verbosity_error()
logging.set_verbosity_warning()
logging.set_verbosity_info()
logging.set_verbosity_debug()
LTXV_DEBUG=1
LTXV_FRAME_LOG_EVERY=8
import os, subprocess, shlex, tempfile
import torch
import json
import numpy as np
import random
import os
import shlex
import yaml
from typing import List, Dict
from pathlib import Path
import imageio
from PIL import Image # Import adicionado para handle_media_upload_for_dims
import tempfile
from huggingface_hub import hf_hub_download
import sys
import subprocess
import gc
import shutil
import contextlib
import time
import traceback
from einops import rearrange
import torch.nn.functional as F
from managers.vae_manager import vae_manager_singleton
from tools.video_encode_tool import video_encode_tool_singleton
DEPS_DIR = Path("/data")
LTX_VIDEO_REPO_DIR = DEPS_DIR / "LTX-Video"

# CORREÇÃO: Movido run_setup para o início para garantir que seja definido antes de ser chamado.
def run_setup():
    setup_script_path = "setup.py"
    if not os.path.exists(setup_script_path):
        print("[DEBUG] 'setup.py' não encontrado. Pulando clonagem de dependências.")
        return
    try:
        print("[DEBUG] Executando setup.py para dependências...")
        subprocess.run([sys.executable, setup_script_path], check=True)
        print("[DEBUG] Setup concluído com sucesso.")
    except subprocess.CalledProcessError as e:
        print(f"[DEBUG] ERRO no setup.py (code {e.returncode}). Abortando.")
        sys.exit(1)
        
if not LTX_VIDEO_REPO_DIR.exists():
    print(f"[DEBUG] Repositório não encontrado em {LTX_VIDEO_REPO_DIR}. Rodando setup...")
    run_setup()

def add_deps_to_path():
    repo_path = str(LTX_VIDEO_REPO_DIR.resolve())
    if str(LTX_VIDEO_REPO_DIR.resolve()) not in sys.path:
        sys.path.insert(0, repo_path)
        print(f"[DEBUG] Repo adicionado ao sys.path: {repo_path}")
def _query_gpu_processes_via_nvml(device_index: int) -> List[Dict]:
    try:
        import psutil
        import pynvml as nvml
        nvml.nvmlInit()
        handle = nvml.nvmlDeviceGetHandleByIndex(device_index)
        try:
            procs = nvml.nvmlDeviceGetComputeRunningProcesses_v3(handle)
        except Exception:
            procs = nvml.nvmlDeviceGetComputeRunningProcesses(handle)
        results = []
        for p in procs:
            pid = int(p.pid)
            used_mb = None
            try:
                if getattr(p, "usedGpuMemory", None) is not None and p.usedGpuMemory not in (0,):
                    used_mb = max(0, int(p.usedGpuMemory) // (1024 * 1024))
            except Exception:
                used_mb = None
            name = "unknown"
            user = "unknown"
            try:
                import psutil
                pr = psutil.Process(pid)
                name = pr.name()
                user = pr.username()
            except Exception:
                pass
            results.append({"pid": pid, "name": name, "user": user, "used_mb": used_mb})
        nvml.nvmlShutdown()
        return results
    except Exception:
        return []
def _query_gpu_processes_via_nvidiasmi(device_index: int) -> List[Dict]:
    cmd = f"nvidia-smi -i {device_index} --query-compute-apps=pid,process_name,used_memory --format=csv,noheader,nounits"
    try:
        out = subprocess.check_output(shlex.split(cmd), stderr=subprocess.STDOUT, text=True, timeout=2.0)
    except Exception:
        return []
    results = []
    for line in out.strip().splitlines():
        parts = [p.strip() for p in line.split(",")]
        if len(parts) >= 3:
            try:
                pid = int(parts[0]); name = parts[1]; used_mb = int(parts[2])
                user = "unknown"
                try:
                    import psutil
                    pr = psutil.Process(pid)
                    user = pr.username()
                except Exception:
                    pass
                results.append({"pid": pid, "name": name, "user": user, "used_mb": used_mb})
            except Exception:
                continue
    return results
def calculate_new_dimensions(orig_w, orig_h, divisor=8):
    if orig_w == 0 or orig_h == 0:
        return 512, 512
    if orig_w >= orig_h:
        aspect_ratio = orig_w / orig_h
        new_h = 512
        new_w = new_h * aspect_ratio
    else:
        aspect_ratio = orig_h / orig_w
        new_w = 512
        new_h = new_w * aspect_ratio
    final_w = int(round(new_w / divisor)) * divisor
    final_h = int(round(new_h / divisor)) * divisor
    final_w = max(divisor, final_w)
    final_h = max(divisor, final_h)
    print(f"[Dimension Calc] Original: {orig_w}x{orig_h} -> Calculado: {new_w:.0f}x{new_h:.0f} -> Final (divisível por {divisor}): {final_w}x{final_h}")
    return final_h, final_w
def handle_media_upload_for_dims(filepath, current_h, current_w):
    # CORREÇÃO: Gradio (`gr`) não deve ser usado no backend. Retornando tupla diretamente.
    if not filepath or not os.path.exists(str(filepath)):
        return current_h, current_w
    try:
        if str(filepath).lower().endswith(('.png', '.jpg', '.jpeg', '.webp')):
            with Image.open(filepath) as img:
                orig_w, orig_h = img.size
        else: 
            with imageio.get_reader(filepath) as reader:
                meta = reader.get_meta_data()
                orig_w, orig_h = meta.get('size', (current_w, current_h))
        new_h, new_w = calculate_new_dimensions(orig_w, orig_h)
        return new_h, new_w
    except Exception as e:
        print(f"Erro ao processar mídia para dimensões: {e}")
        return current_h, current_w
def _gpu_process_table(processes: List[Dict], current_pid: int) -> str:
    if not processes:
        return "  - Processos ativos: (nenhum)\n"
    processes = sorted(processes, key=lambda x: (x.get("used_mb") or 0), reverse=True)
    lines = ["  - Processos ativos (PID | USER | NAME | VRAM MB):"]
    for p in processes:
        star = "*" if p["pid"] == current_pid else " "
        used_str = str(p["used_mb"]) if p.get("used_mb") is not None else "N/A"
        lines.append(f"    {star} {p['pid']} | {p['user']} | {p['name']} | {used_str}")
    return "\n".join(lines) + "\n"
def log_tensor_info(tensor, name="Tensor"):
    if not isinstance(tensor, torch.Tensor):
        print(f"\n[INFO] '{name}' não é tensor.")
        return
    print(f"\n--- Tensor: {name} ---")
    print(f"  - Shape: {tuple(tensor.shape)}")
    print(f"  - Dtype: {tensor.dtype}")
    print(f"  - Device: {tensor.device}")
    if tensor.numel() > 0:
        try:
            print(f"  - Min: {tensor.min().item():.4f}  Max: {tensor.max().item():.4f}  Mean: {tensor.mean().item():.4f}")
        except Exception:
            pass
    print("------------------------------------------\n")
add_deps_to_path()
from ltx_video.pipelines.pipeline_ltx_video import ConditioningItem, LTXMultiScalePipeline
from ltx_video.utils.skip_layer_strategy import SkipLayerStrategy
from ltx_video.models.autoencoders.vae_encode import un_normalize_latents, normalize_latents
from ltx_video.pipelines.pipeline_ltx_video import adain_filter_latent
from api.ltx.inference import (
    create_ltx_video_pipeline,
    create_latent_upsampler,
    load_image_to_tensor_with_resize_and_crop,
    seed_everething,
    calculate_padding,
    load_media_file,
)
class VideoService:
    def __init__(self):
        t0 = time.perf_counter()
        print("[DEBUG] Inicializando VideoService...")
        self.debug = os.getenv("LTXV_DEBUG", "1") == "1"
        self.frame_log_every = int(os.getenv("LTXV_FRAME_LOG_EVERY", "8"))
        self.config = self._load_config()
        print(f"[DEBUG] Config carregada (precision={self.config.get('precision')}, sampler={self.config.get('sampler')})")
        self.device = "cuda" if torch.cuda.is_available() else "cpu"
        print(f"[DEBUG] Device selecionado: {self.device}")
        self.last_memory_reserved_mb = 0.0
        self._tmp_dirs = set(); self._tmp_files = set(); self._last_outputs = []

        self.pipeline, self.latent_upsampler = self._load_models()
        print(f"[DEBUG] Pipeline e Upsampler carregados. Upsampler ativo? {bool(self.latent_upsampler)}")

        print(f"[DEBUG] Movendo modelos para {self.device}...")
        self.pipeline.to(self.device)
        if self.latent_upsampler:
            self.latent_upsampler.to(self.device)

        self._apply_precision_policy()
        print(f"[DEBUG] runtime_autocast_dtype = {getattr(self, 'runtime_autocast_dtype', None)}")

        vae_manager_singleton.attach_pipeline(
            self.pipeline,
            device=self.device,
            autocast_dtype=self.runtime_autocast_dtype
        )
        print(f"[DEBUG] VAE manager conectado: has_vae={hasattr(self.pipeline, 'vae')} device={self.device}")

        if self.device == "cuda":
            torch.cuda.empty_cache()
            self._log_gpu_memory("Após carregar modelos")

        print(f"[DEBUG] VideoService pronto. boot_time={time.perf_counter()-t0:.3f}s")

    def _log_gpu_memory(self, stage_name: str):
        if self.device != "cuda":
            return
        device_index = torch.cuda.current_device() if torch.cuda.is_available() else 0
        current_reserved_b = torch.cuda.memory_reserved(device_index)
        current_reserved_mb = current_reserved_b / (1024 ** 2)
        total_memory_b = torch.cuda.get_device_properties(device_index).total_memory
        total_memory_mb = total_memory_b / (1024 ** 2)
        peak_reserved_mb = torch.cuda.max_memory_reserved(device_index) / (1024 ** 2)
        delta_mb = current_reserved_mb - getattr(self, "last_memory_reserved_mb", 0.0)
        processes = _query_gpu_processes_via_nvml(device_index) or _query_gpu_processes_via_nvidiasmi(device_index)
        print(f"\n--- [LOG GPU] {stage_name} (cuda:{device_index}) ---")
        print(f"  - Reservado: {current_reserved_mb:.2f} MB / {total_memory_mb:.2f} MB  (Δ={delta_mb:+.2f} MB)")
        if peak_reserved_mb > getattr(self, "last_memory_reserved_mb", 0.0):
            print(f"  - Pico reservado (nesta fase): {peak_reserved_mb:.2f} MB")
        print(_gpu_process_table(processes, os.getpid()), end="")
        print("--------------------------------------------------\n")
        self.last_memory_reserved_mb = current_reserved_mb

    def _register_tmp_dir(self, d: str):
        if d and os.path.isdir(d):
            self._tmp_dirs.add(d); print(f"[DEBUG] Registrado tmp dir: {d}")

    def _register_tmp_file(self, f: str):
        if f and os.path.exists(f):
            self._tmp_files.add(f); print(f"[DEBUG] Registrado tmp file: {f}")

    def finalize(self, keep_paths=None, extra_paths=None, clear_gpu=True):
        print("[DEBUG] Finalize: iniciando limpeza...")
        keep = set(keep_paths or []); extras = set(extra_paths or [])
        removed_files = 0
        for f in list(self._tmp_files | extras):
            try:
                if f not in keep and os.path.isfile(f):
                    os.remove(f); removed_files += 1; print(f"[DEBUG] Removido arquivo tmp: {f}")
            except Exception as e:
                print(f"[DEBUG] Falha removendo arquivo {f}: {e}")
            finally:
                self._tmp_files.discard(f)
        removed_dirs = 0
        for d in list(self._tmp_dirs):
            try:
                if d not in keep and os.path.isdir(d):
                    shutil.rmtree(d, ignore_errors=True); removed_dirs += 1; print(f"[DEBUG] Removido diretório tmp: {d}")
            except Exception as e:
                print(f"[DEBUG] Falha removendo diretório {d}: {e}")
            finally:
                self._tmp_dirs.discard(d)
        print(f"[DEBUG] Finalize: arquivos removidos={removed_files}, dirs removidos={removed_dirs}")
        gc.collect()
        try:
            if clear_gpu and torch.cuda.is_available():
                torch.cuda.empty_cache()
                try:
                    torch.cuda.ipc_collect()
                except Exception:
                    pass
        except Exception as e:
            print(f"[DEBUG] Finalize: limpeza GPU falhou: {e}")
        try:
            self._log_gpu_memory("Após finalize")
        except Exception as e:
            print(f"[DEBUG] Log GPU pós-finalize falhou: {e}")

    def _load_config(self):
        base = LTX_VIDEO_REPO_DIR / "configs"
        candidates = [
            base / "ltxv-13b-0.9.8-dev-fp8.yaml",
            base / "ltxv-13b-0.9.8-distilled-fp8.yaml",
            base / "ltxv-13b-0.9.8-distilled.yaml",
        ]
        for cfg in candidates:
            if cfg.exists():
                print(f"[DEBUG] Config selecionada: {cfg}")
                with open(cfg, "r") as file:
                    return yaml.safe_load(file)
        cfg = base / "ltxv-13b-0.9.8-distilled-fp8.yaml"
        print(f"[DEBUG] Config fallback: {cfg}")
        with open(cfg, "r") as file:
            return yaml.safe_load(file)

    def _load_models(self):
        t0 = time.perf_counter()
        LTX_REPO = "Lightricks/LTX-Video"
        print("[DEBUG] Baixando checkpoint principal...")
        distilled_model_path = hf_hub_download(
            repo_id=LTX_REPO,
            filename=self.config["checkpoint_path"],
            local_dir=os.getenv("HF_HOME"),
            cache_dir=os.getenv("HF_HOME_CACHE"),
            token=os.getenv("HF_TOKEN"),
        )
        self.config["checkpoint_path"] = distilled_model_path
        print(f"[DEBUG] Checkpoint em: {distilled_model_path}")

        print("[DEBUG] Baixando upscaler espacial...")
        spatial_upscaler_path = hf_hub_download(
            repo_id=LTX_REPO,
            filename=self.config["spatial_upscaler_model_path"],
            local_dir=os.getenv("HF_HOME"),
            cache_dir=os.getenv("HF_HOME_CACHE"),
            token=os.getenv("HF_TOKEN")
        )
        self.config["spatial_upscaler_model_path"] = spatial_upscaler_path
        print(f"[DEBUG] Upscaler em: {spatial_upscaler_path}")

        print("[DEBUG] Construindo pipeline...")
        pipeline = create_ltx_video_pipeline(
            ckpt_path=self.config["checkpoint_path"],
            precision=self.config["precision"],
            text_encoder_model_name_or_path=self.config["text_encoder_model_name_or_path"],
            sampler=self.config["sampler"],
            device="cpu",
            enhance_prompt=False,
            prompt_enhancer_image_caption_model_name_or_path=self.config["prompt_enhancer_image_caption_model_name_or_path"],
            prompt_enhancer_llm_model_name_or_path=self.config["prompt_enhancer_llm_model_name_or_path"],
        )
        print("[DEBUG] Pipeline pronto.")

        latent_upsampler = None
        if self.config.get("spatial_upscaler_model_path"):
            print("[DEBUG] Construindo latent_upsampler...")
            latent_upsampler = create_latent_upsampler(self.config["spatial_upscaler_model_path"], device="cpu")
            print("[DEBUG] Upsampler pronto.")
        print(f"[DEBUG] _load_models() tempo total={time.perf_counter()-t0:.3f}s")
        return pipeline, latent_upsampler

    def _promote_fp8_weights_to_bf16(self, module):
        if not isinstance(module, torch.nn.Module):
            print("[DEBUG] Promoção FP8→BF16 ignorada: alvo não é nn.Module.")
            return
        f8 = getattr(torch, "float8_e4m3fn", None)
        if f8 is None:
            print("[DEBUG] torch.float8_e4m3fn indisponível.")
            return
        p_cnt = b_cnt = 0
        for _, p in module.named_parameters(recurse=True):
            try:
                if p.dtype == f8:
                    with torch.no_grad():
                        p.data = p.data.to(torch.bfloat16); p_cnt += 1
            except Exception:
                pass
        for _, b in module.named_buffers(recurse=True):
            try:
                if hasattr(b, "dtype") and b.dtype == f8:
                    b.data = b.data.to(torch.bfloat16); b_cnt += 1
            except Exception:
                pass
        print(f"[DEBUG] FP8→BF16: params_promoted={p_cnt}, buffers_promoted={b_cnt}")

    @torch.no_grad()
    def _upsample_latents_internal(self, latents: torch.Tensor) -> torch.Tensor:
        if not self.latent_upsampler:
            raise ValueError("Latent Upsampler não está carregado.")
        self.latent_upsampler.to(self.device)
        self.pipeline.vae.to(self.device)
        print(f"[DEBUG-UPSAMPLE] Shape de entrada: {tuple(latents.shape)}")
        latents = un_normalize_latents(latents, self.pipeline.vae, vae_per_channel_normalize=True)
        upsampled_latents = self.latent_upsampler(latents)
        upsampled_latents = normalize_latents(upsampled_latents, self.pipeline.vae, vae_per_channel_normalize=True)
        print(f"[DEBUG-UPSAMPLE] Shape de saída: {tuple(upsampled_latents.shape)}")
        return upsampled_latents
        
    def _apply_precision_policy(self):
        prec = str(self.config.get("precision", "")).lower()
        self.runtime_autocast_dtype = torch.float32
        print(f"[DEBUG] Aplicando política de precisão: {prec}")
        if prec == "float8_e4m3fn":
            self.runtime_autocast_dtype = torch.bfloat16
            force_promote = os.getenv("LTXV_FORCE_BF16_ON_FP8", "0") == "1"
            print(f"[DEBUG] FP8 detectado. force_promote={force_promote}")
            if force_promote and hasattr(torch, "float8_e4m3fn"):
                try:
                    self._promote_fp8_weights_to_bf16(self.pipeline)
                except Exception as e:
                    print(f"[DEBUG] Promoção FP8→BF16 na pipeline falhou: {e}")
                try:
                    if self.latent_upsampler:
                        self._promote_fp8_weights_to_bf16(self.latent_upsampler)
                except Exception as e:
                    print(f"[DEBUG] Promoção FP8→BF16 no upsampler falhou: {e}")
        elif prec == "bfloat16":
            self.runtime_autocast_dtype = torch.bfloat16
        elif prec == "mixed_precision":
            self.runtime_autocast_dtype = torch.float16
        else:
            self.runtime_autocast_dtype = torch.float32

    def _prepare_conditioning_tensor(self, filepath, height, width, padding_values):
        print(f"[DEBUG] Carregando condicionamento: {filepath}")
        tensor = load_image_to_tensor_with_resize_and_crop(filepath, height, width)
        tensor = torch.nn.functional.pad(tensor, padding_values)
        out = tensor.to(self.device, dtype=self.runtime_autocast_dtype) if self.device == "cuda" else tensor.to(self.device)
        print(f"[DEBUG] Cond shape={tuple(out.shape)} dtype={out.dtype} device={out.device}")
        return out

    def _dividir_latentes_por_tamanho(self, latents_brutos, num_latente_por_chunk: int, overlap: int = 1):
        sum_latent = latents_brutos.shape[2]
        chunks = []
        if num_latente_por_chunk >= sum_latent:
            return [latents_brutos.clone().detach()] # CORREÇÃO: Retornar uma lista e clonar
        # CORREÇÃO: Lógica de chunking simplificada e corrigida para evitar estouro de índice
        start = 0
        while start < sum_latent:
            end = min(start + num_latente_por_chunk, sum_latent)
            # Para o overlap, pegamos um pouco do chunk anterior, exceto para o primeiro
            overlap_start = max(0, start - overlap)
            
            # O chunk a ser processado vai de `overlap_start` até `end`
            # mas o chunk "real" para junção posterior seria de `start` a `end`
            # A lógica atual já faz um overlap simples, vamos refinar
            effective_end = min(start + num_latente_por_chunk, sum_latent)
            chunk = latents_brutos[:, :, start:effective_end, :, :].clone().detach()
            
            # Adiciona overlap no final se não for o último chunk
            if effective_end < sum_latent:
                overlap_end = min(effective_end + overlap, sum_latent)
                chunk = latents_brutos[:, :, start:overlap_end, :, :].clone().detach()

            print(f"[DEBUG] Chunk: start={start}, end={chunk.shape[2]}, total_latents={sum_latent}")
            chunks.append(chunk)
            
            # Avança para o próximo chunk
            if start + num_latente_por_chunk >= sum_latent:
                break
            start += num_latente_por_chunk
            
        return chunks
   
    def _get_total_frames(self, video_path: str) -> int:
        cmd = [
            "ffprobe", "-v", "error", "-select_streams", "v:0", "-count_frames",
            "-show_entries", "stream=nb_read_frames", "-of", "default=nokey=1:noprint_wrappers=1", video_path
        ]
        result = subprocess.run(cmd, capture_output=True, text=True, check=True)
        return int(result.stdout.strip())    

    def _gerar_lista_com_transicoes(self, pasta: str, video_paths: list[str], crossfade_frames: int = 8) -> list[str]:
        # Esta função parece complexa e propensa a erros com nomes de arquivo.
        # Por segurança, mantendo a lógica original, mas corrigindo possíveis bugs de `shell=True`
        # e garantindo que os arquivos existam.
        if len(video_paths) <= 1:
            return video_paths # Não há o que fazer
            
        nova_lista_intermediaria = []
        # Primeiro, cria todos os vídeos podados
        videos_podados = []
        for i, base in enumerate(video_paths):
            video_podado = os.path.join(pasta, f"podado_{i}.mp4")
            total_frames = self._get_total_frames(base)
            
            start_frame = crossfade_frames if i > 0 else 0
            end_frame = total_frames - crossfade_frames if i < len(video_paths) - 1 else total_frames
            
            # Pular poda se não houver frames suficientes
            if start_frame >= end_frame:
                continue

            cmd = [
                'ffmpeg', '-y', '-hide_banner', '-loglevel', 'error', '-i', base,
                '-vf', f'trim=start_frame={start_frame}:end_frame={end_frame},setpts=PTS-STARTPTS',
                '-an', video_podado
            ]
            subprocess.run(cmd, check=True)
            videos_podados.append(video_podado)

        # Agora, cria as transições e monta a lista final
        lista_final = [videos_podados[0]]
        for i in range(len(video_paths) - 1):
            video_anterior = video_paths[i]
            video_seguinte = video_paths[i+1]
            
            # Extrai fade_fim do anterior
            fade_fim_path = os.path.join(pasta, f"fade_fim_{i}.mp4")
            total_frames_anterior = self._get_total_frames(video_anterior)
            cmd_fim = [
                'ffmpeg', '-y', '-hide_banner', '-loglevel', 'error', '-i', video_anterior,
                '-vf', f'trim=start_frame={total_frames_anterior - crossfade_frames},setpts=PTS-STARTPTS',
                '-an', fade_fim_path
            ]
            subprocess.run(cmd_fim, check=True)

            # Extrai fade_ini do seguinte
            fade_ini_path = os.path.join(pasta, f"fade_ini_{i+1}.mp4")
            cmd_ini = [
                'ffmpeg', '-y', '-hide_banner', '-loglevel', 'error', '-i', video_seguinte,
                '-vf', f'trim=end_frame={crossfade_frames},setpts=PTS-STARTPTS', '-an', fade_ini_path
            ]
            subprocess.run(cmd_ini, check=True)

            # Cria a transição
            transicao_path = os.path.join(pasta, f"transicao_{i}_{i+1}.mp4")
            cmd_blend = [
                'ffmpeg', '-y', '-hide_banner', '-loglevel', 'error',
                '-i', fade_fim_path, '-i', fade_ini_path,
                '-filter_complex', f'[0:v][1:v]blend=all_expr=\'A*(1-T/{crossfade_frames})+B*(T/{crossfade_frames})\',format=yuv420p',
                '-frames:v', str(crossfade_frames), transicao_path
            ]
            subprocess.run(cmd_blend, check=True)

            lista_final.append(transicao_path)
            lista_final.append(videos_podados[i+1])
            
        return lista_final
        
    def _concat_mp4s_no_reencode(self, mp4_list: List[str], out_path: str):
        if not mp4_list:
            raise ValueError("A lista de MP4s para concatenar está vazia.")
        # Se houver apenas um vídeo, apenas o copie/mova
        if len(mp4_list) == 1:
            shutil.move(mp4_list[0], out_path)
            print(f"[DEBUG] Apenas um vídeo, movido para: {out_path}")
            return
            
        with tempfile.NamedTemporaryFile("w", delete=False, suffix=".txt") as f:
            for mp4 in mp4_list:
                f.write(f"file '{os.path.abspath(mp4)}'\n")
            list_path = f.name
    
        cmd = f"ffmpeg -y -f concat -safe 0 -i {list_path} -c copy {out_path}"
        print(f"[DEBUG] Concat: {cmd}")
    
        try:
            subprocess.check_call(shlex.split(cmd))
        finally:
            try:
                os.remove(list_path)
            except Exception:
                pass   
   
    def generate(
        self,
        prompt,
        negative_prompt,
        mode="text-to-video",
        start_image_filepath=None,
        middle_image_filepath=None,
        middle_frame_number=None,
        middle_image_weight=1.0,
        end_image_filepath=None,
        end_image_weight=1.0,
        input_video_filepath=None,
        height=512,
        width=704,
        duration=2.0,
        frames_to_use=9, # Parâmetro não utilizado, mas mantido por consistência
        seed=42,
        randomize_seed=True,
        guidance_scale=3.0,
        improve_texture=True,
        progress_callback=None,
        external_decode=True, # Parâmetro não utilizado, mas mantido
    ):
        t_all = time.perf_counter()
        print(f"[DEBUG] generate() begin mode={mode} improve_texture={improve_texture}")
        if self.device == "cuda":
            torch.cuda.empty_cache(); torch.cuda.reset_peak_memory_stats()
        self._log_gpu_memory("Início da Geração")

        if mode == "image-to-video" and not start_image_filepath:
            raise ValueError("A imagem de início é obrigatória para o modo image-to-video")
        used_seed = random.randint(0, 2**32 - 1) if randomize_seed else int(seed)
        seed_everething(used_seed); print(f"[DEBUG] Seed usado: {used_seed}")
        FPS = 24.0; MAX_NUM_FRAMES = 2570
        target_frames_rounded = round(duration * FPS)
        n_val = round((float(target_frames_rounded) - 1.0) / 8.0)
        actual_num_frames = max(9, min(MAX_NUM_FRAMES, int(n_val * 8 + 1)))
        height_padded = ((height - 1) // 8 + 1) * 8
        width_padded = ((width - 1) // 8 + 1) * 8
        padding_values = calculate_padding(height, width, height_padded, width_padded)
        generator = torch.Generator(device=self.device).manual_seed(used_seed)
        
        conditioning_items = []
        if mode == "image-to-video":
            start_tensor = self._prepare_conditioning_tensor(start_image_filepath, height, width, padding_values)
            conditioning_items.append(ConditioningItem(start_tensor, 0, 1.0))
            if middle_image_filepath and middle_frame_number is not None:
                middle_tensor = self._prepare_conditioning_tensor(middle_image_filepath, height, width, padding_values)
                safe_middle_frame = max(0, min(int(middle_frame_number), actual_num_frames - 1))
                conditioning_items.append(ConditioningItem(middle_tensor, safe_middle_frame, float(middle_image_weight)))
            if end_image_filepath:
                end_tensor = self._prepare_conditioning_tensor(end_image_filepath, height, width, padding_values)
                last_frame_index = actual_num_frames - 1
                conditioning_items.append(ConditioningItem(end_tensor, last_frame_index, float(end_image_weight)))
            print(f"[DEBUG] Conditioning items: {len(conditioning_items)}")

        call_kwargs = {
            "prompt": prompt, "negative_prompt": negative_prompt, "height": height_padded, "width": width_padded,
            "num_frames": actual_num_frames, "frame_rate": int(FPS), "generator": generator, "output_type": "latent",
            "conditioning_items": conditioning_items if conditioning_items else None, "media_items": None,
            "decode_timestep": self.config["decode_timestep"], "decode_noise_scale": self.config["decode_noise_scale"],
            "stochastic_sampling": self.config["stochastic_sampling"], "image_cond_noise_scale": 0.01, "is_video": True,
            "vae_per_channel_normalize": True, "mixed_precision": (self.config["precision"] == "mixed_precision"),
            "offload_to_cpu": False, "enhance_prompt": False, "skip_layer_strategy": SkipLayerStrategy.AttentionValues,
        }
        
        # CORREÇÃO: Inicialização de listas
        latents_list = []
        temp_dir = tempfile.mkdtemp(prefix="ltxv_"); self._register_tmp_dir(temp_dir)
        results_dir = "/app/output"; os.makedirs(results_dir, exist_ok=True)
            
        try:
            if improve_texture:
                ctx = torch.autocast(device_type="cuda", dtype=self.runtime_autocast_dtype) if self.device == "cuda" else contextlib.nullcontext()
                with ctx:
                    if not self.latent_upsampler:
                        raise ValueError("Upscaler espacial não carregado, mas 'improve_texture' está ativo.")
                    
                    print("\n--- INICIANDO ETAPA 1: GERAÇÃO BASE (FIRST PASS) ---")
                    t_pass1 = time.perf_counter()
                    first_pass_config = self.config.get("first_pass", {}).copy()
                    first_pass_config.pop("num_inference_steps", None)
                    downscale_factor = self.config.get("downscale_factor", 0.6666666)
                    vae_scale_factor = self.pipeline.vae_scale_factor
                    x_width = int(width_padded * downscale_factor)
                    downscaled_width = x_width - (x_width % vae_scale_factor)
                    x_height = int(height_padded * downscale_factor)
                    downscaled_height = x_height - (x_height % vae_scale_factor)
                    print(f"[DEBUG] First Pass Dims: Original Pad ({width_padded}x{height_padded}) -> Downscaled ({downscaled_width}x{downscaled_height})")
                    
                    first_pass_kwargs = call_kwargs.copy()
                    first_pass_kwargs.update({
                        "output_type": "latent", "width": downscaled_width, "height": downscaled_height,
                        "guidance_scale": float(guidance_scale), **first_pass_config
                    })
                    
                    print(f"[DEBUG] First Pass: Gerando em {downscaled_width}x{downscaled_height}...")
                    # CORREÇÃO: Usar self.pipeline, não a variável deletada 'pipeline'
                    latents = self.pipeline(**first_pass_kwargs).images
                    log_tensor_info(latents, "Latentes Base (First Pass)")
                    print(f"[DEBUG] First Pass concluída em {time.perf_counter() - t_pass1:.2f}s")
                    
                with ctx:
                    print("\n--- INICIANDO ETAPA 2: UPSCALE DOS LATENTES ---")
                    t_upscale = time.perf_counter()
                    upsampled_latents = self._upsample_latents_internal(latents)
                    upsampled_latents = adain_filter_latent(latents=upsampled_latents, reference_latents=latents)
                    print(f"[DEBUG] Upscale de Latentes concluído em {time.perf_counter() - t_upscale:.2f}s")
                    
                    # CORREÇÃO: Manter latentes originais para AdaIN e passar latentes com upscale para o second pass
                    reference_latents_cpu = latents.detach().to("cpu", non_blocking=True)
                    latents_to_refine = upsampled_latents
                    del upsampled_latents; del latents; gc.collect(); torch.cuda.empty_cache()
                
                # CORREÇÃO: Lógica de chunking para o second pass
                latents_parts = self._dividir_latentes_por_tamanho(latents_to_refine, 32, 8) # Exemplo: chunks de 32 frames com 8 de overlap
                del latents_to_refine

                with ctx:
                    for i, latents_chunk in enumerate(latents_parts):
                        print(f"\n--- INICIANDO ETAPA 3.{i+1}: REFINAMENTO DE TEXTURA (SECOND PASS) ---")
                        # CORREÇÃO: AdaIN precisa de latents de referência com mesmo H/W, o que não é o caso aqui.
                        # Vamos aplicar AdaIN com o próprio chunk para normalização, ou pular. Pulando por simplicidade.
                        
                        second_pass_config = self.config.get("second_pass", {}).copy()
                        second_pass_config.pop("num_inference_steps", None)
                        
                        # O tamanho do second pass deve ser o tamanho do latente de entrada (após upscale)
                        second_pass_height, second_pass_width = latents_chunk.shape[3] * 8, latents_chunk.shape[4] * 8

                        print(f"[DEBUG] Second Pass Dims: Target ({second_pass_width}x{second_pass_height})")
                        t_pass2 = time.perf_counter()
                        second_pass_kwargs = call_kwargs.copy()
                        second_pass_kwargs.update({
                           "output_type": "latent", "width": second_pass_width, "height": second_pass_height,
                           "latents": latents_chunk.to(self.device), # Mover chunk para GPU
                           "guidance_scale": float(guidance_scale),
                           "num_frames": latents_chunk.shape[2], # Usar o número de frames do chunk
                           **second_pass_config
                        })
                        print(f"[DEBUG] Second Pass: Refinando chunk {i+1}/{len(latents_parts)}...")
                        final_latents = self.pipeline(**second_pass_kwargs).images
                        log_tensor_info(final_latents, "Latentes Finais (Pós-Second Pass)")
                        print(f"[DEBUG] Second part Pass concluída em {time.perf_counter() - t_pass2:.2f}s")
                        latents_cpu = final_latents.detach().to("cpu", non_blocking=True)
                        latents_list.append(latents_cpu)
                        del final_latents; del latents_chunk; gc.collect(); torch.cuda.empty_cache()
            else: 
                ctx = torch.autocast(device_type="cuda", dtype=self.runtime_autocast_dtype) if self.device == "cuda" else contextlib.nullcontext()
                with ctx:    
                    print("\n--- INICIANDO GERAÇÃO DE ETAPA ÚNICA ---")
                    t_single = time.perf_counter()
                    single_pass_call_kwargs = call_kwargs.copy()
                    # CORREÇÃO: `pipeline_instance` não existe, usar `self.pipeline`.
                    latents_single_pass = self.pipeline(**single_pass_call_kwargs).images
                    log_tensor_info(latents_single_pass, "Latentes Finais (Etapa Única)")
                    print(f"[DEBUG] Etapa única concluída em {time.perf_counter() - t_single:.2f}s")
                    latents_cpu = latents_single_pass.detach().to("cpu", non_blocking=True)
                    latents_list.append(latents_cpu) # CORREÇÃO: aqui deve ser latents_cpu, não latents_single_pass
                    del latents_single_pass; gc.collect(); torch.cuda.empty_cache()

            # --- ETAPA FINAL: DECODIFICAÇÃO E CODIFICAÇÃO MP4 ---
            print("\n--- INICIANDO ETAPA FINAL: DECODIFICAÇÃO E MONTAGEM ---")
            partes_mp4 = []
            for i, latents in enumerate(latents_list):
                print(f"[DEBUG] Decodificando partição {i+1}/{len(latents_list)}: {tuple(latents.shape)}")
                output_video_path = os.path.join(temp_dir, f"output_{used_seed}_{i}.mp4")
                
                pixel_tensor = vae_manager_singleton.decode(
                    latents.to(self.device, non_blocking=True),
                    decode_timestep=float(self.config.get("decode_timestep", 0.05))
                )
                log_tensor_info(pixel_tensor, "Pixel tensor (VAE saída)")

                video_encode_tool_singleton.save_video_from_tensor(
                   pixel_tensor, output_video_path, fps=call_kwargs["frame_rate"], progress_callback=progress_callback
                )
                partes_mp4.append(output_video_path)
                del pixel_tensor; del latents; gc.collect(); torch.cuda.empty_cache()

            final_vid = os.path.join(results_dir, f"final_video_{used_seed}.mp4")
            if len(partes_mp4) > 1:
                # A função _gerar_lista_com_transicoes é complexa, usando uma concatenação direta como fallback robusto.
                # Para usar a transição, a lógica de overlap na divisão de latentes precisa ser perfeita.
                print("[DEBUG] Múltiplas partes geradas, concatenando...")
                partes_mp4_fade = self._gerar_lista_com_transicoes(pasta=temp_dir, video_paths=partes_mp4, crossfade_frames=8)
                self._concat_mp4s_no_reencode(partes_mp4_fade, final_vid)
            else:
                shutil.move(partes_mp4[0], final_vid)

            self._log_gpu_memory("Fim da Geração")
            return final_vid, used_seed
            
        except Exception as e:
            print("[DEBUG] EXCEÇÃO NA GERAÇÃO:")
            print("".join(traceback.format_exception(type(e), e, e.__traceback__)))
            raise
        
        finally:
            gc.collect()
            if torch.cuda.is_available():
                torch.cuda.empty_cache()
                torch.cuda.ipc_collect()
            self.finalize(keep_paths=[]) # O resultado final já foi movido

print("Criando instância do VideoService. O carregamento do modelo começará agora...")
video_generation_service = VideoService()