fix(video-render): Kling PiAPI → Native KlingAI (T11 follow-up)

박재오 발견: Kling 공식 API key 발급 (Access Key + Secret Key).
PiAPI gateway가 아닌 native api.klingai.com 사용.

변경:
- providers/kling.py: JWT 인증 (HS256, iss=access_key, exp=now+1800, nbf=now-5).
  POST /v1/videos/text2video → GET /v1/videos/{kind}/{task_id} 폴링.
  data.task_result.videos[0].url 다운로드.
  text2video / image2video 자동 분기.
- .env.example: PIAPI_API_KEY → KLING_ACCESS_KEY + KLING_SECRET_KEY
- docker-compose: 같은 env 교체
- requirements.txt: + PyJWT>=2.8.0

박재오 측: .env에 두 키 모두 입력.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

services/docker-compose.yml

@@ -63,17 +63,14 @@ services:
       - NAS_BASE_URL=${NAS_BASE_URL:-http://192.168.45.54:18801}
       - INTERNAL_API_KEY=${INTERNAL_API_KEY:-}
       - OPENAI_API_KEY=${OPENAI_API_KEY:-}
-      - GOOGLE_PROJECT_ID=${GOOGLE_PROJECT_ID:-}
-      - GOOGLE_LOCATION=${GOOGLE_LOCATION:-us-central1}
-      - GOOGLE_GCS_BUCKET=${GOOGLE_GCS_BUCKET:-}
-      - GOOGLE_APPLICATION_CREDENTIALS=/app/keys/gcp-sa.json
-      - PIAPI_API_KEY=${PIAPI_API_KEY:-}
+      - GEMINI_API_KEY=${GEMINI_API_KEY:-}
+      - KLING_ACCESS_KEY=${KLING_ACCESS_KEY:-}
+      - KLING_SECRET_KEY=${KLING_SECRET_KEY:-}
       - SEEDANCE_API_KEY=${SEEDANCE_API_KEY:-}
       - VIDEO_MEDIA_ROOT=${VIDEO_MEDIA_ROOT:-/mnt/nas/webpage/data/video}
       - VIDEO_MEDIA_URL_PREFIX=${VIDEO_MEDIA_URL_PREFIX:-/media/video}
     volumes:
       - /mnt/nas/webpage/data/video:/mnt/nas/webpage/data/video
-      - ${GCP_SA_JSON_HOST_PATH:-/etc/webai/gcp-sa.json}:/app/keys/gcp-sa.json:ro
     healthcheck:
       test: ["CMD", "python", "-c", "import urllib.request; urllib.request.urlopen('http://localhost:8000/health')"]
       interval: 60s

services/video-render/.env.example

@@ -10,14 +10,12 @@ INTERNAL_API_KEY=__copy_from_nas_dotenv__
 # Sora 2 (OpenAI)
 OPENAI_API_KEY=__paste_openai_key__
 
-# Veo 3.1 (Google Vertex AI)
-GOOGLE_PROJECT_ID=__paste_gcp_project_id__
-GOOGLE_LOCATION=us-central1
-GOOGLE_GCS_BUCKET=__paste_gcs_bucket_name__
-GOOGLE_APPLICATION_CREDENTIALS=/app/keys/gcp-sa.json
+# Veo (Google Gemini API — ai.google.dev. Vertex AI 경로 아님, GCS bucket 불필요)
+GEMINI_API_KEY=__paste_gemini_key__
 
-# Kling (PiAPI gateway)
-PIAPI_API_KEY=__paste_piapi_key__
+# Kling (Native KlingAI — JWT auth with Access Key + Secret Key)
+KLING_ACCESS_KEY=__paste_kling_access_key__
+KLING_SECRET_KEY=__paste_kling_secret_key__
 
 # Seedance 2.0 (BytePlus)
 SEEDANCE_API_KEY=__paste_seedance_key__

services/video-render/providers/kling.py

@@ -1,6 +1,7 @@
-"""Kling AI video generation — PiAPI gateway 경유.
+"""Kling AI video generation — Native KlingAI API (api.klingai.com).
 
-POST https://api.piapi.ai/api/v1/task → GET /api/v1/task/{id} 폴링 → data.output.video_url 다운로드.
+JWT auth: HS256, payload {iss: ACCESS_KEY, exp: now+1800, nbf: now-5}.
+POST /v1/videos/text2video → GET /v1/videos/text2video/{task_id} → task_result.videos[0].url 다운로드.
 """
 from __future__ import annotations
 
@@ -9,26 +10,41 @@ import os
 import time
 from typing import Optional
 
+import jwt as pyjwt
 import requests
 
 from nas_client import webhook_update_task
 
 logger = logging.getLogger(__name__)
 
-PIAPI_BASE_URL = "https://api.piapi.ai/api/v1"
+KLING_BASE_URL = "https://api.klingai.com"
 VIDEO_MEDIA_ROOT = os.getenv("VIDEO_MEDIA_ROOT", "/mnt/nas/webpage/data/video")
 VIDEO_MEDIA_URL_PREFIX = os.getenv("VIDEO_MEDIA_URL_PREFIX", "/media/video")
 
-POLL_INTERVAL = 10   # Kling은 30~180초
-POLL_MAX_ATTEMPTS = 60   # 최대 10분
+POLL_INTERVAL = 10
+POLL_MAX_ATTEMPTS = 60   # 최대 ~10분
 
-DEFAULT_VERSION = "2.6"
+DEFAULT_MODEL = "kling-v1-6"
+
+JWT_EXP_SECONDS = 1800   # 30분
+JWT_NBF_OFFSET = -5      # 5초 뒤로
+
+
+def _generate_jwt() -> Optional[str]:
+    access_key = os.getenv("KLING_ACCESS_KEY", "")
+    secret_key = os.getenv("KLING_SECRET_KEY", "")
+    if not access_key or not secret_key:
+        return None
+    now = int(time.time())
+    headers = {"alg": "HS256", "typ": "JWT"}
+    payload = {"iss": access_key, "exp": now + JWT_EXP_SECONDS, "nbf": now + JWT_NBF_OFFSET}
+    return pyjwt.encode(payload, secret_key, algorithm="HS256", headers=headers)
 
 
 def _headers() -> dict:
-    api_key = os.getenv("PIAPI_API_KEY", "")
+    token = _generate_jwt()
     return {
-        "x-api-key": api_key,
+        "Authorization": f"Bearer {token}" if token else "",
         "Content-Type": "application/json",
     }
 
@@ -36,80 +52,83 @@ def _headers() -> dict:
 def run_kling_generation(task_id: str, params: dict) -> None:
     """Kling으로 영상 생성 → mp4 → NAS SMB → webhook."""
     try:
-        if not os.getenv("PIAPI_API_KEY"):
-            webhook_update_task(task_id, "failed", 0, "", error="PIAPI_API_KEY 미설정")
+        if not os.getenv("KLING_ACCESS_KEY") or not os.getenv("KLING_SECRET_KEY"):
+            webhook_update_task(task_id, "failed", 0, "",
+                                error="KLING_ACCESS_KEY 또는 KLING_SECRET_KEY 미설정")
             return
 
         webhook_update_task(task_id, "processing", 5, "Kling API 호출 중...")
 
-        input_obj = {
-            "prompt": params["prompt"][:2500],
-            "duration": params.get("duration", 5),
-            "aspect_ratio": params.get("aspect_ratio", "16:9"),
-            "mode": params.get("mode", "std"),
-            "version": params.get("model") or DEFAULT_VERSION,
-        }
-        if params.get("negative_prompt"):
-            input_obj["negative_prompt"] = params["negative_prompt"][:2500]
-        if params.get("cfg_scale") is not None:
-            input_obj["cfg_scale"] = str(params["cfg_scale"])
-        if params.get("image_url"):
-            input_obj["image_url"] = params["image_url"]
+        # image_url 있으면 image2video, 없으면 text2video
+        is_image2video = bool(params.get("image_url"))
+        endpoint_path = "/v1/videos/image2video" if is_image2video else "/v1/videos/text2video"
 
         body = {
-            "model": "kling",
-            "task_type": "video_generation",
-            "input": input_obj,
-            "config": {"service_mode": "public"},
+            "model_name": params.get("model") or DEFAULT_MODEL,
+            "prompt": params["prompt"][:2500],
+            "duration": str(params.get("duration", 5)),
+            "aspect_ratio": params.get("aspect_ratio", "16:9"),
+            "mode": params.get("mode", "std"),
         }
+        if params.get("negative_prompt"):
+            body["negative_prompt"] = params["negative_prompt"][:2500]
+        if params.get("cfg_scale") is not None:
+            body["cfg_scale"] = float(params["cfg_scale"])
+        if is_image2video:
+            body["image"] = params["image_url"]
 
-        resp = requests.post(f"{PIAPI_BASE_URL}/task", headers=_headers(), json=body, timeout=30)
+        resp = requests.post(f"{KLING_BASE_URL}{endpoint_path}",
+                             headers=_headers(), json=body, timeout=30)
         if resp.status_code != 200:
             webhook_update_task(task_id, "failed", 0, "",
-                                error=f"Kling/PiAPI 오류: {resp.status_code} {resp.text[:300]}")
+                                error=f"Kling API 오류: {resp.status_code} {resp.text[:300]}")
             return
 
         body_json = resp.json()
-        if body_json.get("code") != 200:
+        if body_json.get("code") != 0:
             webhook_update_task(task_id, "failed", 0, "",
-                                error=f"Kling/PiAPI 거부: {body_json.get('message', '?')}")
+                                error=f"Kling API 거부: {body_json.get('message', '?')}")
             return
 
-        piapi_task_id = (body_json.get("data") or {}).get("task_id", "")
-        if not piapi_task_id:
-            webhook_update_task(task_id, "failed", 0, "", error="Kling/PiAPI 응답에 task_id 없음")
+        kling_task_id = (body_json.get("data") or {}).get("task_id", "")
+        if not kling_task_id:
+            webhook_update_task(task_id, "failed", 0, "", error="Kling 응답에 task_id 없음")
             return
 
         webhook_update_task(task_id, "processing", 15, "Kling 작업 등록됨")
 
-        # 폴링 — GET /task/{id}
+        # 폴링 — GET /v1/videos/{text2video|image2video}/{task_id}
         video_url = None
         for attempt in range(POLL_MAX_ATTEMPTS):
             time.sleep(POLL_INTERVAL)
-            fetch = requests.get(f"{PIAPI_BASE_URL}/task/{piapi_task_id}",
+            fetch = requests.get(f"{KLING_BASE_URL}{endpoint_path}/{kling_task_id}",
                                  headers=_headers(), timeout=30)
             if fetch.status_code != 200:
                 continue
             fd = fetch.json()
-            data = fd.get("data", {})
-            status = data.get("status", "")
+            if fd.get("code") != 0:
+                continue
+            data = fd.get("data") or {}
+            status = data.get("task_status", "")
             scaled = min(15 + int((attempt / POLL_MAX_ATTEMPTS) * 65), 79)
             webhook_update_task(task_id, "processing", scaled, f"Kling 생성 중... ({status})")
 
-            if status == "Completed":
-                video_url = (data.get("output") or {}).get("video_url", "")
+            if status == "succeed":
+                videos = ((data.get("task_result") or {}).get("videos") or [])
+                if videos:
+                    video_url = videos[0].get("url", "")
                 break
-            elif status in ("Failed", "failed"):
-                err = (data.get("error") or {}).get("message", "Kling 작업 실패")
+            elif status == "failed":
+                err = data.get("task_status_msg") or "Kling 작업 실패"
                 webhook_update_task(task_id, "failed", 0, "", error=err)
                 return
-            # Pending/Processing/Staged → 계속 폴링
+            # submitted/processing → 계속 폴링
         else:
             webhook_update_task(task_id, "failed", 0, "", error="Kling 폴링 timeout (10분)")
             return
 
         if not video_url:
-            webhook_update_task(task_id, "failed", 0, "", error="Kling 완료했으나 video_url 없음")
+            webhook_update_task(task_id, "failed", 0, "", error="Kling 완료했으나 video url 없음")
             return
 
         webhook_update_task(task_id, "processing", 85, "Kling 결과 다운로드 중...")
@@ -117,6 +136,7 @@ def run_kling_generation(task_id: str, params: dict) -> None:
         os.makedirs(VIDEO_MEDIA_ROOT, exist_ok=True)
         file_path = os.path.join(VIDEO_MEDIA_ROOT, filename)
 
+        # Kling 결과 url은 일반적으로 인증 불필요 (signed URL)
         dl = requests.get(video_url, stream=True, timeout=300)
         dl.raise_for_status()
         with open(file_path, "wb") as f:
@@ -127,7 +147,7 @@ def run_kling_generation(task_id: str, params: dict) -> None:
         webhook_update_task(task_id, "succeeded", 100, "Kling 생성 완료", video_url=local_url)
 
     except requests.Timeout:
-        webhook_update_task(task_id, "failed", 0, "", error="Kling/PiAPI 타임아웃")
+        webhook_update_task(task_id, "failed", 0, "", error="Kling API 타임아웃")
     except Exception as e:
         logger.exception("Kling generation error task=%s", task_id)
         webhook_update_task(task_id, "failed", 0, "", error=str(e))

services/video-render/providers/veo.py

@@ -1,13 +1,13 @@
-"""Veo 3.1 video generation — Google Vertex AI.
+"""Veo 3.1 video generation — Gemini API (ai.google.dev).
 
-POST .../models/{MODEL}:predictLongRunning → POST :fetchPredictOperation 폴링 →
-결과 gs://bucket/path/sample_0.mp4 → google-cloud-storage로 다운로드 → NAS SMB.
+POST https://generativelanguage.googleapis.com/v1beta/models/{MODEL}:predictLongRunning
+GET https://generativelanguage.googleapis.com/v1beta/{operation_name}
+→ done=true 시 response.generateVideoResponse.generatedSamples[0].video.uri 다운로드
 """
 from __future__ import annotations
 
 import logging
 import os
-import subprocess
 import time
 from typing import Optional
 
@@ -17,100 +17,57 @@ from nas_client import webhook_update_task
 
 logger = logging.getLogger(__name__)
 
+GEMINI_BASE_URL = "https://generativelanguage.googleapis.com/v1beta"
 VIDEO_MEDIA_ROOT = os.getenv("VIDEO_MEDIA_ROOT", "/mnt/nas/webpage/data/video")
 VIDEO_MEDIA_URL_PREFIX = os.getenv("VIDEO_MEDIA_URL_PREFIX", "/media/video")
 
-POLL_INTERVAL = 12   # Veo는 30~120초 소요
-POLL_MAX_ATTEMPTS = 50   # 최대 ~10분
+POLL_INTERVAL = 10   # Veo는 30~120초 소요
+POLL_MAX_ATTEMPTS = 60   # 최대 ~10분
 
-DEFAULT_MODEL = "veo-3.1-fast-generate-001"
+DEFAULT_MODEL = "veo-3.1-fast-generate-preview"
 
 
-def _gcloud_access_token() -> Optional[str]:
-    """GOOGLE_APPLICATION_CREDENTIALS service account JSON으로 access token 발행.
-
-    google-auth가 컨테이너 안에서 자동 인증 — Bearer 토큰을 GCS SDK가 직접 사용.
-    REST API 호출용으로는 명시적 token이 필요 → google.auth로 발행.
-    """
-    try:
-        from google.auth import default as google_default_auth
-        from google.auth.transport.requests import Request as GoogleAuthRequest
-
-        credentials, _ = google_default_auth(
-            scopes=["https://www.googleapis.com/auth/cloud-platform"],
-        )
-        credentials.refresh(GoogleAuthRequest())
-        return credentials.token
-    except Exception:
-        logger.exception("Google credentials refresh 실패")
-        return None
-
-
-def _download_gcs(gcs_uri: str, local_path: str) -> bool:
-    """gs://bucket/path/file.mp4 → local_path 다운로드. 성공 여부 반환."""
-    try:
-        from google.cloud import storage as gcs_storage
-        if not gcs_uri.startswith("gs://"):
-            return False
-        without_scheme = gcs_uri[len("gs://"):]
-        bucket_name, blob_path = without_scheme.split("/", 1)
-        client = gcs_storage.Client(project=os.getenv("GOOGLE_PROJECT_ID"))
-        bucket = client.bucket(bucket_name)
-        blob = bucket.blob(blob_path)
-        blob.download_to_filename(local_path)
-        return True
-    except Exception:
-        logger.exception("GCS 다운로드 실패: %s", gcs_uri)
-        return False
+def _headers() -> dict:
+    api_key = os.getenv("GEMINI_API_KEY", "")
+    return {
+        "x-goog-api-key": api_key,
+        "Content-Type": "application/json",
+    }
 
 
 def run_veo_generation(task_id: str, params: dict) -> None:
-    """Veo 3.1로 영상 생성 → GCS → NAS SMB → webhook."""
+    """Veo로 영상 생성 → mp4 → NAS SMB → webhook."""
     try:
-        project_id = os.getenv("GOOGLE_PROJECT_ID", "")
-        location = os.getenv("GOOGLE_LOCATION", "us-central1")
-        gcs_bucket = os.getenv("GOOGLE_GCS_BUCKET", "")
-
-        if not project_id or not gcs_bucket:
-            webhook_update_task(task_id, "failed", 0, "",
-                                error="GOOGLE_PROJECT_ID 또는 GOOGLE_GCS_BUCKET 미설정")
+        if not os.getenv("GEMINI_API_KEY"):
+            webhook_update_task(task_id, "failed", 0, "", error="GEMINI_API_KEY 미설정 (Windows .env)")
             return
 
-        token = _gcloud_access_token()
-        if not token:
-            webhook_update_task(task_id, "failed", 0, "",
-                                error="Google access token 발행 실패 (서비스 계정 JSON 확인)")
-            return
-
-        webhook_update_task(task_id, "processing", 5, "Veo API 호출 중...")
+        webhook_update_task(task_id, "processing", 5, "Veo (Gemini API) 호출 중...")
 
         model_id = params.get("model") or DEFAULT_MODEL
-        endpoint_base = (
-            f"https://{location}-aiplatform.googleapis.com/v1/projects/{project_id}"
-            f"/locations/{location}/publishers/google/models/{model_id}"
-        )
-        headers = {
-            "Authorization": f"Bearer {token}",
-            "Content-Type": "application/json",
-        }
         body = {
             "instances": [{"prompt": params["prompt"]}],
             "parameters": {
-                "storageUri": f"gs://{gcs_bucket}/veo/{task_id}/",
-                "sampleCount": 1,
                 "aspectRatio": params.get("aspect_ratio") or "16:9",
+                "numberOfVideos": 1,
             },
         }
         if params.get("duration"):
-            body["parameters"]["duration"] = params["duration"]
+            body["parameters"]["durationSeconds"] = str(params["duration"])
+        if params.get("resolution"):
+            body["parameters"]["resolution"] = params["resolution"]
         if params.get("negative_prompt"):
             body["parameters"]["negativePrompt"] = params["negative_prompt"]
+        if params.get("person_generation"):
+            body["parameters"]["personGeneration"] = params["person_generation"]
 
-        resp = requests.post(f"{endpoint_base}:predictLongRunning",
-                             headers=headers, json=body, timeout=30)
+        resp = requests.post(
+            f"{GEMINI_BASE_URL}/models/{model_id}:predictLongRunning",
+            headers=_headers(), json=body, timeout=30,
+        )
         if resp.status_code != 200:
             webhook_update_task(task_id, "failed", 0, "",
-                                error=f"Veo API 오류: {resp.status_code} {resp.text[:300]}")
+                                error=f"Veo Gemini API 오류: {resp.status_code} {resp.text[:300]}")
             return
 
         op_name = resp.json().get("name", "")
