"""ACE-Step V1.5 REST API istemcisi (port 8001). Endpointler: - POST /release_task → task_id döner - POST /query_result → task durumunu sorgular - GET /v1/audio → üretilen ses dosyasını indirir - GET /health → durum kontrolü """ from __future__ import annotations from pathlib import Path from typing import Optional, Dict, Any, Callable import json import time import urllib.parse import urllib.request import urllib.error class ACEStepClient: """ACE-Step API'yi çağıran hafif istemci.""" def __init__(self, api_url: str = "http://127.0.0.1:8001", api_key: str = ""): self.api_url = api_url.rstrip("/") self.api_key = api_key # ------------------------------------------------------------- yardım def _istek(self, yol: str, method: str = "GET", veri: Optional[Dict[str, Any]] = None, timeout: int = 30, ham: bool = False): url = f"{self.api_url}{yol}" headers = {"Content-Type": "application/json"} if self.api_key: headers["Authorization"] = f"Bearer {self.api_key}" body = json.dumps(veri).encode("utf-8") if veri is not None else None req = urllib.request.Request(url, data=body, headers=headers, method=method) with urllib.request.urlopen(req, timeout=timeout) as r: data = r.read() if ham: return data return json.loads(data.decode("utf-8")) # --------------------------------------------------------- durum kontrol def saglikli_mi(self) -> bool: try: self._istek("/health", timeout=3) return True except Exception: return False # ------------------------------------------------------------- üretim def release_task(self, caption: str, lirikler: str = "", sure: int = 120, guidance_scale: float = 15.0, infer_step: int = 27, seed: int = -1, gorev: str = "text2music", ref_audio: str = "", src_audio: str = "", repaint_bas: float = 0.0, repaint_bit: Optional[float] = None, thinking: bool = False, ek_parametreler: Optional[Dict[str, Any]] = None) -> str: """Yeni üretim görevi başlat. task_id döner.""" payload: Dict[str, Any] = { "global_caption": caption, "prompt": caption, "lyrics": lirikler, "audio_duration": float(sure), "guidance_scale": guidance_scale, "inference_steps": infer_step, "seed": seed, "use_random_seed": seed == -1, "task_type": gorev, "thinking": thinking, "audio_format": "mp3", } if ref_audio: payload["reference_audio_path"] = str(Path(ref_audio).resolve()) if src_audio: payload["src_audio_path"] = str(Path(src_audio).resolve()) if repaint_bit is not None: payload["repainting_start"] = repaint_bas payload["repainting_end"] = repaint_bit if ek_parametreler: payload.update(ek_parametreler) sonuc = self._istek("/release_task", method="POST", veri=payload) task_id = sonuc.get("task_id") or sonuc.get("data", {}).get("task_id", "") if not task_id: raise RuntimeError(f"Task ID alınamadı: {json.dumps(sonuc)[:300]}") return task_id # ACE-Step durum kodları (server_utils.STATUS_MAP): # 0 = queued/running, 1 = succeeded, 2 = failed DURUM_BEKLIYOR = 0 DURUM_BASARILI = 1 DURUM_BASARISIZ = 2 def query_result(self, task_id: str) -> Dict[str, Any]: """Görev durumunu sorgula. Parse edilmiş tek sonuç item'ını döner.""" govde: Dict[str, Any] = {"task_id_list": [task_id]} if self.api_key: govde["ai_token"] = self.api_key sonuc = self._istek("/query_result", method="POST", veri=govde) # wrap_response: {..., "data": [...]} veya direkt {"data": [...]} veri = sonuc.get("data", sonuc) if isinstance(veri, list) and veri: item = veri[0] elif isinstance(veri, dict): item = veri else: return {"status": self.DURUM_BEKLEMEDE, "result": [], "task_id": task_id} # result: JSON string olarak gelir → listeye parse et ham = item.get("result", "[]") try: ic_liste = json.loads(ham) if isinstance(ham, str) else ham except Exception: ic_liste = [] item["result_parsed"] = ic_liste if isinstance(ic_liste, list) else [] return item def bekle_ve_indir(self, task_id: str, hedef_dosya: str | Path, timeout: int = 900, poll: float = 2.0, ilerleme_cb: Optional[Callable[[str, float], None]] = None) -> Optional[str]: """Görev tamamlanana kadar bekle ve sesi indir. Dosya yolu döner.""" bas = time.time() son_asama = "" son_yuzde = -1.0 while time.time() - bas < timeout: try: d = self.query_result(task_id) except Exception as e: if ilerleme_cb: ilerleme_cb(f"Sorgu hatası: {e}", 0.0) time.sleep(poll) continue durum = int(d.get("status", 0) or 0) ic = d.get("result_parsed") or [] ilk = ic[0] if ic and isinstance(ic[0], dict) else {} asama = str(ilk.get("stage") or d.get("progress_text") or "") yuzde_ham = ilk.get("progress", 0) try: yuzde_v = float(yuzde_ham or 0) except (TypeError, ValueError): yuzde_v = 0.0 # Bazı sürümler 0-1 aralığında, bazıları 0-100 döner yuzde = yuzde_v * 100.0 if yuzde_v <= 1.0 else yuzde_v if (asama != son_asama or abs(yuzde - son_yuzde) > 1.0) and ilerleme_cb: ilerleme_cb(asama or "çalışıyor", yuzde) son_asama, son_yuzde = asama, yuzde if durum == self.DURUM_BASARILI: ses_yolu = ilk.get("file", "") or ilk.get("audio_path", "") if ses_yolu: return self._ses_indir(ses_yolu, hedef_dosya) raise RuntimeError(f"Başarılı ama dosya yolu boş: {json.dumps(d)[:300]}") if durum == self.DURUM_BASARISIZ: hata = ilk.get("error") or d.get("progress_text") or "bilinmeyen hata" raise RuntimeError(f"Üretim başarısız: {hata}") time.sleep(poll) raise TimeoutError(f"ACE-Step zaman aşımı ({timeout}s)") def _ses_indir(self, uzak_yol: str, hedef: str | Path) -> str: """Sunucudan ses dosyasını indir.""" hedef = Path(hedef) hedef.parent.mkdir(parents=True, exist_ok=True) if uzak_yol.startswith(("http://", "https://")): url = uzak_yol elif uzak_yol.startswith("/v1/audio"): url = f"{self.api_url}{uzak_yol}" else: encoded = urllib.parse.quote(uzak_yol, safe="") url = f"{self.api_url}/v1/audio?path={encoded}" headers = {} if self.api_key: headers["Authorization"] = f"Bearer {self.api_key}" req = urllib.request.Request(url, headers=headers) with urllib.request.urlopen(req, timeout=120) as r: hedef.write_bytes(r.read()) return str(hedef)