gahusb/web-page-backend
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
ca7a502514ced8a9aca7fadb715ae760a93af790
web-page-backend/docs/superpowers/plans/2026-05-23-video-studio-backend.md

1374 lines
48 KiB
Markdown
Raw Blame History

# Video Studio — Plan 1: 이미지 생성 백엔드 Implementation Plan
> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
**Goal:** image-lab(NAS) + image-render(Windows)를 신설해 `POST /api/image/generate` (provider 선택: gpt_image/nano_banana/flux) → Redis 큐 → Windows 워커 이미지 생성 → NAS webhook → 폴링으로 결과 image_url 확보. curl로 검증 가능한 작동 백엔드.
**Architecture:** video-lab/video-render 복제 패턴. NAS image-lab은 작업 큐 관리(SQLite image_tasks + Redis push + internal webhook), Windows image-render는 Redis BLPOP → provider dispatch → NAS SMB 저장 → webhook. 접근 B(클라이언트 오케스트레이션)의 백엔드 절반.
**Tech Stack:** FastAPI, Redis(aioredis/redis), SQLite, httpx/requests, OpenAI Images API, Gemini 2.5 Flash Image, ComfyUI(FLUX 로컬). 참조 spec: `docs/superpowers/specs/2026-05-23-video-studio-node-canvas-design.md`
---
## File Structure
**web-backend (NAS):**
- `image-lab/app/db.py` — image_tasks 테이블 + CRUD (video-lab 복제)
- `image-lab/app/auth.py` — verify_internal_key (video-lab 복제, 동일)
- `image-lab/app/internal_router.py` — `/api/internal/image/update` (video-lab 복제)
- `image-lab/app/main.py` — generate/tasks/providers 엔드포인트 (video-lab 복제 + provider 메타 교체)
- `image-lab/app/__init__.py`, `Dockerfile`, `requirements.txt`, `tests/`
- `docker-compose.yml` (수정), nginx 설정 (수정)
**web-ai (Windows):**
- `services/image-render/nas_client.py` — webhook 어댑터 (video-render 복제, 엔드포인트 교체)
- `services/image-render/providers/gpt_image.py` — OpenAI Images (신규)
- `services/image-render/providers/nano_banana.py` — Gemini Flash Image (신규)
- `services/image-render/providers/flux.py` — ComfyUI 로컬 (신규)
- `services/image-render/worker.py` — Redis BLPOP dispatch (video-render 복제)
- `services/image-render/main.py`, `Dockerfile`, `requirements.txt`, `tests/`
- `services/docker-compose.yml` (수정)
---
## Phase A — image-lab (NAS)
### Task 1: image-lab DB — image_tasks 테이블
**Files:**
- Create: `image-lab/app/__init__.py` (빈 파일)
- Create: `image-lab/app/db.py`
- Test: `image-lab/tests/test_db.py`
복제 원본: `video-lab/app/db.py`. 치환 규칙: `video_tasks`→`image_tasks`, `video_url`→`image_url`, `VIDEO_DATA_DIR`→`IMAGE_DATA_DIR`, `video.db`→`image.db`.
- [ ] **Step 1: 디렉토리 + 빈 __init__.py 생성**
```bash
mkdir -p image-lab/app image-lab/tests
touch image-lab/app/__init__.py image-lab/tests/__init__.py
```
- [ ] **Step 2: 실패 테스트 작성**
`image-lab/tests/test_db.py`:
```python
import os, tempfile, importlib
def _fresh_db(monkeypatch, tmp):
monkeypatch.setenv("IMAGE_DATA_DIR", tmp)
import app.db as db
importlib.reload(db)
db.init_db()
return db
def test_create_and_get_task(monkeypatch):
with tempfile.TemporaryDirectory() as tmp:
db = _fresh_db(monkeypatch, tmp)
row = db.create_task("t1", "gpt_image", {"prompt": "a cat"})
assert row["id"] == "t1"
assert row["provider"] == "gpt_image"
assert row["status"] == "queued"
got = db.get_task("t1")
assert got["id"] == "t1"
assert db.get_task("nope") is None
def test_update_task_sets_image_url(monkeypatch):
with tempfile.TemporaryDirectory() as tmp:
db = _fresh_db(monkeypatch, tmp)
db.create_task("t2", "nano_banana", {"prompt": "x"})
db.update_task("t2", "succeeded", 100, message="done", image_url="/media/image/t2.png")
got = db.get_task("t2")
assert got["status"] == "succeeded"
assert got["image_url"] == "/media/image/t2.png"
assert got["progress"] == 100
```
- [ ] **Step 3: 테스트 실패 확인**
Run: `cd image-lab && python -m pytest tests/test_db.py -v`
Expected: FAIL (`ModuleNotFoundError: app.db` 또는 import 실패)
- [ ] **Step 4: db.py 작성**
`image-lab/app/db.py` (video-lab/app/db.py 복제 후 치환):
```python
"""SQLite persistence for image_tasks. Single table — task 단위 추적만."""
from __future__ import annotations
import json
import os
import sqlite3
from contextlib import contextmanager
from typing import Any, Dict, Optional
DB_PATH = os.path.join(os.getenv("IMAGE_DATA_DIR", "/app/data"), "image.db")
@contextmanager
def _conn():
os.makedirs(os.path.dirname(DB_PATH), exist_ok=True)
conn = sqlite3.connect(DB_PATH)
conn.row_factory = sqlite3.Row
conn.execute("PRAGMA journal_mode=WAL")
conn.execute("PRAGMA busy_timeout=5000")
try:
yield conn
conn.commit()
finally:
conn.close()
def init_db() -> None:
with _conn() as conn:
conn.execute(
"""
CREATE TABLE IF NOT EXISTS image_tasks (
id TEXT PRIMARY KEY,
provider TEXT NOT NULL,
params TEXT NOT NULL,
status TEXT DEFAULT 'queued',
progress INTEGER DEFAULT 0,
message TEXT DEFAULT '',
image_url TEXT,
error TEXT,
created_at TEXT DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now')),
updated_at TEXT DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now'))
)
"""
)
def _row_to_dict(row) -> Dict[str, Any]:
return {
"id": row["id"], "provider": row["provider"], "params": row["params"],
"status": row["status"], "progress": row["progress"], "message": row["message"],
"image_url": row["image_url"], "error": row["error"],
"created_at": row["created_at"], "updated_at": row["updated_at"],
}
def create_task(task_id: str, provider: str, params: Dict[str, Any]) -> Dict[str, Any]:
with _conn() as conn:
conn.execute(
"INSERT INTO image_tasks (id, provider, params) VALUES (?, ?, ?)",
(task_id, provider, json.dumps(params)),
)
row = conn.execute("SELECT * FROM image_tasks WHERE id = ?", (task_id,)).fetchone()
return _row_to_dict(row)
def update_task(task_id: str, status: str, progress: int, message: str = "",
image_url: Optional[str] = None, error: Optional[str] = None) -> None:
with _conn() as conn:
conn.execute(
"""
UPDATE image_tasks
SET status = ?, progress = ?, message = ?, image_url = ?, error = ?,
updated_at = strftime('%Y-%m-%dT%H:%M:%fZ','now')
WHERE id = ?
""",
(status, progress, message, image_url, error, task_id),
)
def get_task(task_id: str) -> Optional[Dict[str, Any]]:
with _conn() as conn:
row = conn.execute("SELECT * FROM image_tasks WHERE id = ?", (task_id,)).fetchone()
return _row_to_dict(row) if row else None
```
- [ ] **Step 5: 테스트 통과 확인**
Run: `cd image-lab && python -m pytest tests/test_db.py -v`
Expected: PASS (2 passed)
- [ ] **Step 6: 커밋**
```bash
git add image-lab/app/__init__.py image-lab/app/db.py image-lab/tests/__init__.py image-lab/tests/test_db.py
git commit -m "feat(image-lab): image_tasks 테이블 + CRUD (video-lab 복제)"
```
---
### Task 2: image-lab auth — verify_internal_key
**Files:**
- Create: `image-lab/app/auth.py`
- Test: `image-lab/tests/test_auth.py`
복제 원본: `video-lab/app/auth.py` (치환 없음, 그대로).
- [ ] **Step 1: 실패 테스트 작성**
`image-lab/tests/test_auth.py`:
```python
import pytest
from fastapi import HTTPException
from app.auth import verify_internal_key
def test_no_server_key_rejects(monkeypatch):
monkeypatch.delenv("INTERNAL_API_KEY", raising=False)
with pytest.raises(HTTPException) as e:
verify_internal_key("anything")
assert e.value.status_code == 401
def test_wrong_key_rejects(monkeypatch):
monkeypatch.setenv("INTERNAL_API_KEY", "secret")
with pytest.raises(HTTPException) as e:
verify_internal_key("wrong")
assert e.value.status_code == 401
def test_correct_key_passes(monkeypatch):
monkeypatch.setenv("INTERNAL_API_KEY", "secret")
assert verify_internal_key("secret") is None
```
- [ ] **Step 2: 실패 확인**
Run: `cd image-lab && python -m pytest tests/test_auth.py -v`
Expected: FAIL (`ModuleNotFoundError: app.auth`)
- [ ] **Step 3: auth.py 작성 (video-lab 복제)**
`image-lab/app/auth.py`:
```python
"""Windows image-render worker → NAS image-lab internal webhook 인증."""
from __future__ import annotations
import os
from fastapi import Header, HTTPException
def verify_internal_key(x_internal_key: str = Header(...)):
expected = os.getenv("INTERNAL_API_KEY")
if not expected:
raise HTTPException(401, "INTERNAL_API_KEY not configured on server")
if x_internal_key != expected:
raise HTTPException(401, "Invalid X-Internal-Key")
```
- [ ] **Step 4: 통과 확인**
Run: `cd image-lab && python -m pytest tests/test_auth.py -v`
Expected: PASS (3 passed)
- [ ] **Step 5: 커밋**
```bash
git add image-lab/app/auth.py image-lab/tests/test_auth.py
git commit -m "feat(image-lab): verify_internal_key (video-lab 복제)"
```
---
### Task 3: image-lab internal_router — /api/internal/image/update
**Files:**
- Create: `image-lab/app/internal_router.py`
- Test: `image-lab/tests/test_internal_router.py`
복제 원본: `video-lab/app/internal_router.py`. 치환: `/api/internal/video/update`→`/api/internal/image/update`, `video_url`→`image_url`, `video_update`→`image_update`.
- [ ] **Step 1: 실패 테스트 작성**
`image-lab/tests/test_internal_router.py`:
```python
import os, tempfile, importlib
from fastapi import FastAPI
from fastapi.testclient import TestClient
def _client(monkeypatch, tmp):
monkeypatch.setenv("IMAGE_DATA_DIR", tmp)
monkeypatch.setenv("INTERNAL_API_KEY", "secret")
import app.db as db; importlib.reload(db); db.init_db()
import app.internal_router as ir; importlib.reload(ir)
app = FastAPI(); app.include_router(ir.router)
return TestClient(app), db
def test_update_requires_key(monkeypatch):
with tempfile.TemporaryDirectory() as tmp:
client, db = _client(monkeypatch, tmp)
db.create_task("t1", "gpt_image", {"prompt": "x"})
r = client.post("/api/internal/image/update",
json={"task_id": "t1", "status": "succeeded", "progress": 100})
assert r.status_code == 422 or r.status_code == 401 # header 누락
def test_update_succeeds_with_key(monkeypatch):
with tempfile.TemporaryDirectory() as tmp:
client, db = _client(monkeypatch, tmp)
db.create_task("t1", "gpt_image", {"prompt": "x"})
r = client.post("/api/internal/image/update",
headers={"X-Internal-Key": "secret"},
json={"task_id": "t1", "status": "succeeded", "progress": 100,
"image_url": "/media/image/t1.png"})
assert r.status_code == 200
assert db.get_task("t1")["image_url"] == "/media/image/t1.png"
def test_update_unknown_task_404(monkeypatch):
with tempfile.TemporaryDirectory() as tmp:
client, db = _client(monkeypatch, tmp)
r = client.post("/api/internal/image/update",
headers={"X-Internal-Key": "secret"},
json={"task_id": "nope", "status": "failed", "progress": 0})
assert r.status_code == 404
```
- [ ] **Step 2: 실패 확인**
Run: `cd image-lab && python -m pytest tests/test_internal_router.py -v`
Expected: FAIL (import 실패)
- [ ] **Step 3: internal_router.py 작성**
`image-lab/app/internal_router.py`:
```python
"""Windows image-render → NAS image-lab internal webhook."""
from __future__ import annotations
import logging
from typing import Optional
from fastapi import APIRouter, Depends, HTTPException
from pydantic import BaseModel, Field
from . import db
from .auth import verify_internal_key
logger = logging.getLogger(__name__)
router = APIRouter()
class UpdatePayload(BaseModel):
task_id: str
status: str = Field(..., description="processing|succeeded|failed")
progress: int = Field(..., ge=0, le=100)
message: str = ""
image_url: Optional[str] = None
error: Optional[str] = None
@router.post("/api/internal/image/update", dependencies=[Depends(verify_internal_key)])
def image_update(payload: UpdatePayload):
task = db.get_task(payload.task_id)
if task is None:
raise HTTPException(404, f"task not found: {payload.task_id}")
db.update_task(payload.task_id, payload.status, payload.progress,
message=payload.message, image_url=payload.image_url, error=payload.error)
logger.info("internal/image/update task=%s status=%s progress=%d",
payload.task_id, payload.status, payload.progress)
return {"ok": True}
```
- [ ] **Step 4: 통과 확인**
Run: `cd image-lab && python -m pytest tests/test_internal_router.py -v`
Expected: PASS (3 passed)
- [ ] **Step 5: 커밋**
```bash
git add image-lab/app/internal_router.py image-lab/tests/test_internal_router.py
git commit -m "feat(image-lab): /api/internal/image/update webhook (video-lab 복제)"
```
---
### Task 4: image-lab main — generate/tasks/providers 엔드포인트
**Files:**
- Create: `image-lab/app/main.py`
- Test: `image-lab/tests/test_main.py`
복제 원본: `video-lab/app/main.py`. 치환: `queue:video-render`→`queue:image-render`, `/api/video/*`→`/api/image/*`, provider set·메타 교체, `kind:"video"`→`kind:"image"`, job_type `{provider}_generation` 유지.
- [ ] **Step 1: 실패 테스트 작성 (Redis push는 monkeypatch)**
`image-lab/tests/test_main.py`:
```python
import os, tempfile, importlib
from fastapi.testclient import TestClient
def _client(monkeypatch, tmp):
monkeypatch.setenv("IMAGE_DATA_DIR", tmp)
import app.db as db; importlib.reload(db)
import app.main as main; importlib.reload(main)
pushed = []
async def fake_push(task_id, job_type, params):
pushed.append((task_id, job_type, params))
monkeypatch.setattr(main, "_push_render_job", fake_push)
return TestClient(main.app), db, pushed
def test_providers_lists_three(monkeypatch):
with tempfile.TemporaryDirectory() as tmp:
client, _, _ = _client(monkeypatch, tmp)
r = client.get("/api/image/providers")
ids = {p["id"] for p in r.json()["providers"]}
assert ids == {"gpt_image", "nano_banana", "flux"}
def test_generate_rejects_unknown_provider(monkeypatch):
with tempfile.TemporaryDirectory() as tmp:
client, _, _ = _client(monkeypatch, tmp)
r = client.post("/api/image/generate", json={"provider": "midjourney", "prompt": "x"})
assert r.status_code == 400
def test_generate_creates_task_and_pushes(monkeypatch):
with tempfile.TemporaryDirectory() as tmp:
client, db, pushed = _client(monkeypatch, tmp)
r = client.post("/api/image/generate", json={"provider": "gpt_image", "prompt": "a cat"})
assert r.status_code == 200
task_id = r.json()["task_id"]
assert db.get_task(task_id)["status"] == "queued"
assert pushed[0][1] == "gpt_image_generation"
```
- [ ] **Step 2: 실패 확인**
Run: `cd image-lab && python -m pytest tests/test_main.py -v`
Expected: FAIL (import 실패)
- [ ] **Step 3: main.py 작성**
`image-lab/app/main.py`:
```python
"""FastAPI entrypoint for image-lab.
POST /api/image/generate — provider + prompt → Redis push → task_id
GET /api/image/tasks/{id} — DB 조회
GET /api/image/providers — 3 provider 메타
"""
from __future__ import annotations
import json, logging, os, uuid
from datetime import datetime, timedelta, timezone
from typing import Any, Dict, Optional
import redis.asyncio as aioredis
from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel, Field
from . import db
from .internal_router import router as internal_router
logger = logging.getLogger(__name__)
CORS_ALLOW_ORIGINS = os.getenv("CORS_ALLOW_ORIGINS", "http://localhost:3007,http://localhost:8080")
REDIS_URL = os.getenv("REDIS_URL", "redis://redis:6379")
redis_client = aioredis.from_url(REDIS_URL, decode_responses=False)
SUPPORTED_PROVIDERS = {"gpt_image", "nano_banana", "flux"}
app = FastAPI()
app.include_router(internal_router)
app.add_middleware(
CORSMiddleware,
allow_origins=[o.strip() for o in CORS_ALLOW_ORIGINS.split(",")],
allow_credentials=False,
allow_methods=["GET", "POST", "PUT", "DELETE", "OPTIONS", "PATCH"],
allow_headers=["Content-Type"],
)
@app.on_event("startup")
def on_startup():
db.init_db()
@app.get("/health")
def health():
return {"ok": True, "service": "image-lab"}
@app.get("/api/image/providers")
def list_providers():
return {"providers": [
{"id": "gpt_image", "name": "GPT Image 2.0", "models": ["gpt-image-1"],
"sizes": ["1024x1024", "1024x1536", "1536x1024"]},
{"id": "nano_banana", "name": "Nano Banana (Gemini)", "models": ["gemini-2.5-flash-image"],
"sizes": ["1024x1024"]},
{"id": "flux", "name": "FLUX (local)", "models": ["flux-schnell", "flux-dev"],
"sizes": ["1024x1024", "832x1216", "1216x832"]},
]}
class GenerateRequest(BaseModel):
provider: str = Field(..., description="gpt_image|nano_banana|flux")
model: Optional[str] = None
prompt: str
size: Optional[str] = None
negative_prompt: Optional[str] = None
extra: Optional[Dict[str, Any]] = None
class Config:
extra = "allow"
async def _push_render_job(task_id: str, job_type: str, params: dict) -> None:
kst = timezone(timedelta(hours=9))
payload = {
"task_id": task_id, "kind": "image", "job_type": job_type,
"params": params, "submitted_at": datetime.now(kst).isoformat(),
}
await redis_client.rpush("queue:image-render", json.dumps(payload))
@app.post("/api/image/generate")
async def generate_image(req: GenerateRequest):
if req.provider not in SUPPORTED_PROVIDERS:
raise HTTPException(400, f"지원하지 않는 provider: {req.provider} (supported: {sorted(SUPPORTED_PROVIDERS)})")
task_id = str(uuid.uuid4())
params = req.model_dump(exclude_none=True)
db.create_task(task_id, req.provider, params)
job_type = f"{req.provider}_generation"
await _push_render_job(task_id, job_type, params)
return {"task_id": task_id, "provider": req.provider}
@app.get("/api/image/tasks/{task_id}")
def get_task_status(task_id: str):
t = db.get_task(task_id)
if not t:
raise HTTPException(404, "task not found")
return t
```
- [ ] **Step 4: 통과 확인**
Run: `cd image-lab && python -m pytest tests/test_main.py -v`
Expected: PASS (3 passed)
- [ ] **Step 5: 커밋**
```bash
git add image-lab/app/main.py image-lab/tests/test_main.py
git commit -m "feat(image-lab): generate/tasks/providers 엔드포인트 (video-lab 복제)"
```
---
### Task 5: image-lab 컨테이너 + docker-compose + nginx 차단
**Files:**
- Create: `image-lab/Dockerfile`, `image-lab/requirements.txt`, `image-lab/env.example`
- Modify: `docker-compose.yml` (image-lab 서비스 추가)
- Modify: nginx 설정 (`/api/internal/image/` 3-layer 차단 — music/video 패턴 복제)
- [ ] **Step 1: requirements.txt 작성**
`image-lab/requirements.txt`:
```
fastapi==0.115.0
uvicorn[standard]==0.30.6
pydantic==2.9.2
redis==5.0.8
httpx==0.27.2
```
- [ ] **Step 2: Dockerfile 작성 (video-lab Dockerfile 참조, uvicorn --workers 1)**
`image-lab/Dockerfile`:
```dockerfile
FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY app ./app
EXPOSE 8000
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "1"]
```
- [ ] **Step 3: docker-compose.yml에 image-lab 추가 + frontend.depends_on 등재**
`docker-compose.yml` (video-lab 블록 다음에 추가, 포트 18802는 NAS 사용 포트 18000/18500/18600/18700/18800/18801/18850/18900/18950 중 빈 자리로 확인됨):
**+ `frontend.depends_on` 리스트에 `image-lab` 추가** (nginx upstream 보장):
```yaml
frontend:
depends_on:
- lotto
- travel-proxy
# ... 기존 항목들
- video-lab
- image-lab # 신규
```
```yaml
image-lab:
build: ./image-lab
container_name: image-lab
restart: unless-stopped
ports:
- "18802:8000"
environment:
- REDIS_URL=redis://redis:6379
- INTERNAL_API_KEY=${INTERNAL_API_KEY}
- IMAGE_DATA_DIR=/app/data
- CORS_ALLOW_ORIGINS=http://localhost:3007,http://localhost:8080
volumes:
- ${RUNTIME_PATH}/image-data:/app/data
depends_on:
- redis
healthcheck:
test: ["CMD", "python", "-c", "import urllib.request; urllib.request.urlopen('http://localhost:8000/health')"]
interval: 60s
timeout: 5s
retries: 3
```
- [ ] **Step 4: scripts/* 5위치 등재 (NAS rsync + build + health + data dir)**
신규 lab 추가 시 누락하면 NAS deploy 실패(memory `feedback_nas_deploy_paths.md` rule 3). video-lab과 동일하게 다음 5위치에 `image-lab`/`image` 추가:
1. **`scripts/deploy-nas.sh`** — `SERVICES="... video-lab nginx scripts"` → `... video-lab image-lab nginx scripts`
2. **`scripts/deploy.sh`** — `BUILD_TARGETS="... video-lab frontend"` → `... video-lab image-lab frontend`
3. **`scripts/deploy.sh`** — `CONTAINER_NAMES="... video-lab frontend"` → `... video-lab image-lab frontend`
4. **`scripts/deploy.sh`** — `HEALTH_ENDPOINTS="... video-lab redis"` → `... video-lab image-lab redis`
5. **`scripts/deploy.sh`** — `DATA_DIRS="music stock insta realestate agent-office personal video"` → `... personal video image`
(frontend.depends_on은 Step 3에서 처리됨 → 총 6위치 완료)
- [ ] **Step 5: nginx `/api/internal/image/` 3-layer 차단**
기존 video(`/api/internal/video/`) 차단 블록을 찾아 동일 패턴으로 image 블록 추가. 검색:
```bash
grep -rn "api/internal/video" <nginx conf 경로>
```
찾은 블록을 복제해 `video`→`image` 치환 (location deny + 내부 전용 표시). 정확한 파일은 grep 결과로 확정.
- [ ] **Step 6: 빌드 + 헬스 확인**
```bash
docker compose build image-lab
docker compose up -d image-lab redis
curl http://localhost:18802/health
```
Expected: `{"ok":true,"service":"image-lab"}`
- [ ] **Step 7: 커밋**
```bash
git add image-lab/Dockerfile image-lab/requirements.txt image-lab/env.example docker-compose.yml scripts/deploy.sh scripts/deploy-nas.sh
git commit -m "feat(image-lab): Dockerfile + compose entry + scripts 6위치 + nginx 차단"
```
---
## Phase B — image-render (Windows, web-ai repo)
> 이하 작업은 **web-ai repo**에서 진행. `cd C:\Users\jaeoh\Desktop\workspace\web-ai`.
### Task 6: image-render nas_client + worker scaffold
**Files:**
- Create: `services/image-render/nas_client.py`, `services/image-render/providers/__init__.py`
- Test: `services/image-render/tests/test_worker.py` (Task 10에서 채움 — 여기선 nas_client만)
- Test: `services/image-render/tests/test_nas_client.py`
복제 원본: `services/video-render/nas_client.py`. 치환: `/api/internal/video/update`→`/api/internal/image/update`, `video_url`→`image_url`, default `NAS_BASE_URL` 18801→18802.
- [ ] **Step 1: 실패 테스트 작성**
`services/image-render/tests/test_nas_client.py`:
```python
import nas_client
def test_webhook_includes_image_url(monkeypatch):
captured = {}
def fake_post(payload): captured.update(payload)
monkeypatch.setattr(nas_client, "_post", fake_post)
nas_client.webhook_update_task("t1", "succeeded", 100, "done", image_url="/media/image/t1.png")
assert captured["task_id"] == "t1"
assert captured["image_url"] == "/media/image/t1.png"
def test_webhook_omits_none_fields(monkeypatch):
captured = {}
monkeypatch.setattr(nas_client, "_post", lambda p: captured.update(p))
nas_client.webhook_update_task("t2", "processing", 10, "working")
assert "image_url" not in captured and "error" not in captured
```
- [ ] **Step 2: 실패 확인**
Run: `cd services/image-render && python -m pytest tests/test_nas_client.py -v`
Expected: FAIL (import 실패)
- [ ] **Step 3: nas_client.py 작성**
`services/image-render/nas_client.py`:
```python
"""NAS webhook 어댑터 — Windows worker → NAS image-lab HTTP 위임."""
from __future__ import annotations
import logging, os
from typing import Any, Dict, Optional
import httpx
logger = logging.getLogger(__name__)
_TIMEOUT = 10.0
def _post(payload: Dict[str, Any]) -> None:
nas_base_url = os.getenv("NAS_BASE_URL", "http://192.168.45.54:18802")
internal_api_key = os.getenv("INTERNAL_API_KEY", "")
url = f"{nas_base_url}/api/internal/image/update"
try:
r = httpx.post(url, headers={"X-Internal-Key": internal_api_key}, json=payload, timeout=_TIMEOUT)
if r.status_code != 200:
logger.error("webhook %s returned %d: %s", payload.get("task_id"), r.status_code, r.text[:200])
except Exception:
logger.exception("webhook %s 호출 실패", payload.get("task_id"))
def webhook_update_task(task_id: str, status: str, progress: int, message: str = "",
image_url: Optional[str] = None, error: Optional[str] = None) -> None:
payload: Dict[str, Any] = {"task_id": task_id, "status": status, "progress": progress, "message": message}
if image_url is not None:
payload["image_url"] = image_url
if error is not None:
payload["error"] = error
_post(payload)
```
- [ ] **Step 4: 통과 확인 + 커밋**
Run: `cd services/image-render && python -m pytest tests/test_nas_client.py -v` → PASS
```bash
git add services/image-render/nas_client.py services/image-render/providers/__init__.py services/image-render/tests/test_nas_client.py
git commit -m "feat(image-render): nas_client webhook adapter (video-render 복제)"
```
---
### Task 7: provider gpt_image — OpenAI Images API
**Files:**
- Create: `services/image-render/providers/gpt_image.py`
- Create: `services/image-render/providers/_media.py` (b64 → NAS SMB 저장 헬퍼, 3 provider 공용)
- Test: `services/image-render/tests/test_gpt_image.py`
- [ ] **Step 1: 공용 미디어 저장 헬퍼 작성**
`services/image-render/providers/_media.py`:
```python
"""b64 이미지 → NAS SMB 경로 저장 → /media/image URL 반환."""
from __future__ import annotations
import base64, os, uuid
IMAGE_MEDIA_ROOT = os.getenv("IMAGE_MEDIA_ROOT", "/mnt/nas/webpage/data/image")
IMAGE_MEDIA_URL_PREFIX = os.getenv("IMAGE_MEDIA_URL_PREFIX", "/media/image")
def save_b64_png(task_id: str, b64_data: str) -> str:
os.makedirs(IMAGE_MEDIA_ROOT, exist_ok=True)
fname = f"{task_id}-{uuid.uuid4().hex[:8]}.png"
path = os.path.join(IMAGE_MEDIA_ROOT, fname)
with open(path, "wb") as f:
f.write(base64.b64decode(b64_data))
return f"{IMAGE_MEDIA_URL_PREFIX}/{fname}"
```
- [ ] **Step 2: 실패 테스트 작성 (requests + _media monkeypatch)**
`services/image-render/tests/test_gpt_image.py`:
```python
import providers.gpt_image as gi
def test_missing_key_reports_failed(monkeypatch):
monkeypatch.delenv("OPENAI_API_KEY", raising=False)
calls = []
monkeypatch.setattr(gi, "webhook_update_task", lambda *a, **k: calls.append((a, k)))
gi.run_gpt_image_generation("t1", {"prompt": "a cat"})
# 마지막 호출이 failed
assert calls[-1][0][1] == "failed"
def test_success_saves_and_reports_url(monkeypatch):
monkeypatch.setenv("OPENAI_API_KEY", "sk-test")
calls = []
monkeypatch.setattr(gi, "webhook_update_task", lambda *a, **k: calls.append((a, k)))
monkeypatch.setattr(gi, "save_b64_png", lambda tid, b64: "/media/image/t1.png")
class FakeResp:
status_code = 200
def json(self): return {"data": [{"b64_json": "ZmFrZQ=="}]}
def raise_for_status(self): pass
monkeypatch.setattr(gi.requests, "post", lambda *a, **k: FakeResp())
gi.run_gpt_image_generation("t1", {"prompt": "a cat"})
succeeded = [c for c in calls if c[0][1] == "succeeded"]
assert succeeded and succeeded[-1][1]["image_url"] == "/media/image/t1.png"
```
- [ ] **Step 3: 실패 확인**
Run: `cd services/image-render && python -m pytest tests/test_gpt_image.py -v`
Expected: FAIL (import 실패)
- [ ] **Step 4: gpt_image.py 작성**
`services/image-render/providers/gpt_image.py`:
```python
"""GPT Image 2.0 — OpenAI Images API.
POST https://api.openai.com/v1/images/generations
body {model:"gpt-image-1", prompt, size, n:1} → data[0].b64_json
"""
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__)
OPENAI_URL = "https://api.openai.com/v1/images/generations"
DEFAULT_MODEL = "gpt-image-1"
def run_gpt_image_generation(task_id: str, params: dict) -> None:
try:
if not os.getenv("OPENAI_API_KEY"):
webhook_update_task(task_id, "failed", 0, "", error="OPENAI_API_KEY 미설정 (Windows .env)")
return
webhook_update_task(task_id, "processing", 10, "GPT Image 호출 중...")
body = {
"model": params.get("model") or DEFAULT_MODEL,
"prompt": params["prompt"],
"size": params.get("size") or "1024x1024",
"n": 1,
}
resp = requests.post(
OPENAI_URL,
headers={"Authorization": f"Bearer {os.getenv('OPENAI_API_KEY')}", "Content-Type": "application/json"},
json=body, timeout=120,
)
if resp.status_code != 200:
webhook_update_task(task_id, "failed", 0, "", error=f"OpenAI {resp.status_code}: {resp.text[:200]}")
return
b64 = resp.json()["data"][0]["b64_json"]
url = save_b64_png(task_id, b64)
webhook_update_task(task_id, "succeeded", 100, "완료", image_url=url)
except Exception as e:
logger.exception("gpt_image task=%s 실패", task_id)
webhook_update_task(task_id, "failed", 0, "", error=str(e))
```
- [ ] **Step 5: 통과 확인 + 커밋**
Run: `cd services/image-render && python -m pytest tests/test_gpt_image.py -v` → PASS
```bash
git add services/image-render/providers/_media.py services/image-render/providers/gpt_image.py services/image-render/tests/test_gpt_image.py
git commit -m "feat(image-render): gpt_image provider + media helper (SP image)"
```
---
### Task 8: provider nano_banana — Gemini 2.5 Flash Image
**Files:**
- Create: `services/image-render/providers/nano_banana.py`
- Test: `services/image-render/tests/test_nano_banana.py`
- [ ] **Step 1: 실패 테스트 작성**
`services/image-render/tests/test_nano_banana.py`:
```python
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"]
```
- [ ] **Step 2: 실패 확인**
Run: `cd services/image-render && python -m pytest tests/test_nano_banana.py -v`
Expected: FAIL
- [ ] **Step 3: nano_banana.py 작성**
`services/image-render/providers/nano_banana.py`:
```python
"""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))
```
- [ ] **Step 4: 통과 확인 + 커밋**
Run: `cd services/image-render && python -m pytest tests/test_nano_banana.py -v` → PASS
```bash
git add services/image-render/providers/nano_banana.py services/image-render/tests/test_nano_banana.py
git commit -m "feat(image-render): nano_banana (Gemini Flash Image) provider"
```
---
### Task 9: provider flux — ComfyUI 로컬
**Files:**
- Create: `services/image-render/providers/flux.py`
- Create: `services/image-render/providers/flux_workflow.json` (ComfyUI에서 export한 FLUX text2img workflow — engineer가 ComfyUI UI에서 "Save (API Format)"로 생성)
- Test: `services/image-render/tests/test_flux.py`
> **GPU 가드**: FLUX는 RTX 5070 Ti VRAM을 Chronos(장중 7GB)·video-render와 공유. 장중(09:00~15:30)에는 GPU 충돌 위험 → 환경변수 `FLUX_BLOCK_TRADING_HOURS=1`이면 장중 거부(API provider로 폴백 권장).
- [ ] **Step 1: 실패 테스트 작성**
`services/image-render/tests/test_flux.py`:
```python
import providers.flux as fx
def test_blocked_during_trading_hours(monkeypatch):
monkeypatch.setenv("FLUX_BLOCK_TRADING_HOURS", "1")
monkeypatch.setattr(fx, "_is_trading_hours", lambda: True)
calls = []
monkeypatch.setattr(fx, "webhook_update_task", lambda *a, **k: calls.append((a, k)))
fx.run_flux_generation("t1", {"prompt": "a cat"})
assert calls[-1][0][1] == "failed"
assert "장중" in calls[-1][1]["error"]
def test_success_polls_history_and_saves(monkeypatch):
monkeypatch.setattr(fx, "_is_trading_hours", lambda: False)
calls = []
monkeypatch.setattr(fx, "webhook_update_task", lambda *a, **k: calls.append((a, k)))
monkeypatch.setattr(fx, "_load_workflow", lambda prompt, size: {"3": {}})
monkeypatch.setattr(fx, "_submit_prompt", lambda wf: "pid-1")
monkeypatch.setattr(fx, "_poll_image_b64", lambda pid: "ZmFrZQ==")
monkeypatch.setattr(fx, "save_b64_png", lambda tid, b64: "/media/image/t1.png")
fx.run_flux_generation("t1", {"prompt": "a cat"})
assert [c for c in calls if c[0][1] == "succeeded"]
```
- [ ] **Step 2: 실패 확인**
Run: `cd services/image-render && python -m pytest tests/test_flux.py -v`
Expected: FAIL
- [ ] **Step 3: flux.py 작성**
`services/image-render/providers/flux.py`:
```python
"""FLUX 로컬 — ComfyUI HTTP API.
POST {COMFYUI_URL}/prompt (workflow JSON) → prompt_id
GET {COMFYUI_URL}/history/{prompt_id} → outputs → image filename
GET {COMFYUI_URL}/view?filename=... → PNG bytes → b64
"""
from __future__ import annotations
import base64, json, logging, os, time
from datetime import datetime, timezone, timedelta
import requests
from nas_client import webhook_update_task
from providers._media import save_b64_png
logger = logging.getLogger(__name__)
COMFYUI_URL = os.getenv("COMFYUI_URL", "http://127.0.0.1:8188")
WORKFLOW_PATH = os.path.join(os.path.dirname(__file__), "flux_workflow.json")
POLL_INTERVAL = 2
POLL_MAX = 120
def _is_trading_hours() -> bool:
kst = timezone(timedelta(hours=9))
now = datetime.now(kst)
if now.weekday() >= 5:
return False
return (now.hour, now.minute) >= (9, 0) and (now.hour, now.minute) <= (15, 30)
def _load_workflow(prompt: str, size: str) -> dict:
with open(WORKFLOW_PATH, encoding="utf-8") as f:
wf = json.load(f)
# CLIPTextEncode 노드의 text를 prompt로 치환 (workflow에 "%PROMPT%" placeholder 사용)
raw = json.dumps(wf).replace("%PROMPT%", prompt.replace('"', "'"))
return json.loads(raw)
def _submit_prompt(workflow: dict) -> str:
r = requests.post(f"{COMFYUI_URL}/prompt", json={"prompt": workflow}, timeout=30)
r.raise_for_status()
return r.json()["prompt_id"]
def _poll_image_b64(prompt_id: str):
for _ in range(POLL_MAX):
h = requests.get(f"{COMFYUI_URL}/history/{prompt_id}", timeout=10)
data = h.json().get(prompt_id)
if data and data.get("outputs"):
for node_out in data["outputs"].values():
for img in node_out.get("images", []):
view = requests.get(f"{COMFYUI_URL}/view",
params={"filename": img["filename"], "subfolder": img.get("subfolder", ""), "type": img.get("type", "output")},
timeout=30)
view.raise_for_status()
return base64.b64encode(view.content).decode()
time.sleep(POLL_INTERVAL)
return None
def run_flux_generation(task_id: str, params: dict) -> None:
try:
if os.getenv("FLUX_BLOCK_TRADING_HOURS") == "1" and _is_trading_hours():
webhook_update_task(task_id, "failed", 0, "", error="장중 GPU 보호 — FLUX 거부 (API provider 사용 권장)")
return
webhook_update_task(task_id, "processing", 10, "FLUX (ComfyUI) 생성 중...")
wf = _load_workflow(params["prompt"], params.get("size") or "1024x1024")
pid = _submit_prompt(wf)
b64 = _poll_image_b64(pid)
if not b64:
webhook_update_task(task_id, "failed", 0, "", error="ComfyUI 타임아웃 또는 출력 없음")
return
url = save_b64_png(task_id, b64)
webhook_update_task(task_id, "succeeded", 100, "완료", image_url=url)
except Exception as e:
logger.exception("flux task=%s 실패", task_id)
webhook_update_task(task_id, "failed", 0, "", error=str(e))
```
- [ ] **Step 4: 통과 확인**
Run: `cd services/image-render && python -m pytest tests/test_flux.py -v`
Expected: PASS
- [ ] **Step 5: flux_workflow.json placeholder 생성 안내**
`flux_workflow.json`은 ComfyUI UI에서 FLUX text2img 그래프 구성 후 **"Save (API Format)"**로 export한 JSON. CLIPTextEncode 노드의 text 값을 `"%PROMPT%"`로 수동 치환해 저장. (engineer가 ComfyUI 환경에서 직접 생성 — 코드로 생성 불가)
- [ ] **Step 6: 커밋**
```bash
git add services/image-render/providers/flux.py services/image-render/tests/test_flux.py
git commit -m "feat(image-render): flux (ComfyUI 로컬) provider + GPU 장중 가드"
```
---
### Task 10: image-render worker — Redis BLPOP dispatch
**Files:**
- Create: `services/image-render/worker.py`
- Test: `services/image-render/tests/test_worker.py`
복제 원본: `services/video-render/worker.py`. 치환: `queue:video-render`→`queue:image-render`, dispatch table을 3 provider로, provider import 교체.
- [ ] **Step 1: 실패 테스트 작성**
`services/image-render/tests/test_worker.py`:
```python
import worker
def test_dispatch_routes_to_provider(monkeypatch):
called = {}
monkeypatch.setattr(worker, "run_gpt_image_generation", lambda tid, p: called.setdefault("gpt", (tid, p)))
worker._dispatch({"job_type": "gpt_image_generation", "task_id": "t1", "params": {"prompt": "x"}})
assert called["gpt"][0] == "t1"
def test_dispatch_unknown_job_type_reports_failed(monkeypatch):
calls = []
monkeypatch.setattr(worker, "webhook_update_task", lambda *a, **k: calls.append((a, k)))
worker._dispatch({"job_type": "midjourney_generation", "task_id": "t9", "params": {}})
assert calls[-1][0][1] == "failed"
```
- [ ] **Step 2: 실패 확인**
Run: `cd services/image-render && python -m pytest tests/test_worker.py -v`
Expected: FAIL
- [ ] **Step 3: worker.py 작성**
`services/image-render/worker.py`:
```python
"""Redis BLPOP worker — queue:image-render → job_type dispatch → NAS webhook.
queue:paused 가 set이면 대기 (task-watcher가 박재오 활동 감지 시 set).
video-render worker.py 패턴 — string-based dispatch + getattr (테스트 patch 호환).
"""
from __future__ import annotations
import asyncio, json, logging, os, sys
import redis.asyncio as aioredis
from nas_client import webhook_update_task
from providers.gpt_image import run_gpt_image_generation
from providers.nano_banana import run_nano_banana_generation
from providers.flux import run_flux_generation
logger = logging.getLogger(__name__)
REDIS_URL = os.getenv("REDIS_URL", "redis://192.168.45.54:6379")
QUEUE_KEY = "queue:image-render"
PAUSED_KEY = "queue:paused"
_DISPATCH_TABLE = {
"gpt_image_generation": "run_gpt_image_generation",
"nano_banana_generation": "run_nano_banana_generation",
"flux_generation": "run_flux_generation",
}
def _dispatch(payload: dict) -> None:
job_type = payload.get("job_type", "")
task_id = payload.get("task_id", "")
params = payload.get("params", {})
fn_name = _DISPATCH_TABLE.get(job_type)
if fn_name is None:
logger.error("unknown job_type=%s task=%s", job_type, task_id)
webhook_update_task(task_id, "failed", 0, "", error=f"unknown job_type: {job_type}")
return
try:
fn = getattr(sys.modules[__name__], fn_name)
except AttributeError:
webhook_update_task(task_id, "failed", 0, "", error=f"internal dispatch error: {fn_name}")
return
fn(task_id, params)
async def worker_loop():
redis = aioredis.from_url(REDIS_URL, decode_responses=False)
logger.info("image-render worker started (queue=%s)", QUEUE_KEY)
while True:
try:
if await redis.get(PAUSED_KEY):
await asyncio.sleep(2)
continue
item = await redis.blpop(QUEUE_KEY, timeout=5)
if item is None:
continue
payload = json.loads(item[1])
await asyncio.to_thread(_dispatch, payload)
except Exception:
logger.exception("worker_loop iteration 실패")
await asyncio.sleep(1)
if __name__ == "__main__":
logging.basicConfig(level=logging.INFO)
asyncio.run(worker_loop())
```
- [ ] **Step 4: 통과 확인 + 커밋**
Run: `cd services/image-render && python -m pytest tests/ -v` → ALL PASS
```bash
git add services/image-render/worker.py services/image-render/tests/test_worker.py
git commit -m "feat(image-render): Redis BLPOP worker + 3 provider dispatch"
```
---
### Task 11: image-render main + Dockerfile + compose entry
**Files:**
- Create: `services/image-render/main.py`, `Dockerfile`, `requirements.txt`, `env.example`
- Modify: `services/docker-compose.yml` (image-render 서비스 추가)
- [ ] **Step 1: main.py 작성 (video-render main 패턴 — lifespan에서 worker_loop 태스크)**
`services/image-render/main.py`:
```python
"""image-render FastAPI — health + 백그라운드 worker_loop."""
from __future__ import annotations
import asyncio, logging
from contextlib import asynccontextmanager
from fastapi import FastAPI
from worker import worker_loop
logging.basicConfig(level=logging.INFO)
@asynccontextmanager
async def lifespan(app: FastAPI):
task = asyncio.create_task(worker_loop())
yield
task.cancel()
app = FastAPI(lifespan=lifespan)
@app.get("/health")
def health():
return {"ok": True, "service": "image-render"}
```
- [ ] **Step 2: requirements.txt**
`services/image-render/requirements.txt`:
```
fastapi==0.115.0
uvicorn[standard]==0.30.6
redis==5.0.8
httpx==0.27.2
requests==2.32.3
```
- [ ] **Step 3: Dockerfile**
`services/image-render/Dockerfile`:
```dockerfile
FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
EXPOSE 8000
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "1"]
```
- [ ] **Step 4: docker-compose.yml에 image-render 추가**
> ⚠️ **포트 주의**: web-ai/services 현재 점유 — 18710/18711/18712 + **18713(task-watcher)**. image-render는 **18714** 사용 (18713 충돌 회피).
`services/docker-compose.yml` (video-render 블록 다음, GPU·NAS SMB 마운트는 video-render와 동일하게):
```yaml
image-render:
build: ./image-render
container_name: image-render
restart: unless-stopped
ports:
- "18714:8000"
environment:
- REDIS_URL=redis://192.168.45.54:6379
- NAS_BASE_URL=http://192.168.45.54:18802
- INTERNAL_API_KEY=${INTERNAL_API_KEY}
- OPENAI_API_KEY=${OPENAI_API_KEY}
- GEMINI_API_KEY=${GEMINI_API_KEY}
- COMFYUI_URL=${COMFYUI_URL:-http://host.docker.internal:8188}
- FLUX_BLOCK_TRADING_HOURS=1
- IMAGE_MEDIA_ROOT=/mnt/nas/webpage/data/image
volumes:
- /mnt/nas/webpage/data/image:/mnt/nas/webpage/data/image
```
- [ ] **Step 5: 빌드 + 헬스 + 커밋**
```bash
docker compose -f services/docker-compose.yml build image-render
docker compose -f services/docker-compose.yml up -d image-render
curl http://localhost:18714/health
git add services/image-render/main.py services/image-render/Dockerfile services/image-render/requirements.txt services/image-render/env.example services/docker-compose.yml
git commit -m "feat(image-render): main + Dockerfile + compose entry"
```
---
### Task 12: E2E 검증 (curl)
- [ ] **Step 1: 이미지 생성 요청**
```bash
curl -X POST http://localhost:18802/api/image/generate \
-H "Content-Type: application/json" \
-d '{"provider":"gpt_image","prompt":"a cute white desktop robot, big LED eyes, studio lighting"}'
```
Expected: `{"task_id":"<uuid>","provider":"gpt_image"}`
- [ ] **Step 2: 폴링**
```bash
curl http://localhost:18802/api/image/tasks/<task_id>
```
Expected: status가 queued → processing → succeeded로 전이, `image_url` 채워짐
- [ ] **Step 3: 이미지 확인**
```bash
curl -I http://localhost:8080<image_url> # nginx /media/image/ alias
```
Expected: 200 + image/png
- [ ] **Step 4: nano_banana·flux도 동일 검증** (각 provider key·ComfyUI 준비 필요)
---
## Self-Review
**Spec coverage:**
- image-lab (NAS): Task 1~5 ✅ | image-render (Windows): Task 6~11 ✅ | E2E: Task 12 ✅
- provider 3종(gpt_image/nano_banana/flux): Task 7/8/9 ✅
- nginx 3-layer 차단: Task 5 Step 4 ✅
- queue:paused 존중: Task 10 worker_loop ✅
- 비범위(keyframe/1→N/서버저장): 본 plan 미포함 (2·3단계) — spec과 일치 ✅
- 프론트 /studio: **Plan 2에서** (본 plan은 백엔드만)
**Placeholder scan:** flux_workflow.json은 ComfyUI export 산출물로 코드 생성 불가 항목 — Task 9 Step 5에 명확한 생성 절차 명시 (placeholder 아님). nginx 정확 파일은 grep으로 확정 지시.
**Type consistency:** `webhook_update_task(task_id, status, progress, message, image_url=, error=)` — nas_client(Task 6)·3 provider(7/8/9)·worker(10) 전부 동일 시그니처. job_type `{provider}_generation` — image-lab main(Task 4)·worker dispatch table(Task 10) 일치. `save_b64_png(task_id, b64)` — _media(Task 7)·3 provider 일치.
**미해결(구현 중 확정):** image-lab 포트 18802(compose 충돌 확인) / flux_workflow.json(ComfyUI export) / nano_banana 모델명(gemini-2.5-flash-image, API 응답으로 검증).
Reference in New Issue View Git Blame Copy Permalink