"""씬 이미지 생성 모듈 — 장면별 시네마 프레임 생성.""" # Version: 1.4.0 — structured prompt with separated sections (scene/world/character/previous) # prompt_dependency: prototype_prompts/v5 # updated_at: 2026-03-16 import json import logging import uuid from datetime import datetime, timezone from pathlib import Path from typing import Any, Dict, List, Optional, Sequence, Tuple from app.modules.llm.gemini_image_client import GeminiImageClient, ModerationError from app.modules.reference_image_generator import ( _build_world_guide_block, _entity_type_label, _format_bulleted_block, ) logger = logging.getLogger(__name__) PROMPTS_DIR = Path(__file__).resolve().parent.parent.parent.parent / "prompts" / "_base" / "prototype_prompts" / "v5" SCENE_ASPECT_RATIO = "16:9" MAX_SANITIZE_ATTEMPTS = 3 def _load_prompt(filename: str) -> str: return (PROMPTS_DIR / filename).read_text(encoding="utf-8").strip() def _special_entity_guardrail_lines(names: Sequence[str], language_code: str) -> List[str]: """더 이상 하드코딩된 가드레일 없음. 빈 리스트 반환.""" return [] def _build_scene_prompt( still: Dict[str, Any], visible_entities: List[Dict[str, Any]], world_guide: Dict[str, Any], language_code: str, include_previous: bool = False, previous_still: Optional[Dict[str, Any]] = None, review_note: str = "", ) -> str: """Build a structured scene image generation prompt. Structured sections in order: 1. [SCENE DESCRIPTION] — the T2I prompt text (NO angle/color here) 2. [WORLD CONTEXT] — brief world setting summary (1-2 sentences) 3. Reference block — character/prop reference labels for text prompt 4. Camera/lighting instructions 5. Scene rules (guardrails) 6. Optional review notes NOTE: Angle/color instructions are NOT included in the base prompt. Those belong only in A/B i2i variation edits. """ # Brief world context (1-2 sentences max) world_summary = world_guide.get("world_setting_summary", "") if len(world_summary) > 200: world_summary = world_summary[:200].rsplit(".", 1)[0] + "." visible_names = [e.get("name", "") for e in visible_entities] special_guardrails = _special_entity_guardrail_lines(visible_names, language_code) # Classify entities by type for structured prompt characters = [e for e in visible_entities if e.get("entity_type") == "character"] props = [e for e in visible_entities if e.get("entity_type") == "prop"] # NOTE: Location references are intentionally excluded from scene generation. # Background consistency is maintained through same-background scene linking # (previous scene reference) instead of location reference images. ko = language_code == "ko" # Build reference block for text prompt reference_lines = [] ref_idx = 1 if characters: section_title = "[CHARACTER REFERENCES - 이 씬에 등장하는 인물 참조 이미지]" if ko else "[CHARACTER REFERENCES]" reference_lines.append(section_title) for e in characters: name = e.get("name", "") # Extract visual anchor traits for distinguishing characters brief_traits = "" try: traits_raw = e.get("stable_traits", "{}") if isinstance(traits_raw, str): import json as _json traits_data = _json.loads(traits_raw) else: traits_data = traits_raw anchors = traits_data.get("visual_anchor_traits", []) if anchors: brief_traits = " (" + ", ".join(anchors[:3]) + ")" except Exception: pass if ko: reference_lines.append( f"- 참조 이미지 {ref_idx}: {name}{brief_traits} " f"({_entity_type_label('character', 'ko')}) 기준 이미지. " "이 이미지의 인물을 참조하여 정체성과 형태를 고정하고, 장면 상태는 아래 지시를 따른다." ) else: reference_lines.append( f"- Reference image {ref_idx}: {name}{brief_traits} " f"({_entity_type_label('character', 'en')}) anchor. " "Preserve identity and shape; use this scene description for actual wardrobe and state." ) ref_idx += 1 if props: section_title = "[PROP REFERENCES - 이 씬에 등장하는 소품 참조 이미지]" if ko else "[PROP REFERENCES]" reference_lines.append(section_title) for e in props: name = e.get("name", "") if ko: reference_lines.append( f"- 참조 이미지 {ref_idx}: {name} " f"({_entity_type_label('prop', 'ko')}) 기준 이미지. " "형태와 재질을 유지하라." ) else: reference_lines.append( f"- Reference image {ref_idx}: {name} " f"({_entity_type_label('prop', 'en')}) anchor. " "Preserve shape and material." ) ref_idx += 1 if include_previous and previous_still is not None: section_title = "[PREVIOUS SCENE - 직전 관련 씬 이미지 (연속성 유지용)]" if ko else "[PREVIOUS SCENE - continuity anchor]" reference_lines.append(section_title) beat = previous_still.get("beat_title", "") if ko: reference_lines.append( f"- 참조 이미지 {ref_idx}: 바로 앞의 연속 장면 이미지. " f"이전 장면 제목은 '{beat}' 이다. 의상 상태와 위치 연속성을 이어라." ) else: reference_lines.append( f"- Reference image {ref_idx}: previous related scene image from " f"'{beat}'. Carry wardrobe and spatial continuity forward." ) # Parse camera and lighting from JSON camera_json = still.get("camera_json", "{}") if isinstance(camera_json, str): try: camera = json.loads(camera_json) except json.JSONDecodeError: camera = {} else: camera = camera_json lighting_json = still.get("lighting_json", "{}") if isinstance(lighting_json, str): try: lighting = json.loads(lighting_json) except json.JSONDecodeError: lighting = {} else: lighting = lighting_json camera_block = _format_bulleted_block( [f"{k}: {v}" for k, v in camera.items()], "카메라 지시 없음." if ko else "No camera note.", ) lighting_block = _format_bulleted_block( [f"{k}: {v}" for k, v in lighting.items()], "조명 지시 없음." if ko else "No lighting note.", ) # Build structured prompt with clear sections t2i_prompt = still.get("still_frame_prompt", "") scene_heading = still.get("screenplay_scene_heading", "") beat_title = still.get("beat_title", "") sections = [] # Section 1: Scene description (comes first — primary intent) sections.append(f"[SCENE DESCRIPTION]\n{t2i_prompt}") # Section 2: World context (brief) if world_summary: sections.append(f"[WORLD CONTEXT]\n{world_summary}") # Section 3: Scene metadata meta_lines = [] if beat_title: meta_label = "장면 제목" if ko else "Beat title" meta_lines.append(f"- {meta_label}: {beat_title}") if scene_heading: meta_label = "스크린플레이 헤딩" if ko else "Scene heading" meta_lines.append(f"- {meta_label}: {scene_heading}") if meta_lines: sections.append("[SCENE META]\n" + "\n".join(meta_lines)) # Section 4: Reference image labels ref_text = "\n".join(reference_lines + [f"- {item}" for item in special_guardrails]) if ref_text.strip(): sections.append(ref_text) # Section 5: Camera and lighting sections.append(f"[CAMERA]\n{camera_block}") sections.append(f"[LIGHTING]\n{lighting_block}") # Section 6: Scene rules — WorldGuide의 style_rules에서 동적 생성 style_rules = world_guide.get("style_rules", {}) must_maintain = style_rules.get("must_maintain", []) must_avoid = style_rules.get("must_avoid", []) aspect_ratio = style_rules.get("aspect_ratio", SCENE_ASPECT_RATIO) rules_lines = ["[SCENE RULES]"] if must_maintain: for rule in must_maintain: rules_lines.append(f"- 유지: {rule}" if ko else f"- Maintain: {rule}") if must_avoid: for rule in must_avoid: rules_lines.append(f"- 금지: {rule}" if ko else f"- Avoid: {rule}") # 기본 규칙 (항상 적용) rules_lines.append(f"- {'참조 이미지는 정체성과 형태 고정용이다.' if ko else 'Reference images fix identity and form.'}") rules_lines.append(f"- {'이미지 안에 텍스트, 라벨, 참조 번호 금지.' if ko else 'No text, labels, or reference numbers inside image.'}") rules_lines.append(f"- {'최종 이미지는 ' + aspect_ratio + ' 비율 시네마 프레임.' if ko else 'Final image: ' + aspect_ratio + ' cinema frame.'}") sections.append("\n".join(rules_lines)) prompt = "\n\n".join(sections) # Optional: review notes if review_note.strip(): heading = "[수정 우선 메모]" if ko else "[Manual fix notes]" intro = ( "- 아래 검수 메모를 이번 재생성에서 우선 반영하라." if ko else "- Prioritize the following reviewer notes in this regeneration." ) note_lines = [f"- {line.strip()}" for line in review_note.strip().splitlines() if line.strip()] prompt = "\n\n".join([prompt, heading, "\n".join([intro] + note_lines)]) return prompt def _build_labeled_references( visible_entities: List[Dict[str, Any]], reference_image_map: Optional[Dict[str, bytes]], previous_scene_bytes: Optional[bytes], previous_still: Optional[Dict[str, Any]], language_code: str, ) -> List[Tuple[str, bytes]]: """Build labeled (label, image_bytes) tuples for Gemini reference images. Structured format — each reference image is preceded by a clear text label identifying what the image is (character name, prop name, or previous scene). Section ordering: 1. Character references (each labeled with name + role) 2. Prop references (each labeled with name) 3. Previous scene (for same-location continuity) NOTE: Location references are intentionally excluded. Background consistency is maintained through same-background scene linking (previous scene reference) instead of location reference images. NOTE: Angle/color instructions are NOT included here — those belong only to A/B i2i variation edits (already implemented separately). """ labeled: List[Tuple[str, bytes]] = [] if not reference_image_map: reference_image_map = {} characters = [e for e in visible_entities if e.get("entity_type") == "character"] # Locations intentionally skipped — scene-to-scene continuity replaces location refs props = [e for e in visible_entities if e.get("entity_type") == "prop"] ko = language_code == "ko" # Character references — numbered, with distinguishing traits for identity ref_num = 1 for entity in characters: eid = entity.get("id", "") if eid in reference_image_map: name = entity.get("name", "") # Extract visual anchor traits for distinguishing characters brief_traits = "" try: traits_raw = entity.get("stable_traits", "{}") if isinstance(traits_raw, str): import json as _json traits_data = _json.loads(traits_raw) else: traits_data = traits_raw anchors = traits_data.get("visual_anchor_traits", []) if anchors: brief_traits = ". " + ", ".join(anchors[:4]) except Exception: pass label = ( f"Reference image {ref_num}: [{name}]{brief_traits}. " f"Keep exact face, hair, and build from this image." ) labeled.append((label, reference_image_map[eid])) ref_num += 1 # Prop references — numbered, matching [entity_name] markers in T2I prompt for entity in props: eid = entity.get("id", "") if eid in reference_image_map: name = entity.get("name", "") label = ( f"Reference image {ref_num}: [{name}] — object shape and material only. " f"Preserve exact appearance from this image." ) labeled.append((label, reference_image_map[eid])) ref_num += 1 # Previous scene — for same-location visual continuity if previous_scene_bytes is not None and previous_still is not None: beat = previous_still.get("beat_title", "") label = ( f"Reference image {ref_num}: [Previous Scene] — same location, maintain visual continuity. " f"Previous: '{beat}'. Keep environment, wardrobe, spatial layout." ) labeled.append((label, previous_scene_bytes)) return labeled class SceneImageGenerator: """장면 이미지 생성기.""" def __init__( self, gemini_client: GeminiImageClient, language: str = "ko", ) -> None: self._gemini = gemini_client self._language = language def generate_for_still( self, still: Dict[str, Any], visible_entities: List[Dict[str, Any]], world_guide: Dict[str, Any], output_dir: Path, reference_image_map: Optional[Dict[str, bytes]] = None, previous_scene_bytes: Optional[bytes] = None, previous_still: Optional[Dict[str, Any]] = None, review_note: str = "", episode_id: Optional[str] = None, aspect_ratio: str = "16:9", ) -> Dict[str, Any]: """Generate a scene image for a single still. Args: still: Scene still data dict. visible_entities: List of entity dicts visible in this scene. world_guide: World guide dict. output_dir: Directory to save the image. reference_image_map: entity_id -> PNG bytes for reference images. previous_scene_bytes: PNG bytes of the previous scene for continuity. previous_still: Previous still data for prompt context. review_note: Manual reviewer notes for regeneration. episode_id: Episode ID for the asset record. Returns: ImageAsset-like dict. """ include_previous = previous_scene_bytes is not None and previous_still is not None prompt = _build_scene_prompt( still=still, visible_entities=visible_entities, world_guide=world_guide, language_code=self._language, include_previous=include_previous, previous_still=previous_still, review_note=review_note, ) # Build labeled reference images for structured Gemini input labeled_refs = _build_labeled_references( visible_entities=visible_entities, reference_image_map=reference_image_map, previous_scene_bytes=previous_scene_bytes if include_previous else None, previous_still=previous_still if include_previous else None, language_code=self._language, ) image_bytes, _elapsed_ms = self._gemini.generate_image( prompt=prompt, labeled_references=labeled_refs if labeled_refs else None, aspect_ratio=SCENE_ASPECT_RATIO, ) still_id = still.get("id", str(uuid.uuid4())) output_dir.mkdir(parents=True, exist_ok=True) out_path = output_dir / f"{uuid.uuid4()}.png" out_path.write_bytes(image_bytes) return { "id": str(uuid.uuid4()), "asset_type": "scene", "entity_id": None, "still_id": still_id, "episode_id": episode_id, "file_path": str(out_path), "prompt_used": prompt, "generation_model": self._gemini._model, "width": None, "height": None, "status": "generated", "review_notes": "", "created_at": datetime.now(timezone.utc).isoformat(), } def generate_for_still_with_retry( self, still: Dict[str, Any], visible_entities: List[Dict[str, Any]], world_guide: Dict[str, Any], output_dir: Path, sanitizer: Any, tracker: Any, reference_image_map: Optional[Dict[str, bytes]] = None, previous_scene_bytes: Optional[bytes] = None, previous_still: Optional[Dict[str, Any]] = None, review_note: str = "", episode_id: Optional[str] = None, aspect_ratio: str = "16:9", ) -> Dict[str, Any]: """Generate scene image with moderation retry. 1. Try original prompt 2. If ModerationError: sanitize prompt, retry 3. Up to 3 sanitization attempts 4. Track all attempts in GenerationTracker Args: still: Scene still data dict. visible_entities: Entities visible in this scene. world_guide: World guide dict. output_dir: Directory to save the image. sanitizer: PromptSanitizer instance. tracker: GenerationTracker instance. reference_image_map: entity_id -> PNG bytes. previous_scene_bytes: PNG bytes of previous scene. previous_still: Previous still data. review_note: Reviewer notes. episode_id: Episode ID. Returns: ImageAsset-like dict. """ include_previous = previous_scene_bytes is not None and previous_still is not None still_id = still.get("id", str(uuid.uuid4())) prompt = _build_scene_prompt( still=still, visible_entities=visible_entities, world_guide=world_guide, language_code=self._language, include_previous=include_previous, previous_still=previous_still, review_note=review_note, ) labeled_refs = _build_labeled_references( visible_entities=visible_entities, reference_image_map=reference_image_map, previous_scene_bytes=previous_scene_bytes if include_previous else None, previous_still=previous_still if include_previous else None, language_code=self._language, ) current_prompt = prompt prompt_version = "original" last_strategy: Optional[str] = None last_sanitize_note: Optional[str] = None for attempt in range(1, MAX_SANITIZE_ATTEMPTS + 2): # 1 original + up to 3 sanitized try: image_bytes, elapsed_ms = self._gemini.generate_image( prompt=current_prompt, labeled_references=labeled_refs if labeled_refs else None, aspect_ratio=aspect_ratio, ) # Success -- save and record output_dir.mkdir(parents=True, exist_ok=True) out_path = output_dir / f"{uuid.uuid4()}.png" out_path.write_bytes(image_bytes) asset_id = str(uuid.uuid4()) tracker.record( still_id=still_id, image_asset_id=asset_id, attempt_number=attempt, prompt_used=current_prompt, prompt_version=prompt_version, model_name=self._gemini._model, status="success", response_time_ms=elapsed_ms, ) return { "id": asset_id, "asset_type": "scene", "entity_id": None, "still_id": still_id, "episode_id": episode_id, "file_path": str(out_path), "prompt_used": current_prompt, "generation_model": self._gemini._model, "width": None, "height": None, "status": "generated", "review_notes": "", "created_at": datetime.now(timezone.utc).isoformat(), "sanitization_strategy": last_strategy, "original_prompt": prompt if last_strategy else None, "sanitization_note": last_sanitize_note, } except ModerationError as exc: logger.warning( "Scene moderation block: still=%s, attempt=%d, reason=%s", still_id, attempt, exc.block_reason, ) tracker.record( still_id=still_id, attempt_number=attempt, prompt_used=current_prompt, prompt_version=prompt_version, model_name=self._gemini._model, status="moderation_blocked", block_reason=exc.block_reason, block_categories=exc.block_categories, ) # Try sanitization if we haven't exhausted attempts if attempt <= MAX_SANITIZE_ATTEMPTS: try: sanitize_result = sanitizer.sanitize( original_prompt=current_prompt, block_reason=exc.block_reason, block_categories=exc.block_categories, attempt=attempt, ) current_prompt = sanitize_result["sanitized_prompt"] prompt_version = f"sanitized_v{attempt}" last_strategy = sanitize_result.get("strategy") last_sanitize_note = sanitize_result.get("changes") logger.info( "Prompt sanitized for still=%s: strategy=%s", still_id, last_strategy, ) except Exception as san_exc: logger.error("Sanitization failed: %s", san_exc) raise exc from san_exc else: raise except Exception as exc: tracker.record( still_id=still_id, attempt_number=attempt, prompt_used=current_prompt, prompt_version=prompt_version, model_name=self._gemini._model, status="error", ) raise # Should not reach here raise RuntimeError(f"Scene image generation exhausted all retries for still {still_id}")