55 lines
1.4 KiB
Python
55 lines
1.4 KiB
Python
"""NAS webhook 어댑터 — Windows worker → NAS image-lab HTTP 위임.
|
|
|
|
video-render nas_client 복제 (call-time os.getenv으로 테스트 격리).
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import 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)
|