"""Gemini Nano Banana 2 (gemini-3.1-flash-image-preview) 이미지 생성 스크립트. SA(서비스 계정) Bearer 토큰으로 Gemini API REST를 직접 호출합니다(방법 3). 시나리오 A, B, C에 대한 보험 GA 리크루팅 광고 이미지를 생성합니다. """ from __future__ import annotations import base64 import json import time from datetime import datetime from pathlib import Path from typing import Any import requests import gcloud_auth OUTPUT_DIR = Path("/home/jay/workspace/tools/ai-image-gen/output/v2-gemini-nb2") RESULTS_JSON = OUTPUT_DIR / "results.json" ERRORS_LOG = OUTPUT_DIR / "errors.log" MODEL_ID = "gemini-3.1-flash-image-preview" GEMINI_API_BASE = "https://generativelanguage.googleapis.com/v1beta" GEMINI_SCOPE = "https://www.googleapis.com/auth/generative-language" SCENARIOS: dict[str, str] = { "A": ( "Professional Korean insurance general agency recruiting advertisement. " "Modern office environment with confident business professionals. " "Clean, premium design. 1080x1080 square format for Meta ads. " "Text overlay: '당신의 커리어, 새로운 시작' in Korean. " "Color scheme: navy blue and gold. Corporate premium feel." ), "B": ( "Korean insurance consultant personal branding image. " "Confident male professional in business suit, modern Seoul office backdrop. " "Warm lighting, trustworthy atmosphere. 1080x1080 square. " "Subtle text: '보험의 모든 것' in elegant Korean typography." ), "C": ( "Motivational career change advertisement for insurance professionals. " "Split image: left side dark/stressful office, right side bright/modern workspace. " "Aspirational feel. 1080x1080 square format. " "Korean text: '지금이 기회입니다' centered." ), } def log_error(message: str) -> None: """에러 로그를 파일에 기록합니다.""" timestamp = datetime.now().isoformat() with open(ERRORS_LOG, "a", encoding="utf-8") as f: f.write(f"[{timestamp}] {message}\n") print(f"[ERROR] {message}") def generate_image_via_gemini_api( token: str, prompt: str, output_path: Path, scenario: str, ) -> dict[str, Any]: """Gemini API REST 호출로 이미지를 생성하고 PNG 파일로 저장합니다.""" url = f"{GEMINI_API_BASE}/models/{MODEL_ID}:generateContent" headers = { "Authorization": f"Bearer {token}", "Content-Type": "application/json", } payload: dict[str, Any] = { "contents": [{"parts": [{"text": prompt}]}], "generationConfig": {"responseModalities": ["IMAGE", "TEXT"]}, } print(f"[시나리오 {scenario}] 이미지 생성 요청 중... (모델: {MODEL_ID})") start_time = time.time() response = requests.post(url, headers=headers, json=payload, timeout=180) response.raise_for_status() elapsed = time.time() - start_time data: dict[str, Any] = response.json() candidates = data.get("candidates", []) if not candidates: raise RuntimeError(f"응답에 candidates가 없습니다. 응답: {json.dumps(data)[:300]}") parts = candidates[0].get("content", {}).get("parts", []) image_part: dict[str, Any] | None = None for part in parts: if "inlineData" in part: image_part = part break if image_part is None: text_parts = [p.get("text", "") for p in parts if "text" in p] raise RuntimeError(f"이미지 데이터가 응답에 없습니다. " f"텍스트 파트: {text_parts[:2]}") mime_type: str = image_part["inlineData"].get("mimeType", "image/jpeg") image_b64: str = image_part["inlineData"]["data"] image_bytes = base64.b64decode(image_b64) # 실제 MIME 타입에 맞는 확장자로 저장 ext = ".jpg" if "jpeg" in mime_type else ".png" if output_path.suffix.lower() != ext: output_path = output_path.with_suffix(ext) output_path.write_bytes(image_bytes) file_size = output_path.stat().st_size print( f"[시나리오 {scenario}] 완료: {output_path.name} " f"({file_size:,} bytes, {elapsed:.1f}초, mime={mime_type})" ) return { "scenario": scenario, "model": MODEL_ID, "auth_method": "sa_bearer_token_generative-language_scope", "timestamp": datetime.now().isoformat(), "filename": output_path.name, "filepath": str(output_path), "time_seconds": round(elapsed, 2), "file_size_bytes": file_size, "mime_type": mime_type, "error": None, } def main() -> None: """메인: 3개 시나리오 이미지 생성.""" OUTPUT_DIR.mkdir(parents=True, exist_ok=True) print("=" * 60) print("Gemini Nano Banana 2 이미지 생성 시작") print(f"모델: {MODEL_ID}") print(f"출력 디렉토리: {OUTPUT_DIR}") print(f"인증 방법: SA Bearer 토큰 (generative-language scope)") print("=" * 60) # SA Bearer 토큰 획득 print("\n[인증] SA 서비스 계정 토큰 획득 중...") try: token = gcloud_auth.get_service_account_token(GEMINI_SCOPE) print(f"[인증] SA 토큰 획득 성공 (길이: {len(token)} chars)") except Exception as e: error_msg = f"SA 토큰 획득 실패: {type(e).__name__}: {e}" log_error(error_msg) raise SystemExit(1) from e results: list[dict[str, Any]] = [] total_start = time.time() for scenario in ["A", "B", "C"]: prompt = SCENARIOS[scenario] output_path = OUTPUT_DIR / f"gemini_nb2_{scenario}.jpg" result: dict[str, Any] = { "scenario": scenario, "model": MODEL_ID, "auth_method": "sa_bearer_token_generative-language_scope", "timestamp": datetime.now().isoformat(), "filename": f"gemini_nb2_{scenario}.png", "filepath": str(output_path), "time_seconds": None, "file_size_bytes": None, "mime_type": None, "error": None, } print(f"\n[{scenario}] 시나리오 처리 중...") print(f" 프롬프트: {prompt[:80]}...") try: gen_result = generate_image_via_gemini_api(token, prompt, output_path, scenario) result.update(gen_result) except requests.HTTPError as e: error_msg = f"시나리오 {scenario} HTTP 오류: " f"{e.response.status_code} - {e.response.text[:500]}" log_error(error_msg) result["error"] = error_msg except Exception as e: error_msg = f"시나리오 {scenario} 오류: {type(e).__name__}: {e}" log_error(error_msg) result["error"] = error_msg results.append(result) total_elapsed = time.time() - total_start # results.json 저장 results_data: dict[str, Any] = { "run_timestamp": datetime.now().isoformat(), "model": MODEL_ID, "auth_method": "sa_bearer_token_generative-language_scope", "total_time_seconds": round(total_elapsed, 2), "scenarios": results, } with open(RESULTS_JSON, "w", encoding="utf-8") as f: json.dump(results_data, f, ensure_ascii=False, indent=2) # 요약 출력 print("\n" + "=" * 60) print("생성 결과 요약") print("=" * 60) success_count = sum(1 for r in results if r.get("error") is None) fail_count = len(results) - success_count print(f"성공: {success_count}/3, 실패: {fail_count}/3") print(f"총 소요 시간: {total_elapsed:.1f}초") print(f"결과 저장: {RESULTS_JSON}") if ERRORS_LOG.exists(): print(f"에러 로그: {ERRORS_LOG}") for r in results: status = "OK" if r.get("error") is None else "FAIL" size = r.get("file_size_bytes") t = r.get("time_seconds") size_str = f"{size:,} bytes" if size else "N/A" time_str = f"{t:.2f}초" if t else "N/A" print(f" [{r['scenario']}] {status} | {size_str} | {time_str}") if __name__ == "__main__": main()