zio/kintex
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
9987d958db
kintex/tools/nanobanana/worker.py
zio 9987d958db feat(v2.0): 자동전시시스템 하네스 재구성 + PLANNING v2.0 + 백엔드 스캐폴드 + 평면도 자산 
PLANNING v2.0: 자동전시시스템 재정의, 6역할 웹/모바일 분리, 도메인 M10~M18(옥션 M15·관람객 M10·마케팅/공개사이트 M12·BI M16(운영사 ROI 포함)·CMS M17·관리자 M18), WISE/UIWS 공통·시스템관리 레이어 §5B(2FA OTP), 옥션/견적서/역경매, M2/M3 3안 생성→선택/병합. design.md v1.1(Stitch 정합·SCR 매핑).

하네스 재구성: 전문 에이전트 17종(아키텍트 AA·SA·TA·DA·NA / 공통 common-dev / 코어 backend·frontend·db / 도메인 bidding·visitor·cms·bi·admin / AI ai-dev·visualizer / QA·devops) + kintex-impl-orchestrator v2.0(Phase A~E) + IMPLEMENTATION_BACKLOG v2.0 + CLAUDE.md.

백엔드 스캐폴드: Spring Boot 3.2.5(com.zioinfo.kintex) + JWT/RBAC + WebSocket + 버전형 룰엔진 + M2~M5 컨트롤러(M3 사전검증·M4 견적·M5 RenderJob 실구현, 공간경로 501 대기). compileJava SUCCESS.

자산: 평면도 JPG 15장 + 매니페스트(R4 트렌치·CAD 해소). CAD 93MB·build는 gitignore.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-07-11 17:32:40 +09:00

496 lines
20 KiB
Python
Raw Blame History

