"""xAI 직접 이미지 편집(i2i) 클라이언트 — `/v1/images/edits` (2026-09-18). ## 왜 이 경로가 생겼나 (컨트리로드 실측) 콘티 변환은 OpenRouter 의 chat 경로로 `x-ai/grok-imagine-image-quality` 를 불렀다. 2026-09-18 14:16 까지는 되다가 14:18 부터 **전부 404** 가 났다: OpenRouter x-ai/grok-imagine-image-quality → 404 "No endpoints found that support the requested output modalities: image" OpenRouter x-ai/grok-imagine-image → 400 "not a valid model ID" xAI 직접 /v1/images/generations → 200 (그러나 넣은 원본을 **무시**) xAI 직접 /v1/images/edits → 200 · 원본 유지(구조 차이 17~27, 글만 그리면 119 · 무작위 ~128) 즉 grok 이미지는 살아 있고 **xAI 직접 편집 경로**로만 닿는다. 문서상 최신은 `grok-imagine-image-2.0`(그리고 `-quality` 는 2026-11-02 은퇴 예정)인데, 이 계정은 둘 다 404("이 팀에 접근 권한이 없다")라 지금 쓸 수 있는 모델은 `grok-imagine-image` 하나다. 접근이 열리면 `XAI_IMAGE_MODEL` 만 바꾸면 된다. ★`quality` 는 네 값 모두 200 으로 받는다 — 사용자 지시로 `high` 를 보낸다. 눈으로 더 좋아졌는지는 사람이 판단한다(이 파일이 주장하지 않는다). ★헤더를 안 갖춘 요청은 Cloudflare 가 403(1010)으로 끊는다 — httpx 로 보낸다. """ from __future__ import annotations import base64 import logging import time from typing import Dict, List, Optional, Tuple from app.core.config import settings as _settings from app.core.image_call_budget import reserve_current_call from app.modules.llm.gemini_image_client import GeminiImageClient from app.modules.llm.image_format import ensure_png_bytes from app.modules.llm.image_moderation import is_moderation_text from app.modules.llm.image_send_state import classify_http_status from app.modules.llm.llm_logger import log_llm_call from app.services.image_capture.sink import capture_generated_image logger = logging.getLogger(__name__) XAI_EDITS_URL = "https://api.x.ai/v1/images/edits" class XaiImageClient(GeminiImageClient): """cine 슬롯 모양(`generate_image`)으로 xAI 편집 API 를 부른다.""" def __init__(self, api_key: str = None, model: str = None) -> None: super().__init__( api_key=api_key, model=model or str(getattr(_settings, "xai_image_model", "") or ""), ) def _get_api_key(self) -> str: if self._fixed_api_key: return self._fixed_api_key key = (getattr(_settings, "xai_api_key", "") or "").strip() if not key: raise RuntimeError("XAI_API_KEY 가 설정에 없다") return key def generate_image( self, prompt: str, reference_images: Optional[List[bytes]] = None, aspect_ratio: str = "16:9", labeled_references: Optional[List[Tuple[str, bytes]]] = None, ) -> Tuple[bytes, int]: """원본 1장을 고쳐 (PNG bytes, ms) 로 돌려준다. ★원본이 없으면 **선다** — 편집 API 에 원본 없이 보내면 그냥 새 그림을 그려 주는데(실측), 그것은 콘티 변환이 아니다. 조용히 다른 일을 하느니 드러나게 실패한다. """ import httpx refs = [b for _lbl, b in (labeled_references or [])] or list( reference_images or []) if not refs: raise RuntimeError( "xAI 편집 경로에 원본 이미지가 없다 — 변환할 대상이 있어야 한다") ref_labels = [lbl for lbl, _b in (labeled_references or []) if lbl] body: Dict[str, object] = { "model": self._model, "prompt": prompt, "image": { "url": "data:image/png;base64," + base64.b64encode(refs[0]).decode("ascii"), "type": "image_url", }, } quality = str(getattr(_settings, "xai_image_quality", "") or "") if quality: body["quality"] = quality from app.modules.llm.image_tracer import get_image_tracer _tracer = get_image_tracer() timeout = _settings.llm_timeout_image_gen max_retries = _settings.llm_max_retries start_time = time.monotonic() last_error: Exception | None = None for attempt in range(1, max_retries + 2): try: reserve_current_call(source="xai_image_client.edits") with httpx.Client(timeout=timeout) as client: r = client.post( XAI_EDITS_URL, json=body, headers={ "Authorization": f"Bearer {self._get_api_key()}", "Content-Type": "application/json", }, ) if r.status_code >= 400: text = r.text[:500] # ★검열은 다시 보내지 않는다 — 요금만 나간다. if is_moderation_text(text): raise RuntimeError(f"moderation blocked: {text}") verdict = classify_http_status(r.status_code, via="xai") if verdict == "retryable" and attempt <= max_retries: last_error = RuntimeError( f"xAI image API error {r.status_code}: {text}") logger.warning( "xAI 이미지 편집 %d — 재시도 %d/%d: %s", r.status_code, attempt, max_retries, text[:120]) time.sleep(min(2 ** attempt, 20)) continue raise RuntimeError( f"xAI image API error {r.status_code}: {text}") payload = r.json() item = (payload.get("data") or [{}])[0] if item.get("b64_json"): raw = base64.b64decode(item["b64_json"]) elif item.get("url"): with httpx.Client(timeout=timeout) as client: raw = client.get(item["url"]).content else: raise RuntimeError( f"xAI image API 응답에 이미지가 없다: {str(payload)[:200]}") elapsed_ms = int((time.monotonic() - start_time) * 1000) call_id = log_llm_call( model_name=self._model, user_prompt=prompt, output_text="[image edited]", status="success", duration_ms=elapsed_ms, reference_image_ids=ref_labels, **self._log_ctx, ) _tracer.log( step=self._trace_step(), model=self._model, prompt=prompt, ref_image_ids=ref_labels, output_image_id=self._ctx.get("image_id"), duration_ms=elapsed_ms, params={"backend": "xai", "endpoint": "images/edits", "quality": quality}, extra_metadata=self._opik_meta, ) png = ensure_png_bytes( raw, context=f"{self._model}/" f"{self._ctx.get('operation_type') or '?'}", ) _capture_meta = { k: self._ctx.get(k) for k in ("project_id", "episode_id", "operation_type", "step", "scene_index", "shot_index", "still_id", "entity_id", "image_id") if self._ctx.get(k) is not None } if ref_labels: _capture_meta["reference_labels"] = ref_labels capture_generated_image( png, role=self._ctx.get("operation_type") or "xai_image_edit", prompt=prompt, generation_call_id=call_id, pipeline_metadata=_capture_meta, ) return png, elapsed_ms except RuntimeError: raise except Exception as exc: # 네트워크 계열 — 사다리를 탄다 last_error = exc if attempt <= max_retries: logger.warning("xAI 이미지 편집 실패 — 재시도 %d/%d: %s", attempt, max_retries, str(exc)[:120]) time.sleep(min(2 ** attempt, 20)) continue break elapsed_ms = int((time.monotonic() - start_time) * 1000) log_llm_call( model_name=self._model, user_prompt=prompt, status="error", error_message=str(last_error)[:500], duration_ms=elapsed_ms, reference_image_ids=ref_labels, **self._log_ctx, ) raise RuntimeError(f"xAI image API 실패: {last_error}")