"""레퍼런스 이미지 생성 모듈 — 엔티티별 기준 이미지 생성.""" # Version: 1.2.0 — photorealistic style enforcement, v2 reference prompts # prompt_dependency: prototype_prompts/v5 + reference_image/v2 # 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 from app.modules.llm.gemini_image_client import GeminiImageClient, ModerationError logger = logging.getLogger(__name__) MAX_SANITIZE_ATTEMPTS = 3 PROMPTS_DIR = Path(__file__).resolve().parent.parent.parent.parent / "prompts" / "_base" / "prototype_prompts" / "v5" REFERENCE_STYLE_DIR = Path(__file__).resolve().parent.parent.parent.parent / "prompts" / "_base" / "reference_image" / "v2" REFERENCE_ASPECT_RATIO = { "character": "3:4", "location": "16:9", "prop": "1:1", } # SPECIAL_ENTITY_GUARDRAILS 제거됨 — 특정 시나리오 고유명사 하드코딩 금지. # 요소별 가드레일은 entity description과 world_guide의 must_avoid에서 동적으로 결정. def _load_prompt(filename: str) -> str: return (PROMPTS_DIR / filename).read_text(encoding="utf-8").strip() def _load_reference_style(entity_type: str) -> str: """Load photorealistic style requirements for the given entity type.""" style_map = { "character": "character.md", "location": "location.md", "prop": "prop.md", } filename = style_map.get(entity_type) if filename: path = REFERENCE_STYLE_DIR / filename if path.exists(): return path.read_text(encoding="utf-8").strip() return "" def _entity_type_label(entity_type: str, language_code: str) -> str: labels = { "ko": {"character": "인물", "location": "배경/장소", "prop": "중요 소품"}, "en": {"character": "character", "location": "location", "prop": "prop"}, } locale = "ko" if language_code == "ko" else "en" return labels[locale].get(entity_type, entity_type) def _entity_type_reference_rules(entity_type: str, language_code: str) -> List[str]: if language_code == "ko": mapping = { "character": [ "중립적인 기본 자세로 보여 주고, 얼굴과 체형 식별이 우선이다.", "장면성 포즈나 일시적 상태보다 기본 외형과 기본 복장 톤을 우선한다.", "visual_world_rules 의 시대·지역 설정에 맞는 현실적 기본 복장을 사용하라.", ], "location": [ "공간 구조가 한눈에 들어오는 establishing 구도로 표현하라.", "군중이나 주연 인물은 넣지 않거나 최소화하라.", "공간 재질, 조명 구조, 출입 동선, 핵심 설비를 분명히 보여라.", ], "prop": [ "사람 손이나 몸에서 떼어 낸 단독 기준 이미지로 보여라.", "형태, 재질, 장착 구조, 조작부가 선명해야 한다.", "장비의 크기감이 읽히도록 깔끔하게 배치하라.", ], } else: mapping = { "character": [ "Use a neutral baseline pose and prioritize face and body recognition.", "Prefer stable appearance over scene-specific action or transient state.", "Use realistic baseline clothing appropriate to the era and region defined in visual_world_rules.", ], "location": [ "Use an establishing composition that clearly explains the space.", "Avoid crowds and named characters unless absolutely necessary.", "Show material, lighting structure, circulation path, and key built-in devices.", ], "prop": [ "Show it as an isolated reference without hands or a human body.", "Make shape, material, attachment structure, and controls clearly readable.", "Present it cleanly so scale and silhouette are easy to understand.", ], } return mapping.get(entity_type, []) def _special_entity_guardrail_lines(names: Sequence[str], language_code: str) -> List[str]: """더 이상 하드코딩된 가드레일 없음. 빈 리스트 반환.""" return [] def _format_bulleted_block(items: Sequence[str], empty_line: str) -> str: if not items: return f"- {empty_line}" return "\n".join(f"- {item}" for item in items) def _build_world_guide_block(world_guide: Dict[str, Any]) -> str: summary = world_guide.get("world_setting_summary", "") lines = [ f"시대/배경: {world_guide.get('world_time_period', '')}", f"세계 요약: {summary}", f"기술 수준: {world_guide.get('technology_level', '')}", "핵심 시대 규칙:", *[f"- {item}" for item in world_guide.get("era_guardrails", [])], "핵심 복장 규칙:", *[f"- {item}" for item in world_guide.get("costume_guardrails", [])], "금지 오독:", *[f"- {item}" for item in world_guide.get("prohibited_visual_misreads", [])], "연속성 규칙:", *[f"- {item}" for item in world_guide.get("continuity_guardrails", [])], ] return "\n".join(lines) def _reference_description(entity: Dict[str, Any], language_code: str) -> str: name = entity.get("name", "") entity_type = entity.get("entity_type", "") description = entity.get("description", name) if entity_type == "prop": if language_code == "ko": return f"{name}의 기준 형태, 재질, 구조를 보여 주는 중립 설명." return f"Neutral description focused on the stable shape, material, and structure of {name}." if entity_type == "location": if language_code == "ko": return (description or name) + " 사람보다 공간 구조가 우선이다." return (description or name) + " Prioritize the space itself over people." return description or name def _build_reference_prompt( entity: Dict[str, Any], world_guide: Dict[str, Any], language_code: str, ) -> str: """Build a localized entity reference image prompt.""" name = entity.get("name", "") entity_type = entity.get("entity_type", "character") description = _reference_description(entity, language_code) world_block = _build_world_guide_block(world_guide) # Parse stable_traits stable_traits = entity.get("stable_traits", "{}") if isinstance(stable_traits, str): try: traits_data = json.loads(stable_traits) except json.JSONDecodeError: traits_data = {} else: traits_data = stable_traits if isinstance(traits_data, list): visual_traits = traits_data else: visual_traits = traits_data.get("visual_anchor_traits", []) if not visual_traits and isinstance(traits_data, dict): # Flatten all values as trait lines for _k, v in traits_data.items(): if isinstance(v, str): visual_traits.append(v) traits_block = _format_bulleted_block(visual_traits, name) rule_items = _entity_type_reference_rules(entity_type, language_code) + \ _special_entity_guardrail_lines([name], language_code) rules_block = _format_bulleted_block(rule_items, "기본 식별 규칙을 유지하라.") aspect_ratio = REFERENCE_ASPECT_RATIO.get(entity_type, "3:4") template_file = "entity_reference_ko.md" # 현재 한국어 템플릿만 존재 template = _load_prompt(template_file) base_prompt = template.format( entity_name=name, entity_type_label=_entity_type_label(entity_type, language_code), description=description, traits_block=traits_block, world_guide_block=world_block, entity_type_rules=rules_block, aspect_ratio=aspect_ratio, ) # Append photorealistic style requirements from reference_image/v2 style_block = _load_reference_style(entity_type) if style_block: base_prompt = base_prompt + "\n\n" + style_block return base_prompt class ReferenceImageGenerator: """엔티티 레퍼런스 이미지 생성기.""" def __init__( self, gemini_client: GeminiImageClient, language: str = "ko", ) -> None: self._gemini = gemini_client self._language = language def generate_for_entity( self, entity: Dict[str, Any], world_guide: Dict[str, Any], output_dir: Path, ) -> Dict[str, Any]: """Generate a reference image for a single entity. Returns an ImageAsset-like dict. """ entity_id = entity["id"] entity_type = entity.get("entity_type", "character") prompt = _build_reference_prompt(entity, world_guide, self._language) aspect_ratio = REFERENCE_ASPECT_RATIO.get(entity_type, "3:4") # ── 시대 인지 사전 조사 (2026-08-14 사용자 확정 "무조건") ── # 소품·장소 엔티티만 — 인물은 identity 체계 관할이라 제외. 판별이 # "조사 불요"면 기존 경로 그대로, 조사 실패도 생성 비차단(참조 # 없이 그리던 기존 동작으로 진행 + 감사 기록). OFF=byte-identical. era_notes: Dict[str, Any] = {} era_refs: Optional[List[bytes]] = None from app.core.config import settings as _settings if bool(getattr(_settings, "era_research_enabled", False)) \ and entity_type in ("prop", "location"): from app.modules.pipeline.era_research import ( assess_subjects, build_ref_role, research_reference, ) world_block = _build_world_guide_block(world_guide) subject_text = "\n".join(filter(None, [ str(entity.get("name") or ""), str(entity.get("description") or ""), ", ".join(entity.get("visual_traits") or []) if isinstance(entity.get("visual_traits"), list) else "", ])) try: subjects = assess_subjects( step_tag="era_research_entity", subject_text=subject_text, world_facts_block=world_block, ) except Exception as exc: # noqa: BLE001 — 판별 실패 비차단 subjects = [] era_notes = {"error": f"assess: {exc!r}"[:200]} if subjects: meta = research_reference( subject=subjects[0], world_facts_block=world_block, out_path=output_dir / f"{entity_id}_eraref.png", step_tag="era_research_entity", ) if meta: prompt = prompt + "\n\n" + build_ref_role( meta["subject"]) era_refs = [ (output_dir / f"{entity_id}_eraref.png" ).read_bytes()] era_notes = {"researched": True, **meta} else: era_notes = {"researched": False, "subject": subjects[0].get( "subject_native")} elif not era_notes: era_notes = {"researched": False, "reason": "no_subject"} image_bytes, _elapsed_ms = self._gemini.generate_image( prompt=prompt, reference_images=era_refs, aspect_ratio=aspect_ratio, ) output_dir.mkdir(parents=True, exist_ok=True) out_path = output_dir / f"{entity_id}.png" out_path.write_bytes(image_bytes) return { "id": str(uuid.uuid4()), "asset_type": "reference", "entity_id": entity_id, "still_id": None, "file_path": str(out_path), "prompt_used": prompt, "generation_model": self._gemini._model, "width": None, "height": None, "status": "generated", # 시대 조사 감사 — OFF/비대상이면 기존 그대로 빈 문자열 # (byte-identical). 조사 여부·질의·선택 URL·sha 가 남아야 # "이 참조가 어디서 왔나"를 기록만으로 되짚는다. "review_notes": ( json.dumps({"era_research": era_notes}, ensure_ascii=False) if era_notes else ""), "created_at": datetime.now(timezone.utc).isoformat(), } def generate_for_entity_with_retry( self, entity: Dict[str, Any], world_guide: Dict[str, Any], output_dir: Path, sanitizer: Any, tracker: Any, extra_references: Optional[List[tuple]] = None, ) -> Dict[str, Any]: """Generate reference 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: extra_references: Optional list of (label, image_bytes) tuples from dependency entities whose images should be used as visual references for this entity. """ entity_id = entity["id"] entity_type = entity.get("entity_type", "character") prompt = _build_reference_prompt(entity, world_guide, self._language) aspect_ratio = REFERENCE_ASPECT_RATIO.get(entity_type, "3:4") current_prompt = prompt prompt_version = "original" last_strategy: Optional[str] = None last_sanitize_note: Optional[str] = None # Build labeled_references from extra_references (dependency images) labeled_refs = list(extra_references) if extra_references else None for attempt in range(1, MAX_SANITIZE_ATTEMPTS + 2): try: image_bytes, elapsed_ms = self._gemini.generate_image( prompt=current_prompt, reference_images=None, aspect_ratio=aspect_ratio, labeled_references=labeled_refs, ) output_dir.mkdir(parents=True, exist_ok=True) out_path = output_dir / f"{entity_id}.png" out_path.write_bytes(image_bytes) asset_id = str(uuid.uuid4()) tracker.record( entity_id=entity_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": "reference", "entity_id": entity_id, "still_id": None, "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( "Reference moderation block: entity=%s, attempt=%d, reason=%s", entity_id, attempt, exc.block_reason, ) tracker.record( entity_id=entity_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, ) 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 entity=%s: strategy=%s", entity_id, last_strategy, ) except Exception as san_exc: logger.error("Sanitization failed: %s", san_exc) raise exc from san_exc else: raise except Exception: tracker.record( entity_id=entity_id, attempt_number=attempt, prompt_used=current_prompt, prompt_version=prompt_version, model_name=self._gemini._model, status="error", ) raise raise RuntimeError(f"Reference image generation exhausted all retries for entity {entity_id}") def generate_all( self, entities: List[Dict[str, Any]], world_guide: Dict[str, Any], output_dir: Path, ) -> List[Dict[str, Any]]: """Generate reference images for all entities.""" results = [] for entity in entities: result = self.generate_for_entity(entity, world_guide, output_dir) results.append(result) return results