"""나노바나나 RenderJob 워커 사이드카 (스켈레톤).
Spring Boot 백엔드가 Redis 큐에 넣은 RenderJob 을 소비해, PLANNING §6-2 scene 을
사실화 사진(S1~S5,S7 생성형) 또는 배선 오버레이(S6 래스터 합성)로 렌더하고,
결과 이미지 + 메타데이터를 오브젝트 스토리지에 적재한 뒤 완료 이벤트를 발행한다.
설계 기준:
- PLANNING §6(나노바나나 파이프라인 v1.2) · §8(Python 워커 사이드카 아키텍처).
- IMPLEMENTATION_BACKLOG S-4(워커 골격) / M5-2(생성) / M5-3(S6 래스터).
- 계약 단일 출처: `_workspace/01_worker_contract.md`.
- 호출 스택·프롬프트·방어 로직은 `client.py`(ReRoomAI 검증 패턴) 재사용.
핵심 성질(운영 불변식):
1. G1 게이트 — 실 Gemini 호출은 env `NANOBANANA_LIVE=1` + `GEMINI_API_KEY` 있을 때만.
기본은 목/degraded 모드(키·네트워크 없이 플레이스홀더 이미지 + 정상 메타데이터).
2. 지연 연결 — Redis 미기동이어도 import·process_job() 직접 호출은 성립.
3. S6 은 생성형 아님 — 항상 로컬 PIL 결정적 래스터 합성(좌표 정합).
4. 비밀 미노출 — 키/IP/스택트레이스를 이벤트·로그·에러에 기록하지 않는다.
CLI:
python -m tools.nanobanana.worker # 큐 소비 루프(BLPOP)
python -m tools.nanobanana.worker --smoke # 무네트워크 스모크(목 잡 1건 처리)
"""
from __future__ import annotations
import argparse
import base64
import json
import os
import sys
from dataclasses import dataclass, field
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Optional
# ----------------------------------------------------------------------------
# client.py 재사용 — 모듈 실행(`-m`)·직접 실행 양쪽 지원
# ----------------------------------------------------------------------------
try: # 패키지 컨텍스트 (python -m tools.nanobanana.worker)
from .client import ( # type: ignore
MODEL_NAME,
SHOT_PRESETS,
WIRING_COLORS,
GeneratedImage,
NanoBananaAuthError,
NanoBananaError,
NanoBananaQuotaError,
NanoBananaSafetyError,
build_booth_prompt,
build_metadata,
render_wiring_overlay_raster,
)
except ImportError: # 스크립트 컨텍스트 (python worker.py) — 동일 디렉터리 임포트
sys.path.insert(0, str(Path(__file__).resolve().parent))
from client import ( # type: ignore
MODEL_NAME,
SHOT_PRESETS,
WIRING_COLORS,
GeneratedImage,
NanoBananaAuthError,
NanoBananaError,
NanoBananaQuotaError,
NanoBananaSafetyError,
build_booth_prompt,
build_metadata,
render_wiring_overlay_raster,
)
# ----------------------------------------------------------------------------
# 환경 설정 (계약 §7)
# ----------------------------------------------------------------------------
REDIS_URL = os.environ.get("REDIS_URL", "redis://localhost:6379/0")
QUEUE_KEY = os.environ.get("NANOBANANA_QUEUE", "kintex:renderjobs")
EVENT_CHANNEL = os.environ.get("NANOBANANA_EVENT_CHANNEL", "kintex:renderjob:events")
OUTPUT_DIR = os.environ.get("NANOBANANA_OUTPUT_DIR", "output/visualizations")
#: BLPOP 폴링 간격(초) — None 무한 대기 대신 짧은 폴링으로 graceful stop 가능.
BLPOP_TIMEOUT_S = 5
def is_live() -> bool:
"""G1 게이트: 실 Gemini 호출 조건(PLANNING R12).
NANOBANANA_LIVE=1 그리고 GEMINI_API_KEY 가 있을 때만 True.
그 외에는 목/degraded 모드(기본). 키 값 자체는 참조만 하고 반환·로그하지 않는다.
"""
flag = os.environ.get("NANOBANANA_LIVE", "").strip().lower() in ("1", "true", "yes", "on")
has_key = bool(os.environ.get("GEMINI_API_KEY"))
return flag and has_key
# ----------------------------------------------------------------------------
# 목/degraded 플레이스홀더 이미지 (키·네트워크 불필요)
# ----------------------------------------------------------------------------
#: PIL 부재 시 폴백용 1x1 회색 PNG(유효 최소 이미지).
_MINIMAL_PNG = base64.b64decode(
"iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mNk+M9QDwADhgGAWjR9awAAAABJRU5ErkJggg=="
)
def _placeholder_png(lines: list[str], size: tuple[int, int] = (1024, 640)) -> tuple[bytes, str]:
"""플레이스홀더 PNG 생성. PIL 있으면 안내 문구+워터마크 렌더, 없으면 최소 PNG.
생성형이 아니라 파이프라인 구조 검증용 자리표시자다(G1 미승인/목 모드).
"""
try:
import io
from PIL import Image, ImageDraw # type: ignore
except ImportError:
return _MINIMAL_PNG, "image/png"
W, H = size
im = Image.new("RGB", (W, H), (34, 38, 46))
draw = ImageDraw.Draw(im)
# 안내 문구(중앙)
y = H // 2 - 12 * len(lines)
for ln in lines:
draw.text((40, y), ln, fill=(210, 214, 222))
y += 24
# 워터마크(하단) — §6-6 정책과 동일 문구
draw.rectangle([0, H - 34, W, H], fill=(20, 22, 28))
draw.text((12, H - 26), "AI 생성 예상 이미지 — 실제 시공 결과와 다를 수 있음 (MOCK)", fill=(255, 214, 90))
buf = io.BytesIO()
im.save(buf, format="PNG")
return buf.getvalue(), "image/png"
# ============================================================================
# RenderJob / RenderResult 계약 객체 (계약 §2/§3)
# ============================================================================
@dataclass
class RenderJob:
job_id: str
shot_preset: str = "S1"
event_id: Optional[str] = None
booth_id: Optional[str] = None
scene: dict = field(default_factory=dict)
seed: Optional[int] = None
reference_image: Optional[str] = None
layers: Optional[list[str]] = None
wiring: Optional[dict] = None
hall_dims_m: Optional[list[float]] = None
options: dict = field(default_factory=dict)
@classmethod
def from_dict(cls, d: dict) -> "RenderJob":
if not d.get("job_id"):
raise NanoBananaError("RenderJob.job_id 누락")
return cls(
job_id=str(d["job_id"]),
shot_preset=d.get("shot_preset", "S1"),
event_id=d.get("event_id"),
booth_id=d.get("booth_id"),
scene=d.get("scene") or {},
seed=d.get("seed"),
reference_image=d.get("reference_image"),
layers=d.get("layers"),
wiring=d.get("wiring"),
hall_dims_m=d.get("hall_dims_m"),
options=d.get("options") or {},
)
@dataclass
class RenderResult:
job_id: str
status: str # DONE | FAILED
image_ref: Optional[dict] = None
meta: dict = field(default_factory=dict)
error: Optional[dict] = None
# ============================================================================
# 오브젝트 스토리지 (기본=로컬 FS degraded 어댑터, 계약 §4)
# ============================================================================
class ObjectStore:
"""이미지 + 사이드카 메타데이터 적재. 기본 구현은 로컬 파일시스템.
운영 시 S3/GCS 어댑터로 교체하되 save_image() 반환 계약(§4)은 유지한다.
"""
def __init__(self, root: str = OUTPUT_DIR, backend: str = "local"):
self.root = Path(root)
self.backend = backend
@staticmethod
def key_for(job: RenderJob) -> str:
ev = job.event_id or "no-event"
booth = job.booth_id or "no-booth"
return f"{ev}/{booth}/{job.job_id}_{job.shot_preset}.png"
def save_image(self, image: GeneratedImage, key: str) -> dict:
"""GeneratedImage.save() 로 이미지+사이드카(.meta.json) 기록 후 참조 반환."""
path = self.root / key
image.save(path) # 사이드카 .meta.json 동시 기록(결정적, B-03)
return {
"backend": self.backend,
"key": key,
"uri": path.resolve().as_uri(),
"sidecar_key": key + ".meta.json",
"content_type": image.mime_type,
}
# ============================================================================
# 완료 이벤트 발행 (계약 §3) — Redis pub/sub 스텁(지연 연결)
# ============================================================================
class EventPublisher:
"""완료/실패 이벤트를 Redis 채널에 발행. 백엔드가 구독→WebSocket 릴레이.
Redis 미연결 시에도 예외로 파이프라인을 멈추지 않는다(로그만). 비밀 미노출.
"""
def __init__(self, redis_url: str = REDIS_URL, channel: str = EVENT_CHANNEL):
self.redis_url = redis_url
self.channel = channel
self._r = None # 지연 연결
def _redis(self):
if self._r is None:
import redis # type: ignore # 지연 임포트 — 미설치여도 import 성립
self._r = redis.from_url(self.redis_url)
return self._r
def publish(self, event: dict) -> None:
payload = json.dumps(event, ensure_ascii=False)
try:
self._redis().publish(self.channel, payload)
except Exception as e: # noqa: BLE001 — 발행 실패가 렌더를 무효화하지 않음
# 메시지 요약만(키/스택트레이스 미기록)
print(
f"[event] publish skipped (redis unavailable): {type(e).__name__}", file=sys.stderr
)
# 항상 콘솔 관측(백엔드 부재 스모크에서도 이벤트 흐름 확인)
print(f"[event] {event.get('type')} job={event.get('job_id')} status={event.get('status')}")
# ============================================================================
# 워커
# ============================================================================
class RenderWorker:
"""RenderJob 소비 루프 + 단건 처리(process_job). 지연 연결로 구조만으로 성립."""
def __init__(
self,
redis_url: str = REDIS_URL,
queue_key: str = QUEUE_KEY,
store: Optional[ObjectStore] = None,
publisher: Optional[EventPublisher] = None,
):
self.redis_url = redis_url
self.queue_key = queue_key
self.store = store or ObjectStore()
self.publisher = publisher or EventPublisher(redis_url)
self._r = None # 지연 연결
self._stop = False
self._client = None # NanoBananaClient 지연 생성(실 호출 시에만)
# ------------------------------------------------------------------
# 렌더 디스패치
# ------------------------------------------------------------------
def _render(self, job: RenderJob) -> GeneratedImage:
"""샷 프리셋에 따라 (a) S6 래스터 합성, (b) 생성형/목 렌더로 분기."""
preset = SHOT_PRESETS.get(job.shot_preset)
if preset is None:
raise NanoBananaError(f"알 수 없는 샷 프리셋: {job.shot_preset} (S1~S7)")
# (a) S6 배선 오버레이 — 생성형 아님, 항상 로컬 PIL 결정적 합성(M5-3)
if not preset.get("generative", True):
return self._render_wiring(job)
# (b) 생성형 샷(S1~S5,S7)
if is_live():
return self._render_live(job) # G1 승인 경로
return self._render_mock(job) # 기본: 목/degraded
def _render_wiring(self, job: RenderJob) -> GeneratedImage:
"""S6: scene.wiring/최상위 wiring → 좌표 정합 래스터 오버레이(client 재사용)."""
scene_inner = job.scene.get("scene", job.scene)
wiring = job.wiring or scene_inner.get("wiring") or {}
if not wiring:
raise NanoBananaError("S6 배선 오버레이에 wiring 데이터가 필요합니다.")
# 좌표계 기준(m): job.hall_dims_m → booth.size_m → hall.dims_m → 기본
booth = scene_inner.get("booth", {}) or {}
hall = scene_inner.get("hall", {}) or {}
dims = job.hall_dims_m or booth.get("size_m") or hall.get("dims_m") or [10, 10]
hall_dims_m = (float(dims[0]), float(dims[1]))
return render_wiring_overlay_raster(
wiring,
hall_dims_m=hall_dims_m,
kinds=job.options.get("kinds"),
base_image=job.reference_image,
px_per_m=int(job.options.get("px_per_m", 40)),
)
def _render_live(self, job: RenderJob) -> GeneratedImage:
"""G1 승인 경로 — 실 Gemini 호출(NanoBananaClient 지연 생성)."""
if self._client is None:
try: # 패키지/스크립트 양쪽 임포트
from .client import NanoBananaClient # type: ignore
except ImportError:
from client import NanoBananaClient # type: ignore
self._client = NanoBananaClient(model=MODEL_NAME)
img = self._client.render_shot(
job.scene,
shot_preset=job.shot_preset,
reference_image=job.reference_image,
seed=job.seed,
layers=job.layers,
)
img.metadata.setdefault("render_path", "generative")
img.metadata["live"] = True
img.metadata["degraded"] = False
return img
def _render_mock(self, job: RenderJob) -> GeneratedImage:
"""기본 경로 — 목/degraded. 프롬프트·메타데이터는 실제와 동일 경로로 생성."""
# 프롬프트는 실제 빌더로 조립(구조 검증 목적) — 실패해도 잡은 진행
try:
prompt = build_booth_prompt(job.scene, shot_preset=job.shot_preset, layers=job.layers)
except Exception: # noqa: BLE001
prompt = f"[mock] {job.shot_preset} prompt build skipped"
meta = build_metadata(job.scene, "mock", job.shot_preset, job.seed)
meta["render_path"] = "generative"
meta["live"] = False
meta["degraded"] = True
preset = SHOT_PRESETS.get(job.shot_preset, {})
data, mime = _placeholder_png(
[
"KINTEX 나노바나나 — MOCK / DEGRADED",
f"shot: {job.shot_preset} {preset.get('label', '')}",
f"booth: {job.booth_id or '-'} event: {job.event_id or '-'}",
"G1 미승인/목 모드 — 실 Gemini 호출 없이 생성한 자리표시자",
]
)
return GeneratedImage(data=data, mime_type=mime, prompt_used=prompt, metadata=meta)
# ------------------------------------------------------------------
# 단건 처리 (테스트·스모크에서 직접 호출 가능 — Redis 불필요)
# ------------------------------------------------------------------
def process_job(self, job: RenderJob) -> RenderResult:
"""RenderJob 1건 처리: 렌더 → 스토리지 적재 → 완료 이벤트 발행."""
try:
image = self._render(job)
key = ObjectStore.key_for(job)
image_ref = self.store.save_image(image, key)
result = RenderResult(
job_id=job.job_id, status="DONE", image_ref=image_ref, meta=image.metadata
)
self.publisher.publish(self._event("renderjob.completed", job, result))
return result
except Exception as e: # noqa: BLE001 — 분류 후 친화 에러(비밀 미노출)
error = self._classify_error(e)
result = RenderResult(job_id=job.job_id, status="FAILED", error=error)
self.publisher.publish(self._event("renderjob.failed", job, result))
return result
@staticmethod
def _classify_error(e: Exception) -> dict:
"""예외 → {code,message}. 스택트레이스·키 미포함(PLANNING §6-5)."""
if isinstance(e, NanoBananaAuthError):
code = "AUTH"
elif isinstance(e, NanoBananaQuotaError):
code = "QUOTA"
elif isinstance(e, NanoBananaSafetyError):
code = "SAFETY"
elif isinstance(e, (ValueError, KeyError, TypeError)):
code = "BAD_REQUEST"
else:
code = "RENDER_ERROR"
return {"code": code, "message": str(e)[:300]}
@staticmethod
def _event(etype: str, job: RenderJob, result: RenderResult) -> dict:
return {
"type": etype,
"job_id": job.job_id,
"event_id": job.event_id,
"booth_id": job.booth_id,
"shot_preset": job.shot_preset,
"status": result.status,
"image_ref": result.image_ref,
"meta": result.meta,
"error": result.error,
"emitted_at": datetime.now(timezone.utc).isoformat(),
}
# ------------------------------------------------------------------
# 소비 루프 (지연 연결)
# ------------------------------------------------------------------
def _redis(self):
if self._r is None:
import redis # type: ignore # 지연 임포트
self._r = redis.from_url(self.redis_url)
return self._r
def stop(self) -> None:
self._stop = True
def run(self) -> None:
"""BLPOP 폴링 소비 루프. Redis 연결은 최초 진입 시점에 시도(지연)."""
mode = "LIVE(Gemini)" if is_live() else "MOCK/degraded"
print(f"[worker] start queue={self.queue_key} mode={mode} model={MODEL_NAME}")
r = self._redis()
while not self._stop:
item = r.blpop(self.queue_key, timeout=BLPOP_TIMEOUT_S)
if item is None:
continue # 폴링 타임아웃 — stop 플래그 재확인
_, raw = item
try:
payload = json.loads(raw)
job = RenderJob.from_dict(payload)
except Exception as e: # noqa: BLE001 — 파싱 실패 잡은 스킵(무한루프 방지)
print(f"[worker] drop malformed job: {type(e).__name__}", file=sys.stderr)
continue
self.process_job(job)
# ============================================================================
# 스모크 / CLI
# ============================================================================
def _smoke() -> int:
"""무네트워크 스모크: 목 모드로 생성형 잡 1건 + S6 잡 1건 처리, 사이드카 확인."""
import tempfile
tmp = Path(tempfile.mkdtemp(prefix="nanobanana_smoke_"))
worker = RenderWorker(store=ObjectStore(root=str(tmp)))
scene = {
"hall": {"id": "제1전시장 7홀", "dims_m": [126, 90], "ceiling_m": 12},
"booth": {"id": "A-102", "size_m": [6, 3], "type": "independent"},
"design": {"signage": {"text": "주식회사 가디아"}, "brand_color": "#0052A5"},
"lighting": {"mode": "night", "color_temp_k": 4000},
"wiring": {
"power": [{"from_trench": [1, 1], "to": [5, 2], "kw": 3}],
"network": [{"path": [[0, 0], [3, 1], [5, 2]]}],
},
"render_hints": {"style": "tech"},
}
ok = True
# (1) 생성형 샷 S2 (목 모드)
job1 = RenderJob(
job_id="smoke-s2", event_id="evt_smoke", booth_id="A-102", shot_preset="S2", scene=scene
)
r1 = worker.process_job(job1)
sidecar1 = tmp / (r1.image_ref["key"] + ".meta.json") if r1.image_ref else None
print(f"[smoke] S2 status={r1.status} live={r1.meta.get('live')} degraded={r1.meta.get('degraded')}")
if r1.status != "DONE" or not (sidecar1 and sidecar1.exists()):
ok = False
print("[smoke] FAIL: S2 결과/사이드카 누락")
# (2) S6 배선 오버레이(래스터 합성 — G1 무관)
job2 = RenderJob(
job_id="smoke-s6", event_id="evt_smoke", booth_id="A-102", shot_preset="S6", scene=scene
)
r2 = worker.process_job(job2)
sidecar2 = tmp / (r2.image_ref["key"] + ".meta.json") if r2.image_ref else None
rp = r2.meta.get("render_path")
print(f"[smoke] S6 status={r2.status} render_path={rp}")
# PIL 부재 환경에서는 S6 렌더가 실패할 수 있음(설계상 Pillow 필요) — 그 경우 관대 처리
if r2.status == "DONE":
if rp != "backend_raster_composite" or not (sidecar2 and sidecar2.exists()):
ok = False
print("[smoke] FAIL: S6 결과/사이드카/경로 불일치")
else:
print(f"[smoke] S6 skipped (환경 제약): {r2.error}")
print(f"[smoke] output dir: {tmp}")
print("[smoke] RESULT:", "PASS" if ok else "FAIL")
return 0 if ok else 1
def main(argv: Optional[list[str]] = None) -> int:
parser = argparse.ArgumentParser(description="나노바나나 RenderJob 워커")
parser.add_argument("--smoke", action="store_true", help="무네트워크 스모크(목 잡 처리)")
args = parser.parse_args(argv)
if args.smoke:
return _smoke()
RenderWorker().run()
return 0
if __name__ == "__main__":
raise SystemExit(main())
Reference in New Issue View Git Blame Copy Permalink