"""QThread tabanlı video üretim pipeline'ı. GUI'den başlatılır, her sahneyi LTX API ile üretir, sinyallerle bildirir. """ from __future__ import annotations from pathlib import Path from typing import Optional, List import shutil import time from PySide6.QtCore import QThread, Signal from model import Proje, Sahne, en_yakin_izinli_sure from .ltx_client import LTXClient from .face_swap import yuz_degistir, facefusion_kurulu_mu from . import ffmpeg_yardimci as ffm class VideoIslemHatti(QThread): """Tek bir projenin tüm (veya seçili) sahnelerini üretir.""" log = Signal(str, str) # (seviye, mesaj): bilgi/uyari/hata/basari ilerleme = Signal(int, int, str) # (mevcut, toplam, aciklama) sahne_basladi = Signal(str) # sahne_id sahne_guncellendi = Signal(str) # sahne_id (durum değişti) sahne_tamamlandi = Signal(str, str) # (sahne_id, video_yolu) hata = Signal(str) tamamlandi = Signal(str) # nihai video yolu def __init__(self, proje: Proje, client: LTXClient, calisma_dizini: Path, sadece_sahne_idleri: Optional[List[str]] = None, otomatik_birlestir: bool = True, ses_ekle: bool = True): super().__init__() self.proje = proje self.client = client self.calisma_dizini = Path(calisma_dizini) self.sadece = set(sadece_sahne_idleri or []) self.otomatik_birlestir = otomatik_birlestir self.ses_ekle = ses_ekle self._iptal = False # ------------------------------------------------------------- yardım def iptal_et(self) -> None: self._iptal = True self.log.emit("uyari", "İptal sinyali gönderildi, mevcut segmentten sonra duracak...") def _ilerleme(self, mevcut: int, toplam: int, aciklama: str) -> None: self.ilerleme.emit(mevcut, toplam, aciklama) def _bilgi(self, mesaj: str) -> None: self.log.emit("bilgi", mesaj) def _uyari(self, mesaj: str) -> None: self.log.emit("uyari", mesaj) def _hata(self, mesaj: str) -> None: self.log.emit("hata", mesaj) def _basari(self, mesaj: str) -> None: self.log.emit("basari", mesaj) # ---------------------------------------------------------------- run def run(self) -> None: try: self._calis() except Exception as e: self._hata(f"Beklenmeyen hata: {e}") self.hata.emit(str(e)) def _calis(self) -> None: if not self.proje.sahneler: self._hata("Üretilecek sahne yok.") self.hata.emit("Sahne yok") return self.calisma_dizini.mkdir(parents=True, exist_ok=True) sahneler = [s for s in self.proje.sahneler if not self.sadece or s.id in self.sadece] toplam = len(sahneler) self._bilgi(f"Pipeline başladı: {toplam} sahne üretilecek") self._ilerleme(0, toplam, "Hazırlanıyor") # LTX'e bağlan if not self.client.baglan(): self._hata("LTX Desktop backend bulunamadı. Uygulamayı açın.") self.hata.emit("LTX bulunamadı") return self._basari(f"LTX bağlandı: {self.client.api_url}") face_swap_aktif_global = ( self.proje.ayarlar.video.face_swap_aktif and self.proje.ayarlar.video.face_swap_kaynak and facefusion_kurulu_mu() ) # İlk sahnenin giriş görseli boşsa projenin kaynak_gorseli ata if sahneler and not sahneler[0].giris_gorsel: sahneler[0].giris_gorsel = self.proje.kaynak_gorsel onceki_video: Optional[str] = None for i, s in enumerate(sahneler): if self._iptal: self._uyari("İptal edildi.") return self._ilerleme(i, toplam, f"Sahne {i+1}/{toplam}: {s.prompt[:40]}...") self.sahne_basladi.emit(s.id) s.durum = "uretiliyor" s.baslama_zamani = time.time() self.sahne_guncellendi.emit(s.id) # Giriş görselini ayarla: kendi giris_gorsel'i varsa onu, # yoksa önceki videonun son karesi giris = s.giris_gorsel if not giris and onceki_video: son_kare = self.calisma_dizini / f"son_kare_{s.sira:03d}.jpg" if ffm.son_kare_cikar(onceki_video, son_kare): giris = str(son_kare) s.giris_gorsel = giris if not giris or not Path(giris).exists(): s.durum = "hata" s.hata_mesaji = "Giriş görseli yok" self._hata(f"Sahne {i+1}: giriş görseli yok") self.sahne_guncellendi.emit(s.id) continue # Face swap (giriş görseline) face_kaynak = s.face_swap_kaynak or self.proje.ayarlar.video.face_swap_kaynak if s.face_swap_aktif and face_kaynak and facefusion_kurulu_mu(): swap_cikis = self.calisma_dizini / f"swap_giris_{s.sira:03d}.jpg" if yuz_degistir(face_kaynak, giris, swap_cikis): giris = str(swap_cikis) self._bilgi(f"Sahne {i+1}: giriş görseli face-swap'landı") # Süreyi çözünürlüğe göre snap'le cozunurluk = self.proje.ayarlar.video.cozunurluk sure = en_yakin_izinli_sure(s.sure, cozunurluk) try: self._bilgi(f"Sahne {i+1}: LTX'e istek gönderiliyor ({sure}s @ {cozunurluk})") def _ilerleme(faz: str, yuzde: float, _i=i, _t=toplam, _p=s.prompt): self._ilerleme(_i, _t, f"Sahne {_i+1}/{_t} [{faz}] %{yuzde:.0f}") self.log.emit("bilgi", f"LTX: [{faz}] %{yuzde:.0f}") video_yolu = self.client.video_uret( prompt=s.prompt or self.proje.ayarlar.video.global_prompt, gorsel=giris, ses=s.ses_yolu, cozunurluk=cozunurluk, oran=self.proje.ayarlar.video.oran, sure=sure, fps=self.proje.ayarlar.video.fps, guidance_scale=s.guidance_scale, steps=s.steps, negatif=s.negatif_prompt or self.proje.ayarlar.video.global_negatif, model=self.proje.ayarlar.video.model, ilerleme_cb=_ilerleme, ) except Exception as e: s.durum = "hata" s.hata_mesaji = str(e) self._hata(f"Sahne {i+1} hata: {e}") self.sahne_guncellendi.emit(s.id) continue if not video_yolu or not Path(video_yolu).exists(): s.durum = "hata" s.hata_mesaji = "Video üretilemedi" self._hata(f"Sahne {i+1}: video üretilemedi") self.sahne_guncellendi.emit(s.id) continue # Segmenti çalışma dizinine kopyala hedef = self.calisma_dizini / f"segment_{s.sira:03d}.mp4" shutil.copy2(video_yolu, hedef) s.cikis_video = str(hedef) # Son kareyi hazırla (sonraki sahneye giriş olacak) son_kare = self.calisma_dizini / f"son_kare_{s.sira:03d}.jpg" if ffm.son_kare_cikar(hedef, son_kare): s.son_kare = str(son_kare) onceki_video = str(hedef) s.durum = "tamamlandi" s.bitis_zamani = time.time() self.sahne_guncellendi.emit(s.id) self.sahne_tamamlandi.emit(s.id, s.cikis_video) self._basari(f"Sahne {i+1}/{toplam} tamamlandı ({s.bitis_zamani - s.baslama_zamani:.1f}s)") self._ilerleme(toplam, toplam, "Sahneler bitti") # Birleştirme & ses ekleme if self.otomatik_birlestir: basarilar = [s for s in self.proje.sahneler if s.durum == "tamamlandi" and s.cikis_video] if not basarilar: self._hata("Birleştirilecek başarılı sahne yok.") self.hata.emit("Sahne yok") return ham = self.calisma_dizini / "birlesmis_ham.mp4" if len(basarilar) == 1: shutil.copy2(basarilar[0].cikis_video, ham) else: self._bilgi(f"{len(basarilar)} segment birleştiriliyor...") if not ffm.videolari_birlestir([s.cikis_video for s in basarilar], ham): self._hata("FFmpeg birleştirme başarısız.") self.hata.emit("Birleştirme hatası") return # Ses ekle cikis = self.calisma_dizini / f"{self.proje.ad}_nihai.mp4" if self.ses_ekle and self.proje.kaynak_ses and Path(self.proje.kaynak_ses).exists(): self._bilgi("Orijinal ses ekleniyor...") if ffm.videoya_ses_ekle(ham, self.proje.kaynak_ses, cikis): self.proje.cikis_video = str(cikis) self._basari(f"Nihai video: {cikis}") self.tamamlandi.emit(str(cikis)) return else: self._uyari("Ses eklenemedi, sessiz video kullanılıyor.") shutil.copy2(ham, cikis) self.proje.cikis_video = str(cikis) self._basari(f"Nihai video: {cikis}") self.tamamlandi.emit(str(cikis)) else: self._basari("Sahneler üretildi, birleştirme kullanıcıdan bekleniyor.") self.tamamlandi.emit("")