feat(image-lab): /api/internal/image/update webhook (video-lab 복제)

image-lab/app/internal_router.py

@@ -0,0 +1,52 @@
+"""Windows image-render → NAS image-lab internal webhook.
+
+POST /api/internal/image/update
+- X-Internal-Key 인증 필수
+- image_tasks row update (status, progress, message, image_url, error)
+"""
+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}

image-lab/tests/test_internal_router.py

@@ -0,0 +1,38 @@
+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