"""WanGP Köprü Servisi - FastAPI uygulaması. Flutter masaüstü uygulamasının WanGP'nin in-process Python API'sine HTTP/SSE üzerinden erişebilmesi için sarmalayıcı. Mimari: - Tek WanGP session, lazy init (ilk job geldiğinde) - Job kayıt sözlüğü (asyncio.Lock korumalı) - SSE event stream (her job için iterator queue'su) - Eski jobların 1 saatlik otomatik temizliği """ from __future__ import annotations import asyncio import base64 import io import json import logging import mimetypes import os import sys import time import uuid from contextlib import asynccontextmanager from pathlib import Path from typing import Any, AsyncIterator from fastapi import ( FastAPI, File, HTTPException, Query, Request, UploadFile, ) from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import JSONResponse, StreamingResponse from kopru_modeller import ( GpuBilgi, IlerlemeKaydi, IptalYanit, JobIstek, JobOzet, JobYanit, LoraKaydi, ModelKaydi, OlayKaydi, SaglikYanit, SilYanit, SonucYanit, YuklemeYanit, ) # ---------------------------------------------------------------------------- # Logger # ---------------------------------------------------------------------------- LOGGER = logging.getLogger("fiko-kopru") def _logger_kur() -> None: if LOGGER.handlers: return LOGGER.setLevel(logging.INFO) formatci = logging.Formatter( "[%(levelname)s] %(asctime)s | %(name)s | %(message)s", datefmt="%H:%M:%S", ) konsol = logging.StreamHandler() konsol.setFormatter(formatci) LOGGER.addHandler(konsol) _logger_kur() # ---------------------------------------------------------------------------- # Genel ayarlar / durum # ---------------------------------------------------------------------------- WANGP_ROOT: Path | None = None WANGP_OUTPUT_DIR: Path | None = None YUKLEME_DIZINI: Path = Path(__file__).parent / "yuklenenler" # WanGP session referansı (lazy init) _wangp_session: Any | None = None _wangp_session_lock = asyncio.Lock() _wangp_init_hatasi: str | None = None # Job kayıt deposu _jobs: dict[str, "JobKayit"] = {} _jobs_lock = asyncio.Lock() # Otomatik temizlik (saat) JOB_TEMIZLIK_SURESI_SN = 3600 # 1 saat PERIYODIK_TEMIZLIK_ARALIGI_SN = 300 # 5 dakika # ---------------------------------------------------------------------------- # JobKayit - tek bir üretim işinin tüm bilgileri # ---------------------------------------------------------------------------- class JobKayit: """Bir WanGP üretim işinin köprü tarafındaki kaydı.""" def __init__( self, job_id: str, wangp_job: Any, settings: dict[str, Any], etiket: str | None = None, ) -> None: self.job_id = job_id self.wangp_job = wangp_job self.settings = settings self.etiket = etiket self.olusturma_zamani = time.time() self.baslangic_zamani: float | None = None self.bitis_zamani: float | None = None self.son_olay_zamani: float | None = None self.durum: str = "kuyruk" # kuyruk | calisiyor | tamam | iptal | hata self.ilerleme: int = 0 self.faz: str = "" self.mesaj: str = "" self.su_anki_adim: int | None = None self.toplam_adim: int | None = None # Sonuç bilgisi self.sonuc: dict[str, Any] | None = None self.hata_mesajlari: list[str] = [] # Tüm olayların log'u (son N adet kullanıcıya state için yeterli) self.son_olay: dict[str, Any] | None = None # Bu jobın olaylarını dinleyen aktif SSE abonelerinin queue'ları self.aboneler: list[asyncio.Queue[dict[str, Any]]] = [] # WanGP'den event okuma task'i self.dinleyici_task: asyncio.Task | None = None # WanGP olaylarını publish için ana event loop referansı self.ana_loop: asyncio.AbstractEventLoop | None = None def model_type(self) -> str | None: if isinstance(self.settings, dict): mt = self.settings.get("model_type") return str(mt) if mt else None return None def ozet(self) -> JobOzet: return JobOzet( job_id=self.job_id, durum=self.durum, ilerleme=self.ilerleme, faz=self.faz, olusturma_zamani=self.olusturma_zamani, bitis_zamani=self.bitis_zamani, etiket=self.etiket, model_type=self.model_type(), ) def ilerleme_kaydi(self) -> IlerlemeKaydi: return IlerlemeKaydi( job_id=self.job_id, durum=self.durum, # type: ignore[arg-type] ilerleme=self.ilerleme, faz=self.faz, mesaj=self.mesaj, su_anki_adim=self.su_anki_adim, toplam_adim=self.toplam_adim, olusturma_zamani=self.olusturma_zamani, baslangic_zamani=self.baslangic_zamani, bitis_zamani=self.bitis_zamani, son_olay_zamani=self.son_olay_zamani, etiket=self.etiket, ) async def aboneye_yayinla(self, paket: dict[str, Any]) -> None: """Tüm aktif SSE abonelerine olayı gönder.""" olu_aboneler: list[asyncio.Queue[dict[str, Any]]] = [] for kuyruk in list(self.aboneler): try: kuyruk.put_nowait(paket) except asyncio.QueueFull: olu_aboneler.append(kuyruk) for kuyruk in olu_aboneler: try: self.aboneler.remove(kuyruk) except ValueError: pass # ---------------------------------------------------------------------------- # WanGP Session yönetimi # ---------------------------------------------------------------------------- def _wangp_modul_yukle() -> Any: """shared.api modülünü import et (sys.path'e WanGP root eklenmiş olmalı).""" if WANGP_ROOT is None: raise RuntimeError( "WanGP kök dizini ayarlanmamış. baslat.py --wangp-root ile başlatın." ) kok_str = str(WANGP_ROOT.resolve()) if kok_str not in sys.path: sys.path.insert(0, kok_str) LOGGER.info("sys.path'e eklendi: %s", kok_str) try: from shared import api as wangp_api # type: ignore return wangp_api except ImportError as exc: raise RuntimeError( f"WanGP shared.api modülü yüklenemedi: {exc}. " f"--wangp-root parametresinin doğru olduğundan emin olun: {WANGP_ROOT}" ) from exc async def _wangp_session_al() -> Any: """WanGP session'ını lazy olarak başlat ve döndür.""" global _wangp_session, _wangp_init_hatasi if _wangp_session is not None: return _wangp_session async with _wangp_session_lock: if _wangp_session is not None: return _wangp_session LOGGER.info("WanGP session başlatılıyor (ilk job geldi)...") loop = asyncio.get_running_loop() def _baslat() -> Any: wangp_api = _wangp_modul_yukle() kw: dict[str, Any] = {"root": WANGP_ROOT} if WANGP_OUTPUT_DIR is not None: kw["output_dir"] = WANGP_OUTPUT_DIR return wangp_api.init(**kw) try: _wangp_session = await loop.run_in_executor(None, _baslat) LOGGER.info("WanGP session hazır.") _wangp_init_hatasi = None except Exception as exc: _wangp_init_hatasi = str(exc) LOGGER.exception("WanGP session başlatılamadı: %s", exc) raise return _wangp_session # ---------------------------------------------------------------------------- # WanGP olaylarını dinle ve abonelere yayınla # ---------------------------------------------------------------------------- def _olay_serilestir(kind: str, data: Any) -> Any: """WanGP olay verisini JSON-uyumlu hale getir.""" if data is None: return None # PreviewUpdate - PIL Image içerir if kind == "preview": return _onizleme_paketle(data) # StreamMessage if kind == "stream": if hasattr(data, "stream") and hasattr(data, "text"): return {"stream": data.stream, "text": data.text} return {"stream": "stdout", "text": str(data)} # ProgressUpdate if kind == "progress": if hasattr(data, "phase"): return { "phase": getattr(data, "phase", ""), "status": getattr(data, "status", ""), "progress": getattr(data, "progress", 0), "current_step": getattr(data, "current_step", None), "total_steps": getattr(data, "total_steps", None), "raw_phase": getattr(data, "raw_phase", None), "unit": getattr(data, "unit", None), } # GenerationResult - completed if kind == "completed": if hasattr(data, "success"): return { "success": bool(getattr(data, "success", False)), "generated_files": [str(p) for p in getattr(data, "generated_files", []) or []], "errors": [str(e) for e in getattr(data, "errors", []) or []], "cancelled": bool(getattr(data, "cancelled", False)), "total_tasks": int(getattr(data, "total_tasks", 0) or 0), "successful_tasks": int(getattr(data, "successful_tasks", 0) or 0), "failed_tasks": int(getattr(data, "failed_tasks", 0) or 0), } # GenerationError - error if kind == "error": if hasattr(data, "message"): return { "message": str(getattr(data, "message", "")), "task_index": getattr(data, "task_index", None), "stage": getattr(data, "stage", None), "cancelled": bool(getattr(data, "cancelled", False)), } return {"message": str(data)} # Düz dict/str/sayılar if isinstance(data, (dict, list, str, int, float, bool)): return data # Dataclass benzeri if hasattr(data, "__dict__"): try: d = {} for k, v in vars(data).items(): if k.startswith("_"): continue try: json.dumps(v) d[k] = v except (TypeError, ValueError): d[k] = str(v) return d except Exception: return str(data) return str(data) def _onizleme_paketle(preview: Any) -> dict[str, Any]: """PreviewUpdate -> base64 PNG + meta veri.""" paket: dict[str, Any] = { "phase": getattr(preview, "phase", ""), "status": getattr(preview, "status", ""), "progress": getattr(preview, "progress", 0), "current_step": getattr(preview, "current_step", None), "total_steps": getattr(preview, "total_steps", None), "image_b64": None, "format": "png", } img = getattr(preview, "image", None) if img is None: return paket try: buf = io.BytesIO() img.save(buf, format="PNG", optimize=False) paket["image_b64"] = base64.b64encode(buf.getvalue()).decode("ascii") except Exception as exc: # PIL Image olmayabilir LOGGER.debug("Önizleme görüntüsü kodlanamadı: %s", exc) return paket def _job_durum_guncelle(job: JobKayit, kind: str, veri: Any) -> None: """Olaya göre job state'ini güncelle.""" job.son_olay_zamani = time.time() if kind == "started": job.durum = "calisiyor" job.baslangic_zamani = time.time() job.faz = "loading_model" elif kind == "progress": if isinstance(veri, dict): job.faz = veri.get("phase") or job.faz job.ilerleme = int(veri.get("progress") or job.ilerleme) job.mesaj = veri.get("status") or "" job.su_anki_adim = veri.get("current_step") job.toplam_adim = veri.get("total_steps") elif kind == "completed": job.durum = "tamam" job.ilerleme = 100 job.bitis_zamani = time.time() if isinstance(veri, dict): job.sonuc = dict(veri) if veri.get("cancelled"): job.durum = "iptal" elif kind == "error": # Hata illa ki job'u iptal etmiyor (recoverable olabilir), # ama completed gelmediyse bunu işaretle if isinstance(veri, dict): mesaj = veri.get("message") or "" if mesaj: job.hata_mesajlari.append(mesaj) job.mesaj = mesaj if veri.get("cancelled"): job.durum = "iptal" async def _wangp_dinleyici(job: JobKayit) -> None: """WanGP'nin job.events SSE stream'ini oku ve abonelere yayınla. WanGP'nin events.iter() bloklayan bir generator olduğu için onu run_in_executor ile sürekli besleyen bir thread/loop kullanırız. """ loop = asyncio.get_running_loop() job.ana_loop = loop olay_kuyrugu: asyncio.Queue[dict[str, Any] | None] = asyncio.Queue() def _thread_okuyucu() -> None: """Background thread - WanGP queue'sundan event okur, asyncio queue'ya iter.""" try: for olay in job.wangp_job.events.iter(timeout=1.0): if olay is None: if job.wangp_job.events.closed: break continue paket = { "kind": str(getattr(olay, "kind", "info")), "data": _olay_serilestir(getattr(olay, "kind", "info"), getattr(olay, "data", None)), "ts": float(getattr(olay, "timestamp", time.time())), } try: asyncio.run_coroutine_threadsafe(olay_kuyrugu.put(paket), loop).result(timeout=5.0) except Exception as exc: LOGGER.warning("Olay kuyruğa yazılamadı: %s", exc) break except Exception as exc: LOGGER.exception("WanGP olay okuyucu thread hatası: %s", exc) finally: try: asyncio.run_coroutine_threadsafe(olay_kuyrugu.put(None), loop) except Exception: pass # Background thread başlat okuma_task = loop.run_in_executor(None, _thread_okuyucu) # Asyncio kuyruğundan paketleri tüket try: while True: paket = await olay_kuyrugu.get() if paket is None: break try: _job_durum_guncelle(job, paket["kind"], paket["data"]) job.son_olay = paket await job.aboneye_yayinla(paket) except Exception as exc: LOGGER.warning("Job durum güncelleme hatası: %s", exc) finally: # Eğer henüz tamam/iptal/hata atanmadıysa (mesela stream yarıda kesildi) if job.durum in {"kuyruk", "calisiyor"}: job.durum = "hata" if not job.wangp_job.cancel_requested else "iptal" job.bitis_zamani = time.time() son_paket = { "kind": "completed", "data": { "success": False, "generated_files": [], "errors": ["Stream kapatıldı"], "cancelled": bool(job.wangp_job.cancel_requested), }, "ts": time.time(), } await job.aboneye_yayinla(son_paket) # Aboneleri kapat for kuyruk in list(job.aboneler): try: kuyruk.put_nowait({"kind": "_kapali", "data": None, "ts": time.time()}) except Exception: pass try: await okuma_task except Exception: pass # ---------------------------------------------------------------------------- # Otomatik temizlik task'i # ---------------------------------------------------------------------------- async def _periyodik_temizlik_dongusu() -> None: """1 saatten eski tamamlanmış işleri temizle.""" while True: try: await asyncio.sleep(PERIYODIK_TEMIZLIK_ARALIGI_SN) await _eski_jobu_temizle() except asyncio.CancelledError: break except Exception as exc: LOGGER.warning("Periyodik temizlik hatası: %s", exc) async def _eski_jobu_temizle() -> int: """Süresi dolmuş tamamlanmış jobları sil. Sayıyı döndürür.""" simdi = time.time() silinen = 0 async with _jobs_lock: eskiler = [ jid for jid, job in _jobs.items() if job.durum in {"tamam", "iptal", "hata"} and (job.bitis_zamani or job.olusturma_zamani) < simdi - JOB_TEMIZLIK_SURESI_SN ] for jid in eskiler: _jobs.pop(jid, None) silinen += 1 if silinen > 0: LOGGER.info("Otomatik temizlik: %d eski job silindi.", silinen) return silinen # ---------------------------------------------------------------------------- # FastAPI uygulaması # ---------------------------------------------------------------------------- @asynccontextmanager async def yasamdongusu(uygulama: FastAPI) -> AsyncIterator[None]: LOGGER.info("Köprü servisi başlıyor.") YUKLEME_DIZINI.mkdir(parents=True, exist_ok=True) temizlik_task = asyncio.create_task(_periyodik_temizlik_dongusu()) try: yield finally: LOGGER.info("Köprü servisi kapanıyor.") temizlik_task.cancel() try: await temizlik_task except asyncio.CancelledError: pass # Aktif jobları iptal et for job in list(_jobs.values()): try: if job.durum in {"kuyruk", "calisiyor"}: job.wangp_job.cancel() except Exception: pass uygulama = FastAPI( title="Fiko Film Stüdyosu - WanGP Köprü", description="WanGP'nin in-process Python API'sini Flutter masaüstüne HTTP/SSE üzerinden açan servis.", version="1.0.0", lifespan=yasamdongusu, ) # CORS - sadece localhost, Flutter desktop için uygulama.add_middleware( CORSMiddleware, allow_origins=[ "http://localhost", "http://127.0.0.1", "http://localhost:*", "http://127.0.0.1:*", ], allow_origin_regex=r"^http://(localhost|127\.0\.0\.1)(:\d+)?$", allow_credentials=False, allow_methods=["*"], allow_headers=["*"], ) # ---------------------------------------------------------------------------- # Yardımcı: settings doğrulama # ---------------------------------------------------------------------------- def _settings_dogrula(settings: dict[str, Any]) -> None: if not isinstance(settings, dict): raise HTTPException(status_code=400, detail="settings bir sözlük olmalı.") eksik = [] if not settings.get("model_type"): eksik.append("model_type") if "prompt" not in settings: eksik.append("prompt") if eksik: raise HTTPException( status_code=400, detail=f"settings içinde zorunlu alanlar eksik: {', '.join(eksik)}", ) # ---------------------------------------------------------------------------- # Endpoint: POST /jobs - yeni iş başlat # ---------------------------------------------------------------------------- @uygulama.post("/jobs", response_model=JobYanit) async def job_olustur(istek: JobIstek) -> JobYanit: _settings_dogrula(istek.settings) try: session = await _wangp_session_al() except Exception as exc: raise HTTPException(status_code=500, detail=f"WanGP başlatılamadı: {exc}") from exc job_id = uuid.uuid4().hex loop = asyncio.get_running_loop() def _submit() -> Any: return session.submit_task(istek.settings) try: wangp_job = await loop.run_in_executor(None, _submit) except RuntimeError as exc: # WanGP "already in progress" hatası raise HTTPException(status_code=409, detail=str(exc)) from exc except Exception as exc: raise HTTPException(status_code=500, detail=f"submit_task başarısız: {exc}") from exc job = JobKayit( job_id=job_id, wangp_job=wangp_job, settings=istek.settings, etiket=istek.etiket, ) async with _jobs_lock: _jobs[job_id] = job job.dinleyici_task = asyncio.create_task(_wangp_dinleyici(job)) LOGGER.info( "Yeni job: id=%s model=%s etiket=%s", job_id, job.model_type(), istek.etiket, ) return JobYanit( job_id=job_id, durum="kuyruk", olusturma_zamani=job.olusturma_zamani, ) # ---------------------------------------------------------------------------- # Endpoint: GET /jobs - aktif/son N iş özeti # ---------------------------------------------------------------------------- @uygulama.get("/jobs", response_model=list[JobOzet]) async def job_listesi(limit: int = Query(default=50, ge=1, le=500)) -> list[JobOzet]: async with _jobs_lock: kayitlar = list(_jobs.values()) # En yeni en üstte kayitlar.sort(key=lambda j: j.olusturma_zamani, reverse=True) return [j.ozet() for j in kayitlar[:limit]] # ---------------------------------------------------------------------------- # Endpoint: GET /jobs/{id}/state - anlık durum # ---------------------------------------------------------------------------- @uygulama.get("/jobs/{job_id}/state", response_model=IlerlemeKaydi) async def job_durum(job_id: str) -> IlerlemeKaydi: async with _jobs_lock: job = _jobs.get(job_id) if job is None: raise HTTPException(status_code=404, detail="Job bulunamadı") return job.ilerleme_kaydi() # ---------------------------------------------------------------------------- # Endpoint: GET /jobs/{id}/result - tamamlanmışsa sonuç # ---------------------------------------------------------------------------- @uygulama.get("/jobs/{job_id}/result", response_model=SonucYanit) async def job_sonuc(job_id: str) -> SonucYanit: async with _jobs_lock: job = _jobs.get(job_id) if job is None: raise HTTPException(status_code=404, detail="Job bulunamadı") if job.durum not in {"tamam", "iptal", "hata"}: raise HTTPException(status_code=409, detail=f"Job henüz bitmedi (durum: {job.durum})") sonuc = job.sonuc or {} return SonucYanit( job_id=job.job_id, success=bool(sonuc.get("success", False)), generated_files=list(sonuc.get("generated_files") or []), errors=list(sonuc.get("errors") or []) + list(job.hata_mesajlari), cancelled=bool(sonuc.get("cancelled", job.durum == "iptal")), total_tasks=int(sonuc.get("total_tasks") or 0), successful_tasks=int(sonuc.get("successful_tasks") or 0), failed_tasks=int(sonuc.get("failed_tasks") or 0), ) # ---------------------------------------------------------------------------- # Endpoint: POST /jobs/{id}/cancel # ---------------------------------------------------------------------------- @uygulama.post("/jobs/{job_id}/cancel", response_model=IptalYanit) async def job_iptal(job_id: str) -> IptalYanit: async with _jobs_lock: job = _jobs.get(job_id) if job is None: raise HTTPException(status_code=404, detail="Job bulunamadı") if job.durum in {"tamam", "iptal", "hata"}: return IptalYanit(ok=False, mesaj=f"Job zaten bitmiş (durum: {job.durum})") try: job.wangp_job.cancel() except Exception as exc: raise HTTPException(status_code=500, detail=f"Cancel hatası: {exc}") from exc LOGGER.info("Job iptal istendi: %s", job_id) return IptalYanit(ok=True, mesaj="İptal sinyali gönderildi.") # ---------------------------------------------------------------------------- # Endpoint: DELETE /jobs/{id} # ---------------------------------------------------------------------------- @uygulama.delete("/jobs/{job_id}", response_model=SilYanit) async def job_sil(job_id: str) -> SilYanit: async with _jobs_lock: job = _jobs.pop(job_id, None) if job is None: raise HTTPException(status_code=404, detail="Job bulunamadı") if job.durum in {"kuyruk", "calisiyor"}: try: job.wangp_job.cancel() except Exception: pass return SilYanit(ok=True, mesaj="Job silindi.") # ---------------------------------------------------------------------------- # Endpoint: GET /jobs/{id}/events (SSE) # ---------------------------------------------------------------------------- @uygulama.get("/jobs/{job_id}/events") async def job_olaylari(job_id: str, request: Request) -> StreamingResponse: async with _jobs_lock: job = _jobs.get(job_id) if job is None: raise HTTPException(status_code=404, detail="Job bulunamadı") abone_kuyrugu: asyncio.Queue[dict[str, Any]] = asyncio.Queue(maxsize=1024) job.aboneler.append(abone_kuyrugu) # Eğer zaten gelen son olay varsa hemen başta gönder (geç gelen istemci için) if job.son_olay is not None: try: abone_kuyrugu.put_nowait(job.son_olay) except Exception: pass async def _yayim() -> AsyncIterator[bytes]: son_ping = time.time() try: while True: if await request.is_disconnected(): break # 5 saniye timeout - keepalive ping yapmak için try: paket = await asyncio.wait_for(abone_kuyrugu.get(), timeout=5.0) except asyncio.TimeoutError: paket = None simdi = time.time() if paket is not None: if paket.get("kind") == "_kapali": # Stream sonu kapanis = json.dumps({"kind": "completed", "data": {"closed": True}, "ts": simdi}) yield f"data: {kapanis}\n\n".encode("utf-8") break veri = json.dumps(paket, ensure_ascii=False, default=str) yield f"data: {veri}\n\n".encode("utf-8") son_ping = simdi # Keepalive (15 sn) if simdi - son_ping >= 15.0: yield b": ping\n\n" son_ping = simdi # İşin bittiyse ve queue boşsa stream'i kapatabiliriz if job.durum in {"tamam", "iptal", "hata"} and abone_kuyrugu.empty(): # Job kapanmış, son durumu da gönderdiysek çık await asyncio.sleep(0.1) if abone_kuyrugu.empty(): break finally: try: job.aboneler.remove(abone_kuyrugu) except ValueError: pass basliklar = { "Cache-Control": "no-cache", "Connection": "keep-alive", "X-Accel-Buffering": "no", } return StreamingResponse(_yayim(), media_type="text/event-stream", headers=basliklar) # ---------------------------------------------------------------------------- # Endpoint: GET /file - dosya servis et # ---------------------------------------------------------------------------- @uygulama.get("/file") async def dosya_oku(path: str = Query(..., description="Tam dosya yolu")) -> StreamingResponse: p = Path(path) if not p.is_file(): raise HTTPException(status_code=404, detail="Dosya bulunamadı") icerik_tipi, _ = mimetypes.guess_type(str(p)) if not icerik_tipi: icerik_tipi = "application/octet-stream" async def _akis() -> AsyncIterator[bytes]: chunk_size = 64 * 1024 loop = asyncio.get_running_loop() with p.open("rb") as h: while True: veri = await loop.run_in_executor(None, h.read, chunk_size) if not veri: break yield veri basliklar = { "Content-Disposition": f'inline; filename="{p.name}"', "Content-Length": str(p.stat().st_size), } return StreamingResponse(_akis(), media_type=icerik_tipi, headers=basliklar) # ---------------------------------------------------------------------------- # Endpoint: POST /upload - multipart dosya yükle # ---------------------------------------------------------------------------- @uygulama.post("/upload", response_model=YuklemeYanit) async def dosya_yukle(file: UploadFile = File(...)) -> YuklemeYanit: YUKLEME_DIZINI.mkdir(parents=True, exist_ok=True) isim_guvenli = (file.filename or "dosya").replace("..", "_").replace("/", "_").replace("\\", "_") if not isim_guvenli: isim_guvenli = "dosya" benzersiz = f"{int(time.time() * 1000)}_{uuid.uuid4().hex[:8]}_{isim_guvenli}" hedef = YUKLEME_DIZINI / benzersiz boyut = 0 try: with hedef.open("wb") as h: while True: parca = await file.read(1024 * 1024) if not parca: break h.write(parca) boyut += len(parca) finally: await file.close() LOGGER.info("Dosya yüklendi: %s (%d bayt)", hedef.name, boyut) return YuklemeYanit( path=str(hedef.resolve()), isim=isim_guvenli, boyut_byte=boyut, icerik_tipi=str(file.content_type or ""), ) # ---------------------------------------------------------------------------- # Endpoint: GET /modeller - WanGP defaults klasörü # ---------------------------------------------------------------------------- def _model_tipi_tahmin(json_payload: dict[str, Any], dosya_adi: str) -> str: """JSON içeriğine ve dosya adına bakarak model tipini belirle.""" model = json_payload.get("model") or {} arch = str(model.get("architecture") or "").lower() image_outputs = bool(model.get("image_outputs")) video_outputs = bool(model.get("video_outputs")) audio_outputs = bool(model.get("audio_outputs")) nm = (dosya_adi + " " + str(model.get("name") or "")).lower() if image_outputs: return "image" if audio_outputs or arch in {"chatterbox", "ace_step", "f5_tts"} or "tts" in nm or "ace_step" in nm: return "audio_tts" if video_outputs or arch in {"hunyuan", "wan", "ltx", "wan2gp"} or "video" in nm or "_t2v" in nm or "_i2v" in nm: if "_i2v" in nm or "i2v" in arch: return "video_i2v" if "lipsync" in nm or "talk" in nm or "animate" in nm: return "video_lipsync" return "video_t2v" if "upsamp" in nm or "post" in nm or "refine" in nm: return "postprocess" # Varsayılan: image (FLUX ailesi gibi) if "flux" in arch or "qwen" in arch or "sd" in arch or "image" in nm: return "image" return "diger" @uygulama.get("/modeller", response_model=list[ModelKaydi]) async def modeller_listesi() -> list[ModelKaydi]: if WANGP_ROOT is None: raise HTTPException(status_code=500, detail="WanGP kök dizini ayarlı değil") defaults_dir = WANGP_ROOT / "defaults" if not defaults_dir.is_dir(): raise HTTPException(status_code=500, detail=f"defaults klasörü yok: {defaults_dir}") sonuc: list[ModelKaydi] = [] for json_dosya in sorted(defaults_dir.glob("*.json")): try: with json_dosya.open("r", encoding="utf-8") as h: payload = json.load(h) except Exception as exc: LOGGER.warning("JSON okunamadı %s: %s", json_dosya.name, exc) continue if not isinstance(payload, dict): continue model = payload.get("model") or {} if not isinstance(model, dict): continue model_id = json_dosya.stem tip = _model_tipi_tahmin(payload, json_dosya.stem) sonuc.append( ModelKaydi( id=model_id, ad=str(model.get("name") or model_id), tip=tip, # type: ignore[arg-type] aciklama=str(model.get("description") or ""), architecture=str(model.get("architecture") or ""), resolution=payload.get("resolution"), video_length=payload.get("video_length"), num_inference_steps=payload.get("num_inference_steps"), guidance_scale=payload.get("guidance_scale"), flow_shift=payload.get("flow_shift"), image_outputs=bool(model.get("image_outputs")), video_outputs=bool(model.get("video_outputs")), audio_outputs=bool(model.get("audio_outputs")), lock_inference_steps=bool(model.get("lock_inference_steps")), urls=[str(u) for u in (model.get("URLs") or []) if u], ) ) return sonuc # ---------------------------------------------------------------------------- # Endpoint: GET /loralar # ---------------------------------------------------------------------------- @uygulama.get("/loralar", response_model=list[LoraKaydi]) async def loralar_listesi() -> list[LoraKaydi]: if WANGP_ROOT is None: raise HTTPException(status_code=500, detail="WanGP kök dizini ayarlı değil") loras_dir = WANGP_ROOT / "loras" metadata_dir = WANGP_ROOT / "loras_metadata" if not loras_dir.is_dir(): return [] # Tüm metadata JSON'ları okuyup dosya adı -> meta map'ı oluştur meta_map: dict[str, dict[str, Any]] = {} if metadata_dir.is_dir(): for json_dosya in metadata_dir.rglob("*.json"): try: with json_dosya.open("r", encoding="utf-8") as h: veri = json.load(h) if isinstance(veri, dict): # metadata dosyalarında ya tek lora ya da liste olabilir if "filename" in veri or "name" in veri: anahtar = str(veri.get("filename") or veri.get("name") or json_dosya.stem) meta_map[anahtar.lower()] = veri except Exception: continue sonuc: list[LoraKaydi] = [] for safetensors in sorted(loras_dir.rglob("*.safetensors")): rel_klasor = safetensors.parent.relative_to(loras_dir) klasor_str = "" if str(rel_klasor) == "." else str(rel_klasor) anahtar = safetensors.stem.lower() meta = meta_map.get(anahtar) or meta_map.get(safetensors.name.lower()) or {} try: boyut = safetensors.stat().st_size except Exception: boyut = 0 tetik = meta.get("trigger_words") or meta.get("activation") or meta.get("triggers") or [] if isinstance(tetik, str): tetik = [t.strip() for t in tetik.split(",") if t.strip()] agirlik = meta.get("recommended_weight") or meta.get("default_weight") or meta.get("weight") try: agirlik_f = float(agirlik) if agirlik is not None else None except (TypeError, ValueError): agirlik_f = None sonuc.append( LoraKaydi( dosya=str(safetensors.resolve()), ad=safetensors.stem, klasor=klasor_str, boyut_byte=boyut, tetik_kelimeler=[str(t) for t in tetik] if isinstance(tetik, list) else [], onerilen_agirlik=agirlik_f, aciklama=str(meta.get("description") or ""), ) ) return sonuc # ---------------------------------------------------------------------------- # Endpoint: GET /saglik # ---------------------------------------------------------------------------- def _gpu_bilgisi() -> GpuBilgi | None: try: import torch # type: ignore except Exception: return None if not torch.cuda.is_available(): return None try: idx = torch.cuda.current_device() ad = torch.cuda.get_device_name(idx) toplam = float(torch.cuda.get_device_properties(idx).total_memory) / (1024 ** 2) kullanim = float(torch.cuda.memory_allocated(idx)) / (1024 ** 2) return GpuBilgi( ad=str(ad), vram_kullanim_mb=round(kullanim, 1), vram_toplam_mb=round(toplam, 1), ) except Exception: return None @uygulama.get("/saglik", response_model=SaglikYanit) async def saglik() -> SaglikYanit: aktif = sum(1 for j in _jobs.values() if j.durum in {"kuyruk", "calisiyor"}) toplam = len(_jobs) model_aktif: str | None = None try: if _wangp_session is not None: wangp_api = _wangp_modul_yukle() modul = getattr(wangp_api, "_RUNTIME", None) if modul is not None and hasattr(modul, "module"): m = modul.module ad = getattr(m, "transformer_filename", None) or getattr(m, "model_filename", None) if ad: model_aktif = str(Path(str(ad)).name) except Exception: pass return SaglikYanit( durum="ok", wangp_yuklu=_wangp_session is not None, model_aktif=model_aktif, aktif_job_sayisi=aktif, toplam_job_sayisi=toplam, gpu=_gpu_bilgisi(), ) # ---------------------------------------------------------------------------- # Endpoint: GET / - basit ana sayfa # ---------------------------------------------------------------------------- @uygulama.get("/") async def kok() -> dict[str, Any]: return { "servis": "Fiko Film Stüdyosu - WanGP Köprü", "versiyon": "1.0.0", "wangp_root": str(WANGP_ROOT) if WANGP_ROOT else None, "wangp_yuklu": _wangp_session is not None, "init_hatasi": _wangp_init_hatasi, "endpoint_listesi": [ "POST /jobs", "GET /jobs", "GET /jobs/{id}/state", "GET /jobs/{id}/result", "GET /jobs/{id}/events (SSE)", "POST /jobs/{id}/cancel", "DELETE /jobs/{id}", "GET /file?path=...", "POST /upload (multipart)", "GET /modeller", "GET /loralar", "GET /saglik", ], } # ---------------------------------------------------------------------------- # Hata yakalayıcı # ---------------------------------------------------------------------------- @uygulama.exception_handler(Exception) async def beklenmedik_hata(_: Request, exc: Exception) -> JSONResponse: LOGGER.exception("Beklenmedik hata: %s", exc) return JSONResponse( status_code=500, content={"detail": f"Sunucu hatası: {exc}"}, ) # ---------------------------------------------------------------------------- # Programatik başlatma için ayar fonksiyonu (baslat.py'den çağrılır) # ---------------------------------------------------------------------------- def ayarlari_kur( wangp_root: str | os.PathLike, cikti_dizini: str | os.PathLike | None = None, yukleme_dizini: str | os.PathLike | None = None, ) -> None: """baslat.py içinden çağrılır - global yolları ayarlar.""" global WANGP_ROOT, WANGP_OUTPUT_DIR, YUKLEME_DIZINI WANGP_ROOT = Path(wangp_root).resolve() if cikti_dizini is not None: WANGP_OUTPUT_DIR = Path(cikti_dizini).resolve() if yukleme_dizini is not None: YUKLEME_DIZINI = Path(yukleme_dizini).resolve() LOGGER.info("Yapılandırma: WANGP_ROOT=%s OUTPUT=%s UPLOADS=%s", WANGP_ROOT, WANGP_OUTPUT_DIR, YUKLEME_DIZINI)