From e91715bf2cf1e265a397f3883ebe182394614030 Mon Sep 17 00:00:00 2001 From: gahusb Date: Sat, 23 May 2026 11:28:21 +0900 Subject: [PATCH] =?UTF-8?q?docs(plan):=20video-studio=20Plan=201=20?= =?UTF-8?q?=E2=80=94=20image-render=20=ED=8F=AC=ED=8A=B8=2018714(task-watc?= =?UTF-8?q?her=20=EC=B6=A9=EB=8F=8C=20=ED=9A=8C=ED=94=BC)=20+=20scripts=20?= =?UTF-8?q?6=EC=9C=84=EC=B9=98=20=EB=93=B1=EC=9E=AC=20step=20=EC=B6=94?= =?UTF-8?q?=EA=B0=80?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../plans/2026-05-23-video-studio-backend.md | 1373 +++++++++++++++++ 1 file changed, 1373 insertions(+) create mode 100644 docs/superpowers/plans/2026-05-23-video-studio-backend.md diff --git a/docs/superpowers/plans/2026-05-23-video-studio-backend.md b/docs/superpowers/plans/2026-05-23-video-studio-backend.md new file mode 100644 index 0000000..1d53037 --- /dev/null +++ b/docs/superpowers/plans/2026-05-23-video-studio-backend.md @@ -0,0 +1,1373 @@ +# 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" +``` +찾은 블록을 복제해 `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":"","provider":"gpt_image"}` + +- [ ] **Step 2: 폴링** + +```bash +curl http://localhost:18802/api/image/tasks/ +``` +Expected: status가 queued → processing → succeeded로 전이, `image_url` 채워짐 + +- [ ] **Step 3: 이미지 확인** + +```bash +curl -I http://localhost:8080 # 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 응답으로 검증).