"""변형 추천 모듈 v2 — GPT Vision으로 생성된 원본 이미지를 분석하여 A/B 변형 추천.""" import base64 import json import logging import socket import time import urllib.error import urllib.request from pathlib import Path from typing import Any, Dict from app.core.config import settings from app.modules.llm.image_tracer import traced_call logger = logging.getLogger(__name__) PROMPTS_DIR = ( Path(__file__).resolve().parent.parent.parent.parent / "prompts" / "_base" / "variation_recommender" / "v2" ) OPENAI_API_URL = "https://api.openai.com/v1/responses" _VARIATION_ITEM_SCHEMA: Dict[str, Any] = { "type": "object", "additionalProperties": False, "properties": { "type": { "type": "string", "description": "angle | color | angle+color | none", }, "angle": { "type": ["object", "null"], "description": "Camera position only (sent as 3D diagram to i2i). null if type has no angle.", "properties": { "horizontal": {"type": "number", "description": "0-360 degrees rotation"}, "vertical": {"type": "number", "description": "-30 to 30 degrees tilt"}, "zoom": {"type": "number", "description": "0.8 to 1.5 zoom factor"}, }, "required": ["horizontal", "vertical", "zoom"], "additionalProperties": False, }, "composition": { "type": "string", "description": "Lens/framing note for i2i (NOT angle). Derive from scene cinematic intent and camera_direction reference. Empty if no composition change.", }, "color": { "type": "string", "description": "Lighting/color change only. Derive from scene cinematic intent. Empty if no color change.", }, "reason": {"type": "string"}, }, "required": ["type", "angle", "composition", "color", "reason"], } _RESPONSE_SCHEMA: Dict[str, Any] = { "type": "object", "additionalProperties": False, "properties": { "variation_a": _VARIATION_ITEM_SCHEMA, "variation_b": _VARIATION_ITEM_SCHEMA, "recommended": {"type": "string"}, "reasoning": {"type": "string"}, }, "required": ["variation_a", "variation_b", "recommended", "reasoning"], } class VariationRecommenderV2: """GPT Vision 기반 변형 추천기 — 생성된 원본 이미지를 분석하여 A/B 추천.""" def __init__(self, api_key: str | None = None, model: str | None = None) -> None: # ★생성 시점 복사 금지 — 이후 슬롯 전환이 이 인스턴스에 닿지 않는다. self._explicit_key = api_key self._model = model or settings.openai_model @traced_call(operation="variation_recommend", provider="openai", model_of=lambda self, *a, **k: getattr(self, "_model", "unknown"), prompt_of=lambda self, image_bytes, scene_description="", *a, **k: scene_description, output_text="[variation recommended]") def recommend_from_image( self, image_bytes: bytes, scene_description: str = "", beat_title: str = "", _api_key: str | None = None, ) -> Dict[str, Any]: """생성된 원본 이미지를 GPT Vision으로 분석하여 A/B 변형을 추천. ★기록: urllib 직접 호출이라 litellm 콜백 밖이다 (2026-08-07). Args: image_bytes: 원본 씬 이미지 PNG bytes. scene_description: 씬 T2I 프롬프트 (참고용). beat_title: 씬 제목 (참고용). Returns: {"variation_a": {...}, "variation_b": {...}, "recommended": "A"|"B"|"original", "reasoning": "..."} """ if _api_key is None: # 슬롯은 브로커가 정하고, 키 수준 실패면 다음 슬롯으로 다시 부른다. from app.core.openai_keys import call_with_key_failover return call_with_key_failover( lambda _k: self.recommend_from_image( image_bytes, scene_description=scene_description, beat_title=beat_title, _api_key=_k), where="variation_recommender.vision", fixed_key=self._explicit_key) system_prompt = (PROMPTS_DIR / "system.md").read_text(encoding="utf-8").strip() user_text = "Analyze this scene image and recommend A/B variations.\n" if beat_title: user_text += f"Scene: {beat_title}\n" if scene_description: user_text += f"T2I prompt used: {scene_description}\n" b64_data = base64.b64encode(image_bytes).decode("ascii") body = { "model": self._model, "input": [ { "type": "message", "role": "user", "content": [ {"type": "input_text", "text": system_prompt + "\n\n" + user_text}, { "type": "input_image", "image_url": f"data:image/png;base64,{b64_data}", }, ], }, ], "text": { "format": { "type": "json_schema", "name": "variation_recommendation", "strict": True, "schema": _RESPONSE_SCHEMA, } }, "temperature": 0.3, "store": False, } req = urllib.request.Request( OPENAI_API_URL, data=json.dumps(body).encode("utf-8"), headers={ "Authorization": f"Bearer {_api_key}", "Content-Type": "application/json", }, method="POST", ) timeout = settings.llm_timeout_validation max_retries = settings.llm_max_retries last_error: Exception | None = None for attempt in range(1, max_retries + 2): try: with urllib.request.urlopen(req, timeout=timeout) as resp: payload = json.loads(resp.read().decode("utf-8")) break except urllib.error.HTTPError as exc: error_text = exc.read().decode("utf-8", errors="replace") last_error = RuntimeError(f"GPT Vision API error {exc.code}: {error_text}") # urllib 은 provider 를 안 실어 브로커가 OpenAI 호출인지 모른다 — # 올라가는 예외에 표시를 붙여야 기존 판정 계약이 적용된다. from app.core.openai_keys import mark_openai_failure mark_openai_failure( last_error, status=exc.code, body=error_text) # ★키 수준 실패는 같은 키로 재시도해도 풀리지 않는다 — quota 소진이 # 429 로 올 때 죽은 키를 max_retries 번 더 태우고 backoff 까지 # 걸었다(실측 13.8초). 전환은 브로커가 판단하게 넘긴다. from app.core.openai_keys import is_key_level_failure if (exc.code in {429, 500, 502, 503, 504} and attempt <= max_retries and not is_key_level_failure(last_error)): time.sleep(2 * attempt) continue raise last_error from exc except (urllib.error.URLError, socket.timeout) as exc: last_error = exc if attempt <= max_retries: time.sleep(2 * attempt) continue raise RuntimeError(f"GPT Vision API failed after retries: {exc}") from exc else: raise RuntimeError(f"GPT Vision API failed: {last_error}") # Extract result output_text = payload.get("output_text") if isinstance(output_text, str) and output_text.strip(): return json.loads(output_text) output = payload.get("output") if isinstance(output, list): for item in output: if not isinstance(item, dict): continue content = item.get("content") if not isinstance(content, list): continue for part in content: if isinstance(part, dict) and part.get("type") == "output_text": text = part.get("text") if isinstance(text, str) and text.strip(): return json.loads(text) raise RuntimeError("GPT Vision response did not include result.") def _active_openai_key() -> str: """활성 키 슬롯 — 1차 필드를 직접 보면 보조 키가 안 쓰인다.""" from app.core.openai_keys import active_key return active_key()