#!/usr/bin/env python3 """ 일괄 이미지 생성 스크립트 Gemini gemini-2.5-flash-image (나노바나나) 모델로 스팟 이미지 일괄 생성 사용법: # 모든 카테고리 폴백 이미지 생성 python scripts/batch_generate_images.py --fallback # 특정 카테고리 스팟 이미지 생성 (최대 5개) python scripts/batch_generate_images.py --category beach --limit 5 # generated_image_url이 없는 모든 스팟 (최대 10개) python scripts/batch_generate_images.py --limit 10 # 특정 스팟 ID로 생성 python scripts/batch_generate_images.py --spot-ids spot1 spot2 spot3 # 드라이런 (실제 생성 없이 대상 목록만 확인) python scripts/batch_generate_images.py --dry-run --limit 20 환경변수: GEMINI_API_KEY 또는 GOOGLE_API_KEY: Gemini API 키 SUPABASE_URL: Supabase 프로젝트 URL SUPABASE_SERVICE_ROLE_KEY: Supabase service role 키 """ import argparse import asyncio import base64 import logging import os import sys import time # Add parent directory to path for imports sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from dotenv import load_dotenv load_dotenv() from db import get_supabase logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s", ) logger = logging.getLogger(__name__) # Import prompts from the router from routers.image_gen import CATEGORY_PROMPTS, DEFAULT_PROMPT, NANO_BANANA_MODEL, STORAGE_BUCKET def get_genai_client(): """Get Google GenAI client""" from google import genai api_key = os.getenv("GEMINI_API_KEY") or os.getenv("GOOGLE_API_KEY") if not api_key: raise ValueError("GEMINI_API_KEY or GOOGLE_API_KEY environment variable is required") return genai.Client(api_key=api_key) def generate_image(client, prompt: str) -> tuple[bytes | None, str]: """Generate image with Gemini and return (image_bytes, mime_type)""" from google.genai import types response = client.models.generate_content( model=NANO_BANANA_MODEL, contents=[prompt], config=types.GenerateContentConfig( response_modalities=["IMAGE", "TEXT"], ), ) if response.candidates and response.candidates[0].content: for part in response.candidates[0].content.parts: if hasattr(part, "inline_data") and part.inline_data: data = part.inline_data.data mime = part.inline_data.mime_type or "image/webp" image_bytes = data if isinstance(data, bytes) else base64.b64decode(data) return image_bytes, mime return None, "" def upload_to_storage(supabase, path: str, image_bytes: bytes, mime_type: str) -> str: """Upload image to Supabase Storage and return public URL""" try: supabase.storage.from_(STORAGE_BUCKET).remove([path]) except Exception: pass supabase.storage.from_(STORAGE_BUCKET).upload( path=path, file=image_bytes, file_options={"content-type": mime_type, "upsert": "true"}, ) return supabase.storage.from_(STORAGE_BUCKET).get_public_url(path) def generate_fallback_images(dry_run: bool = False): """Generate category fallback images""" logger.info(f"Generating fallback images for {len(CATEGORY_PROMPTS)} categories...") if dry_run: for category in CATEGORY_PROMPTS: logger.info(f" [DRY-RUN] Would generate fallback for: {category}") return client = get_genai_client() supabase = get_supabase() success = 0 failed = 0 for category, prompt_text in CATEGORY_PROMPTS.items(): try: full_prompt = ( f"Generate a representative photograph for the '{category}' category " f"of places in Jeju Aewol area. {prompt_text} " f"No text overlays, no watermarks, no people's faces." ) logger.info(f" Generating fallback for: {category}") image_bytes, mime_type = generate_image(client, full_prompt) if not image_bytes: logger.warning(f" No image data for {category}") failed += 1 continue ext = "webp" if "png" in mime_type: ext = "png" elif "jpeg" in mime_type or "jpg" in mime_type: ext = "jpg" path = f"fallback/{category}.{ext}" url = upload_to_storage(supabase, path, image_bytes, mime_type) logger.info(f" ✅ {category} → {url}") success += 1 # Rate limiting time.sleep(2) except Exception as e: logger.error(f" ❌ {category} failed: {e}") failed += 1 logger.info(f"\nFallback generation complete: {success} success, {failed} failed") def generate_spot_images( spot_ids: list[str] | None = None, category: str | None = None, limit: int = 10, dry_run: bool = False, ): """Generate images for spots""" supabase = get_supabase() # Build query if spot_ids: result = supabase.table("story_spots").select( "id, name, category" ).in_("id", spot_ids).execute() elif category: result = supabase.table("story_spots").select( "id, name, category" ).eq("category", category).is_("generated_image_url", "null").limit(limit).execute() else: result = supabase.table("story_spots").select( "id, name, category" ).is_("generated_image_url", "null").limit(limit).execute() spots = result.data or [] logger.info(f"Found {len(spots)} spots to generate images for") if not spots: logger.info("No spots found matching criteria") return if dry_run: for spot in spots: logger.info(f" [DRY-RUN] Would generate: {spot['id']} ({spot['name']}) [{spot['category']}]") return client = get_genai_client() success = 0 failed = 0 for i, spot in enumerate(spots, 1): try: cat = spot.get("category", "coastline") base_prompt = CATEGORY_PROMPTS.get(cat, DEFAULT_PROMPT) full_prompt = ( f"Generate a high-quality photograph of {spot['name']} in Jeju Aewol area. " f"{base_prompt} " f"No text overlays, no watermarks, no people's faces." ) logger.info(f" [{i}/{len(spots)}] Generating: {spot['name']} ({cat})") image_bytes, mime_type = generate_image(client, full_prompt) if not image_bytes: logger.warning(f" No image data for {spot['id']}") failed += 1 continue ext = "webp" if "png" in mime_type: ext = "png" elif "jpeg" in mime_type or "jpg" in mime_type: ext = "jpg" path = f"{spot['id']}.{ext}" url = upload_to_storage(supabase, path, image_bytes, mime_type) # Update DB supabase.table("story_spots").update({ "generated_image_url": url }).eq("id", spot["id"]).execute() logger.info(f" ✅ {spot['name']} → {url}") success += 1 # Rate limiting time.sleep(2) except Exception as e: logger.error(f" ❌ {spot['name']} failed: {e}") failed += 1 logger.info(f"\nSpot image generation complete: {success} success, {failed} failed out of {len(spots)}") def main(): parser = argparse.ArgumentParser(description="Batch generate spot images with Gemini Nano Banana") parser.add_argument("--fallback", action="store_true", help="Generate category fallback images") parser.add_argument("--category", type=str, help="Generate for specific category") parser.add_argument("--spot-ids", nargs="+", help="Generate for specific spot IDs") parser.add_argument("--limit", type=int, default=10, help="Max spots to generate (default: 10)") parser.add_argument("--dry-run", action="store_true", help="Show what would be generated without doing it") args = parser.parse_args() if args.fallback: generate_fallback_images(dry_run=args.dry_run) else: generate_spot_images( spot_ids=args.spot_ids, category=args.category, limit=args.limit, dry_run=args.dry_run, ) if __name__ == "__main__": main()