diff --git a/services/image-render/providers/nano_banana.py b/services/image-render/providers/nano_banana.py new file mode 100644 index 0000000..f932f6c --- /dev/null +++ b/services/image-render/providers/nano_banana.py @@ -0,0 +1,52 @@ +"""Nano Banana — Gemini 2.5 Flash Image (generativelanguage API). + +POST /v1beta/models/{MODEL}:generateContent +→ candidates[0].content.parts[*].inlineData.data (b64 png) +""" +from __future__ import annotations + +import logging, os +import requests + +from nas_client import webhook_update_task +from providers._media import save_b64_png + +logger = logging.getLogger(__name__) +GEMINI_BASE = "https://generativelanguage.googleapis.com/v1beta" +DEFAULT_MODEL = "gemini-2.5-flash-image" + + +def _extract_b64(data: dict): + for cand in data.get("candidates", []): + for part in cand.get("content", {}).get("parts", []): + inline = part.get("inlineData") or part.get("inline_data") + if inline and inline.get("data"): + return inline["data"] + return None + + +def run_nano_banana_generation(task_id: str, params: dict) -> None: + try: + if not os.getenv("GEMINI_API_KEY"): + webhook_update_task(task_id, "failed", 0, "", error="GEMINI_API_KEY 미설정 (Windows .env)") + return + webhook_update_task(task_id, "processing", 10, "Nano Banana (Gemini) 호출 중...") + model_id = params.get("model") or DEFAULT_MODEL + body = {"contents": [{"parts": [{"text": params["prompt"]}]}]} + resp = requests.post( + f"{GEMINI_BASE}/models/{model_id}:generateContent", + headers={"x-goog-api-key": os.getenv("GEMINI_API_KEY"), "Content-Type": "application/json"}, + json=body, timeout=120, + ) + if resp.status_code != 200: + webhook_update_task(task_id, "failed", 0, "", error=f"Gemini {resp.status_code}: {resp.text[:200]}") + return + b64 = _extract_b64(resp.json()) + if not b64: + webhook_update_task(task_id, "failed", 0, "", error="Gemini 응답에 이미지 없음") + return + url = save_b64_png(task_id, b64) + webhook_update_task(task_id, "succeeded", 100, "완료", image_url=url) + except Exception as e: + logger.exception("nano_banana task=%s 실패", task_id) + webhook_update_task(task_id, "failed", 0, "", error=str(e)) diff --git a/services/image-render/tests/test_nano_banana.py b/services/image-render/tests/test_nano_banana.py new file mode 100644 index 0000000..b57e10f --- /dev/null +++ b/services/image-render/tests/test_nano_banana.py @@ -0,0 +1,25 @@ +import providers.nano_banana as nb + +def test_missing_key_reports_failed(monkeypatch): + monkeypatch.delenv("GEMINI_API_KEY", raising=False) + calls = [] + monkeypatch.setattr(nb, "webhook_update_task", lambda *a, **k: calls.append((a, k))) + nb.run_nano_banana_generation("t1", {"prompt": "a cat"}) + assert calls[-1][0][1] == "failed" + +def test_success_extracts_inline_data(monkeypatch): + monkeypatch.setenv("GEMINI_API_KEY", "g-test") + calls = [] + monkeypatch.setattr(nb, "webhook_update_task", lambda *a, **k: calls.append((a, k))) + monkeypatch.setattr(nb, "save_b64_png", lambda tid, b64: "/media/image/t1.png") + + class FakeResp: + status_code = 200 + def json(self): + return {"candidates": [{"content": {"parts": [ + {"inlineData": {"mimeType": "image/png", "data": "ZmFrZQ=="}} + ]}}]} + monkeypatch.setattr(nb.requests, "post", lambda *a, **k: FakeResp()) + + nb.run_nano_banana_generation("t1", {"prompt": "a cat"}) + assert [c for c in calls if c[0][1] == "succeeded"]