@@ -118,16 +75,15 @@ def run_veo_generation(task_id: str, params: dict) -> None:
             webhook_update_task(task_id, "failed", 0, "", error="Veo 응답에 operation name 없음")
             return
 
-        webhook_update_task(task_id, "processing", 15, f"Veo 작업 시작됨")
+        webhook_update_task(task_id, "processing", 15, "Veo 작업 시작됨")
 
-        # 폴링 — fetchPredictOperation
-        gcs_uri = None
+        # 폴링 — GET /v1beta/{operation_name}
+        video_uri = None
         for attempt in range(POLL_MAX_ATTEMPTS):
             time.sleep(POLL_INTERVAL)
-            fetch = requests.post(
-                f"{endpoint_base}:fetchPredictOperation",
-                headers=headers,
-                json={"operationName": op_name},
+            fetch = requests.get(
+                f"{GEMINI_BASE_URL}/{op_name}",
+                headers=_headers(),
                 timeout=30,
             )
             if fetch.status_code != 200:
@@ -142,29 +98,34 @@ def run_veo_generation(task_id: str, params: dict) -> None:
                     webhook_update_task(task_id, "failed", 0, "",
                                         error=f"Veo 작업 실패: {fd['error'].get('message','?')}")
                     return
-                videos = (fd.get("response") or {}).get("videos") or []
-                if not videos:
-                    webhook_update_task(task_id, "failed", 0, "", error="Veo 완료했으나 videos 비어 있음")
+                # response.generateVideoResponse.generatedSamples[0].video.uri
+                response = fd.get("response") or {}
+                gen = response.get("generateVideoResponse") or {}
+                samples = gen.get("generatedSamples") or []
+                if not samples:
+                    webhook_update_task(task_id, "failed", 0, "", error="Veo 완료했으나 generatedSamples 비어 있음")
                     return
-                gcs_uri = videos[0].get("gcsUri", "")
+                video_uri = (samples[0].get("video") or {}).get("uri", "")
                 break
         else:
             webhook_update_task(task_id, "failed", 0, "", error="Veo 폴링 timeout (10분)")
             return
 
-        if not gcs_uri:
-            webhook_update_task(task_id, "failed", 0, "", error="Veo 응답에 gcsUri 없음")
+        if not video_uri:
+            webhook_update_task(task_id, "failed", 0, "", error="Veo 응답에 video.uri 없음")
             return
 
-        webhook_update_task(task_id, "processing", 85, "GCS에서 mp4 다운로드 중...")
+        webhook_update_task(task_id, "processing", 85, "Veo 결과 다운로드 중...")
         filename = f"{task_id}.mp4"
         os.makedirs(VIDEO_MEDIA_ROOT, exist_ok=True)
         file_path = os.path.join(VIDEO_MEDIA_ROOT, filename)
 
-        ok = _download_gcs(gcs_uri, file_path)
-        if not ok:
-            webhook_update_task(task_id, "failed", 0, "", error=f"GCS 다운로드 실패: {gcs_uri}")
-            return
+        # 다운로드 — x-goog-api-key 헤더 그대로 사용 (Gemini API가 인증 처리)
+        dl = requests.get(video_uri, headers=_headers(), stream=True, timeout=300)
+        dl.raise_for_status()
+        with open(file_path, "wb") as f:
+            for chunk in dl.iter_content(chunk_size=8192):
+                f.write(chunk)
 
         local_url = f"{VIDEO_MEDIA_URL_PREFIX}/{filename}"
         webhook_update_task(task_id, "succeeded", 100, "Veo 생성 완료", video_url=local_url)

services/video-render/requirements.txt

@@ -4,7 +4,7 @@ requests==2.32.3
 redis>=5.0
 httpx>=0.27
 openai>=1.50.0
-google-cloud-storage>=2.18.0
+PyJWT>=2.8.0
 pytest>=8.0
 pytest-asyncio>=0.24
 respx>=0.21