"""FikoYT Görsel Üretim Borusu (Pipeline) ==================================================== SDXL + IP-Adapter (yüz/ürün referansı) + InsightFace face-swap. Boru tek kez başlatılır, modeller lazy-load edilir, GPU üzerinde tek üretim yapılır (kuyruk server.py tarafında). VRAM tasarrufu için model offload kapatılabilir, ancak <=12GB için önerilir. """ from __future__ import annotations import io import os import gc import json import time import math import random import threading import traceback from dataclasses import dataclass, field, asdict from pathlib import Path from typing import Optional, List, Callable, Dict, Any import torch import numpy as np from PIL import Image, ImageOps # Lazy import — sadece üretim sırasında load _diffusers_loaded = False _insightface_loaded = False # ===================== Model Kayıt Defteri ===================== # Hugging Face repo veya yerel klasör. Anahtar = UI'da gösterilen isim. SDXL_MODELLERI: Dict[str, Dict[str, Any]] = { # ---------- Foto-gerçekçi ---------- "RealVisXL v4.0": { "repo": "SG161222/RealVisXL_V4.0", "tip": "sdxl", "varsayilan_steps": 30, "varsayilan_cfg": 6.5, }, "RealVisXL v5.0": { "repo": "SG161222/RealVisXL_V5.0", "tip": "sdxl", "varsayilan_steps": 30, "varsayilan_cfg": 6.5, }, "Juggernaut XL v9": { "repo": "RunDiffusion/Juggernaut-XL-v9", "tip": "sdxl", "varsayilan_steps": 30, "varsayilan_cfg": 6.0, }, "Juggernaut XL Lightning": { "repo": "RunDiffusion/Juggernaut-XL-Lightning", "tip": "sdxl", "varsayilan_steps": 6, "varsayilan_cfg": 1.5, }, "epiCRealism XL v8": { "repo": "misri/epicrealismXL_v8Kiss", "tip": "sdxl", "varsayilan_steps": 30, "varsayilan_cfg": 5.0, }, "CyberRealistic XL": { "repo": "misri/cyberrealisticXL_v3", "tip": "sdxl", "varsayilan_steps": 30, "varsayilan_cfg": 6.0, }, # ---------- Genel amaçlı / sanat ---------- "SDXL Base 1.0": { "repo": "stabilityai/stable-diffusion-xl-base-1.0", "tip": "sdxl", "varsayilan_steps": 30, "varsayilan_cfg": 7.0, }, "DreamShaper XL Turbo": { "repo": "Lykon/dreamshaper-xl-v2-turbo", "tip": "sdxl-turbo", "varsayilan_steps": 8, "varsayilan_cfg": 2.0, }, "Copax TimelessXL v13": { "repo": "misri/copaxTimelessxlSDXL1_v13", "tip": "sdxl", "varsayilan_steps": 30, "varsayilan_cfg": 6.5, }, # ---------- Hızlı (Lightning / Turbo) ---------- "SDXL Lightning 4-step": { "repo": "ByteDance/SDXL-Lightning", "tip": "sdxl-lightning", "varsayilan_steps": 4, "varsayilan_cfg": 1.0, "lora_dosya": "sdxl_lightning_4step_lora.safetensors", "base_repo": "stabilityai/stable-diffusion-xl-base-1.0", }, "SDXL Lightning 8-step": { "repo": "ByteDance/SDXL-Lightning", "tip": "sdxl-lightning", "varsayilan_steps": 8, "varsayilan_cfg": 1.0, "lora_dosya": "sdxl_lightning_8step_lora.safetensors", "base_repo": "stabilityai/stable-diffusion-xl-base-1.0", }, "SDXL Turbo": { "repo": "stabilityai/sdxl-turbo", "tip": "sdxl-turbo", "varsayilan_steps": 4, "varsayilan_cfg": 0.0, }, # ---------- Anime / sanatsal ---------- "Animagine XL 3.1": { "repo": "cagliostrolab/animagine-xl-3.1", "tip": "sdxl", "varsayilan_steps": 28, "varsayilan_cfg": 7.0, }, "Pony Diffusion XL v6": { "repo": "AstraliteHeart/pony-diffusion-v6", "tip": "sdxl", "varsayilan_steps": 28, "varsayilan_cfg": 6.0, }, } VARSAYILAN_MODEL = "SDXL Base 1.0" IP_ADAPTER_REPO = "h94/IP-Adapter" IP_ADAPTER_AGIRLIK_DOSYA = "ip-adapter-plus_sdxl_vit-h.safetensors" IP_ADAPTER_ALT_KLASOR = "sdxl_models" IP_ADAPTER_IMAGE_ENCODER = "h94/IP-Adapter" IP_ADAPTER_IMAGE_ENCODER_SUBFOLDER = "models/image_encoder" # IP-Adapter FaceID — sadece kimlik (InsightFace embedding), kompozisyon prompt'tan IP_ADAPTER_FACEID_REPO = "h94/IP-Adapter-FaceID" IP_ADAPTER_FACEID_WEIGHT = "ip-adapter-faceid_sdxl.bin" IP_ADAPTER_FACEID_LORA = "ip-adapter-faceid_sdxl_lora.safetensors" # Face-swap modeli (InsightFace) INSWAPPER_MODELI = "inswapper_128.onnx" # ===================== Yardımcılar ===================== def _device() -> str: return "cuda" if torch.cuda.is_available() else "cpu" def _dtype(): return torch.float16 if torch.cuda.is_available() else torch.float32 def _pillow_resmi_yukle(yol: str, max_kenar: int = 1024) -> Image.Image: img = Image.open(yol).convert("RGB") img = ImageOps.exif_transpose(img) w, h = img.size if max(w, h) > max_kenar: oran = max_kenar / max(w, h) img = img.resize((int(w * oran), int(h * oran)), Image.LANCZOS) return img def _yuzu_kirp(img: Image.Image, face_app, marj: float = 1.6) -> Image.Image: """InsightFace ile yüzü tespit et, marj ile kare kırp. Yüz bulunamazsa orijinal döner.""" try: np_bgr = np.array(img)[:, :, ::-1].copy() yuzler = face_app.get(np_bgr) if not yuzler: return img # En büyük yüzü seç yuzler.sort( key=lambda y: (y.bbox[2] - y.bbox[0]) * (y.bbox[3] - y.bbox[1]), reverse=True, ) x1, y1, x2, y2 = yuzler[0].bbox.astype(int) cx, cy = (x1 + x2) / 2, (y1 + y2) / 2 w, h = x2 - x1, y2 - y1 boyut = max(w, h) * marj x1n = int(max(0, cx - boyut / 2)) y1n = int(max(0, cy - boyut / 2)) x2n = int(min(img.width, cx + boyut / 2)) y2n = int(min(img.height, cy + boyut / 2)) return img.crop((x1n, y1n, x2n, y2n)) except Exception as e: print(f"[Boru] Yüz kırpma hatası: {e}") return img def _pillow_resmi_kaydet(img: Image.Image, hedef: str, kalite: int = 95): Path(hedef).parent.mkdir(parents=True, exist_ok=True) if hedef.lower().endswith((".jpg", ".jpeg")): img.save(hedef, quality=kalite, optimize=True) else: img.save(hedef, optimize=True) # ===================== Üretim İstek/Sonuç ===================== @dataclass class UretimIstegi: prompt: str negative_prompt: str = "" model_ismi: str = VARSAYILAN_MODEL genislik: int = 1024 yukseklik: int = 1024 steps: int = 30 cfg: float = 7.0 seed: int = -1 batch: int = 1 scheduler: str = "auto" # auto | euler | dpm++ | unipc # Referans görseller (IP-Adapter) referans_yuz_yolu: Optional[str] = None # IP-Adapter face mode yuz_referans_olcek: float = 0.7 yuz_modu: str = "faceid" # "faceid" (sadece kimlik) | "plus" (kimlik+stil) referans_urun_yolu: Optional[str] = None # IP-Adapter style mode urun_referans_olcek: float = 0.6 # Çoklu referans — saç stili + kıyafet referans_sac_yolu: Optional[str] = None sac_referans_olcek: float = 0.6 referans_elbise_yolu: Optional[str] = None elbise_referans_olcek: float = 0.7 referans_arkaplan_yolu: Optional[str] = None arkaplan_referans_olcek: float = 0.4 # Face swap (post-process, InsightFace) face_swap_kaynak_yuz: Optional[str] = None # bu yüzü çıktıya yapıştır face_swap_etkin: bool = False # img2img — verilen görsel üzerinde değişiklik kaynak_resim_yolu: Optional[str] = None degisiklik_gucu: float = 0.7 # 0.0=aynı kalır, 1.0=yeniden çiz # Mod: "text2img" | "img2img" | "sadece_yuz_degis" | "kiyafet_degis" | "sac_degis" # | "buyut" (upscale 2x) | "genislet" (outpainting) gorev_tipi: str = "text2img" # Outpainting parametreleri genislet_yon: str = "all" # "all" | "left" | "right" | "top" | "bottom" genislet_oran: float = 1.5 # 1.0=aynı, 2.0=iki kat alan # Multi-face swap: birden çok kaynak yüz, tespit edilen N yüze sırayla atanır coklu_yuz_kaynaklari: Optional[List[str]] = None # Akış: sıralı adım listesi (her adım önceki çıktıyı kaynak olarak alır) # Her adım dict: { tip, prompt, negative, gucu, ref_yuz, ref_sac, ref_elbise, ... } adimlar: Optional[List[Dict[str, Any]]] = None # Aktif LoRA listesi: [{ "yol": "...", "olcek": 0.8, "tetikleyici": "zwx_xxx" }] aktif_loralar: Optional[List[Dict[str, Any]]] = None # ADetailer — yüz restorasyonu (üretim sonrası) adetailer_etkin: bool = False adetailer_denoise: float = 0.4 # 0.3-0.5 önerilir; orijinali bozma # Upscaler — çıktıyı büyüt (1=kapalı, 2=2x, 4=4x) upscale_faktor: int = 1 upscale_model: str = "auto" # "auto" | "realesrgan" | "lanczos" # Çıkış cikti_klasoru: str = "outputs" cikti_format: str = "png" # png | jpg | webp @dataclass class UretimSonucu: yollar: List[str] = field(default_factory=list) seed: int = 0 sure_sn: float = 0.0 # ===================== Boru ===================== class GorselBorusu: def __init__(self): self._kilit = threading.RLock() self._aktif_model: Optional[str] = None self._pipe = None self._ip_adapter_modu: Optional[str] = None # 'plus' | 'faceid' | None self._plus_slot_sayisi: int = 1 self._aktif_lora_isimleri: List[str] = [] self._aktif_lora_agirliklari: List[float] = [] self._face_app = None # insightface app self._face_swapper = None # inswapper self._cuda = torch.cuda.is_available() @property def _ip_adapter_yuklu(self) -> bool: return self._ip_adapter_modu is not None # ---------- Model yükleme ---------- def _modelin_borusunu_kur(self, model_ismi: str): if self._aktif_model == model_ismi and self._pipe is not None: return global _diffusers_loaded if not _diffusers_loaded: _diffusers_loaded = True from diffusers import ( StableDiffusionXLPipeline, EulerDiscreteScheduler, DPMSolverMultistepScheduler, AutoencoderKL, ) # Eski boruyu temizle if self._pipe is not None: del self._pipe self._pipe = None self._ip_adapter_modu = None self._plus_slot_sayisi = 1 self._aktif_lora_isimleri = [] self._aktif_lora_agirliklari = [] gc.collect() if self._cuda: torch.cuda.empty_cache() cfg = SDXL_MODELLERI[model_ismi] tip = cfg["tip"] repo = cfg["repo"] dtype = _dtype() if tip in ("sdxl", "sdxl-turbo"): pipe = StableDiffusionXLPipeline.from_pretrained( repo, torch_dtype=dtype, use_safetensors=True, variant=("fp16" if dtype == torch.float16 else None) ) elif tip == "sdxl-lightning": from huggingface_hub import hf_hub_download base_repo = cfg["base_repo"] lora = cfg["lora_dosya"] pipe = StableDiffusionXLPipeline.from_pretrained( base_repo, torch_dtype=dtype, use_safetensors=True, variant=("fp16" if dtype == torch.float16 else None) ) ckpt_yolu = hf_hub_download(repo, lora) pipe.load_lora_weights(ckpt_yolu) pipe.fuse_lora() pipe.scheduler = EulerDiscreteScheduler.from_config( pipe.scheduler.config, timestep_spacing="trailing" ) else: raise ValueError(f"Bilinmeyen model tipi: {tip}") # SDXL fp16 VAE NaN bug → fp16-fix VAE yükle (siyah çıktıyı önler) try: fp16_vae = AutoencoderKL.from_pretrained( "madebyollin/sdxl-vae-fp16-fix", torch_dtype=dtype, ) pipe.vae = fp16_vae print("[Boru] fp16-fix VAE yüklendi (siyah çıktı düzeltmesi)") except Exception as e: print(f"[Boru] fp16-fix VAE yüklenemedi: {e}") # Scheduler ön-ayarı if tip == "sdxl": pipe.scheduler = DPMSolverMultistepScheduler.from_config( pipe.scheduler.config, use_karras_sigmas=True ) # VRAM optimizasyonu if self._cuda: try: pipe.enable_model_cpu_offload() except Exception: pipe = pipe.to("cuda") try: pipe.enable_vae_tiling() except Exception: pass try: pipe.enable_xformers_memory_efficient_attention() except Exception: pass # FreeU DEVRE DIŞI — fp16 + low-strength img2img'de FFT NaN üretiyor # (ComplexHalf fp16 yetersiz). Yüksek strength text2img'de fark # ihmal edilebilir, alttaki bug riski daha kötü. # try: # pipe.enable_freeu(s1=0.9, s2=0.2, b1=1.3, b2=1.4) # print("[Boru] FreeU aktif (kalite boost)") # except Exception: # pass self._pipe = pipe self._aktif_model = model_ismi print(f"[Boru] Model yüklendi: {model_ismi}") def _aktif_loralari_uygula(self, loralar: Optional[List[Dict[str, Any]]]): """Verilen LoRA listesini pipe'a yükler (her seferinde yenilenir). loralar: [{ 'yol': '/path/to/lora.safetensors', 'olcek': 0.8 }, ...] Not: FaceID LoRA varsa onu unload etmez — sadece kullanıcı LoRA'larını yönetir. Adapter isimleri 'lora_N' prefix'iyle, FaceID 'faceid_lora'.""" if self._pipe is None: self._aktif_lora_isimleri = [] self._aktif_lora_agirliklari = [] return # Sadece kullanıcı LoRA'larını kaldır (FaceID korunur) try: mevcut = list(getattr(self, "_aktif_lora_isimleri", []) or []) for ad in mevcut: try: self._pipe.delete_adapters(ad) except Exception: pass except Exception: pass self._aktif_lora_isimleri = [] self._aktif_lora_agirliklari = [] if not loralar: return for i, lora in enumerate(loralar): yol = lora.get("yol") olcek = float(lora.get("olcek", 0.8)) if not yol or not Path(yol).exists(): print(f"[Boru] LoRA bulunamadı: {yol}") continue ad = f"lora_{i}" try: self._pipe.load_lora_weights( yol, adapter_name=ad, ) self._aktif_lora_isimleri.append(ad) self._aktif_lora_agirliklari.append(olcek) print(f"[Boru] LoRA yüklendi: {Path(yol).name} × {olcek}") except Exception as e: print(f"[Boru] LoRA yükleme hatası ({yol}): {e}") def _ip_adapter_yenile(self, hedef_mod: Optional[str], plus_slot_sayisi: int = 1): """hedef_mod: 'plus' (IP-Adapter Plus, N slot), 'faceid' (FaceID), None. plus_slot_sayisi: 'plus' modunda kaç slot yüklensin (1-4).""" # Plus için slot sayısı değişirse yeniden yükle if self._ip_adapter_modu == hedef_mod and \ getattr(self, "_plus_slot_sayisi", 1) == plus_slot_sayisi: return if self._pipe is None: return # Önce mevcut adapter'ı kaldır if self._ip_adapter_modu is not None: try: self._pipe.unload_ip_adapter() except Exception: pass # FaceID LoRA varsa sadece o adapter'ı sil (kullanıcı LoRA'ları # kalmalı). delete_adapters yoksa unload_lora_weights fallback. if self._ip_adapter_modu == "faceid": try: self._pipe.delete_adapters("faceid_lora") except Exception: try: self._pipe.unload_lora_weights() # Kullanıcı LoRA isimlerini sıfırla (yeniden yüklenmeli) self._aktif_lora_isimleri = [] self._aktif_lora_agirliklari = [] except Exception: pass self._ip_adapter_modu = None print("[Boru] IP-Adapter kaldırıldı") if hedef_mod == "plus": try: n = max(1, min(4, plus_slot_sayisi)) weights = [IP_ADAPTER_AGIRLIK_DOSYA] * n self._pipe.load_ip_adapter( IP_ADAPTER_REPO, subfolder=IP_ADAPTER_ALT_KLASOR, weight_name=weights if n > 1 else IP_ADAPTER_AGIRLIK_DOSYA, image_encoder_folder="models/image_encoder", ) if self._cuda and hasattr(self._pipe, "image_encoder") \ and self._pipe.image_encoder is not None: self._pipe.image_encoder = self._pipe.image_encoder.to( "cuda", dtype=_dtype() ) self._ip_adapter_modu = "plus" self._plus_slot_sayisi = n print(f"[Boru] IP-Adapter Plus yüklendi ({n} slot)") except Exception as e: print(f"[Boru] IP-Adapter Plus yüklenemedi: {e}") import traceback traceback.print_exc() elif hedef_mod == "faceid": try: from huggingface_hub import hf_hub_download # Eski faceid_lora adapter'ı varsa önce sil # (ADetailer/Outpaint sonrası restore çakışmasını önler) try: self._pipe.delete_adapters("faceid_lora") except Exception: pass # FaceID LoRA önce yüklenmeli lora_yolu = hf_hub_download( IP_ADAPTER_FACEID_REPO, IP_ADAPTER_FACEID_LORA ) self._pipe.load_lora_weights(lora_yolu, adapter_name="faceid_lora") # FaceID weights — image_encoder yok (InsightFace embedding) self._pipe.load_ip_adapter( IP_ADAPTER_FACEID_REPO, subfolder=None, weight_name=IP_ADAPTER_FACEID_WEIGHT, image_encoder_folder=None, ) self._ip_adapter_modu = "faceid" print("[Boru] IP-Adapter FaceID yüklendi (sadece kimlik)") except Exception as e: print(f"[Boru] FaceID yüklenemedi: {e}") import traceback traceback.print_exc() def _yuz_degis_uygula( self, hedef_resim: Image.Image, kaynak_yollari: List[str], ) -> Image.Image: """Hedefteki tüm yüzleri tespit eder ve sırayla kaynak yüzlerle değiştirir. kaynak_yollari listesinde tek eleman varsa hepsine aynı kaynak uygulanır. Liste daha kısaysa son eleman tekrarlanır.""" if not kaynak_yollari: return hedef_resim self._face_app_yukle() self._face_swapper_yukle() if self._face_app is None or self._face_swapper is None: return hedef_resim # Kaynak yüzleri yükle ve embed çıkar kaynak_yuzleri = [] for yol in kaynak_yollari: try: k_img = _pillow_resmi_yukle(yol, 1024) k_np = np.array(k_img)[:, :, ::-1].copy() k_yuzler = self._face_app.get(k_np) if k_yuzler: k_yuzler.sort( key=lambda y: (y.bbox[2] - y.bbox[0]) * (y.bbox[3] - y.bbox[1]), reverse=True, ) kaynak_yuzleri.append(k_yuzler[0]) else: print(f"[Boru] Kaynakta yüz yok: {yol}") except Exception as e: print(f"[Boru] Kaynak yüz yükleme hatası: {e}") if not kaynak_yuzleri: return hedef_resim np_bgr = np.array(hedef_resim)[:, :, ::-1].copy() hedef_yuzler = self._face_app.get(np_bgr) if not hedef_yuzler: return hedef_resim # Hedef yüzleri sol-sağ sırala (kullanıcının yüklediği sıraya eşle) hedef_yuzler.sort(key=lambda y: y.bbox[0]) print(f"[Boru] {len(hedef_yuzler)} yüz tespit edildi, " f"{len(kaynak_yuzleri)} kaynak yüz var") for i, hedef_yuz in enumerate(hedef_yuzler): kaynak_idx = min(i, len(kaynak_yuzleri) - 1) kaynak_yuz = kaynak_yuzleri[kaynak_idx] np_bgr = self._face_swapper.get( np_bgr, hedef_yuz, kaynak_yuz, paste_back=True, ) return Image.fromarray(np_bgr[:, :, ::-1]) def _akisi_calistir( self, istek: UretimIstegi, ilerleme_cb: Optional[Callable[[int, int, str], None]], t0: float, ) -> UretimSonucu: """Sıralı adımları çalıştırır. Her adımın çıktısı bir sonrakine kaynak.""" if not istek.adimlar: raise ValueError("Adım listesi boş") # Çıktı klasörü → akış için alt klasör kok_klasor = Path(istek.cikti_klasoru) kok_klasor.mkdir(parents=True, exist_ok=True) damga = int(time.time()) akis_klasor = kok_klasor / f"akis_{damga}" akis_klasor.mkdir(parents=True, exist_ok=True) # İlk kaynak — istek.kaynak_resim_yolu varsa onu kullan, yoksa text2img son_cikti = istek.kaynak_resim_yolu tum_yollar = [] toplam = len(istek.adimlar) for idx, adim in enumerate(istek.adimlar): tip = adim.get("tip", "img2img") ad_isim = adim.get("aciklama") or tip print(f"[Akış] Adım {idx + 1}/{toplam}: {tip} — {ad_isim}") if ilerleme_cb: ilerleme_cb( idx, toplam, f"Adım {idx + 1}/{toplam}: {ad_isim}", ) # Bu adım için sub-istek üret (ana istekten kopya, alanları override) sub = UretimIstegi( prompt=adim.get("prompt") or istek.prompt or "", negative_prompt=adim.get("negative_prompt") or istek.negative_prompt, model_ismi=istek.model_ismi, genislik=istek.genislik, yukseklik=istek.yukseklik, steps=int(adim.get("steps") or istek.steps or 30), cfg=float(adim.get("cfg") or istek.cfg or 6.0), seed=adim.get("seed", -1), batch=1, gorev_tipi=tip, kaynak_resim_yolu=son_cikti, degisiklik_gucu=adim.get("gucu", 0.55), referans_yuz_yolu=adim.get("ref_yuz"), yuz_referans_olcek=adim.get("yuz_olcek", 0.45), yuz_modu=adim.get("yuz_modu", "faceid"), referans_sac_yolu=adim.get("ref_sac"), sac_referans_olcek=adim.get("sac_olcek", 0.7), referans_elbise_yolu=adim.get("ref_elbise"), elbise_referans_olcek=adim.get("elbise_olcek", 0.75), referans_arkaplan_yolu=adim.get("ref_arkaplan"), arkaplan_referans_olcek=adim.get("arkaplan_olcek", 0.4), face_swap_kaynak_yuz=adim.get("face_swap_kaynak"), face_swap_etkin=bool(adim.get("face_swap_etkin", False) or tip == "sadece_yuz_degis"), coklu_yuz_kaynaklari=adim.get("coklu_yuz_kaynaklari"), cikti_klasoru=str(akis_klasor), cikti_format=istek.cikti_format, ) # Bu adımı tek başına çalıştır (akış flag'ini iç çağrıda sıfırla) sub.adimlar = None try: # Her adım için kilidi tutmaya gerek yok (zaten bizdeyiz) # Doğrudan iç metodları çağıralım if tip == "sadece_yuz_degis": sonuc = self._sadece_yuz_degis(sub, None, time.time()) else: sonuc = self._tek_uret(sub, None) except Exception as e: print(f"[Akış] Adım {idx + 1} hatası: {e}") import traceback traceback.print_exc() continue if not sonuc.yollar: continue # Adım çıktısını yeniden adlandır → adim_N_.png kaynak_dosya = sonuc.yollar[0] ext = Path(kaynak_dosya).suffix yeni_ad = akis_klasor / f"adim_{idx + 1:02d}_{tip}{ext}" try: Path(kaynak_dosya).rename(yeni_ad) son_cikti = str(yeni_ad) tum_yollar.append(str(yeni_ad)) except Exception: tum_yollar.append(kaynak_dosya) son_cikti = kaynak_dosya if ilerleme_cb: ilerleme_cb(toplam, toplam, "Akış tamamlandı") return UretimSonucu( yollar=tum_yollar, seed=0, sure_sn=time.time() - t0, ) def _tek_uret( self, istek: UretimIstegi, ilerleme_cb: Optional[Callable[[int, int, str], None]], ) -> UretimSonucu: """uret()'in akış için iç versiyonu — kilidi tutmaz, akış değil.""" # Geçici hile: adimları None yap, gorev_tipi sadece_yuz_degis değilse # ana akışa gir. uret() içindeki bu kontroller atlanmış olur. # Bu metod uret()'in body'sini içermeli; kod tekrarını önlemek için # uret()'in iç akışını ayrı çıkarmadık — yine uret() çağırıyoruz ama # kilidi atomik olarak yönetmek lazım. Bunun yerine inline çalıştıralım. # Basitlik adına: yeni Threading.RLock kullanılmıyor; aynı thread re-entry # nedeniyle threading.Lock yerine RLock seçilmediyse hata verir. # Çözüm: _kilit RLock'a alındı (init'te). return self.uret(istek, ilerleme_cb) def _goresi_buyut( self, istek: UretimIstegi, ilerleme_cb: Optional[Callable[[int, int, str], None]], t0: float, ) -> UretimSonucu: """Görseli 2x büyüt — Real-ESRGAN ile (kalite ↑, NaN risk yok). FreeU+fp16+low-strength img2img kombosunda NaN olduğu için artık sadece Real-ESRGAN/LANCZOS+refine yolunu kullanıyoruz. """ if not istek.kaynak_resim_yolu: raise ValueError("Büyütülecek görsel yok (kaynak_resim_yolu)") if ilerleme_cb: ilerleme_cb(0, 2, "Real-ESRGAN ile yükseltiliyor") kaynak = _pillow_resmi_yukle(istek.kaynak_resim_yolu, max_kenar=2048) # Modelin yüklü olduğundan emin ol (img2img refine fallback için) self._modelin_borusunu_kur(istek.model_ismi) try: sonuc = self._upscaler_uygula(kaynak, faktor=2, model="auto") except Exception as e: print(f"[Boru] Upscaler hata, LANCZOS fallback: {e}") sonuc = kaynak.resize( (kaynak.size[0] * 2, kaynak.size[1] * 2), Image.LANCZOS, ) seed = istek.seed if (istek.seed is not None and istek.seed >= 0) else random.randint( 0, 2**31 - 1) if ilerleme_cb: ilerleme_cb(1, 2, "Kaydediliyor") klasor = Path(istek.cikti_klasoru or "outputs") klasor.mkdir(parents=True, exist_ok=True) damga = int(time.time() * 1000) ext = (istek.cikti_format or "png").lower() if ext not in ("png", "jpg", "jpeg", "webp"): ext = "png" hedef = klasor / f"buyutulmus_{damga}_{seed}.{ext}" _pillow_resmi_kaydet(sonuc, str(hedef)) if ilerleme_cb: ilerleme_cb(2, 2, "Tamamlandı") return UretimSonucu(yollar=[str(hedef)], seed=seed, sure_sn=time.time() - t0) def _sahneyi_genislet( self, istek: UretimIstegi, ilerleme_cb: Optional[Callable[[int, int, str], None]], t0: float, ) -> UretimSonucu: """Outpainting PRO — gerçek SDXL Inpainting checkpoint + color match + adaptive feather + 2-pass refine. Pipeline: 1. Edge-extend pre-fill + color match (orijinalin renk dağılımı) 2. SDXL Inpainting (resmi diffusers checkpoint) ile genişletme 3. Soft blend (alpha mask) ile orijinali geri kopyala 4. Final low-strength refine — seam yumuşatma """ if not istek.kaynak_resim_yolu: raise ValueError("Genişletilecek görsel yok (kaynak_resim_yolu)") if ilerleme_cb: ilerleme_cb(0, 4, "Outpaint hazırlık") # Daha yüksek başlangıç çözünürlüğü (1280) — pro detail kaynak = _pillow_resmi_yukle(istek.kaynak_resim_yolu, max_kenar=1280) w, h = kaynak.size oran = max(1.0, min(istek.genislet_oran, 3.0)) yon = istek.genislet_yon if yon == "all": yeni_w = int(w * oran) yeni_h = int(h * oran) ofs_x = (yeni_w - w) // 2 ofs_y = (yeni_h - h) // 2 elif yon == "left": yeni_w = int(w * oran) yeni_h = h ofs_x = yeni_w - w ofs_y = 0 elif yon == "right": yeni_w = int(w * oran) yeni_h = h ofs_x = 0 ofs_y = 0 elif yon == "top": yeni_w = w yeni_h = int(h * oran) ofs_x = 0 ofs_y = yeni_h - h else: # bottom yeni_w = w yeni_h = int(h * oran) ofs_x = 0 ofs_y = 0 # 8'in katı (SDXL gereksinimi) yeni_w = (yeni_w // 8) * 8 yeni_h = (yeni_h // 8) * 8 # Maks SDXL boyutu — VRAM güvenliği için 1536 cap if yeni_w > 1536 or yeni_h > 1536: olcek = 1536 / max(yeni_w, yeni_h) yeni_w = (int(yeni_w * olcek) // 8) * 8 yeni_h = (int(yeni_h * olcek) // 8) * 8 # Kaynak da aynı oranda küçülür yeni_kaynak_w = (int(w * olcek) // 8) * 8 yeni_kaynak_h = (int(h * olcek) // 8) * 8 kaynak = kaynak.resize((yeni_kaynak_w, yeni_kaynak_h), Image.LANCZOS) w, h = kaynak.size ofs_x = max(0, min(yeni_w - w, ofs_x)) ofs_y = max(0, min(yeni_h - h, ofs_y)) print(f"[Outpaint] {yeni_w}x{yeni_h}'e indirildi (VRAM)") from PIL import ImageDraw, ImageFilter # ===== ADAPTIVE FEATHER ===== # Outpaint oranı büyükse daha geniş feather (daha doğal geçiş) # 1.2x → 15px, 1.5x → 25px, 2.0x → 35px, 3.0x → 50px feather_inner = int(8 + (oran - 1.0) * 14) feather_inner = max(8, min(35, feather_inner)) feather_blur = int(15 + (oran - 1.0) * 20) feather_blur = max(15, min(50, feather_blur)) if ilerleme_cb: ilerleme_cb(1, 4, "Edge-extend + color match") # ===== EDGE-EXTEND PRE-FILL + COLOR MATCH ===== canvas = self._kenarlari_genislet(kaynak, yeni_w, yeni_h, ofs_x, ofs_y) # Color match: kenar bölgelerinin tonu orijinale yakın olsun try: canvas = self._renk_eslesme(canvas, kaynak, ofs_x, ofs_y, w, h) except Exception as e: print(f"[Outpaint] color match atlandı: {e}") # ===== MASK (adaptive feather) ===== mask = Image.new("L", (yeni_w, yeni_h), 255) ic_x1 = ofs_x + feather_inner ic_y1 = ofs_y + feather_inner ic_x2 = ofs_x + w - feather_inner ic_y2 = ofs_y + h - feather_inner ImageDraw.Draw(mask).rectangle( [ic_x1, ic_y1, ic_x2, ic_y2], fill=0, ) mask = mask.filter(ImageFilter.GaussianBlur(radius=feather_blur)) # ===== SDXL INPAINTING PIPE ===== # IP-Adapter / image_proj config'i temizle (siyah çıktı fix) self._modelin_borusunu_kur(istek.model_ismi) eski_ip_modu = self._ip_adapter_modu eski_plus_slot = self._plus_slot_sayisi ip_yuklu_idi = eski_ip_modu is not None try: self._pipe.unload_ip_adapter() except Exception: pass try: if hasattr(self._pipe.unet, "config") and \ getattr(self._pipe.unet.config, "encoder_hid_dim_type", None) == "ip_image_proj": self._pipe.unet.config.encoder_hid_dim_type = None except Exception: pass self._ip_adapter_modu = None if ilerleme_cb: ilerleme_cb(2, 4, "Inpainting pipe yükleniyor") # PRO: Gerçek SDXL Inpainting checkpoint kullan (varsa) # Cache'lenmiş ya da ilk indirme. Memory için from_pipe ile current # pipe komponentlerini paylaşır (sadece UNet farklı). pipe = None try: pipe = self._inpainting_pipe_yukle() if pipe is None: # Fallback: base'den türetilmiş inpaint wrapper from diffusers import AutoPipelineForInpainting pipe = AutoPipelineForInpainting.from_pipe(self._pipe) print("[Outpaint] base→inpaint wrapper kullanılıyor") else: print("[Outpaint] resmi SDXL Inpainting checkpoint kullanılıyor") except Exception as e: if ip_yuklu_idi: try: self._ip_adapter_yenile(eski_ip_modu, eski_plus_slot) except Exception: pass raise RuntimeError(f"Inpainting pipe yüklenemedi: {e}") seed = istek.seed if istek.seed is not None and istek.seed >= 0 else random.randint(0, 2**31 - 1) gen = torch.Generator(device=_device()).manual_seed(seed) # Daha betimleyici default prompt — pro kalite ekleri kullanici_prompt = (istek.prompt or "").strip() if kullanici_prompt: prompt = ( f"{kullanici_prompt}, seamless extended scene, same lighting " "and color palette, photorealistic, sharp focus, 8k, " "professional photography, natural composition" ) else: prompt = ( "seamless extended natural scene, same composition style " "and lighting, photorealistic, sharp focus, 8k, " "professional photography, no border, no frame" ) neg_prompt = (istek.negative_prompt or "") + ( ", border, frame, edge artifact, seam, discontinuity, " "watermark, text, signature, low quality, blurry" ) if ilerleme_cb: ilerleme_cb(3, 4, "Genişletme") try: sonuc = pipe( prompt=prompt, negative_prompt=neg_prompt, image=canvas, mask_image=mask, num_inference_steps=max(30, istek.steps or 35), guidance_scale=istek.cfg or 7.5, strength=0.92, # gerçek inpainting modeli için biraz yüksek OK generator=gen, width=yeni_w, height=yeni_h, ).images[0] except Exception as e: print(f"[Outpaint] inpainting hatası: {e} — fallback img2img") try: from diffusers import AutoPipelineForImage2Image pipe2 = AutoPipelineForImage2Image.from_pipe(self._pipe) sonuc = pipe2( prompt=prompt, negative_prompt=neg_prompt, image=canvas, num_inference_steps=istek.steps or 30, guidance_scale=istek.cfg or 7.0, strength=0.55, generator=gen, ).images[0] if sonuc.size != (yeni_w, yeni_h): sonuc = sonuc.resize((yeni_w, yeni_h), Image.LANCZOS) except Exception as e2: raise RuntimeError(f"Outpaint başarısız: {e} / {e2}") if ilerleme_cb: ilerleme_cb(4, 4, "Soft blend + final refine") # ===== ORİJİNAL ALANI ADAPTIVE SOFT BLEND İLE GERİ KOPYALA ===== # Outpaint oranına göre blend feather adapte soft_blend_pad = max(20, min(50, int(20 + (oran - 1.0) * 18))) soft_blend_blur = max(10, min(30, int(10 + (oran - 1.0) * 12))) try: from PIL import Image as _Img, ImageDraw as _Draw from PIL import ImageFilter as _Filt alpha = _Img.new("L", (yeni_w, yeni_h), 0) ic_x1 = ofs_x + soft_blend_pad ic_y1 = ofs_y + soft_blend_pad ic_x2 = ofs_x + w - soft_blend_pad ic_y2 = ofs_y + h - soft_blend_pad _Draw.Draw(alpha).rectangle( [ic_x1, ic_y1, ic_x2, ic_y2], fill=255, ) alpha = alpha.filter(_Filt.GaussianBlur(radius=soft_blend_blur)) kaynak_canvas = _Img.new("RGB", (yeni_w, yeni_h), (128, 128, 128)) kaynak_canvas.paste(kaynak, (ofs_x, ofs_y)) sonuc = _Img.composite(kaynak_canvas, sonuc, alpha) except Exception as e: print(f"[Outpaint] soft-blend uyarı: {e}") try: sonuc.paste(kaynak, (ofs_x, ofs_y)) except Exception: pass # ===== FINAL REFINE — pro 2-pass seam smoothing ===== try: from diffusers import AutoPipelineForImage2Image refine_pipe = AutoPipelineForImage2Image.from_pipe(self._pipe) refine_seed = (seed + 1) % (2**31 - 1) refine_gen = torch.Generator( device=_device()).manual_seed(refine_seed) sonuc_refined = refine_pipe( prompt=prompt, negative_prompt=neg_prompt, image=sonuc, num_inference_steps=20, guidance_scale=5.0, strength=0.20, # hafif — seam yumuşatma + detay zenginleştirme generator=refine_gen, ).images[0] if sonuc_refined.size == sonuc.size: # 2. soft blend — orijinal alan refine'da bozulmuşsa restore try: from PIL import Image as _Img2, ImageDraw as _D2 from PIL import ImageFilter as _F2 alpha2 = _Img2.new("L", (yeni_w, yeni_h), 0) pad2 = max(15, soft_blend_pad // 2) _D2.Draw(alpha2).rectangle( [ofs_x + pad2, ofs_y + pad2, ofs_x + w - pad2, ofs_y + h - pad2], fill=255, ) alpha2 = alpha2.filter(_F2.GaussianBlur(radius=8)) kc2 = _Img2.new("RGB", (yeni_w, yeni_h), (128, 128, 128)) kc2.paste(kaynak, (ofs_x, ofs_y)) sonuc = _Img2.composite(kc2, sonuc_refined, alpha2) except Exception: sonuc = sonuc_refined print("[Outpaint] Final refine + 2.blend uygulandı") except Exception as e: print(f"[Outpaint] Final refine atlandı: {e}") # IP-Adapter restore (outpaint sonrası) if ip_yuklu_idi: try: self._ip_adapter_yenile(eski_ip_modu, eski_plus_slot) except Exception as e: print(f"[Outpaint] IP-Adapter restore uyarı: {e}") klasor = Path(istek.cikti_klasoru or "outputs") klasor.mkdir(parents=True, exist_ok=True) damga = int(time.time() * 1000) ext = (istek.cikti_format or "png").lower() if ext not in ("png", "jpg", "jpeg", "webp"): ext = "png" hedef = klasor / f"genislemis_{damga}_{seed}.{ext}" _pillow_resmi_kaydet(sonuc, str(hedef)) if ilerleme_cb: ilerleme_cb(1, 1, "Tamamlandı") print(f"[Outpaint] {w}x{h} → {yeni_w}x{yeni_h} (yon={yon})") return UretimSonucu(yollar=[str(hedef)], seed=seed, sure_sn=time.time() - t0) # =========== Outpaint pro yardımcılar =========== _inpaint_pipe_cache = None def _inpainting_pipe_yukle(self): """Resmi SDXL Inpainting checkpoint'i yükle (cache'lenmiş). Yüklenemezse None döner — caller fallback yapar. Bu model gerçek inpainting için trained — base text2img'den çok daha tutarlı outpaint sonucu verir. """ if self.__class__._inpaint_pipe_cache is not None: return self.__class__._inpaint_pipe_cache try: from diffusers import StableDiffusionXLInpaintPipeline print("[Outpaint] SDXL Inpainting checkpoint yükleniyor " "(diffusers/stable-diffusion-xl-1.0-inpainting-0.1)...") pipe = StableDiffusionXLInpaintPipeline.from_pretrained( "diffusers/stable-diffusion-xl-1.0-inpainting-0.1", torch_dtype=_dtype(), use_safetensors=True, variant="fp16" if _dtype() == torch.float16 else None, ) # Model swap için cuda offload — VRAM güvenliği try: if _device() == "cuda": pipe.enable_model_cpu_offload() else: pipe.to("cpu") except Exception: pipe.to(_device()) # fp16 VAE fix (siyah çıktı önleme) try: from diffusers import AutoencoderKL vae_fix = AutoencoderKL.from_pretrained( "madebyollin/sdxl-vae-fp16-fix", torch_dtype=_dtype(), ) pipe.vae = vae_fix except Exception: pass self.__class__._inpaint_pipe_cache = pipe print("[Outpaint] SDXL Inpainting hazır (cache'lendi)") return pipe except Exception as e: print(f"[Outpaint] Inpainting checkpoint yüklenemedi: {e}") return None def _renk_eslesme(self, canvas, kaynak, ofs_x: int, ofs_y: int, kw: int, kh: int): """Canvas'ın orijinal alan dışındaki bölgelerinin renk dağılımını orijinaline yaklaştır (basit histogram match per-channel). Edge transition'ı doğallaştırır. """ try: import numpy as _np from PIL import Image as _Img arr = _np.array(canvas).astype(_np.float32) ka = _np.array(kaynak).astype(_np.float32) # Kaynak alanın istatistikleri for c in range(3): ka_c = ka[:, :, c] ka_mean = ka_c.mean() ka_std = ka_c.std() + 1e-6 # Tüm canvas'ı orijinale benzet (orijinal alan zaten doğru, # blend mask sayesinde kenarda smooth geçer) ext_c = arr[:, :, c] ext_mean = ext_c.mean() ext_std = ext_c.std() + 1e-6 # Sadece dış alana uygula — orijinal alan korunsun tasinmis = (ext_c - ext_mean) * (ka_std / ext_std) + ka_mean # Orijinal alanı tekrar yerine koy yeni = tasinmis.copy() yeni[ofs_y:ofs_y + kh, ofs_x:ofs_x + kw] = ka_c arr[:, :, c] = yeni arr = _np.clip(arr, 0, 255).astype(_np.uint8) return _Img.fromarray(arr) except Exception as e: print(f"[Outpaint] renk eşleşme hatası: {e}") return canvas def _kenarlari_genislet(self, kaynak, yeni_w: int, yeni_h: int, ofs_x: int, ofs_y: int): """Canvas'ı orijinalin kenarlarını mirror/extend ile doldur. Gri canvas yerine doğal başlangıç → SDXL inpainting daha iyi devam eder.""" from PIL import Image as _Img w, h = kaynak.size canvas = _Img.new("RGB", (yeni_w, yeni_h), (128, 128, 128)) canvas.paste(kaynak, (ofs_x, ofs_y)) try: # Sol kenar if ofs_x > 0: sol_strip = kaynak.crop((0, 0, 1, h)).resize((ofs_x, h), Image.NEAREST) # Mirror efekt için: kaynaktan biraz sol şeridi alıp ayna ayna_geni = min(ofs_x, w // 4) if ayna_geni > 0: ayna = kaynak.crop((0, 0, ayna_geni, h)).transpose( Image.FLIP_LEFT_RIGHT) canvas.paste(ayna, (ofs_x - ayna_geni, ofs_y)) canvas.paste(sol_strip, (0, ofs_y)) # Sağ kenar sag_baslangic = ofs_x + w if sag_baslangic < yeni_w: sag_geni = yeni_w - sag_baslangic ayna_geni = min(sag_geni, w // 4) if ayna_geni > 0: ayna = kaynak.crop((w - ayna_geni, 0, w, h)).transpose( Image.FLIP_LEFT_RIGHT) canvas.paste(ayna, (sag_baslangic, ofs_y)) # Kalan alana son sütun şeridi kalan = sag_geni - ayna_geni if kalan > 0: sag_strip = kaynak.crop((w - 1, 0, w, h)).resize( (kalan, h), Image.NEAREST) canvas.paste(sag_strip, (sag_baslangic + ayna_geni, ofs_y)) # Üst kenar (canvas tamamı için) if ofs_y > 0: ust_geni_h = min(ofs_y, h // 4) if ust_geni_h > 0: ayna = canvas.crop((0, ofs_y, yeni_w, ofs_y + ust_geni_h)).transpose( Image.FLIP_TOP_BOTTOM) canvas.paste(ayna, (0, ofs_y - ust_geni_h)) # Kalan üst alan kalan = ofs_y - ust_geni_h if kalan > 0: strip = canvas.crop((0, ofs_y - ust_geni_h, yeni_w, ofs_y - ust_geni_h + 1)).resize( (yeni_w, kalan), Image.NEAREST) canvas.paste(strip, (0, 0)) # Alt kenar alt_baslangic = ofs_y + h if alt_baslangic < yeni_h: alt_geni_h = yeni_h - alt_baslangic ayna_h = min(alt_geni_h, h // 4) if ayna_h > 0: ayna = canvas.crop((0, alt_baslangic - ayna_h, yeni_w, alt_baslangic)).transpose( Image.FLIP_TOP_BOTTOM) canvas.paste(ayna, (0, alt_baslangic)) kalan = alt_geni_h - ayna_h if kalan > 0: strip = canvas.crop((0, alt_baslangic + ayna_h - 1, yeni_w, alt_baslangic + ayna_h)).resize( (yeni_w, kalan), Image.NEAREST) canvas.paste(strip, (0, alt_baslangic + ayna_h)) # Hafif blur — sert geçişleri yumuşat (mirror seam'leri) from PIL import ImageFilter yumusatilmis = canvas.filter(ImageFilter.GaussianBlur(radius=3)) # Orijinali tekrar yerine koy (yumuşatılmış kenarlar kalsın) yumusatilmis.paste(kaynak, (ofs_x, ofs_y)) return yumusatilmis except Exception as e: print(f"[Outpaint] edge-extend hatası: {e}, gri canvas kullanılıyor") return canvas def _sadece_yuz_degis( self, istek: UretimIstegi, ilerleme_cb: Optional[Callable[[int, int, str], None]], t0: float, ) -> UretimSonucu: """Promptsuz mod — sadece InsightFace face swap, diffusion yok.""" if not istek.kaynak_resim_yolu: raise ValueError("Hedef görsel gerekli (kaynak_resim_yolu)") kaynak_yollari = list(istek.coklu_yuz_kaynaklari or []) if istek.face_swap_kaynak_yuz and not kaynak_yollari: kaynak_yollari = [istek.face_swap_kaynak_yuz] if not kaynak_yollari: raise ValueError("En az bir kaynak yüz gerekli") if ilerleme_cb: ilerleme_cb(0, 1, "Yüz değişimi") hedef = _pillow_resmi_yukle(istek.kaynak_resim_yolu, max_kenar=2048) sonuc = self._yuz_degis_uygula(hedef, kaynak_yollari) if ilerleme_cb: ilerleme_cb(1, 1, "Kaydediliyor") # Kaydet — ms hassasiyet + random suffix (aynı saniye çakışma engeli) klasor = Path(istek.cikti_klasoru or "outputs") klasor.mkdir(parents=True, exist_ok=True) damga = int(time.time() * 1000) rnd = random.randint(0, 9999) ext = (istek.cikti_format or "png").lower() if ext not in ("png", "jpg", "jpeg", "webp"): ext = "png" hedef_yol = klasor / f"yuzdegis_{damga}_{rnd:04d}.{ext}" _pillow_resmi_kaydet(sonuc, str(hedef_yol)) return UretimSonucu( yollar=[str(hedef_yol)], seed=0, sure_sn=time.time() - t0, ) def _face_embedding_olustur(self, yuz_yolu: str, batch: int = 1): """InsightFace ile face embedding üret. Yüz yoksa None döner. Diffusers CFG için [2*batch, 1, 512] formatı (negative+positive).""" self._face_app_yukle() if self._face_app is None: return None img = _pillow_resmi_yukle(yuz_yolu, 1024) np_bgr = np.array(img)[:, :, ::-1].copy() yuzler = self._face_app.get(np_bgr) if not yuzler: return None # En büyük yüzü al yuzler.sort( key=lambda y: (y.bbox[2] - y.bbox[0]) * (y.bbox[3] - y.bbox[1]), reverse=True, ) positive = torch.from_numpy(yuzler[0].normed_embedding) \ .unsqueeze(0).unsqueeze(0) # [1, 1, 512] # Batch için tekrarla if batch > 1: positive = positive.repeat_interleave(batch, dim=0) # CFG için negative (sıfır) + positive negative = torch.zeros_like(positive) embeds = torch.cat([negative, positive], dim=0) # [2*batch, 1, 512] return embeds.to(_device(), dtype=_dtype()) # ---------- ADetailer (yüz restorasyon) ---------- def _adetailer_yuz_duzelt( self, sonuc_resim: Image.Image, prompt: str, denoise: float = 0.4, negative_prompt: Optional[str] = None, ) -> Image.Image: """ADetailer benzeri yüz restorasyonu. InsightFace ile yüzleri tespit et, her yüzü padding ile inpaint et.""" try: self._face_app_yukle() if self._face_app is None or self._pipe is None: return sonuc_resim np_bgr = np.array(sonuc_resim)[:, :, ::-1].copy() yuzler = self._face_app.get(np_bgr) if not yuzler: print("[ADetailer] Yüz tespit edilmedi, atlanıyor") return sonuc_resim # IP-Adapter / image_proj config'i her durumda temizle (siyah çıktı fix) eski_ip_modu = self._ip_adapter_modu eski_plus_slot = self._plus_slot_sayisi ip_yuklu_idi = eski_ip_modu is not None try: self._pipe.unload_ip_adapter() except Exception: pass try: if hasattr(self._pipe.unet, "config") and \ getattr(self._pipe.unet.config, "encoder_hid_dim_type", None) == "ip_image_proj": self._pipe.unet.config.encoder_hid_dim_type = None except Exception: pass self._ip_adapter_modu = None from diffusers import AutoPipelineForInpainting from PIL import ImageDraw, ImageFilter try: inpaint_pipe = AutoPipelineForInpainting.from_pipe(self._pipe) except Exception as e: print(f"[ADetailer] Inpaint pipe açılamadı: {e}") # IP-Adapter'ı geri yükle if ip_yuklu_idi: try: self._ip_adapter_yenile(eski_ip_modu, eski_plus_slot) except Exception: pass return sonuc_resim denoise = max(0.2, min(0.6, float(denoise))) W, H = sonuc_resim.size calisan = sonuc_resim.copy() duzeltilen = 0 for yuz in yuzler: try: x1, y1, x2, y2 = yuz.bbox.astype(int) yw, yh = x2 - x1, y2 - y1 if yw < 24 or yh < 24: continue # çok küçük yüzler atlanır # %20 padding px = int(yw * 0.20) py = int(yh * 0.20) rx1 = max(0, x1 - px) ry1 = max(0, y1 - py) rx2 = min(W, x2 + px) ry2 = min(H, y2 + py) # Inpaint kutusunu 8'in katına yuvarla bw = ((rx2 - rx1) // 8) * 8 bh = ((ry2 - ry1) // 8) * 8 if bw < 64 or bh < 64: continue rx2 = rx1 + bw ry2 = ry1 + bh # Mask: tüm görsel siyah, yüz alanı beyaz (yumuşatılmış) mask = Image.new("L", (W, H), 0) ImageDraw.Draw(mask).rectangle( [rx1 + bw // 10, ry1 + bh // 10, rx2 - bw // 10, ry2 - bh // 10], fill=255, ) mask = mask.filter(ImageFilter.GaussianBlur(radius=8)) yuz_prompt = (prompt or "") + \ ", detailed face, sharp eyes, natural skin, high quality" yuz_neg = negative_prompt or \ "blurry, distorted face, deformed eyes, low quality" seed = random.randint(0, 2**31 - 1) gen = torch.Generator(device=_device()).manual_seed(seed) duz = inpaint_pipe( prompt=yuz_prompt, negative_prompt=yuz_neg, image=calisan, mask_image=mask, strength=denoise, num_inference_steps=25, guidance_scale=6.0, generator=gen, width=W, height=H, ).images[0] # Yumuşak blend (mask'a göre composite) calisan = Image.composite(duz, calisan, mask) duzeltilen += 1 except Exception as ye: print(f"[ADetailer] Yüz işleme hatası (atlanıyor): {ye}") continue print(f"[ADetailer] {duzeltilen}/{len(yuzler)} yüz düzeltildi") # IP-Adapter'ı geri yükle (eski state) if ip_yuklu_idi: try: self._ip_adapter_yenile(eski_ip_modu, eski_plus_slot) print(f"[ADetailer] IP-Adapter restore edildi " f"(modu={eski_ip_modu})") except Exception as e: print(f"[ADetailer] IP-Adapter restore uyarı: {e}") return calisan except Exception as e: print(f"[ADetailer] Genel hata, orijinal döndü: {e}") # IP-Adapter durumunu geri al try: if 'ip_yuklu_idi' in locals() and ip_yuklu_idi: self._ip_adapter_yenile(eski_ip_modu, eski_plus_slot) except Exception: pass return sonuc_resim # ---------- Upscaler ---------- def _upscaler_uygula( self, resim: Image.Image, faktor: int = 2, model: str = "auto", ) -> Image.Image: """Görseli faktör kadar büyütür. Real-ESRGAN → LANCZOS+refine fallback.""" if faktor <= 1: return resim faktor = max(2, min(4, int(faktor))) eski_w, eski_h = resim.size yeni_w, yeni_h = eski_w * faktor, eski_h * faktor # 1) Real-ESRGAN dene if model in ("auto", "realesrgan"): try: buyutulmus = self._upscaler_realesrgan(resim, faktor) if buyutulmus is not None: print(f"[Upscale] RealESRGAN: {eski_w}x{eski_h} → {buyutulmus.size[0]}x{buyutulmus.size[1]}") return buyutulmus except Exception as e: print(f"[Upscale] RealESRGAN başarısız, fallback: {e}") if model == "realesrgan": # Açıkça istendi ama olmadı → LANCZOS'a düş pass # 2) Fallback: LANCZOS + düşük güç img2img refine try: buyutulmus = resim.resize((yeni_w, yeni_h), Image.LANCZOS) if self._pipe is not None: from diffusers import AutoPipelineForImage2Image refine_pipe = AutoPipelineForImage2Image.from_pipe(self._pipe) seed = random.randint(0, 2**31 - 1) gen = torch.Generator(device=_device()).manual_seed(seed) # 8 katı yap rw = (yeni_w // 8) * 8 rh = (yeni_h // 8) * 8 if (rw, rh) != buyutulmus.size: buyutulmus = buyutulmus.resize((rw, rh), Image.LANCZOS) buyutulmus = refine_pipe( prompt="highly detailed, sharp focus, photorealistic, 8k", negative_prompt="blurry, low quality", image=buyutulmus, strength=0.20, guidance_scale=5.0, num_inference_steps=20, generator=gen, ).images[0] print(f"[Upscale] LANCZOS+refine: {eski_w}x{eski_h} → {buyutulmus.size[0]}x{buyutulmus.size[1]}") return buyutulmus except Exception as e: print(f"[Upscale] Refine başarısız, sadece LANCZOS: {e}") try: return resim.resize((yeni_w, yeni_h), Image.LANCZOS) except Exception: return resim def _upscaler_realesrgan(self, resim: Image.Image, faktor: int) -> Optional[Image.Image]: """Real-ESRGAN ile büyütme. Tile mode (VRAM dostu). None döner = başarısız.""" try: from realesrgan import RealESRGANer from basicsr.archs.rrdbnet_arch import RRDBNet except ImportError: return None # Model seçimi (x4plus 4x, x2plus 2x) try: from huggingface_hub import hf_hub_download except ImportError: return None try: if faktor >= 4: model_yolu = hf_hub_download( "ai-forever/Real-ESRGAN", "RealESRGAN_x4.pth", ) rrdb = RRDBNet(num_in_ch=3, num_out_ch=3, num_feat=64, num_block=23, num_grow_ch=32, scale=4) model_scale = 4 else: # 2x için x4 modelini kullan, sonra küçült (daha kaliteli) model_yolu = hf_hub_download( "ai-forever/Real-ESRGAN", "RealESRGAN_x2.pth", ) rrdb = RRDBNet(num_in_ch=3, num_out_ch=3, num_feat=64, num_block=23, num_grow_ch=32, scale=2) model_scale = 2 except Exception as e: print(f"[Upscale] Model indirilemedi: {e}") return None upsampler = RealESRGANer( scale=model_scale, model_path=model_yolu, model=rrdb, tile=512, # VRAM tasarrufu için tile mode tile_pad=10, pre_pad=0, half=self._cuda, # fp16 sadece CUDA'da device=_device(), ) np_rgb = np.array(resim) np_bgr = np_rgb[:, :, ::-1].copy() cikti_bgr, _ = upsampler.enhance(np_bgr, outscale=faktor) cikti_rgb = cikti_bgr[:, :, ::-1].copy() return Image.fromarray(cikti_rgb) def _face_app_yukle(self): if self._face_app is not None: return global _insightface_loaded if not _insightface_loaded: _insightface_loaded = True from insightface.app import FaceAnalysis # CPU yeterince hızlı (~50ms/resim) ve onnxruntime-gpu cuDNN sorunsuz değil try: app = FaceAnalysis( name="buffalo_l", providers=["CPUExecutionProvider"] ) app.prepare(ctx_id=-1, det_size=(640, 640)) self._face_app = app print("[Boru] InsightFace yüklendi (CPU)") except Exception as e: print(f"[Boru] InsightFace yüklenemedi: {e}") self._face_app = None def _face_swapper_yukle(self): if self._face_swapper is not None: return from insightface.model_zoo import get_model # GitHub'daki orijinal kaldırıldı; HuggingFace mirror'lardan indir kok = Path.home() / ".insightface" / "models" kok.mkdir(parents=True, exist_ok=True) hedef = kok / "inswapper_128.onnx" if not hedef.exists(): mirror_urller = [ "https://huggingface.co/ezioruan/inswapper_128.onnx/resolve/main/inswapper_128.onnx", "https://huggingface.co/MonsterMMORPG/tools/resolve/main/inswapper_128.onnx", "https://huggingface.co/deepinsight/inswapper/resolve/main/inswapper_128.onnx", ] import urllib.request indirildi = False for url in mirror_urller: try: print(f"[Boru] inswapper_128.onnx indiriliyor: {url}") with urllib.request.urlopen(url, timeout=60) as r, \ open(hedef, "wb") as f: f.write(r.read()) if hedef.stat().st_size > 100_000_000: # ~530MB beklenir indirildi = True print(f"[Boru] inswapper indi: {hedef.stat().st_size / 1e6:.1f} MB") break else: hedef.unlink() except Exception as e: print(f"[Boru] Mirror başarısız ({url}): {e}") if not indirildi: raise RuntimeError( "inswapper_128.onnx indirilemedi. " "Manuel indirip şuraya koy: " + str(hedef) ) try: self._face_swapper = get_model( str(hedef), providers=["CPUExecutionProvider"], ) except TypeError: self._face_swapper = get_model(str(hedef)) # ---------- Üretim ---------- def uret( self, istek: UretimIstegi, ilerleme_cb: Optional[Callable[[int, int, str], None]] = None, ) -> UretimSonucu: with self._kilit: t0 = time.time() # AKIŞ — sıralı adım listesi if istek.adimlar and len(istek.adimlar) > 0: return self._akisi_calistir(istek, ilerleme_cb, t0) # SADECE YÜZ DEĞİŞTİRME — diffusion'a gerek yok if istek.gorev_tipi == "sadece_yuz_degis": return self._sadece_yuz_degis(istek, ilerleme_cb, t0) # GÖRSELİ BÜYÜT (upscale 2x) — img2img low-strength + büyük boyut if istek.gorev_tipi == "buyut": return self._goresi_buyut(istek, ilerleme_cb, t0) # SAHNEYİ GENİŞLET (outpainting) — kareyi büyüt + boş alanı doldur if istek.gorev_tipi == "genislet": return self._sahneyi_genislet(istek, ilerleme_cb, t0) # img2img zorunluluğu if istek.gorev_tipi == "img2img" and not istek.kaynak_resim_yolu: raise ValueError( "img2img modu için kaynak_resim_yolu zorunlu", ) # Boş prompt + text2img → uyarı + minimal default if (istek.gorev_tipi or "text2img") == "text2img" and \ not (istek.prompt or "").strip(): print("[Boru] Uyarı: text2img için prompt boş, jenerik kullanılıyor") istek.prompt = "high quality photograph, sharp focus, detailed" self._modelin_borusunu_kur(istek.model_ismi) assert self._pipe is not None # Aktif LoRA'ları yükle (her uret çağrısında temiz başla) self._aktif_loralari_uygula(istek.aktif_loralar) # Seed if istek.seed is None or istek.seed < 0: seed = random.randint(0, 2**31 - 1) else: seed = int(istek.seed) generator = torch.Generator(device=_device()).manual_seed(seed) # IP-Adapter mod yönlendirmesi (FaceID veya çoklu Plus slot) face_embed = None ip_adapter_images: List[Image.Image] = [] ip_olcekleri: List[float] = [] hedef_mod: Optional[str] = None # Plus slot adayları (kategorı, yol, ölçek, yüz kırp?) plus_adaylari = [] if istek.referans_yuz_yolu and istek.yuz_modu == "plus": plus_adaylari.append(( "yuz", istek.referans_yuz_yolu, istek.yuz_referans_olcek, True, )) if istek.referans_sac_yolu: plus_adaylari.append(( "sac", istek.referans_sac_yolu, istek.sac_referans_olcek, False, )) if istek.referans_elbise_yolu: plus_adaylari.append(( "elbise", istek.referans_elbise_yolu, istek.elbise_referans_olcek, False, )) if istek.referans_urun_yolu: plus_adaylari.append(( "urun", istek.referans_urun_yolu, istek.urun_referans_olcek, False, )) if istek.referans_arkaplan_yolu: plus_adaylari.append(( "arkaplan", istek.referans_arkaplan_yolu, istek.arkaplan_referans_olcek, False, )) # Maksimum 4 slot (VRAM) plus_adaylari = plus_adaylari[:4] faceid_aktif = (istek.referans_yuz_yolu and istek.yuz_modu == "faceid") if faceid_aktif: hedef_mod = "faceid" elif plus_adaylari: hedef_mod = "plus" self._ip_adapter_yenile(hedef_mod, plus_slot_sayisi=len(plus_adaylari) or 1) if hedef_mod == "faceid" and self._ip_adapter_modu == "faceid": face_embed = self._face_embedding_olustur( istek.referans_yuz_yolu, batch=max(1, istek.batch), ) if face_embed is None: print("[Boru] Yüz tespit edilemedi, FaceID atlandı") self._ip_adapter_yenile(None) else: self._pipe.set_ip_adapter_scale(istek.yuz_referans_olcek) # FaceID LoRA + kullanıcı LoRA'larını birleştir tum_adapter = ["faceid_lora"] + list(self._aktif_lora_isimleri) tum_agirlik = [istek.yuz_referans_olcek] + \ list(self._aktif_lora_agirliklari) try: self._pipe.set_adapters( tum_adapter, adapter_weights=tum_agirlik, ) except Exception as e: print(f"[Boru] set_adapters (faceid+lora) hatası: {e}") elif hedef_mod == "plus" and self._ip_adapter_modu == "plus": for kategori, yol, olcek, yuz_kirp in plus_adaylari: img = _pillow_resmi_yukle(yol, 1024) if yuz_kirp: self._face_app_yukle() if self._face_app is not None: img = _yuzu_kirp(img, self._face_app, marj=1.6) ip_adapter_images.append(img) ip_olcekleri.append(olcek) print(f"[Boru] slot[{kategori}] = {Path(yol).name} × {olcek}") if len(ip_olcekleri) == 1: self._pipe.set_ip_adapter_scale(ip_olcekleri[0]) else: self._pipe.set_ip_adapter_scale(ip_olcekleri) # Kullanıcı LoRA'ları varsa aktive et if self._aktif_lora_isimleri: try: self._pipe.set_adapters( list(self._aktif_lora_isimleri), adapter_weights=list(self._aktif_lora_agirliklari), ) except Exception as e: print(f"[Boru] set_adapters (plus+lora) hatası: {e}") else: # Ne FaceID ne Plus — sadece kullanıcı LoRA'ları varsa aktive et if self._aktif_lora_isimleri: try: self._pipe.set_adapters( list(self._aktif_lora_isimleri), adapter_weights=list(self._aktif_lora_agirliklari), ) except Exception as e: print(f"[Boru] set_adapters (lora) hatası: {e}") # Diffusers ilerleme callback'i iceriden_son_step = {"i": 0} def cb_on_step(pipe, step_index, timestep, callback_kwargs): iceriden_son_step["i"] = step_index if ilerleme_cb is not None: ilerleme_cb(step_index, istek.steps, "Üretim") return callback_kwargs # img2img modu — kaynak görsel verilmişse pipe'ı dönüştür img2img_modu = bool(istek.kaynak_resim_yolu) kullanilan_pipe = self._pipe if img2img_modu: from diffusers import AutoPipelineForImage2Image kullanilan_pipe = AutoPipelineForImage2Image.from_pipe(self._pipe) kaynak_resmi = _pillow_resmi_yukle( istek.kaynak_resim_yolu, max_kenar=1536, ) # Boyut: kaynak görsel boyutunu kullan (en yakın 8'in katı) w, h = kaynak_resmi.size w = (w // 8) * 8 h = (h // 8) * 8 kaynak_resmi = kaynak_resmi.resize((w, h), Image.LANCZOS) print(f"[Boru] img2img modu: {w}×{h}, güç={istek.degisiklik_gucu}") kwargs = dict( prompt=istek.prompt, negative_prompt=istek.negative_prompt or None, num_inference_steps=istek.steps, guidance_scale=istek.cfg, generator=generator, num_images_per_prompt=max(1, istek.batch), callback_on_step_end=cb_on_step, ) if img2img_modu: kwargs["image"] = kaynak_resmi kwargs["strength"] = max(0.05, min(1.0, istek.degisiklik_gucu)) else: kwargs["width"] = istek.genislik kwargs["height"] = istek.yukseklik if face_embed is not None and self._ip_adapter_modu == "faceid": # FaceID: image_embeds (InsightFace), image yok kwargs["ip_adapter_image_embeds"] = [face_embed] elif ip_adapter_images and self._ip_adapter_modu == "plus": if len(ip_adapter_images) == 1: kwargs["ip_adapter_image"] = ip_adapter_images[0] else: # Multi-slot: her slot için tek bir görsel listesi kwargs["ip_adapter_image"] = ip_adapter_images if ilerleme_cb: ilerleme_cb(0, istek.steps, "Başlıyor") sonuclar = kullanilan_pipe(**kwargs).images # Face swap (post-process) — multi-source destekli kaynak_yollari: List[str] = list(istek.coklu_yuz_kaynaklari or []) if istek.face_swap_kaynak_yuz and not kaynak_yollari: kaynak_yollari = [istek.face_swap_kaynak_yuz] if istek.face_swap_etkin and not kaynak_yollari: print("[Boru] Uyarı: face_swap_etkin=True ama kaynak yüz yok") if istek.face_swap_etkin and kaynak_yollari: try: if ilerleme_cb: ilerleme_cb(istek.steps, istek.steps, "Yüz değişimi") sonuclar = [ self._yuz_degis_uygula(img, kaynak_yollari) for img in sonuclar ] except Exception as e: print(f"[Boru] Face swap hatası (atlanıyor): {e}") import traceback traceback.print_exc() # ADetailer — yüz restorasyon if istek.adetailer_etkin: try: if ilerleme_cb: ilerleme_cb(istek.steps, istek.steps, "ADetailer (yüz)") sonuclar = [ self._adetailer_yuz_duzelt( img, istek.prompt, istek.adetailer_denoise, istek.negative_prompt, ) for img in sonuclar ] except Exception as e: print(f"[ADetailer] Hata (atlanıyor): {e}") import traceback traceback.print_exc() # Upscaler — büyütme if istek.upscale_faktor > 1: try: if ilerleme_cb: ilerleme_cb( istek.steps, istek.steps, f"Upscale {istek.upscale_faktor}x", ) sonuclar = [ self._upscaler_uygula( img, istek.upscale_faktor, istek.upscale_model, ) for img in sonuclar ] except Exception as e: print(f"[Upscale] Hata (atlanıyor): {e}") import traceback traceback.print_exc() # Kaydet yollar = [] klasor = Path(istek.cikti_klasoru or "outputs") klasor.mkdir(parents=True, exist_ok=True) damga = int(time.time() * 1000) ext = (istek.cikti_format or "png").lower() if ext not in ("png", "jpg", "jpeg", "webp"): ext = "png" for i, img in enumerate(sonuclar): hedef = klasor / f"gorsel_{damga}_{seed}_{i:02d}.{ext}" _pillow_resmi_kaydet(img, str(hedef)) yollar.append(str(hedef)) # Yan dosya — metadata try: meta = { "prompt": istek.prompt, "negative_prompt": istek.negative_prompt, "model": istek.model_ismi, "boyut": [istek.genislik, istek.yukseklik], "steps": istek.steps, "cfg": istek.cfg, "seed": seed, "scheduler": istek.scheduler, "yuz_ref": istek.referans_yuz_yolu, "urun_ref": istek.referans_urun_yolu, "face_swap": istek.face_swap_kaynak_yuz if istek.face_swap_etkin else None, "tarih": time.strftime("%Y-%m-%d %H:%M:%S"), } (klasor / f"gorsel_{damga}_{seed}.json").write_text( json.dumps(meta, ensure_ascii=False, indent=2), encoding="utf-8" ) except Exception: pass sure = time.time() - t0 return UretimSonucu(yollar=yollar, seed=seed, sure_sn=sure)