"""이미지 생성 서비스 — 레퍼런스/씬 이미지 생성 파이프라인.""" import hashlib import json import logging import uuid from concurrent.futures import ThreadPoolExecutor, as_completed from datetime import datetime, timezone from pathlib import Path from typing import Any, Dict, List, Optional from sqlalchemy.orm import Session as OrmSession from app.core.config import settings from app.core.errors import AppError from app.i18n.loader import t from app.logging.activity_logger import ActivityLogger from app.core.version_registry import get_module_info from app.models.project import ( CharacterOutlook, Episode, EntityCanon, EntityEpisodeLink, GenerationTrace, ImageAsset, ProjectSettings, RelationFact, RelationParticipant, SceneStill, WorldGuide, ) from app.modules.generation_tracker import GenerationTracker from app.modules.image_validator import ImageValidator from app.modules.progress_tracker import ProgressTracker from app.modules.llm.gemini_image_client import GeminiImageClient, ModerationError from app.modules.llm.gemini_key_pool import key_count as gemini_key_count, get_next_key from app.modules.llm.openai_client import OpenAIClient from app.modules.prompt_sanitizer import PromptSanitizer from app.modules.provenance import ProvenanceRecorder from app.modules.entity_dependency import ( build_visual_dependency_graph, topological_sort_entities, build_scene_dependency_graph, topological_sort_scenes, ) from app.modules.reference_image_generator import ReferenceImageGenerator from app.modules.scene_image_generator import SceneImageGenerator from app.modules.t2i_prompt_composer import T2IPromptComposer from app.modules.world_guide_generator import WorldGuideGenerator logger = logging.getLogger(__name__) def _now() -> str: return datetime.now(timezone.utc).isoformat() def _new_id() -> str: return str(uuid.uuid4()) def _build_final_scene_prompt( t2i_prompt: str, labeled_refs: list, style_context: str, tracer=None, scene_index: int = 0, ) -> str: """T2I 프롬프트를 Gemini Flash로 변환. 고유명사 제거 + 영어 변환 + 참조 이미지 번호 기반 역할 지시 생성. 이미지 모델은 이름을 모르므로, 참조 이미지 번호로만 지시해야 함. """ from app.modules.llm.gemini_text_client import GeminiTextClient from pathlib import Path import re # 참조 이미지 역할 + 사용 지시 구성 ref_roles = [] ref_instructions = [] for i, (label, _) in enumerate(labeled_refs, 1): label_lower = label.lower() if "wearing" in label_lower or "outfit" in label_lower: ref_roles.append(f"Reference image {i}: {label}") ref_instructions.append(f"- keep the face, hair, and identity from image {i}") ref_instructions.append(f"- use the outfit from image {i}") elif "face" in label_lower or "character" in label_lower: # 원본 라벨 유지 (캐릭터 구별 특징 포함) ref_roles.append(f"Reference image {i}: {label}") ref_instructions.append(f"- keep the face, hair, and identity from image {i}") elif "prop" in label_lower or "object" in label_lower: ref_roles.append(f"Reference image {i}: {label}") ref_instructions.append(f"- include the object shown in image {i}") elif "background" in label_lower or "location" in label_lower: ref_roles.append(f"Reference image {i}: background/environment mood only.") ref_instructions.append(f"- use the lighting and environment mood from image {i}") else: ref_roles.append(f"Reference image {i}: {label}") ref_instructions.append(f"- reference image {i}: {label}") ref_instructions.append("- do not copy poses or compositions from reference images") ref_instructions.append("- do not alter character identities from their reference images") ref_instructions.append("- NEVER change the face, age, skin, or hair from reference images regardless of outfit or scene description") ref_roles_text = "\n".join(ref_roles) if ref_roles else "No reference images." ref_instructions_text = "\n".join(ref_instructions) # 프롬프트 템플릿 로드 prompt_dir = Path(__file__).resolve().parent.parent.parent.parent / "prompts" / "_base" / "scene_image" versions = sorted([d.name for d in prompt_dir.iterdir() if d.is_dir()], reverse=True) template_path = prompt_dir / versions[0] / "translate_prompt.md" translation_prompt = template_path.read_text(encoding="utf-8").strip().format( ref_roles_text=ref_roles_text, ref_instructions=ref_instructions_text, style_context=style_context, t2i_prompt=t2i_prompt, ) try: import time as _time _t0 = _time.time() client = GeminiTextClient() result = client.send(user_message=translation_prompt, temperature=0.1) _elapsed = int((_time.time() - _t0) * 1000) if isinstance(result, str) and result.strip(): final = result.strip() # Check for suspiciously short translation (absolute threshold) if len(final) < 30: logger.warning("Scene prompt translation suspiciously short (%d chars), may have failed: %s", len(final), final[:100]) # 트레이싱 if tracer: tracer.log( stage="prompt_translation", model="gemini-3-flash-preview", input_data={"translation_prompt": translation_prompt[:500] + "...", "ref_count": len(labeled_refs)}, output_data=final, metadata={"scene_index": scene_index}, elapsed_ms=_elapsed, ) tracer.log_ref_matching(scene_index, t2i_prompt, labeled_refs, final) return final except Exception as exc: logger.warning("Gemini Flash prompt translation failed: %s", exc) # Fallback: regex 치환 + 참조 지시 직접 구성 cleaned = t2i_prompt cleaned = re.sub(r'\[\[([^\]]+)\]\+\[([^\]]+)\]\]', 'a person', cleaned) cleaned = re.sub(r'\[\[([^\]]+)\]\]', r'\1', cleaned) cleaned = re.sub(r'\[([^\]]+?):\s*([^\]]+)\]', r'\2', cleaned) ref_block = "\n".join(ref_roles + ["", "Generate one image:"] + ref_instructions) return f"Photorealistic cinematic still.\n{cleaned}\n\n{ref_block}" # --------------------------------------------------------------------------- # fal.ai angle helpers (module-level) # --------------------------------------------------------------------------- def _select_and_recommend_angle( image_bytes_list: list, beat_title: str, t2i_prompt: str, ) -> Optional[dict]: """GPT Vision: N개 이미지 중 앵글 적용할 이미지 선택 + 앵글 추천 (최소 20도). Returns: {"best_for_angle": 0, "horizontal_angle": 45, "vertical_angle": 20, "zoom": 5, "reason": "..."} or None on failure. """ import base64 import urllib.request import urllib.error if not image_bytes_list: return None n = len(image_bytes_list) prompt = f"""You are a cinematographer. You have {n} scene images. Scene: {beat_title} Prompt: {t2i_prompt[:200]} Task: 1. Choose which image would benefit MOST from a camera angle adjustment. 2. Recommend the camera angle adjustment for that image. Rules: - horizontal_angle: 0-360 (0=front, 90=right side, 180=back, 270=left side) - vertical_angle: -30 to 90 (-30=low angle looking up, 0=eye level, 90=bird's eye) - zoom: 0-10 (0=wide shot, 5=normal, 10=extreme closeup) - IMPORTANT: horizontal_angle >= 20 OR abs(vertical_angle) >= 20. No trivial adjustments. Return JSON: {{"best_for_angle": N, "horizontal_angle": N, "vertical_angle": N, "zoom": N, "reason": "Korean explanation"}} best_for_angle is 0-based index (0 to {n-1}).""" try: # Build GPT Vision input with multiple images content_parts = [{"type": "input_text", "text": prompt}] for i, img_bytes in enumerate(image_bytes_list): img_b64 = base64.b64encode(img_bytes).decode() content_parts.append({"type": "input_text", "text": f"Image {i+1}:"}) content_parts.append({"type": "input_image", "image_url": f"data:image/png;base64,{img_b64}"}) body = { "model": settings.openai_model, "instructions": "You are a cinematographer selecting the best angle adjustment.", "input": [{"role": "user", "content": content_parts}], "text": {"format": {"type": "json_object"}}, } req = urllib.request.Request( "https://api.openai.com/v1/responses", data=json.dumps(body).encode(), headers={"Authorization": f"Bearer {settings.openai_api_key}", "Content-Type": "application/json"}, ) for attempt in range(3): try: with urllib.request.urlopen(req, timeout=60) as resp: payload = json.loads(resp.read()) break except urllib.error.HTTPError as e: if e.code == 429 and attempt < 2: import time; time.sleep(3 * (attempt + 1)) continue raise output_text = payload.get("output_text", "") if not output_text: for item in payload.get("output", []): if isinstance(item, dict): for part in item.get("content", []): if isinstance(part, dict) and part.get("type") == "output_text": output_text = part.get("text", "") break if output_text: break result = json.loads(output_text) idx = max(0, min(n - 1, int(result.get("best_for_angle", 0)))) h = max(0, min(360, float(result.get("horizontal_angle", 0)))) v = max(-30, min(90, float(result.get("vertical_angle", 0)))) z = max(0, min(10, float(result.get("zoom", 5)))) if h < 20 and abs(v) < 20: if h >= abs(v): h = 20.0 else: v = 20.0 if v >= 0 else -20.0 logger.info("Angle recommendation boosted to meet 20° minimum: H=%.0f V=%.0f", h, v) return { "best_for_angle": idx, "horizontal_angle": h, "vertical_angle": v, "zoom": z, "reason": result.get("reason", ""), } except Exception as exc: logger.warning("Angle selection+recommendation failed: %s", exc) return None def _select_final_best( image_bytes_list: list, beat_title: str, ) -> int: """GPT Vision: N+1개 이미지 중 최종 대표 이미지 선택. Returns 0-based index of the best image. Falls back to 0 on failure. """ import base64 import urllib.request import urllib.error if not image_bytes_list: return 0 n = len(image_bytes_list) if n == 1: return 0 prompt = f"""You are a film director selecting the final hero image for a scene. You have {n} images. The last image may be an angle-adjusted variant. Scene: {beat_title} Select the BEST image that: 1. Has the most cinematic composition and visual impact 2. Best represents the scene's mood and narrative moment 3. Has good lighting, focus, and overall quality 4. Feels like a professional film still frame Return JSON: {{"selected_index": N, "reason": "Korean explanation"}} selected_index is 1-based (1 to {n}).""" try: content_parts = [{"type": "input_text", "text": prompt}] for i, img_bytes in enumerate(image_bytes_list): img_b64 = base64.b64encode(img_bytes).decode() content_parts.append({"type": "input_text", "text": f"Image {i+1}:"}) content_parts.append({"type": "input_image", "image_url": f"data:image/png;base64,{img_b64}"}) body = { "model": settings.openai_model, "instructions": "You are a film director selecting the best cinematic still frame.", "input": [{"role": "user", "content": content_parts}], "text": {"format": {"type": "json_object"}}, } req = urllib.request.Request( "https://api.openai.com/v1/responses", data=json.dumps(body).encode(), headers={"Authorization": f"Bearer {settings.openai_api_key}", "Content-Type": "application/json"}, ) for attempt in range(3): try: with urllib.request.urlopen(req, timeout=60) as resp: payload = json.loads(resp.read()) break except urllib.error.HTTPError as e: if e.code == 429 and attempt < 2: import time; time.sleep(3 * (attempt + 1)) continue raise output_text = payload.get("output_text", "") if not output_text: for item in payload.get("output", []): if isinstance(item, dict): for part in item.get("content", []): if isinstance(part, dict) and part.get("type") == "output_text": output_text = part.get("text", "") break if output_text: break result = json.loads(output_text) selected = int(result.get("selected_index", 1)) reason = result.get("reason", "") idx = max(0, min(selected - 1, n - 1)) logger.info("Final best selection: image %d/%d — %s", selected, n, reason) return idx except Exception as exc: logger.warning("Final best selection failed, defaulting to first: %s", exc) return 0 def _apply_fal_angle( img_bytes: bytes, horizontal: float, vertical: float, zoom: float, ) -> tuple: """fal.ai로 이미지에 앵글 적용. Returns (result_bytes, elapsed_ms) or (None, 0).""" import base64 import time import urllib.error import urllib.request img_b64 = base64.b64encode(img_bytes).decode() body = { "image_urls": [f"data:image/png;base64,{img_b64}"], "horizontal_angle": horizontal, "vertical_angle": vertical, "zoom": zoom, "output_format": "png", "num_images": 1, } req = urllib.request.Request( "https://fal.run/fal-ai/qwen-image-edit-2511-multiple-angles", data=json.dumps(body).encode(), headers={ "Authorization": f"Key {settings.fal_key}", "Content-Type": "application/json", }, ) t0 = time.time() try: with urllib.request.urlopen(req, timeout=120) as resp: result = json.loads(resp.read()) elapsed = int((time.time() - t0) * 1000) images = result.get("images", []) if not images: return None, 0 result_url = images[0].get("url", "") if not result_url: return None, 0 with urllib.request.urlopen(result_url, timeout=30) as dl: return dl.read(), elapsed except Exception as exc: logger.warning("fal.ai angle apply failed: %s", exc) return None, 0 class ImageService: """이미지 생성 및 관리 서비스.""" def __init__( self, db: OrmSession, project_id: str, actor_id: str, ) -> None: self._db = db self._logger = ActivityLogger(db) self._project_id = project_id self._actor_id = actor_id def _get_episode(self, episode_id: str) -> Episode: ep = ( self._db.query(Episode) .filter(Episode.id == episode_id, Episode.project_id == self._project_id) .first() ) if not ep: raise AppError( code="episode.not_found", message=t("episode.not_found"), status_code=404, ) return ep def _get_project_dir(self) -> Path: return Path(settings.projects_dir) / self._project_id def _get_style_context(self) -> str: """프로젝트 세계관을 T2I 프롬프트 앞에 붙일 컨텍스트로 변환. 화풍 아닌 세계관 설명.""" ps = self._db.query(ProjectSettings).filter( ProjectSettings.project_id == self._project_id).first() if not ps or not ps.style_rules_json: return "Photorealistic cinematic still." sr = json.loads(ps.style_rules_json) return ( f"Photorealistic cinematic still. " f"Setting: {sr.get('era','')}, {sr.get('region','')}. " f"Avoid: {sr.get('must_avoid','')}." ) def generate_reference_images_only( self, episode_id: str, ip: Optional[str] = None, mode: str = "resume", ) -> Dict[str, Any]: """엔티티 참조 이미지만 생성 (씬 이미지 제외). mode: "resume" — 이미 참조 이미지 있는 엔티티 스킵 "full" — 기존 참조 이미지 삭제 후 전체 재생성 """ episode = self._get_episode(episode_id) if not episode.fulltext: raise AppError(code="analysis.no_text", message=t("analysis.no_text"), status_code=400) # ── 전 단계 검증: 분석이 완료되어야 함 ── if episode.status != "analyzed": raise AppError( code="image.analysis_not_complete", message=f"분석이 완료되지 않았습니다 (현재: {episode.status}). 분석을 먼저 완료하세요.", status_code=400, ) # 요소가 하나도 없으면 차단 entity_count = self._db.query(EntityCanon).filter( EntityCanon.project_id == self._project_id, ).count() if entity_count == 0: raise AppError( code="image.no_entities", message="추출된 요소가 없습니다. 분석을 먼저 실행하세요.", status_code=400, ) if not settings.gemini_api_key and gemini_key_count() == 0: raise AppError(code="image.gemini_key_missing", message=t("image.gemini_key_missing"), status_code=400) language = episode.language or "ko" project_dir = self._get_project_dir() reference_dir = project_dir / "images" / episode_id / "reference" # Get entities links = self._db.query(EntityEpisodeLink).filter( EntityEpisodeLink.project_id == self._project_id, EntityEpisodeLink.episode_id == episode_id, ).all() canon_ids = [link.canon_id for link in links] entities_orm = self._db.query(EntityCanon).filter( EntityCanon.id.in_(canon_ids) ).all() if canon_ids else [] entities = [ {"id": e.id, "name": e.name, "entity_type": e.entity_type, "description": e.description or "", "stable_traits": e.stable_traits or "{}"} for e in entities_orm ] # full 모드: 해당 에피소드의 참조 이미지만 삭제 + 체크포인트 초기화 if mode == "full": self._db.query(ImageAsset).filter( ImageAsset.project_id == self._project_id, ImageAsset.episode_id == episode_id, ImageAsset.asset_type == "reference", ImageAsset.entity_id.in_([e["id"] for e in entities]), ).delete(synchronize_session="fetch") self._db.commit() # 체크포인트는 아래에서 초기화 (cp_dir 아직 미생성) # resume 모드: 이미 참조 이미지가 있는 엔티티 파악 + primary 이미지 로드 ref_image_map: Dict[str, bytes] = {} already_done: set = set() if mode == "resume": for entity in entities: existing = ( self._db.query(ImageAsset) .filter( ImageAsset.project_id == self._project_id, ImageAsset.entity_id == entity["id"], ImageAsset.asset_type == "reference", ImageAsset.is_primary == 1, ) .order_by(ImageAsset.created_at.desc()) .first() ) if existing: already_done.add(entity["id"]) fp = Path(existing.file_path) if fp.exists(): ref_image_map[entity["id"]] = fp.read_bytes() # Get or create world guide openai_client = OpenAIClient() stills_orm = self._db.query(SceneStill).filter( SceneStill.project_id == self._project_id, SceneStill.episode_id == episode_id, ).all() stills = [{"still_frame_prompt": s.still_frame_prompt or ""} for s in stills_orm] _ft = episode.fulltext or "" wg_hash = hashlib.md5(f"{_ft[:500]}:{len(entities)}:{len(stills)}".encode()).hexdigest() existing_wg = self._db.query(WorldGuide).filter( WorldGuide.project_id == self._project_id, WorldGuide.episode_id == episode_id, ).order_by(WorldGuide.created_at.desc()).first() if existing_wg and existing_wg.source_hash == wg_hash: world_guide = json.loads(existing_wg.guide_json) else: wg_gen = WorldGuideGenerator(llm_client=openai_client) world_guide = wg_gen.generate( fulltext=episode.fulltext, language=language, source_file=episode.source_filename or "episode", entities=entities, stills=stills, ) wg_record = WorldGuide( id=_new_id(), project_id=self._project_id, episode_id=episode_id, guide_json=json.dumps(world_guide, ensure_ascii=False), source_hash=wg_hash, created_at=_now(), ) self._db.add(wg_record) self._db.commit() entity_lookup = {e["id"]: e for e in entities} gemini_client = GeminiImageClient(model=settings.gemini_image_model) from app.modules.pipeline.ref_image_pipeline import generate_and_validate_reference from app.modules.image_checkpoint import ImageCheckpointManager # 체크포인트 초기화 cp_dir = project_dir / "checkpoints" / "images" / episode_id ref_cp = ImageCheckpointManager(cp_dir, "reference") if mode == "full": ref_cp.clear() else: # 체크포인트에서 완료된 항목도 already_done에 병합 already_done |= ref_cp.get_completed_ids() # Dependency graph relations_orm = self._db.query(RelationFact).filter(RelationFact.project_id == self._project_id).all() rel_ids = [r.id for r in relations_orm] participants_orm = self._db.query(RelationParticipant).filter( RelationParticipant.relation_id.in_(rel_ids) ).all() if rel_ids else [] deps = build_visual_dependency_graph( entities, [{"id": r.id, "relation_family": r.relation_family} for r in relations_orm], [{"relation_id": p.relation_id, "canon_id": p.canon_id} for p in participants_orm], ) batches = topological_sort_entities(entities, deps) entity_by_id = {e["id"]: e for e in entities} generated_count = 0 failed_count = 0 skipped_count = len(already_done) max_concurrent = settings.max_concurrent_image_gen total_to_gen = len(entities) - skipped_count progress = ProgressTracker(self._db, episode_id, "reference_image_generation", self._project_id) for batch_idx, batch in enumerate(batches): # resume: 이미 완료된 엔티티 제외 batch_entities = [ entity_by_id[eid] for eid in batch if eid in entity_by_id and eid not in already_done ] if not batch_entities: continue progress.update( f"참조 이미지 {generated_count + skipped_count}/{len(entities)}", generated_count, total_to_gen, ) def _gen_ref(entity): entity_deps = deps.get(entity["id"], set()) dep_refs = [(f"Reference: {entity_by_id.get(d, {}).get('name', '')}", ref_image_map[d]) for d in entity_deps if d in ref_image_map] # v2 파이프라인: T2I + GPT LVM 검증 + 비교 선택 t2i = entity.get("t2i_prompt") or entity.get("description") or entity["name"] # 참조 이미지: T2I에 이미 "Photorealistic..." 포함, style_context 불필요 _style_ctx = "" pipe_result = generate_and_validate_reference( gemini_client=gemini_client, entity_name=entity["name"], entity_description=entity.get("description", ""), entity_type=entity.get("entity_type", "character"), t2i_prompt=t2i, output_dir=reference_dir, extra_references=dep_refs if dep_refs else None, style_context=_style_ctx, ) # 기존 코드 호환 형식으로 변환 result = { "id": _new_id(), "entity_id": entity["id"], "asset_type": "reference", "still_id": None, "file_path": pipe_result["file_path"], "prompt_used": t2i, "generation_model": pipe_result["generation_model"], "width": None, "height": None, "status": "generated", "review_notes": json.dumps(pipe_result.get("validation", {}), ensure_ascii=False), "validation_score": pipe_result.get("validation", {}).get("score"), "created_at": _now(), } return entity["id"], result with ThreadPoolExecutor(max_workers=min(max_concurrent, len(batch_entities))) as executor: futures = {executor.submit(_gen_ref, e): e for e in batch_entities} for future in as_completed(futures): try: eid, result = future.result() fp = Path(result.get("file_path", "")) if fp.exists(): ref_image_map[eid] = fp.read_bytes() # 즉시 DB 저장 + 커밋 (파이프라인에서 이미 GPT LVM 검증 완료) self._db.query(ImageAsset).filter( ImageAsset.project_id == self._project_id, ImageAsset.entity_id == result["entity_id"], ImageAsset.is_primary == 1, ).update({"is_primary": 0}) asset = ImageAsset( id=result["id"], project_id=self._project_id, asset_type="reference", entity_id=result["entity_id"], episode_id=episode_id, file_path=result["file_path"], prompt_used=result["prompt_used"], generation_model=result["generation_model"], status=result["status"], review_notes=result.get("review_notes"), validation_score=result.get("validation_score"), is_primary=1, created_at=result["created_at"], ) self._db.add(asset) self._db.commit() generated_count += 1 # 체크포인트 기록 ref_cp.mark_completed(eid, {"asset_id": result["id"], "file_path": result["file_path"]}) progress.update( f"참조 이미지 {generated_count + skipped_count}/{len(entities)}: {entity_by_id.get(eid, {}).get('name', '')}", generated_count, total_to_gen, ) except Exception as exc: failed_entity = futures[future] eid_failed = failed_entity.get("id", "") logger.warning("Ref image failed (%s): %s", failed_entity.get("name", eid_failed), exc) ref_cp.mark_failed(eid_failed, str(exc)[:500]) failed_count += 1 # 전부 실패 시 에러 if generated_count > 0 and failed_count == generated_count: progress.error("참조 이미지 생성 전부 실패 (API 오류)") raise AppError( code="image.all_ref_failed", message=f"참조 이미지 {failed_count}개 모두 실패했습니다. API 상태를 확인하세요.", status_code=500, ) # ── Character+Outlook composite reference images (병렬) ── outlook_generated = 0 outlook_skipped = 0 try: char_outlooks = ( self._db.query(CharacterOutlook) .filter(CharacterOutlook.project_id == self._project_id) .all() ) # 생성 대상 수집 (DB 조회는 메인 스레드, char+outlook 쌍 중복 제거) outlook_tasks = [] _seen_pairs = set() for co in char_outlooks: pair_key = (co.character_id, co.outlook_id) if pair_key in _seen_pairs: outlook_skipped += 1 continue _seen_pairs.add(pair_key) existing_composite = ( self._db.query(ImageAsset) .filter( ImageAsset.project_id == self._project_id, ImageAsset.entity_id == co.character_id, ImageAsset.asset_type == "reference", ImageAsset.prompt_used.like(f"%outlook_id:{co.outlook_id}%"), ) .first() ) if existing_composite: outlook_skipped += 1 continue char_ref_asset = ( self._db.query(ImageAsset) .filter( ImageAsset.project_id == self._project_id, ImageAsset.entity_id == co.character_id, ImageAsset.asset_type == "reference", ImageAsset.is_primary == 1, ) .order_by(ImageAsset.created_at.desc()) .first() ) if not char_ref_asset: continue char_face_path = Path(char_ref_asset.file_path) if not char_face_path.exists(): continue char_face_bytes = char_face_path.read_bytes() outlook_entity = self._db.query(EntityCanon).filter(EntityCanon.id == co.outlook_id).first() if not outlook_entity: continue outlook_desc = outlook_entity.description or outlook_entity.name char_entity = entity_by_id.get(co.character_id, {}) char_name = char_entity.get("name", "") if not char_name: char_entity_orm = self._db.query(EntityCanon).filter(EntityCanon.id == co.character_id).first() char_name = char_entity_orm.name if char_entity_orm else "character" char_desc = char_entity.get("description", "") if not char_desc: char_entity_orm2 = self._db.query(EntityCanon).filter(EntityCanon.id == co.character_id).first() char_desc = char_entity_orm2.description if char_entity_orm2 else "" outlook_tasks.append({ "co": co, "char_name": char_name, "outlook_name": outlook_entity.name, "outlook_desc": outlook_desc, "char_face_bytes": char_face_bytes, "char_id": co.character_id, "outlook_id": co.outlook_id, "char_desc": char_desc, }) if outlook_tasks: logger.info("Generating %d outlook composites (parallel)", len(outlook_tasks)) def _gen_composite(task): from app.modules.pipeline.ref_image_pipeline import _load_ref_image_prompt labeled_refs = [("Reference face image", task["char_face_bytes"])] # Use type-specific prompt template for outlook ref_prompt = _load_ref_image_prompt("outlook", entity_description=task["char_desc"], outlook_description=task["outlook_desc"]) current_prompt = ref_prompt if ref_prompt else task["outlook_desc"] pipe_result = generate_and_validate_reference( gemini_client=gemini_client, entity_name=f"{task['char_name']} ({task['outlook_name']})", entity_description=task["outlook_desc"], entity_type="outlook", t2i_prompt=current_prompt, output_dir=reference_dir, extra_references=labeled_refs, style_context="", ) return task, pipe_result with ThreadPoolExecutor(max_workers=min(max_concurrent, len(outlook_tasks))) as executor: futures = {executor.submit(_gen_composite, t): t for t in outlook_tasks} for future in as_completed(futures): task = futures[future] try: _, pipe_result = future.result() composite_id = _new_id() composite_asset = ImageAsset( id=composite_id, project_id=self._project_id, asset_type="reference", entity_id=task["char_id"], episode_id=episode_id, file_path=pipe_result["file_path"], prompt_used=f"[outlook_id:{task['outlook_id']}] {task['outlook_desc']}", generation_model=pipe_result["generation_model"], status="generated", review_notes=json.dumps(pipe_result.get("validation", {}), ensure_ascii=False), validation_score=pipe_result.get("validation", {}).get("score"), is_primary=0, created_at=_now(), ) self._db.add(composite_asset) self._db.commit() outlook_generated += 1 logger.info("Outlook composite generated: %s + %s", task["char_name"], task["outlook_name"]) except Exception as exc: logger.warning("Outlook composite failed for %s + %s: %s", task["char_name"], task["outlook_name"], exc) except Exception as exc: logger.warning("CharacterOutlook composite generation error: %s", exc) progress.complete() self._logger.log( actor_id=self._actor_id, action="episode.generate_reference_images", resource_type="episode", resource_id=episode_id, project_id=self._project_id, detail={ "generated": generated_count, "skipped": skipped_count, "outlook_generated": outlook_generated, "outlook_skipped": outlook_skipped, }, ip_address=ip, ) return { "reference_count": generated_count, "skipped": skipped_count, "outlook_generated": outlook_generated, "outlook_skipped": outlook_skipped, } def generate_images(self, episode_id: str, ip: Optional[str] = None, mode: str = "resume") -> None: """Full image generation pipeline for an episode. mode: "resume" — 이미 씬 이미지 있는 still_id 스킵 "full" — 기존 씬 이미지 삭제 후 전체 재생성 """ episode = self._get_episode(episode_id) if not episode.fulltext: raise AppError( code="analysis.no_text", message=t("analysis.no_text"), status_code=400, ) if not settings.gemini_api_key and gemini_key_count() == 0: raise AppError( code="image.gemini_key_missing", message=t("image.gemini_key_missing"), status_code=400, ) # ── 전 단계 검증: 분석 완료 + 참조 이미지 존재 ── if episode.status != "analyzed": raise AppError( code="image.analysis_not_complete", message=f"분석이 완료되지 않았습니다 (현재: {episode.status}). 분석을 먼저 완료하세요.", status_code=400, ) # 씬이 있는지 확인 scene_count = self._db.query(SceneStill).filter( SceneStill.project_id == self._project_id, SceneStill.episode_id == episode_id, ).count() if scene_count == 0: raise AppError( code="image.no_scenes", message="씬이 없습니다. 분석을 먼저 완료하세요.", status_code=400, ) # 참조 이미지가 최소한 있는지 확인 (캐릭터 기준) char_count = self._db.query(EntityCanon).filter( EntityCanon.project_id == self._project_id, EntityCanon.entity_type == "character", ).count() ref_primary_count = self._db.query(ImageAsset).filter( ImageAsset.project_id == self._project_id, ImageAsset.episode_id == episode_id, ImageAsset.asset_type == "reference", ImageAsset.is_primary == 1, ).count() if ref_primary_count == 0: raise AppError( code="image.no_reference_images", message="참조 이미지가 없습니다. 참조 이미지를 먼저 생성하세요.", status_code=400, ) # 캐릭터 수 대비 참조 이미지가 50% 미만이면 경고 (차단은 안 함) if char_count > 0 and ref_primary_count < char_count * 0.5: logger.warning( "참조 이미지 부족: %d/%d 캐릭터만 참조 이미지 있음. 일부 씬에서 품질 저하 가능.", ref_primary_count, char_count, ) language = episode.language or "ko" project_dir = self._get_project_dir() images_dir = project_dir / "images" / episode_id scene_dir = images_dir / "scene" # 씬 이미지 체크포인트 from app.modules.image_checkpoint import ImageCheckpointManager cp_dir = project_dir / "checkpoints" / "images" / episode_id scene_cp = ImageCheckpointManager(cp_dir, "scene") if mode == "full": scene_cp.clear() # Get entities for this episode links = ( self._db.query(EntityEpisodeLink) .filter( EntityEpisodeLink.project_id == self._project_id, EntityEpisodeLink.episode_id == episode_id, ) .all() ) canon_ids = [link.canon_id for link in links] entities_orm = ( self._db.query(EntityCanon) .filter(EntityCanon.id.in_(canon_ids)) .all() ) if canon_ids else [] entities = [ { "id": e.id, "name": e.name, "entity_type": e.entity_type, "description": e.description or "", "stable_traits": e.stable_traits or "{}", } for e in entities_orm ] # Get scene stills for this episode stills_orm = ( self._db.query(SceneStill) .filter(SceneStill.project_id == self._project_id, SceneStill.episode_id == episode_id) .order_by(SceneStill.still_index) .all() ) stills = [ { "id": s.id, "still_index": s.still_index, "screenplay_scene_heading": s.screenplay_scene_heading or "", "beat_title": s.beat_title or "", "still_frame_prompt": s.still_frame_prompt or "", "camera_json": s.camera_json or "{}", "lighting_json": s.lighting_json or "{}", "visible_entities_json": s.visible_entities_json or "[]", "dependent_scene_id": s.dependent_scene_id, } for s in stills_orm ] # full 모드: 기존 씬 이미지 전부 삭제 if mode == "full": self._db.query(ImageAsset).filter( ImageAsset.project_id == self._project_id, ImageAsset.episode_id == episode_id, ImageAsset.asset_type == "scene", ).delete(synchronize_session="fetch") # 변형 이미지도 삭제 self._db.query(ImageAsset).filter( ImageAsset.project_id == self._project_id, ImageAsset.episode_id == episode_id, ImageAsset.variant_type.in_(["variant_a", "variant_b"]), ).delete(synchronize_session="fetch") self._db.commit() # resume 모드: 이미 씬 이미지 있는 still_id 파악 already_done_stills: set = set() if mode == "resume": existing_scene_stills = ( self._db.query(ImageAsset.still_id) .filter( ImageAsset.project_id == self._project_id, ImageAsset.episode_id == episode_id, ImageAsset.asset_type == "scene", ) .distinct() .all() ) already_done_stills = {r[0] for r in existing_scene_stills if r[0]} # 체크포인트에서 완료된 항목도 병합 already_done_stills |= scene_cp.get_completed_ids() # Initialize provenance recorder and progress tracker provenance = ProvenanceRecorder(self._db, self._project_id) stills_to_gen = len(stills) - len(already_done_stills) total_items = stills_to_gen + 1 # +1 for world guide progress = ProgressTracker(self._db, episode_id, "image_generation", self._project_id) # 1. Generate world guide (full 모드면 재생성, resume이면 기존 재사용) progress.update("세계관 가이드 생성 중", 0, total_items) _ft = episode.fulltext or "" wg_hash = hashlib.md5(f"{_ft[:500]}:{len(entities)}:{len(stills)}".encode()).hexdigest() existing_wg = None if mode != "full": existing_wg = self._db.query(WorldGuide).filter( WorldGuide.project_id == self._project_id, WorldGuide.episode_id == episode_id, ).order_by(WorldGuide.created_at.desc()).first() if existing_wg and existing_wg.source_hash == wg_hash: world_guide = json.loads(existing_wg.guide_json) else: openai_client = OpenAIClient() wg_gen = WorldGuideGenerator(llm_client=openai_client) with provenance.start_operation( "image_generation", "world_guide_generator", episode_id=episode_id, ) as op: op.set_input({"fulltext_chars": len(episode.fulltext), "entities": len(entities), "stills": len(stills)}) world_guide = wg_gen.generate( fulltext=episode.fulltext, language=language, source_file=episode.source_filename or "episode", entities=entities, stills=stills, ) op.set_output({"world_setting_summary_len": len(world_guide.get("world_setting_summary", ""))}) # Save world guide to DB wg_record = WorldGuide( id=_new_id(), project_id=self._project_id, episode_id=episode_id, guide_json=json.dumps(world_guide, ensure_ascii=False), source_hash=wg_hash, created_at=_now(), ) self._db.add(wg_record) self._db.flush() # 프로젝트 스타일 규칙을 world_guide에 주입 proj_settings = self._db.query(ProjectSettings).filter( ProjectSettings.project_id == self._project_id, ).first() if proj_settings and proj_settings.style_rules_json: proj_style = json.loads(proj_settings.style_rules_json) # WorldGuide의 style_rules가 있으면 병합, 없으면 프로젝트 스타일 주입 existing_sr = world_guide.get("style_rules", {}) if isinstance(existing_sr, dict) and existing_sr.get("must_maintain"): # WorldGuide가 이미 잘 구조화된 style_rules를 가지고 있음 — 프로젝트 스타일은 메타데이터로 추가 world_guide["project_style"] = proj_style else: world_guide["style_rules"] = proj_style # Build entity lookup by ID entity_lookup = {e["id"]: e for e in entities} # Initialize tracker and sanitizer openai_client = OpenAIClient() tracker = GenerationTracker(self._db, self._project_id) sanitizer = PromptSanitizer(openai_client) # 2. Load existing reference images (절대 새로 생성하지 않음) gemini_client = GeminiImageClient(model=settings.gemini_image_model) ref_image_map: Dict[str, bytes] = {} max_concurrent = settings.max_concurrent_image_gen progress.update("기존 참조 이미지 로드 중", 1, total_items) for entity in entities: # DB에서 해당 엔티티의 primary 참조 이미지 조회 primary_asset = ( self._db.query(ImageAsset) .filter( ImageAsset.project_id == self._project_id, ImageAsset.entity_id == entity["id"], ImageAsset.asset_type == "reference", ImageAsset.is_primary == 1, ) .order_by(ImageAsset.created_at.desc()) .first() ) if primary_asset: fp = Path(primary_asset.file_path) if fp.exists(): ref_image_map[entity["id"]] = fp.read_bytes() logger.info( "Loaded %d existing reference images for %d entities", len(ref_image_map), len(entities), ) # 참조 이미지가 하나도 없으면 씬 생성 차단 if not ref_image_map: raise AppError( code="image.no_reference_images", message=t("image.no_reference_images") if t("image.no_reference_images") != "image.no_reference_images" else "참조 이미지가 없습니다. 엔티티 참조 이미지를 먼저 생성하세요.", status_code=400, ) # Initialize validator for scene image validation validator = self._create_validator() # T2I 프롬프트 + variations 로드 — 분석에서 이미 생성됨. 없으면 런타임 변환 has_missing_t2i = False for si, still_data in enumerate(stills): still_orm = stills_orm[si] still_data["t2i_prompt_cinematic"] = still_orm.t2i_prompt_cinematic or "" still_data["t2i_prompt_closeup"] = still_orm.t2i_prompt_closeup or "" # v5: N개 T2I variations 로드 try: still_data["t2i_variations"] = json.loads(still_orm.t2i_variations_json) if still_orm.t2i_variations_json else [] except (json.JSONDecodeError, TypeError): still_data["t2i_variations"] = [] if not still_data["t2i_prompt_cinematic"]: has_missing_t2i = True if has_missing_t2i: logger.warning("T2I prompts missing for some stills — generating at runtime") from app.modules.t2i_visual_converter import T2IVisualConverter converter = T2IVisualConverter(llm_client=openai_client) entity_t2i_map = {e["id"]: e.get("t2i_prompt", "") for e in entities} scene_dicts = [ {"id": s["id"], "still_frame_prompt": s.get("still_frame_prompt", ""), "screenplay_scene_heading": s.get("screenplay_scene_heading", ""), "visible_entities_json": s.get("visible_entities_json", "[]")} for s in stills if not s.get("t2i_prompt_cinematic") ] scene_t2i_map = converter.convert_scenes(scene_dicts, entity_t2i_map) for si, still_data in enumerate(stills): if not still_data["t2i_prompt_cinematic"] and still_data["id"] in scene_t2i_map: t2i = scene_t2i_map[still_data["id"]] still_data["t2i_prompt_cinematic"] = t2i["a"] still_data["t2i_prompt_closeup"] = t2i["b"] stills_orm[si].t2i_prompt_cinematic = t2i["a"] stills_orm[si].t2i_prompt_closeup = t2i["b"] stills_orm[si].t2i_composer_version = "t2i_visual_converter/v1" self._db.commit() # Fallback: 여전히 없으면 원본 프롬프트 사용 for still_data in stills: if not still_data.get("t2i_prompt_cinematic"): still_data["t2i_prompt_cinematic"] = still_data.get("still_frame_prompt", "") if not still_data.get("t2i_prompt_closeup"): still_data["t2i_prompt_closeup"] = still_data.get("still_frame_prompt", "") # 3. Generate scene images — dependency-ordered, concurrent within batches # Build scene ref_image_map: character + prop refs + outlook composites scene_ref_image_map: Dict[str, bytes] = {} for eid, img_bytes in ref_image_map.items(): entity_info = entity_lookup.get(eid, {}) if entity_info.get("entity_type") == "location": continue scene_ref_image_map[eid] = img_bytes # 인물+아웃룩 합성 이미지도 ref_map에 추가 (키: "outlook:{char_id}:{outlook_id}") try: composite_assets = ( self._db.query(ImageAsset) .filter( ImageAsset.project_id == self._project_id, ImageAsset.asset_type == "reference", ImageAsset.prompt_used.like("%outlook_id:%"), ) .all() ) for ca in composite_assets: # prompt_used 형식: "[outlook_id:xxx] description" import re as _re_ oid_match = _re_.search(r'outlook_id:([a-f0-9-]+)', ca.prompt_used or "") if oid_match and ca.entity_id: outlook_id = oid_match.group(1) composite_key = f"outlook:{ca.entity_id}:{outlook_id}" fp = Path(ca.file_path) if fp.exists(): scene_ref_image_map[composite_key] = fp.read_bytes() logger.info("Loaded %d outlook composite refs into scene_ref_image_map", len(composite_assets)) except Exception as exc: logger.warning("Failed to load outlook composites: %s", exc) from app.modules.pipeline.scene_image_pipeline import generate_and_validate_scene # Track generated scene images per location for same-background linking location_scene_history: Dict[str, tuple] = {} # location_id -> (scene_bytes, still_data) # Track the last generated scene result per scene index (for variations) scene_results_by_index: Dict[int, Dict[str, Any]] = {} scene_paths_by_index: Dict[int, Path] = {} scene_paths_by_index_by_id: Dict[str, Path] = {} # still_id -> Path (연관 씬 참조용) # resume: 이미 완료된 씬의 location_scene_history + scene_paths 복원 if already_done_stills: for si, still_data in enumerate(stills): still_id = still_data.get("id") if still_id not in already_done_stills: continue # DB에서 기존 씬 이미지 조회 (v5: primary 우선, 없으면 최신) existing_asset = ( self._db.query(ImageAsset) .filter( ImageAsset.project_id == self._project_id, ImageAsset.still_id == still_id, ImageAsset.asset_type == "scene", ImageAsset.is_primary == 1, ) .order_by(ImageAsset.created_at.desc()) .first() ) if not existing_asset: existing_asset = ( self._db.query(ImageAsset) .filter( ImageAsset.project_id == self._project_id, ImageAsset.still_id == still_id, ImageAsset.asset_type == "scene", ) .order_by(ImageAsset.created_at.desc()) .first() ) if not existing_asset: continue fp = Path(existing_asset.file_path) if not fp.exists(): continue scene_paths_by_index[si] = fp scene_paths_by_index_by_id[still_id] = fp scene_results_by_index[si] = { "id": existing_asset.id, "file_path": existing_asset.file_path, "prompt_used": existing_asset.prompt_used or "", "status": existing_asset.status, "review_notes": existing_asset.review_notes or "", } # location history 복원 try: vis_ids = json.loads(still_data.get("visible_entities_json", "[]")) except json.JSONDecodeError: vis_ids = [] for v in vis_ids: if isinstance(v, dict): eid = v.get("entity_id", "") if eid in entity_lookup and entity_lookup[eid].get("entity_type") == "location": location_scene_history[eid] = (fp.read_bytes(), still_data) # Build scene dependency graph and topological sort into batches scene_deps = build_scene_dependency_graph(stills, entity_lookup) scene_batches = topological_sort_scenes(len(stills), scene_deps) logger.info( "Scene generation: %d scenes in %d batches (max_concurrent=%d)", len(stills), len(scene_batches), max_concurrent, ) scenes_generated = 0 # 스타일 컨텍스트를 메인 스레드에서 한번만 로드 (스레드 안에서 DB 접근 방지) _cached_style_context = self._get_style_context() for batch_idx, scene_batch in enumerate(scene_batches): # resume: 이미 완료된 still_id 제외 scene_batch = [si for si in scene_batch if stills[si].get("id") not in already_done_stills] if not scene_batch: continue progress.update( f"씬 이미지 {scenes_generated}/{stills_to_gen}", scenes_generated + 1, total_items, ) def _resolve_refs_for_prompt(t2i_prompt: str, visible_entities: list) -> list: """T2I 프롬프트에서 참조 이미지 매칭 (재사용 헬퍼).""" import re as _re labeled_refs = [] _used_ref_ids: set = set() for match in _re.finditer(r'\[\[([^\]]+)\]\+\[([^\]]+)\]\]', t2i_prompt): char_name, outlook_name = match.group(1), match.group(2) char_id = next( (ve["id"] for ve in visible_entities if ve.get("name") == char_name and ve.get("entity_type") == "character"), None, ) if not char_id: continue if outlook_name == "미지정": if char_id in scene_ref_image_map and char_id not in _used_ref_ids: labeled_refs.append(("character identity", scene_ref_image_map[char_id])) _used_ref_ids.add(char_id) continue outlook_id = next( (ve["id"] for ve in visible_entities if ve.get("name") == outlook_name and ve.get("entity_type") == "outlook" and ve.get("id")), None, ) if not outlook_id: outlook_id = next( (eid for eid, info in entity_lookup.items() if info.get("name") == outlook_name and info.get("entity_type") == "outlook"), None, ) if outlook_id: composite_key = f"outlook:{char_id}:{outlook_id}" if composite_key in scene_ref_image_map: labeled_refs.append(("character in outfit", scene_ref_image_map[composite_key])) _used_ref_ids.add(composite_key) continue if char_id in scene_ref_image_map and char_id not in _used_ref_ids: labeled_refs.append(("character identity", scene_ref_image_map[char_id])) _used_ref_ids.add(char_id) for ve in visible_entities: eid = ve.get("id", "") etype = ve.get("entity_type", "") if etype == "prop" and eid in scene_ref_image_map and eid not in _used_ref_ids: labeled_refs.append(("object appearance", scene_ref_image_map[eid])) _used_ref_ids.add(eid) return labeled_refs def _generate_one_variation( si: int, var_t2i: str, var_theme: str, var_theme_label: str, visible_entities: list, best_prev_bytes, labeled_refs: list, ) -> dict: """Generate a single variation image for a scene (runs in thread pool). Returns dict with result data or None on failure. """ still_data = stills[si] try: _full_prompt = _build_final_scene_prompt( var_t2i, labeled_refs, _cached_style_context, ) except Exception as _prompt_exc: logger.warning("Prompt build failed for variation %s: %s", var_theme, _prompt_exc) _full_prompt = f"{_cached_style_context}\n\n{var_t2i}" if _cached_style_context else var_t2i try: pipe_result = generate_and_validate_scene( gemini_client=gemini_client, t2i_prompt=_full_prompt, beat_title=still_data.get("beat_title", ""), output_dir=scene_dir, reference_images=labeled_refs if labeled_refs else None, previous_scene_bytes=best_prev_bytes, ) validation_score = None validation_result_str = None v_status = "generated" v_review_notes = json.dumps(pipe_result.get("validation", {}), ensure_ascii=False) if validator: entity_names = [e.get("name", "") for e in visible_entities] scene_info = { "scene_heading": still_data.get("screenplay_scene_heading", ""), "beat_title": still_data.get("beat_title", ""), "still_frame_prompt": still_data.get("still_frame_prompt", ""), "world_context": world_guide.get("world_setting_summary", ""), } validation_score, validation_result_str, v_status, v_review_notes = ( self._validate_scene( validator, pipe_result["file_path"], scene_info, entity_names, v_status, v_review_notes, ) ) return { "id": _new_id(), "asset_type": "scene", "entity_id": None, "still_id": still_data.get("id"), "episode_id": episode_id, "file_path": pipe_result["file_path"], "prompt_used": _full_prompt, "generation_model": pipe_result["generation_model"], "width": None, "height": None, "status": v_status, "review_notes": v_review_notes, "validation_score": validation_score, "validation_result": validation_result_str, "variant_type": var_theme, "theme_label": var_theme_label, "created_at": _now(), } except ModerationError as exc: logger.warning("Scene %s variation %s blocked: %s", still_data.get("id"), var_theme, exc.block_reason) return None except Exception as exc: logger.error("Scene %s variation %s failed: %s", still_data.get("id"), var_theme, exc) return None def _generate_one_scene(si: int) -> tuple: """Generate N variation images for one still (runs in thread pool). Returns (scene_index, list_of_results, visible_entities, current_location_ids). Each result is a dict or None. """ still_data = stills[si] try: visible_ids = json.loads(still_data["visible_entities_json"]) except json.JSONDecodeError: visible_ids = [] visible_entities = [] for v in visible_ids: if isinstance(v, dict): eid = v.get("entity_id", "") if eid in entity_lookup: visible_entities.append(entity_lookup[eid]) elif isinstance(v, str): if v in entity_lookup: visible_entities.append(entity_lookup[v]) # 시각적 연관 씬 참조 이미지 (v5: dependent_scene_id + location_scene_history 통합) best_prev_bytes = None current_location_ids = [] # 1순위: dependent_scene_id (씬 분석에서 결정된 시각적 연관 씬) dep_scene_id = still_data.get("dependent_scene_id") if dep_scene_id and dep_scene_id in scene_paths_by_index_by_id: dep_path = scene_paths_by_index_by_id[dep_scene_id] if dep_path.exists(): best_prev_bytes = dep_path.read_bytes() # 2순위: 같은 장소 이전 씬 (location_scene_history) if not best_prev_bytes: current_location_ids = [ e["id"] for e in visible_entities if e.get("entity_type") == "location" ] for loc_id in current_location_ids: if loc_id in location_scene_history: best_prev_bytes, _ = location_scene_history[loc_id] break # v5: N개 변형 로드 — t2i_variations가 있으면 사용, 없으면 기본 1개 variations = still_data.get("t2i_variations", []) if not variations: # fallback: 기본 프롬프트 1개 base_t2i = still_data.get("t2i_prompt_cinematic") or still_data.get("still_frame_prompt", "") variations = [{"theme": "base", "theme_label": "base", "t2i_prompt": base_t2i}] # 참조 이미지는 첫 번째 변형 기준으로 생성 (모든 변형에 동일 참조 사용) first_t2i = variations[0].get("t2i_prompt", "") labeled_refs = _resolve_refs_for_prompt(first_t2i, visible_entities) # N개 변형 병렬 생성 var_results = [] with ThreadPoolExecutor(max_workers=min(len(variations), 3)) as var_executor: var_futures = {} for vi, var in enumerate(variations): var_t2i = var.get("t2i_prompt", "") var_theme = var.get("variant_label", var.get("theme", f"var_{vi}")) var_theme_label = var.get("camera_effect", var.get("theme_label", var_theme)) if not var_t2i: continue var_futures[var_executor.submit( _generate_one_variation, si, var_t2i, var_theme, var_theme_label, visible_entities, best_prev_bytes, labeled_refs, )] = vi for future in as_completed(var_futures): try: result = future.result() if result: var_results.append(result) except Exception as exc: logger.error("Variation generation thread error: %s", exc) return (si, var_results, visible_entities, current_location_ids) # Run batch concurrently with ThreadPoolExecutor workers = min(max_concurrent, len(scene_batch)) batch_results: List[tuple] = [] with ThreadPoolExecutor(max_workers=workers) as executor: futures = { executor.submit(_generate_one_scene, si): si for si in scene_batch } for future in as_completed(futures): si = futures[future] try: result = future.result() batch_results.append(result) scenes_generated += 1 except Exception as exc: logger.error( "Scene generation failed: scene_index=%d, error=%s", si, exc, ) # 체크포인트에 실패 기록 still_id_failed = stills[si].get("id", "") if si < len(stills) else "" scene_cp.mark_failed(still_id_failed, str(exc)[:500]) scenes_generated += 1 # Post-process batch results: save assets, GPT select best, update location history # Process in scene_index order for deterministic DB writes for result in sorted(batch_results, key=lambda r: r[0]): si, var_results_list, visible_entities, current_location_ids = result still_data = stills[si] if not var_results_list: continue scene_ref_ids = [e["id"] for e in visible_entities] scene_lineage = self._build_lineage_fields( "scene_image_generator", ref_entity_ids=scene_ref_ids, prompt_type="cinematic", ) # Save all N variation images to DB saved_asset_ids = [] saved_paths = [] for vr in var_results_list: asset = ImageAsset( id=vr["id"], project_id=self._project_id, asset_type=vr["asset_type"], entity_id=vr["entity_id"], still_id=vr["still_id"], episode_id=vr["episode_id"], file_path=vr["file_path"], prompt_used=vr["prompt_used"], generation_model=vr["generation_model"], width=vr["width"], height=vr["height"], status=vr["status"], review_notes=vr["review_notes"], validation_score=vr.get("validation_score"), validation_result=vr.get("validation_result"), variant_type=vr.get("variant_type", "base"), theme_label=vr.get("theme_label"), prompt_type=scene_lineage["prompt_type"], code_version=scene_lineage["code_version"], prompt_file_version=scene_lineage["prompt_file_version"], reference_image_ids=scene_lineage["reference_image_ids"], is_primary=0, created_at=vr["created_at"], ) self._db.add(asset) saved_asset_ids.append(vr["id"]) saved_paths.append(Path(vr["file_path"])) self._db.commit() # ── v6: 새 파이프라인 ── # 1) N개 변형 저장 (is_primary=0) — 위에서 이미 완료 # 2) Gemini Vision: 앵글 적용할 이미지 선택 + 앵글 추천 (최소 20도) # 3) fal.ai 앵글 적용 → N+1번째 이미지 # 4) Gemini Vision: 전체 N+1개 중 최종 대표 이미지 선택 → is_primary=1 fal_generated = False if settings.fal_key and len(var_results_list) > 0: try: # Collect image bytes from all N variations angle_image_bytes = [] for vr in var_results_list: fp = Path(vr["file_path"]) if vr.get("file_path") else None if fp and fp.exists(): angle_image_bytes.append(fp.read_bytes()) if angle_image_bytes: angle_selection = _select_and_recommend_angle( angle_image_bytes, still_data.get("beat_title", ""), still_data.get("t2i_prompt_cinematic", ""), ) if angle_selection: sel_idx = angle_selection["best_for_angle"] logger.info( "Scene %d: angle selection — image %d/%d, H=%.0f V=%.0f Z=%.0f — %s", si, sel_idx + 1, len(var_results_list), angle_selection["horizontal_angle"], angle_selection["vertical_angle"], angle_selection["zoom"], angle_selection.get("reason", ""), ) # Apply fal.ai angle to selected image sel_result = var_results_list[sel_idx] if sel_idx < len(var_results_list) else var_results_list[0] sel_path = Path(sel_result["file_path"]) if sel_result.get("file_path") else None sel_asset_id = saved_asset_ids[sel_idx] if sel_idx < len(saved_asset_ids) else saved_asset_ids[0] if sel_path and sel_path.exists(): fal_bytes, fal_elapsed = _apply_fal_angle( sel_path.read_bytes(), angle_selection["horizontal_angle"], angle_selection["vertical_angle"], angle_selection["zoom"], ) if fal_bytes: fal_id = _new_id() fal_path = scene_dir / f"{fal_id}.png" fal_path.write_bytes(fal_bytes) fal_asset = ImageAsset( id=fal_id, project_id=self._project_id, asset_type="scene", still_id=still_data["id"], episode_id=episode_id, file_path=str(fal_path), prompt_used=( f"[fal.ai angle] H={angle_selection['horizontal_angle']}" f" V={angle_selection['vertical_angle']}" f" Z={angle_selection['zoom']}" ), generation_model="fal-ai/qwen-image-edit-2511-multiple-angles", status="generated", is_primary=0, variant_type="angle_fal", source_image_id=sel_asset_id, created_at=_now(), ) self._db.add(fal_asset) self._db.commit() # Append fal result to tracking lists saved_asset_ids.append(fal_id) var_results_list.append({ "id": fal_id, "file_path": str(fal_path), "asset_type": "scene", "still_id": still_data["id"], "episode_id": episode_id, "variant_type": "angle_fal", }) fal_generated = True logger.info( "Scene %d: fal.ai angle image saved (%d ms) → %s", si, fal_elapsed, fal_id, ) except Exception as exc: logger.warning("fal.ai angle pipeline failed for scene %d: %s", si, exc) # 4) Gemini Vision: 전체 N(+1) 이미지 중 최종 대표 선택 best_idx = 0 try: # 유효한 이미지만 수집 + 원래 인덱스 매핑 valid_pairs = [] for vi, vr in enumerate(var_results_list): fp = Path(vr["file_path"]) if vr.get("file_path") else None if fp and fp.exists(): valid_pairs.append((vi, fp.read_bytes())) if len(valid_pairs) > 1: valid_indices = [vi for vi, _ in valid_pairs] valid_bytes = [b for _, b in valid_pairs] raw_best = _select_final_best(valid_bytes, still_data.get("beat_title", "")) best_idx = valid_indices[min(raw_best, len(valid_indices) - 1)] logger.info( "Scene %d: final best selection — image %d/%d (fal=%s) for still %s", si, best_idx + 1, len(var_results_list), fal_generated, still_data.get("id"), ) except Exception as exc: logger.warning("Final best selection failed for scene %d, using first: %s", si, exc) best_idx = 0 # Validation score override: if selected has very low score, prefer alternatives def _vscore(r): s = r.get("validation_score") return s if s is not None else 50 selected_result = var_results_list[best_idx] if best_idx < len(var_results_list) else var_results_list[0] if _vscore(selected_result) < 40 and len(var_results_list) > 1: # Only consider original N variations for validation override (not fal result) orig_results = var_results_list[:len(saved_asset_ids) - (1 if fal_generated else 0)] if orig_results: alternatives = sorted(orig_results, key=_vscore, reverse=True) if _vscore(alternatives[0]) > _vscore(selected_result) + 20: override_idx = var_results_list.index(alternatives[0]) logger.info("Scene %d: overriding selection with higher-scored alternative (score %s->%s)", si, _vscore(selected_result), _vscore(alternatives[0])) best_idx = override_idx # Set is_primary on the final best image selected_id = saved_asset_ids[best_idx] if best_idx < len(saved_asset_ids) else saved_asset_ids[0] self._db.query(ImageAsset).filter( ImageAsset.id == selected_id, ).update({"is_primary": 1}) self._db.commit() # Track primary result for location history + scene_paths primary_result = var_results_list[best_idx] if best_idx < len(var_results_list) else var_results_list[0] scene_results_by_index[si] = primary_result primary_path = Path(primary_result["file_path"]) if primary_path.exists(): scene_paths_by_index[si] = primary_path scene_paths_by_index_by_id[still_data.get("id", "")] = primary_path # 체크포인트 기록 scene_cp.mark_completed(still_data.get("id", ""), { "asset_ids": saved_asset_ids, "primary_id": selected_id, "primary_path": str(primary_path), }) progress.update( f"씬 이미지 {scenes_generated}/{stills_to_gen}: {still_data.get('beat_title', '')}", scenes_generated + 1, total_items, ) # Update location_scene_history for same-background linking if primary_path.exists(): scene_bytes = primary_path.read_bytes() for loc_id in current_location_ids: location_scene_history[loc_id] = (scene_bytes, still_data) # v5: I2I 변형 비활성화 — N개 원본 생성으로 대체 if False: # 4. GPT Vision 기반 A/B 변형 추천 + 생성 (원본 이미지 분석 후) from app.modules.gemini_i2i_editor import GeminiI2IEditor from app.modules.variation_recommender_v2 import VariationRecommenderV2 var_recommender = VariationRecommenderV2() i2i_editor = GeminiI2IEditor( api_key=get_next_key(), model=settings.gemini_image_model, ) # ── 4a. 변형 추천 (GPT Vision) — 병렬 실행 ── var_max_workers = settings.max_concurrent_variation recommend_targets = [] for si, still_data in enumerate(stills): still_orm = stills_orm[si] scene_path = scene_paths_by_index.get(si) scene_result = scene_results_by_index.get(si) if not scene_path or not scene_result or not scene_path.exists(): continue if still_orm.variation_a_type: continue recommend_targets.append((si, still_data, still_orm, scene_path, scene_result)) if recommend_targets: logger.info("Variation recommendation: %d scenes, max_workers=%d", len(recommend_targets), var_max_workers) def _recommend_one(item): si, still_data, _still_orm, scene_path, _scene_result = item original_bytes = scene_path.read_bytes() rec = var_recommender.recommend_from_image( image_bytes=original_bytes, scene_description=still_data.get("t2i_prompt_cinematic", ""), beat_title=still_data.get("beat_title", ""), ) return si, rec with ThreadPoolExecutor(max_workers=min(var_max_workers, len(recommend_targets))) as executor: futures = {executor.submit(_recommend_one, item): item for item in recommend_targets} for future in as_completed(futures): item = futures[future] si = item[0] still_orm = item[2] still_id = item[1]["id"] try: _, rec = future.result() var_a = rec.get("variation_a", {}) var_b = rec.get("variation_b", {}) still_orm.variation_a_type = var_a.get("type", "none") angle_a = var_a.get("angle") if angle_a: angle_a["composition"] = var_a.get("composition", "") still_orm.variation_a_angle = json.dumps(angle_a) if angle_a else None still_orm.variation_a_color = var_a.get("color") or None still_orm.variation_a_reason = var_a.get("reason", "") still_orm.variation_b_type = var_b.get("type", "none") angle_b = var_b.get("angle") if angle_b: angle_b["composition"] = var_b.get("composition", "") still_orm.variation_b_angle = json.dumps(angle_b) if angle_b else None still_orm.variation_b_color = var_b.get("color") or None still_orm.variation_b_reason = var_b.get("reason", "") still_orm.recommended_variant = rec.get("recommended", "original") self._db.commit() logger.info("Variation recommendation for still %s: A=%s, B=%s, rec=%s", still_id, var_a.get("type"), var_b.get("type"), rec.get("recommended")) except Exception as exc: logger.warning("Variation recommend failed for still %s: %s", still_id, exc) # ── 4b. 변형 이미지 생성 (I2I) — 병렬 실행 ── i2i_targets = [] for si, still_data in enumerate(stills): still_orm = stills_orm[si] scene_path = scene_paths_by_index.get(si) scene_result = scene_results_by_index.get(si) if not scene_path or not scene_result or not scene_path.exists(): continue still_id = still_data["id"] for var_label, var_type_val, var_angle_val, var_color_val in [ ("variant_a", still_orm.variation_a_type, still_orm.variation_a_angle, still_orm.variation_a_color), ("variant_b", still_orm.variation_b_type, still_orm.variation_b_angle, still_orm.variation_b_color), ]: if not var_type_val or var_type_val == "none": continue existing = self._db.query(ImageAsset).filter( ImageAsset.still_id == still_id, ImageAsset.variant_type == var_label, ).first() if existing: continue i2i_targets.append((si, still_id, scene_path, scene_result, var_label, var_type_val, var_angle_val, var_color_val)) if i2i_targets: logger.info("I2I variation generation: %d items, max_workers=%d", len(i2i_targets), var_max_workers) def _generate_i2i(item): si, still_id, scene_path, scene_result, var_label, var_type_val, var_angle_val, var_color_val = item original_bytes = scene_path.read_bytes() angle_data = json.loads(var_angle_val) if var_angle_val else None color_prompt = var_color_val or None composition = angle_data.pop("composition", "") if angle_data else "" edited = None if var_type_val == "angle" and angle_data: edited = i2i_editor.edit_angle( original_bytes, angle_data.get("horizontal", 0), angle_data.get("vertical", 0), angle_data.get("zoom", 1.0), prompt=composition, ) elif var_type_val == "color" and color_prompt: edited = i2i_editor.edit_color(original_bytes, color_prompt) elif var_type_val == "angle+color" and angle_data and color_prompt: edited = i2i_editor.edit_combined( original_bytes, angle_data.get("horizontal", 0), angle_data.get("vertical", 0), angle_data.get("zoom", 1.0), color_prompt=color_prompt, composition=composition, ) if edited is None: return None var_path = scene_dir / f"{_new_id()}.png" var_path.parent.mkdir(parents=True, exist_ok=True) var_path.write_bytes(edited) return { "id": _new_id(), "still_id": still_id, "episode_id": episode_id, "file_path": str(var_path), "prompt_used": scene_result.get("prompt_used", ""), "var_label": var_label, "var_angle_val": var_angle_val, "var_color_val": var_color_val, "source_image_id": scene_result["id"], } with ThreadPoolExecutor(max_workers=min(var_max_workers, len(i2i_targets))) as executor: futures = {executor.submit(_generate_i2i, item): item for item in i2i_targets} for future in as_completed(futures): item = futures[future] still_id = item[1] var_label = item[4] try: result = future.result() if result is None: continue var_asset = ImageAsset( id=result["id"], project_id=self._project_id, asset_type="scene", still_id=result["still_id"], episode_id=result["episode_id"], file_path=result["file_path"], prompt_used=result["prompt_used"], generation_model=settings.gemini_image_model, status="generated", variant_type=result["var_label"], angle_applied=result["var_angle_val"], color_applied=result["var_color_val"], source_image_id=result["source_image_id"], is_primary=0, created_at=_now(), ) self._db.add(var_asset) self._db.commit() logger.info("Variation %s generated for still %s", var_label, still_id) except Exception as exc: logger.warning("Variation %s failed for still %s: %s", var_label, still_id, exc) progress.complete() self._logger.log( actor_id=self._actor_id, action="episode.generate_images", resource_type="episode", resource_id=episode_id, project_id=self._project_id, detail={ "reference_loaded": len(ref_image_map), "scene_count": len(stills), }, ip_address=ip, ) def generate_variations_only(self, episode_id: str) -> None: """기존 원본 이미지 기반 A/B 변형 추천 + 생성 (별도 실행).""" from app.modules.gemini_i2i_editor import GeminiI2IEditor from app.modules.variation_recommender_v2 import VariationRecommenderV2 stills_orm = ( self._db.query(SceneStill) .filter(SceneStill.project_id == self._project_id, SceneStill.episode_id == episode_id) .order_by(SceneStill.still_index) .all() ) if not stills_orm: return project_dir = self._get_project_dir() scene_dir = project_dir / "images" / episode_id / "scene" var_recommender = VariationRecommenderV2() i2i_editor = GeminiI2IEditor( api_key=get_next_key(), model=settings.gemini_image_model, ) recommended = 0 generated = 0 for si, still_orm in enumerate(stills_orm): # 원본 이미지 DB에서 조회 original_asset = ( self._db.query(ImageAsset) .filter( ImageAsset.still_id == still_orm.id, ImageAsset.asset_type == "scene", ImageAsset.variant_type == "original", ) .order_by(ImageAsset.created_at.desc()) .first() ) if not original_asset: continue original_path = Path(original_asset.file_path) if not original_path.exists(): continue original_bytes = original_path.read_bytes() # 추천이 안 된 씬만 추천 if not still_orm.variation_a_type: try: rec = var_recommender.recommend_from_image( image_bytes=original_bytes, scene_description=still_orm.t2i_prompt_cinematic or "", beat_title=still_orm.beat_title or "", ) var_a = rec.get("variation_a", {}) var_b = rec.get("variation_b", {}) still_orm.variation_a_type = var_a.get("type", "none") angle_a = var_a.get("angle") if angle_a: angle_a["composition"] = var_a.get("composition", "") still_orm.variation_a_angle = json.dumps(angle_a) if angle_a else None still_orm.variation_a_color = var_a.get("color") or None still_orm.variation_a_reason = var_a.get("reason", "") still_orm.variation_b_type = var_b.get("type", "none") angle_b = var_b.get("angle") if angle_b: angle_b["composition"] = var_b.get("composition", "") still_orm.variation_b_angle = json.dumps(angle_b) if angle_b else None still_orm.variation_b_color = var_b.get("color") or None still_orm.variation_b_reason = var_b.get("reason", "") still_orm.recommended_variant = rec.get("recommended", "original") self._db.commit() recommended += 1 logger.info("Variation %d/%d recommended: A=%s B=%s", si + 1, len(stills_orm), var_a.get("type"), var_b.get("type")) except Exception as exc: logger.warning("Variation recommend %d failed: %s", si + 1, exc) continue # 변형 이미지 생성 (추천된 것만) for var_idx, (var_label, var_type_val, var_angle_val, var_color_val) in enumerate([ ("variant_a", still_orm.variation_a_type, still_orm.variation_a_angle, still_orm.variation_a_color), ("variant_b", still_orm.variation_b_type, still_orm.variation_b_angle, still_orm.variation_b_color), ]): if not var_type_val or var_type_val == "none": continue # 이미 생성된 변형 스킵 existing_var = ( self._db.query(ImageAsset) .filter( ImageAsset.still_id == still_orm.id, ImageAsset.variant_type == var_label, ) .first() ) if existing_var: continue try: angle_data = json.loads(var_angle_val) if var_angle_val else None color_prompt = var_color_val or None composition = angle_data.pop("composition", "") if angle_data else "" if var_type_val == "angle" and angle_data: edited = i2i_editor.edit_angle( original_bytes, angle_data.get("horizontal", 0), angle_data.get("vertical", 0), angle_data.get("zoom", 1.0), prompt=composition, ) elif var_type_val == "color" and color_prompt: edited = i2i_editor.edit_color(original_bytes, color_prompt) elif var_type_val == "angle+color" and angle_data and color_prompt: edited = i2i_editor.edit_combined( original_bytes, angle_data.get("horizontal", 0), angle_data.get("vertical", 0), angle_data.get("zoom", 1.0), color_prompt=color_prompt, composition=composition, ) else: continue var_path = scene_dir / f"{_new_id()}.png" var_path.parent.mkdir(parents=True, exist_ok=True) var_path.write_bytes(edited) var_asset = ImageAsset( id=_new_id(), project_id=self._project_id, asset_type="scene", still_id=still_orm.id, episode_id=episode_id, file_path=str(var_path), prompt_used=original_asset.prompt_used or "", generation_model=settings.gemini_image_model, status="generated", variant_type=var_label, angle_applied=var_angle_val, color_applied=var_color_val, source_image_id=original_asset.id, is_primary=0, created_at=_now(), ) self._db.add(var_asset) self._db.commit() generated += 1 logger.info("Variation %s generated for still %d", var_label, si + 1) except Exception as exc: logger.warning("Variation %s generation failed for still %d: %s", var_label, si + 1, exc) logger.info("Variations complete: recommended=%d, generated=%d", recommended, generated) def list_images( self, episode_id: Optional[str] = None, asset_type: Optional[str] = None, entity_id: Optional[str] = None, still_id: Optional[str] = None, outlook_id: Optional[str] = None, ) -> List[Dict[str, Any]]: """List images, optionally filtered by type, episode, entity, or still. outlook_id: when provided, returns images whose prompt_used contains 'outlook_id:{outlook_id}' (composite images referencing this outlook). """ query = self._db.query(ImageAsset).filter(ImageAsset.project_id == self._project_id) if asset_type: query = query.filter(ImageAsset.asset_type == asset_type) if episode_id: query = query.filter(ImageAsset.episode_id == episode_id) if entity_id: query = query.filter(ImageAsset.entity_id == entity_id) if still_id: query = query.filter(ImageAsset.still_id == still_id) if outlook_id: query = query.filter(ImageAsset.prompt_used.like(f"%outlook_id:{outlook_id}%")) images = query.order_by(ImageAsset.created_at.desc()).all() # Enrich with parsed angle/recommendation fields when still context available still_cache: Dict[str, Optional[SceneStill]] = {} results = [] for img in images: d = self._image_to_dict(img) if img.still_id: if img.still_id not in still_cache: still_cache[img.still_id] = ( self._db.query(SceneStill) .filter(SceneStill.id == img.still_id) .first() ) d = self._enrich_image_dict(d, img, still_cache.get(img.still_id)) results.append(d) return results @staticmethod def _image_to_dict(img: ImageAsset) -> Dict[str, Any]: """Convert an ImageAsset ORM object to a dict.""" return { "id": img.id, "asset_type": img.asset_type, "entity_id": img.entity_id, "still_id": img.still_id, "episode_id": img.episode_id, "file_path": img.file_path, "prompt_used": img.prompt_used, "generation_model": img.generation_model, "width": img.width, "height": img.height, "status": img.status, "review_notes": img.review_notes or "", "validation_score": img.validation_score, "validation_result": img.validation_result, "sanitization_strategy": img.sanitization_strategy, "original_prompt": img.original_prompt, "sanitization_note": img.sanitization_note, "variant_type": img.variant_type, "angle_applied": img.angle_applied, "color_applied": img.color_applied, "source_image_id": img.source_image_id, "is_primary": bool(img.is_primary), "prompt_type": img.prompt_type, "code_version": img.code_version, "prompt_file_version": img.prompt_file_version, "reference_image_ids": img.reference_image_ids or "[]", "theme_label": img.theme_label, "created_at": img.created_at, } @staticmethod def _enrich_image_dict( d: Dict[str, Any], img: ImageAsset, still: Optional["SceneStill"] = None, ) -> Dict[str, Any]: """Add parsed angle/color fields and recommendation info to an image dict.""" # Parse angle_applied JSON to extract horizontal/vertical/zoom if img.angle_applied: try: angle = json.loads(img.angle_applied) d["angle_horizontal"] = angle.get("horizontal") d["angle_vertical"] = angle.get("vertical") d["angle_zoom"] = angle.get("zoom") except (json.JSONDecodeError, TypeError): pass # Parse color_applied for convenience d.setdefault("color_prompt", img.color_applied) # Add recommendation / selection info from the still if still: rec = (still.recommended_variant or "").upper() sel = (still.selected_variant or still.recommended_variant or "original").upper() vt = (img.variant_type or "original") vt_key = vt.replace("variant_", "").upper() if vt.startswith("variant_") else vt.upper() d["is_recommended"] = (rec == vt_key) d["selected_for_pdf"] = (sel == vt_key) # Recommendation reason reason_attr = None if vt == "variant_a": reason_attr = "variation_a_reason" elif vt == "variant_b": reason_attr = "variation_b_reason" d["recommendation_reason"] = getattr(still, reason_attr, None) if reason_attr else None return d def get_image(self, image_id: str) -> ImageAsset: """Get a single image asset by ID.""" img = ( self._db.query(ImageAsset) .filter(ImageAsset.id == image_id, ImageAsset.project_id == self._project_id) .first() ) if not img: raise AppError( code="image.not_found", message=t("image.not_found"), status_code=404, ) return img def update_review( self, image_id: str, status: str, notes: str, ip: Optional[str] = None, ) -> ImageAsset: """Update review status of an image.""" if status not in ("approved", "needs_fix"): raise AppError( code="image.invalid_review_status", message=t("image.invalid_review_status"), status_code=400, ) img = self.get_image(image_id) img.status = status img.review_notes = notes self._db.commit() self._logger.log( actor_id=self._actor_id, action="image.review", resource_type="image", resource_id=image_id, project_id=self._project_id, detail={"status": status, "notes": notes}, ip_address=ip, ) return img def regenerate_image( self, image_id: str, ip: Optional[str] = None, ) -> None: """Mark an image for regeneration.""" img = self.get_image(image_id) img.status = "regenerating" self._db.commit() self._logger.log( actor_id=self._actor_id, action="image.regenerate", resource_type="image", resource_id=image_id, project_id=self._project_id, ip_address=ip, ) def regenerate_needs_fix( self, episode_id: str, ip: Optional[str] = None, ) -> int: """Mark all needs_fix images for an episode for regeneration.""" images = ( self._db.query(ImageAsset) .filter( ImageAsset.project_id == self._project_id, ImageAsset.episode_id == episode_id, ImageAsset.status == "needs_fix", ) .all() ) if not images: raise AppError( code="image.no_needs_fix", message=t("image.no_needs_fix"), status_code=400, ) for img in images: img.status = "regenerating" self._db.commit() self._logger.log( actor_id=self._actor_id, action="image.regenerate_batch", resource_type="episode", resource_id=episode_id, project_id=self._project_id, detail={"count": len(images)}, ip_address=ip, ) return len(images) # ------------------------------------------------------------------ # Primary image management # ------------------------------------------------------------------ def set_primary_image( self, image_id: str, ip: Optional[str] = None, ) -> Dict[str, Any]: """Set an image as the primary image for its entity/still, unsetting others.""" img = self.get_image(image_id) # Unset primary for all other images with the same entity_id or still_id if img.entity_id: siblings = ( self._db.query(ImageAsset) .filter( ImageAsset.project_id == self._project_id, ImageAsset.entity_id == img.entity_id, ImageAsset.id != image_id, ) .all() ) for sib in siblings: sib.is_primary = 0 if img.still_id: siblings = ( self._db.query(ImageAsset) .filter( ImageAsset.project_id == self._project_id, ImageAsset.still_id == img.still_id, ImageAsset.id != image_id, ) .all() ) for sib in siblings: sib.is_primary = 0 img.is_primary = 1 self._db.commit() self._logger.log( actor_id=self._actor_id, action="image.set_primary", resource_type="image", resource_id=image_id, project_id=self._project_id, detail={"entity_id": img.entity_id, "still_id": img.still_id}, ip_address=ip, ) return self._image_to_dict(img) def _auto_set_primary(self, img: ImageAsset) -> None: """Auto-set the newly created image as primary, unsetting others.""" if img.entity_id: siblings = ( self._db.query(ImageAsset) .filter( ImageAsset.project_id == self._project_id, ImageAsset.entity_id == img.entity_id, ImageAsset.id != img.id, ) .all() ) for sib in siblings: sib.is_primary = 0 if img.still_id: siblings = ( self._db.query(ImageAsset) .filter( ImageAsset.project_id == self._project_id, ImageAsset.still_id == img.still_id, ImageAsset.id != img.id, ) .all() ) for sib in siblings: sib.is_primary = 0 img.is_primary = 1 # ------------------------------------------------------------------ # Single image generation # ------------------------------------------------------------------ def generate_single_entity_image( self, entity_id: str, custom_prompt: Optional[str] = None, ip: Optional[str] = None, ) -> Dict[str, Any]: """Generate a single reference image for an entity.""" if not settings.gemini_api_key and gemini_key_count() == 0: raise AppError( code="image.gemini_key_missing", message=t("image.gemini_key_missing"), status_code=400, ) entity = ( self._db.query(EntityCanon) .filter(EntityCanon.id == entity_id, EntityCanon.project_id == self._project_id) .first() ) if not entity: raise AppError( code="entity.not_found", message=t("entity.not_found"), status_code=404, ) # Find an episode this entity appears in (for directory structure) link = ( self._db.query(EntityEpisodeLink) .filter(EntityEpisodeLink.canon_id == entity_id) .first() ) episode_id = link.episode_id if link else "shared" project_dir = self._get_project_dir() reference_dir = project_dir / "images" / episode_id / "reference" entity_data = { "id": entity.id, "name": entity.name, "entity_type": entity.entity_type, "description": entity.description or "", "stable_traits": entity.stable_traits or "{}", } # Get world guide if available world_guide = self._get_latest_world_guide(episode_id) language = "ko" if episode_id != "shared": ep = self._db.query(Episode).filter(Episode.id == episode_id).first() if ep: language = ep.language or "ko" # Override prompt if custom_prompt provided if custom_prompt: entity_data["description"] = custom_prompt gemini_client = GeminiImageClient(model=settings.gemini_image_model) ref_gen = ReferenceImageGenerator(gemini_client=gemini_client, language=language) openai_client = OpenAIClient() sanitizer = PromptSanitizer(openai_client) tracker = GenerationTracker(self._db, self._project_id) try: result = ref_gen.generate_for_entity_with_retry( entity=entity_data, world_guide=world_guide, output_dir=reference_dir, sanitizer=sanitizer, tracker=tracker, ) except ModerationError as exc: raise AppError( code="image.generation_blocked", message=f"Image generation blocked: {exc.block_reason}", status_code=400, ) lineage = self._build_lineage_fields("reference_image_generator") asset = ImageAsset( id=result["id"], project_id=self._project_id, asset_type=result["asset_type"], entity_id=result["entity_id"], still_id=result["still_id"], episode_id=episode_id, file_path=result["file_path"], prompt_used=result["prompt_used"], generation_model=result["generation_model"], width=result["width"], height=result["height"], status=result["status"], review_notes=result["review_notes"], sanitization_strategy=result.get("sanitization_strategy"), original_prompt=result.get("original_prompt"), sanitization_note=result.get("sanitization_note"), code_version=lineage["code_version"], prompt_file_version=lineage["prompt_file_version"], reference_image_ids=lineage["reference_image_ids"], created_at=result["created_at"], ) self._db.add(asset) self._auto_set_primary(asset) self._db.commit() self._logger.log( actor_id=self._actor_id, action="image.generate_single_entity", resource_type="image", resource_id=result["id"], project_id=self._project_id, detail={"entity_id": entity_id}, ip_address=ip, ) return self._image_to_dict(asset) def generate_single_scene_image( self, still_id: str, custom_prompt: Optional[str] = None, ip: Optional[str] = None, ) -> Dict[str, Any]: """Generate a single scene image for a still.""" if not settings.gemini_api_key and gemini_key_count() == 0: raise AppError( code="image.gemini_key_missing", message=t("image.gemini_key_missing"), status_code=400, ) still = ( self._db.query(SceneStill) .filter(SceneStill.id == still_id, SceneStill.project_id == self._project_id) .first() ) if not still: raise AppError( code="still.not_found", message=t("still.not_found"), status_code=404, ) episode_id = still.episode_id project_dir = self._get_project_dir() scene_dir = project_dir / "images" / episode_id / "scene" still_data = { "id": still.id, "still_index": still.still_index, "screenplay_scene_heading": still.screenplay_scene_heading or "", "beat_title": still.beat_title or "", "still_frame_prompt": custom_prompt or still.still_frame_prompt or "", "camera_json": still.camera_json or "{}", "lighting_json": still.lighting_json or "{}", "visible_entities_json": still.visible_entities_json or "[]", } # Get world guide world_guide = self._get_latest_world_guide(episode_id) # Get visible entities visible_entities = self._get_visible_entities(still.visible_entities_json) # Get reference image map for visible entities (exclude locations for scene gen) all_ref_image_map = self._get_reference_image_map(visible_entities) ref_image_map = { eid: img_bytes for eid, img_bytes in all_ref_image_map.items() if not any( e["id"] == eid and e.get("entity_type") == "location" for e in visible_entities ) } ep = self._db.query(Episode).filter(Episode.id == episode_id).first() language = ep.language if ep else "ko" gemini_client = GeminiImageClient(model=settings.gemini_image_model) openai_client = OpenAIClient() sanitizer = PromptSanitizer(openai_client) tracker = GenerationTracker(self._db, self._project_id) if custom_prompt: # raw_prompt 모드: 사용자가 편집한 최종 프롬프트를 래핑 없이 Gemini에 직접 전달 from app.modules.pipeline.scene_image_pipeline import generate_and_validate_scene # 참조 이미지 구성 labeled_refs = [] try: entity_lookup = {e["id"]: e for e in visible_entities} for e in visible_entities: eid = e["id"] etype = e.get("entity_type", "") if etype != "location" and eid in ref_image_map: labeled_refs.append(("character identity" if etype == "character" else "object appearance", ref_image_map[eid])) except Exception: pass try: pipe_result = generate_and_validate_scene( gemini_client=gemini_client, t2i_prompt=custom_prompt, beat_title=still_data.get("beat_title", ""), output_dir=scene_dir, reference_images=labeled_refs if labeled_refs else None, previous_scene_bytes=None, ) scene_result = { "id": _new_id(), "asset_type": "scene", "entity_id": None, "still_id": still_id, "episode_id": episode_id, "file_path": pipe_result["file_path"], "prompt_used": custom_prompt, "generation_model": pipe_result.get("generation_model", settings.gemini_image_model), "width": None, "height": None, "status": "generated", "review_notes": json.dumps(pipe_result.get("validation", {}), ensure_ascii=False), "created_at": _now(), } except ModerationError as exc: raise AppError( code="image.generation_blocked", message=f"Image generation blocked: {exc.block_reason}", status_code=400, ) else: # 기존 경로: still_frame_prompt → scene pipeline 래핑 scene_gen = SceneImageGenerator(gemini_client=gemini_client, language=language) try: scene_result = scene_gen.generate_for_still_with_retry( still=still_data, visible_entities=visible_entities, world_guide=world_guide, output_dir=scene_dir, sanitizer=sanitizer, tracker=tracker, reference_image_map=ref_image_map, previous_scene_bytes=None, previous_still=None, episode_id=episode_id, ) except ModerationError as exc: raise AppError( code="image.generation_blocked", message=f"Image generation blocked: {exc.block_reason}", status_code=400, ) # Build lineage fields ref_entity_ids = [e["id"] for e in visible_entities] lineage = self._build_lineage_fields( "scene_image_generator", ref_entity_ids=ref_entity_ids, prompt_type="original", ) asset = ImageAsset( id=scene_result["id"], project_id=self._project_id, asset_type=scene_result["asset_type"], entity_id=scene_result["entity_id"], still_id=scene_result["still_id"], episode_id=scene_result["episode_id"], file_path=scene_result["file_path"], prompt_used=scene_result["prompt_used"], generation_model=scene_result["generation_model"], width=scene_result["width"], height=scene_result["height"], status=scene_result["status"], review_notes=scene_result["review_notes"], sanitization_strategy=scene_result.get("sanitization_strategy"), original_prompt=scene_result.get("original_prompt"), sanitization_note=scene_result.get("sanitization_note"), prompt_type=lineage["prompt_type"], code_version=lineage["code_version"], prompt_file_version=lineage["prompt_file_version"], reference_image_ids=lineage["reference_image_ids"], created_at=scene_result["created_at"], ) self._db.add(asset) self._auto_set_primary(asset) self._db.commit() self._logger.log( actor_id=self._actor_id, action="image.generate_single_scene", resource_type="image", resource_id=scene_result["id"], project_id=self._project_id, detail={"still_id": still_id}, ip_address=ip, ) return self._image_to_dict(asset) # ------------------------------------------------------------------ # Upload custom image # ------------------------------------------------------------------ def upload_custom_image( self, file_bytes: bytes, filename: str, entity_id: Optional[str] = None, still_id: Optional[str] = None, episode_id: Optional[str] = None, ip: Optional[str] = None, ) -> Dict[str, Any]: """Upload a custom image file and create an ImageAsset record.""" # Validate ownership: entity/still must belong to this project if entity_id: owner_check = ( self._db.query(EntityCanon) .filter(EntityCanon.id == entity_id, EntityCanon.project_id == self._project_id) .first() ) if not owner_check: raise AppError( code="image.entity_not_found", message=t("image.not_found"), status_code=404, ) if still_id: owner_check = ( self._db.query(SceneStill) .filter(SceneStill.id == still_id, SceneStill.project_id == self._project_id) .first() ) if not owner_check: raise AppError( code="image.still_not_found", message=t("image.not_found"), status_code=404, ) # Determine asset type asset_type = "reference" if entity_id else "scene" # Determine episode_id from entity or still if not provided if not episode_id: if entity_id: link = ( self._db.query(EntityEpisodeLink) .filter(EntityEpisodeLink.canon_id == entity_id) .first() ) episode_id = link.episode_id if link else "shared" elif still_id: still = ( self._db.query(SceneStill) .filter(SceneStill.id == still_id) .first() ) episode_id = still.episode_id if still else "shared" else: episode_id = "shared" # Save file project_dir = self._get_project_dir() sub_dir = "reference" if entity_id else "scene" images_dir = project_dir / "images" / episode_id / sub_dir images_dir.mkdir(parents=True, exist_ok=True) image_id = _new_id() ext = Path(filename).suffix or ".png" file_path = images_dir / f"{image_id}{ext}" file_path.write_bytes(file_bytes) asset = ImageAsset( id=image_id, project_id=self._project_id, asset_type=asset_type, entity_id=entity_id, still_id=still_id, episode_id=episode_id, file_path=str(file_path), prompt_used="uploaded", generation_model="manual_upload", width=None, height=None, status="generated", review_notes="", created_at=_now(), ) self._db.add(asset) self._auto_set_primary(asset) self._db.commit() self._logger.log( actor_id=self._actor_id, action="image.upload", resource_type="image", resource_id=image_id, project_id=self._project_id, detail={"entity_id": entity_id, "still_id": still_id, "filename": filename}, ip_address=ip, ) return self._image_to_dict(asset) # ------------------------------------------------------------------ # Variation pipeline # ------------------------------------------------------------------ def recommend_variations( self, still_id: str, ip: Optional[str] = None, ) -> Dict[str, Any]: """LLM에게 A/B 변형 추천 받기.""" if not settings.openai_api_key: raise AppError( code="image.openai_key_missing", message=t("image.openai_key_missing"), status_code=400, ) still = ( self._db.query(SceneStill) .filter(SceneStill.id == still_id, SceneStill.project_id == self._project_id) .first() ) if not still: raise AppError( code="still.not_found", message=t("still.not_found"), status_code=404, ) camera_angle = still.camera_json or "{}" lighting_mood = still.lighting_json or "{}" from app.modules.variation_recommender import VariationRecommender openai_client = OpenAIClient() recommender = VariationRecommender(llm_client=openai_client) result = recommender.recommend( scene_description=still.still_frame_prompt or "", beat_title=still.beat_title or "", camera_angle=camera_angle, lighting_mood=lighting_mood, ) # Save recommendations to still var_a = result.get("variation_a", {}) still.variation_a_type = var_a.get("type", "none") still.variation_a_angle = json.dumps(var_a.get("angle")) if var_a.get("angle") else None still.variation_a_color = var_a.get("color") or None still.variation_a_reason = var_a.get("reason") or None var_b = result.get("variation_b", {}) still.variation_b_type = var_b.get("type", "none") still.variation_b_angle = json.dumps(var_b.get("angle")) if var_b.get("angle") else None still.variation_b_color = var_b.get("color") or None still.variation_b_reason = var_b.get("reason") or None still.recommended_variant = result.get("recommended", "original") self._db.commit() self._logger.log( actor_id=self._actor_id, action="still.recommend_variations", resource_type="still", resource_id=still_id, project_id=self._project_id, detail={ "variation_a_type": still.variation_a_type, "variation_b_type": still.variation_b_type, "recommended": still.recommended_variant, }, ip_address=ip, ) return result def generate_scene_with_variations( self, still_id: str, ip: Optional[str] = None, ) -> Dict[str, Any]: """원본 + A변형 + B변형 생성. 1. Generate original (existing scene image generation) 2. If variation_a has angle/color -> i2i from original 3. If variation_b has angle/color -> i2i from original 4. Set recommended as PDF default """ if not settings.gemini_api_key and gemini_key_count() == 0: raise AppError( code="image.gemini_key_missing", message=t("image.gemini_key_missing"), status_code=400, ) still = ( self._db.query(SceneStill) .filter(SceneStill.id == still_id, SceneStill.project_id == self._project_id) .first() ) if not still: raise AppError( code="still.not_found", message=t("still.not_found"), status_code=404, ) episode_id = still.episode_id project_dir = self._get_project_dir() scene_dir = project_dir / "images" / episode_id / "scene" scene_dir.mkdir(parents=True, exist_ok=True) # Step 1: Generate original scene image original_result = self.generate_single_scene_image(still_id, ip=ip) original_id = original_result["id"] # Mark original as variant_type="original" original_asset = ( self._db.query(ImageAsset) .filter(ImageAsset.id == original_id) .first() ) if original_asset: original_asset.variant_type = "original" self._db.flush() # Read original image bytes for i2i original_path = Path(original_result["file_path"]) if not original_path.exists(): return { "original": original_result, "variant_a": None, "variant_b": None, "recommended": still.recommended_variant or "original", } original_bytes = original_path.read_bytes() from app.modules.gemini_i2i_editor import GeminiI2IEditor i2i_editor = GeminiI2IEditor( api_key=get_next_key(), model=settings.gemini_image_model, ) lineage = self._build_lineage_fields("gemini_i2i_editor") # Step 2: Generate variation A variant_a_result = None if still.variation_a_type and still.variation_a_type != "none": variant_a_result = self._generate_variation( i2i_editor=i2i_editor, original_bytes=original_bytes, original_id=original_id, still=still, variant_label="variant_a", var_type=still.variation_a_type, angle_json=still.variation_a_angle, color_prompt=still.variation_a_color, scene_dir=scene_dir, episode_id=episode_id, lineage=lineage, ) # Step 3: Generate variation B variant_b_result = None if still.variation_b_type and still.variation_b_type != "none": variant_b_result = self._generate_variation( i2i_editor=i2i_editor, original_bytes=original_bytes, original_id=original_id, still=still, variant_label="variant_b", var_type=still.variation_b_type, angle_json=still.variation_b_angle, color_prompt=still.variation_b_color, scene_dir=scene_dir, episode_id=episode_id, lineage=lineage, ) # Step 4: Set recommended variant as primary for PDF recommended = still.recommended_variant or "original" if recommended == "A" and variant_a_result: self._set_variant_primary(variant_a_result["id"], still_id) elif recommended == "B" and variant_b_result: self._set_variant_primary(variant_b_result["id"], still_id) self._db.commit() self._logger.log( actor_id=self._actor_id, action="still.generate_with_variations", resource_type="still", resource_id=still_id, project_id=self._project_id, detail={ "original_id": original_id, "variant_a_id": variant_a_result["id"] if variant_a_result else None, "variant_b_id": variant_b_result["id"] if variant_b_result else None, "recommended": recommended, }, ip_address=ip, ) return { "original": original_result, "variant_a": variant_a_result, "variant_b": variant_b_result, "recommended": recommended, } def _generate_variation( self, i2i_editor: Any, original_bytes: bytes, original_id: str, still: SceneStill, variant_label: str, var_type: str, angle_json: Optional[str], color_prompt: Optional[str], scene_dir: Path, episode_id: str, lineage: Dict[str, Any], ) -> Optional[Dict[str, Any]]: """Generate a single variation via i2i editing.""" try: angle_params = json.loads(angle_json) if angle_json else None except json.JSONDecodeError: angle_params = None try: if var_type == "angle" and angle_params: edited_bytes = i2i_editor.edit_angle( image_bytes=original_bytes, horizontal=int(angle_params.get("horizontal", 0)), vertical=int(angle_params.get("vertical", 0)), zoom=float(angle_params.get("zoom", 1.0)), ) elif var_type == "color" and color_prompt: edited_bytes = i2i_editor.edit_color( image_bytes=original_bytes, color_prompt=color_prompt, ) elif var_type == "angle+color" and angle_params and color_prompt: edited_bytes = i2i_editor.edit_combined( original_bytes=original_bytes, horizontal=int(angle_params.get("horizontal", 0)), vertical=int(angle_params.get("vertical", 0)), zoom=float(angle_params.get("zoom", 1.0)), color_prompt=color_prompt, ) else: logger.warning("Variation %s has type=%s but missing params", variant_label, var_type) return None except Exception as exc: logger.error("Variation %s i2i failed: %s", variant_label, exc) return None # Save variation image file image_id = _new_id() file_path = scene_dir / f"{image_id}.png" file_path.write_bytes(edited_bytes) asset = ImageAsset( id=image_id, project_id=self._project_id, asset_type="scene", still_id=still.id, episode_id=episode_id, file_path=str(file_path), prompt_used=f"i2i_{var_type}", generation_model=settings.gemini_image_model, status="generated", review_notes="", variant_type=variant_label, angle_applied=angle_json, color_applied=color_prompt, source_image_id=original_id, code_version=lineage["code_version"], prompt_file_version=lineage.get("prompt_file_version"), created_at=_now(), ) self._db.add(asset) self._db.flush() return self._image_to_dict(asset) def _set_variant_primary(self, image_id: str, still_id: str) -> None: """Set a variant as the primary image for a still.""" siblings = ( self._db.query(ImageAsset) .filter( ImageAsset.project_id == self._project_id, ImageAsset.still_id == still_id, ImageAsset.id != image_id, ) .all() ) for sib in siblings: sib.is_primary = 0 target = self._db.query(ImageAsset).filter(ImageAsset.id == image_id).first() if target: target.is_primary = 1 def regenerate_variation( self, image_id: str, angle_json: Optional[str] = None, color_prompt: Optional[str] = None, ip: Optional[str] = None, ) -> Dict[str, Any]: """사용자가 수정한 앵글/색감으로 재생성 (원본 기반 i2i).""" if not settings.gemini_api_key and gemini_key_count() == 0: raise AppError( code="image.gemini_key_missing", message=t("image.gemini_key_missing"), status_code=400, ) img = self.get_image(image_id) # Find source (original) image source_id = img.source_image_id or image_id source_img = ( self._db.query(ImageAsset) .filter(ImageAsset.id == source_id, ImageAsset.project_id == self._project_id) .first() ) if not source_img: raise AppError( code="image.not_found", message=t("image.not_found"), status_code=404, ) source_path = Path(source_img.file_path) if not source_path.exists(): raise AppError( code="image.not_found", message=t("image.not_found"), status_code=404, ) original_bytes = source_path.read_bytes() angle_params = None if angle_json: try: angle_params = json.loads(angle_json) except json.JSONDecodeError: pass if not angle_params: raise AppError( code="image.no_edit_params", message="No angle parameters provided", status_code=400, ) # fal.ai로 앵글 편집 if not settings.fal_key: raise AppError( code="image.fal_key_missing", message="fal.ai API 키가 설정되지 않았습니다", status_code=400, ) h = float(angle_params.get("horizontal", 0)) v = float(angle_params.get("vertical", 0)) z = float(angle_params.get("zoom", 1.0)) fal_bytes, fal_elapsed = _apply_fal_angle(original_bytes, h, v, z) if not fal_bytes: raise AppError( code="image.fal_failed", message="fal.ai 앵글 편집에 실패했습니다", status_code=500, ) logger.info("fal.ai angle edit: H=%.0f V=%.0f Z=%.0f (%d ms)", h, v, z, fal_elapsed) # Save edited image scene_dir = Path(source_img.file_path).parent new_id = _new_id() file_path = scene_dir / f"{new_id}.png" file_path.write_bytes(fal_bytes) asset = ImageAsset( id=new_id, project_id=self._project_id, asset_type="scene", still_id=img.still_id, episode_id=img.episode_id, file_path=str(file_path), prompt_used=f"[fal.ai angle] H={h} V={v} Z={z}", generation_model="fal-ai/qwen-image-edit-2511-multiple-angles", status="generated", review_notes="", variant_type="angle_fal", angle_applied=angle_json, source_image_id=source_id, created_at=_now(), ) self._db.add(asset) self._db.commit() self._logger.log( actor_id=self._actor_id, action="image.regenerate_variation", resource_type="image", resource_id=new_id, project_id=self._project_id, detail={ "source_image_id": source_id, "edit_type": edit_type, }, ip_address=ip, ) return self._image_to_dict(asset) def select_variant( self, still_id: str, variant: str, ip: Optional[str] = None, ) -> Dict[str, Any]: """PDF 대표 이미지 선택 (original/A/B).""" if variant not in ("original", "A", "B"): raise AppError( code="image.invalid_variant", message="Variant must be 'original', 'A', or 'B'", status_code=400, ) still = ( self._db.query(SceneStill) .filter(SceneStill.id == still_id, SceneStill.project_id == self._project_id) .first() ) if not still: raise AppError( code="still.not_found", message=t("still.not_found"), status_code=404, ) still.selected_variant = variant self._db.commit() # Set primary image accordingly variant_type_map = {"original": "original", "A": "variant_a", "B": "variant_b"} target_variant_type = variant_type_map[variant] target_img = ( self._db.query(ImageAsset) .filter( ImageAsset.project_id == self._project_id, ImageAsset.still_id == still_id, ImageAsset.variant_type == target_variant_type, ) .order_by(ImageAsset.created_at.desc()) .first() ) if target_img: self._set_variant_primary(target_img.id, still_id) self._db.commit() self._logger.log( actor_id=self._actor_id, action="still.select_variant", resource_type="still", resource_id=still_id, project_id=self._project_id, detail={"variant": variant}, ip_address=ip, ) return { "still_id": still_id, "selected_variant": variant, "primary_image_id": target_img.id if target_img else None, } def select_original( self, still_id: str, image_id: str, ip: Optional[str] = None, ) -> Dict[str, Any]: """원본 이미지 선택 (여러 장 중).""" still = ( self._db.query(SceneStill) .filter(SceneStill.id == still_id, SceneStill.project_id == self._project_id) .first() ) if not still: raise AppError( code="still.not_found", message=t("still.not_found"), status_code=404, ) img = ( self._db.query(ImageAsset) .filter( ImageAsset.id == image_id, ImageAsset.project_id == self._project_id, ImageAsset.still_id == still_id, ) .first() ) if not img: raise AppError( code="image.not_found", message=t("image.not_found"), status_code=404, ) self._set_variant_primary(image_id, still_id) self._db.commit() self._logger.log( actor_id=self._actor_id, action="still.select_original", resource_type="still", resource_id=still_id, project_id=self._project_id, detail={"image_id": image_id}, ip_address=ip, ) return self._image_to_dict(img) def get_still_images_grouped( self, still_id: str, ) -> Dict[str, Any]: """씬의 모든 이미지를 original/A/B 그룹으로 반환.""" still = ( self._db.query(SceneStill) .filter(SceneStill.id == still_id, SceneStill.project_id == self._project_id) .first() ) if not still: raise AppError( code="still.not_found", message=t("still.not_found"), status_code=404, ) images = ( self._db.query(ImageAsset) .filter( ImageAsset.project_id == self._project_id, ImageAsset.still_id == still_id, ) .order_by(ImageAsset.created_at.desc()) .all() ) groups: Dict[str, list] = {"original": [], "variant_a": [], "variant_b": [], "other": []} for img in images: vt = img.variant_type or "other" d = self._image_to_dict(img) d = self._enrich_image_dict(d, img, still) if vt in groups: groups[vt].append(d) else: groups["other"].append(d) return { "still_id": still_id, "recommended_variant": still.recommended_variant, "selected_variant": still.selected_variant, "groups": groups, } # ------------------------------------------------------------------ # Helper methods # ------------------------------------------------------------------ def _get_latest_world_guide(self, episode_id: str) -> Dict[str, Any]: """Get the latest world guide for an episode.""" wg = ( self._db.query(WorldGuide) .filter(WorldGuide.project_id == self._project_id, WorldGuide.episode_id == episode_id) .order_by(WorldGuide.created_at.desc()) .first() ) if wg: try: return json.loads(wg.guide_json) except json.JSONDecodeError: pass return {} def _get_visible_entities(self, visible_entities_json: Optional[str]) -> List[Dict[str, Any]]: """Parse visible entities JSON and look up entity data.""" if not visible_entities_json: return [] try: visible_ids = json.loads(visible_entities_json) except json.JSONDecodeError: return [] visible_entities = [] for v in visible_ids: if isinstance(v, dict): eid = v.get("entity_id", "") elif isinstance(v, str): eid = v else: continue entity = ( self._db.query(EntityCanon) .filter(EntityCanon.id == eid) .first() ) if entity: visible_entities.append({ "id": entity.id, "name": entity.name, "entity_type": entity.entity_type, "description": entity.description or "", "stable_traits": entity.stable_traits or "{}", }) return visible_entities def _get_reference_image_map(self, visible_entities: List[Dict[str, Any]]) -> Dict[str, bytes]: """Build a map of entity_id -> reference image bytes for visible entities.""" ref_image_map: Dict[str, bytes] = {} for entity_data in visible_entities: eid = entity_data["id"] # Get primary reference image, or latest if none is primary ref_img = ( self._db.query(ImageAsset) .filter( ImageAsset.project_id == self._project_id, ImageAsset.entity_id == eid, ImageAsset.asset_type == "reference", ImageAsset.is_primary == 1, ) .first() ) if not ref_img: ref_img = ( self._db.query(ImageAsset) .filter( ImageAsset.project_id == self._project_id, ImageAsset.entity_id == eid, ImageAsset.asset_type == "reference", ) .order_by(ImageAsset.created_at.desc()) .first() ) if ref_img: fp = Path(ref_img.file_path) if fp.exists(): ref_image_map[eid] = fp.read_bytes() return ref_image_map # ------------------------------------------------------------------ # Generation trace queries # ------------------------------------------------------------------ def get_generation_traces( self, offset: int = 0, limit: int = 50, ) -> List[Dict[str, Any]]: """Get all generation traces (paginated).""" traces = ( self._db.query(GenerationTrace) .filter(GenerationTrace.project_id == self._project_id) .order_by(GenerationTrace.created_at.desc()) .offset(offset) .limit(limit) .all() ) return [GenerationTracker._trace_to_dict(tr) for tr in traces] def get_rejection_stats(self) -> Dict[str, Any]: """Get aggregated rejection statistics.""" tracker = GenerationTracker(self._db, self._project_id) return tracker.get_rejection_stats() def get_image_traces(self, image_id: str) -> List[Dict[str, Any]]: """Get traces for a specific image asset.""" tracker = GenerationTracker(self._db, self._project_id) return tracker.get_traces(image_asset_id=image_id) # ------------------------------------------------------------------ # Validation helpers # ------------------------------------------------------------------ def _create_validator(self) -> Optional[ImageValidator]: if not settings.openai_api_key: logger.info("Skipping image validation -- OpenAI API key not configured.") return None return ImageValidator() def _validate_reference( self, validator: ImageValidator, file_path: str, entity_info: Dict[str, Any], status: str, review_notes: str, ) -> tuple: try: fp = Path(file_path) if not fp.exists(): return None, None, status, review_notes image_bytes = fp.read_bytes() result = validator.validate_reference_image(image_bytes, entity_info) score = result.get("score", 0) result_json = json.dumps(result, ensure_ascii=False) if score < 60: issues = result.get("issues", []) issue_text = "; ".join(issues) if issues else "Low validation score" status = "needs_fix" review_notes = f"[auto-validation] score={score}: {issue_text}" logger.info("Reference image validated: score=%d, passed=%s", score, result.get("passed")) return score, result_json, status, review_notes except Exception as exc: logger.warning("Reference image validation failed: %s", exc) return None, None, status, review_notes def _validate_scene( self, validator: ImageValidator, file_path: str, scene_info: Dict[str, Any], entity_names: List[str], status: str, review_notes: str, ) -> tuple: try: fp = Path(file_path) if not fp.exists(): return None, None, status, review_notes image_bytes = fp.read_bytes() result = validator.validate_scene_image(image_bytes, scene_info, entity_names) score = result.get("score", 0) result_json = json.dumps(result, ensure_ascii=False) if score < 60: issues = result.get("issues", []) issue_text = "; ".join(issues) if issues else "Low validation score" status = "needs_fix" review_notes = f"[auto-validation] score={score}: {issue_text}" logger.info("Scene image validated: score=%d, passed=%s", score, result.get("passed")) return score, result_json, status, review_notes except Exception as exc: logger.warning("Scene image validation failed: %s", exc) return None, None, status, review_notes # ------------------------------------------------------------------ # Public validation endpoints # ------------------------------------------------------------------ def validate_image( self, image_id: str, ip: Optional[str] = None, ) -> Dict[str, Any]: if not settings.openai_api_key: raise AppError( code="image.openai_key_missing", message=t("image.openai_key_missing"), status_code=400, ) img = self.get_image(image_id) fp = Path(img.file_path) if not fp.exists(): raise AppError( code="image.not_found", message=t("image.not_found"), status_code=404, ) image_bytes = fp.read_bytes() validator = ImageValidator() if img.asset_type == "reference": entity_info: Dict[str, Any] = {} if img.entity_id: entity = ( self._db.query(EntityCanon) .filter(EntityCanon.id == img.entity_id) .first() ) if entity: entity_info = { "name": entity.name, "entity_type": entity.entity_type, "description": entity.description or "", "stable_traits": entity.stable_traits or "{}", } result = validator.validate_reference_image(image_bytes, entity_info) else: scene_info: Dict[str, Any] = {} entity_names: List[str] = [] if img.still_id: still = ( self._db.query(SceneStill) .filter(SceneStill.id == img.still_id) .first() ) if still: scene_info = { "scene_heading": still.screenplay_scene_heading or "", "beat_title": still.beat_title or "", "still_frame_prompt": still.still_frame_prompt or "", } try: visible_ids = json.loads(still.visible_entities_json or "[]") for v in visible_ids: eid = v.get("entity_id", "") if isinstance(v, dict) else v entity = ( self._db.query(EntityCanon) .filter(EntityCanon.id == eid) .first() ) if entity: entity_names.append(entity.name) except (json.JSONDecodeError, TypeError): pass result = validator.validate_scene_image(image_bytes, scene_info, entity_names) score = result.get("score", 0) img.validation_score = score img.validation_result = json.dumps(result, ensure_ascii=False) if score < 60: issues = result.get("issues", []) issue_text = "; ".join(issues) if issues else "Low validation score" img.status = "needs_fix" img.review_notes = f"[auto-validation] score={score}: {issue_text}" self._db.commit() self._logger.log( actor_id=self._actor_id, action="image.validate", resource_type="image", resource_id=image_id, project_id=self._project_id, detail={"score": score, "passed": result.get("passed", False)}, ip_address=ip, ) return result def get_validation(self, image_id: str) -> Dict[str, Any]: img = self.get_image(image_id) result: Dict[str, Any] = { "image_id": image_id, "score": img.validation_score, "passed": None, "issues": [], "description": "", } if img.validation_result: try: parsed = json.loads(img.validation_result) result["passed"] = parsed.get("passed") result["issues"] = parsed.get("issues", []) result["description"] = parsed.get("description", "") except json.JSONDecodeError: pass return result # ------------------------------------------------------------------ # T2I Prompt Composer # ------------------------------------------------------------------ def compose_prompts( self, still_id: str, ip: Optional[str] = None, ) -> Dict[str, Any]: """Run T2I composer on a still — save cinematic+closeup prompts.""" if not settings.openai_api_key: raise AppError( code="image.openai_key_missing", message=t("image.openai_key_missing"), status_code=400, ) still = ( self._db.query(SceneStill) .filter(SceneStill.id == still_id, SceneStill.project_id == self._project_id) .first() ) if not still: raise AppError( code="still.not_found", message=t("still.not_found"), status_code=404, ) # Load project-level prompt overrides if any proj_settings = ( self._db.query(ProjectSettings) .filter(ProjectSettings.project_id == self._project_id) .first() ) system_override = proj_settings.composer_system_prompt if proj_settings else None user_override = proj_settings.composer_user_prompt if proj_settings else None openai_client = OpenAIClient() composer = T2IPromptComposer( llm_client=openai_client, prompt_version="v1", system_prompt_override=system_override, user_prompt_override=user_override, ) # Get visible entities and their traits visible_entities = self._get_visible_entities(still.visible_entities_json) entity_visual_traits: Dict[str, Any] = {} for entity in visible_entities: eid = entity["id"] traits = entity.get("stable_traits", "{}") if isinstance(traits, str): try: entity_visual_traits[eid] = json.loads(traits) except json.JSONDecodeError: entity_visual_traits[eid] = {} else: entity_visual_traits[eid] = traits # Get world guide world_guide = self._get_latest_world_guide(still.episode_id) # Parse camera and lighting try: camera = json.loads(still.camera_json or "{}") except json.JSONDecodeError: camera = {} try: lighting = json.loads(still.lighting_json or "{}") except json.JSONDecodeError: lighting = {} result = composer.compose( scene_description=still.still_frame_prompt or "", camera_json=camera, lighting_json=lighting, visible_entities=visible_entities, entity_visual_traits=entity_visual_traits, world_guide=world_guide, ) still.t2i_prompt_cinematic = result["cinematic"] still.t2i_prompt_closeup = result["closeup"] still.t2i_composer_version = result["composer_version"] self._db.commit() self._logger.log( actor_id=self._actor_id, action="still.compose_prompts", resource_type="still", resource_id=still_id, project_id=self._project_id, detail={ "composer_version": result["composer_version"], "key_visual_elements": result.get("key_visual_elements", []), }, ip_address=ip, ) return { "still_id": still_id, "t2i_prompt_cinematic": result["cinematic"], "t2i_prompt_closeup": result["closeup"], "composer_version": result["composer_version"], "key_visual_elements": result.get("key_visual_elements", []), "removed_narrative_elements": result.get("removed_narrative_elements", []), } # ------------------------------------------------------------------ # Project-level composer prompt settings # ------------------------------------------------------------------ def get_composer_prompt(self) -> Dict[str, Any]: """Get current composer prompt settings for this project.""" proj_settings = ( self._db.query(ProjectSettings) .filter(ProjectSettings.project_id == self._project_id) .first() ) return { "composer_system_prompt": proj_settings.composer_system_prompt if proj_settings else None, "composer_user_prompt": proj_settings.composer_user_prompt if proj_settings else None, } def update_composer_prompt( self, system_prompt: Optional[str] = None, user_prompt: Optional[str] = None, ip: Optional[str] = None, ) -> Dict[str, Any]: """Update project-level composer prompt override.""" proj_settings = ( self._db.query(ProjectSettings) .filter(ProjectSettings.project_id == self._project_id) .first() ) if not proj_settings: proj_settings = ProjectSettings( id=_new_id(), project_id=self._project_id, updated_at=_now(), ) self._db.add(proj_settings) if system_prompt is not None: proj_settings.composer_system_prompt = system_prompt if user_prompt is not None: proj_settings.composer_user_prompt = user_prompt proj_settings.updated_at = _now() self._db.commit() self._logger.log( actor_id=self._actor_id, action="project.update_composer_prompt", resource_type="project_settings", resource_id=proj_settings.id, project_id=self._project_id, ip_address=ip, ) return { "composer_system_prompt": proj_settings.composer_system_prompt, "composer_user_prompt": proj_settings.composer_user_prompt, } # ------------------------------------------------------------------ # v5: GPT LVM — N개 이미지에서 최적 1개 선택 # ------------------------------------------------------------------ def _select_best_image_gpt( self, var_results: List[Dict[str, Any]], beat_title: str = "", representative_moment: str = "", ) -> int: """Use GPT 5.4 to select the best image from N generated variations. Args: var_results: List of result dicts, each with 'file_path'. beat_title: Scene beat title for context. representative_moment: Scene representative moment for context. Returns: 0-based index of the selected best image. """ import base64 import urllib.error import urllib.request if not settings.openai_api_key: return 0 # Load prompt template prompt_dir = ( Path(__file__).resolve().parent.parent.parent.parent / "prompts" / "_base" / "lvm_prompts" ) versions = sorted([d.name for d in prompt_dir.iterdir() if d.is_dir()], reverse=True) template_path = prompt_dir / versions[0] / "select_best_from_n.md" prompt_template = template_path.read_text(encoding="utf-8").strip() image_count = len(var_results) text_prompt = prompt_template.format( image_count=image_count, beat_title=beat_title or "(no title)", representative_moment=representative_moment or "(no description)", ) # Build multi-image content for OpenAI vision content_parts = [{"type": "input_text", "text": text_prompt}] for i, vr in enumerate(var_results): fp = Path(vr["file_path"]) if not fp.exists(): continue b64_data = base64.b64encode(fp.read_bytes()).decode("ascii") content_parts.append({ "type": "input_image", "image_url": f"data:image/png;base64,{b64_data}", }) selection_schema = { "type": "object", "additionalProperties": False, "properties": { "selected_index": { "type": "integer", "description": "1-based index of the best image", }, "reason": { "type": "string", "description": "Brief reason for the selection", }, }, "required": ["selected_index", "reason"], } body = { "model": settings.openai_model, "input": [ { "type": "message", "role": "user", "content": content_parts, }, ], "text": { "format": { "type": "json_schema", "name": "image_selection", "strict": True, "schema": selection_schema, } }, "temperature": 0.1, "store": False, } req = urllib.request.Request( "https://api.openai.com/v1/responses", data=json.dumps(body).encode("utf-8"), headers={ "Authorization": f"Bearer {settings.openai_api_key}", "Content-Type": "application/json", }, method="POST", ) import socket import time as _time timeout = settings.llm_timeout_validation max_retries = settings.llm_max_retries last_error = 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"OpenAI Vision API error {exc.code}: {error_text}") if exc.code in {429, 500, 502, 503, 504} and attempt <= max_retries: _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 selection failed after retries: {exc}") from exc else: raise RuntimeError(f"GPT selection failed: {last_error}") # Parse response output_text = payload.get("output_text") if not output_text: output = payload.get("output", []) for item in output: if isinstance(item, dict): for part in (item.get("content") or []): if isinstance(part, dict) and part.get("type") == "output_text": output_text = part.get("text") break if output_text: break if not output_text: logger.warning("GPT selection: no output_text, defaulting to first image") return 0 result = json.loads(output_text) selected = result.get("selected_index", 1) reason = result.get("reason", "") logger.info("GPT selected image %d/%d: %s", selected, image_count, reason) # Convert 1-based to 0-based, clamp to valid range idx = max(0, min(selected - 1, image_count - 1)) return idx # ------------------------------------------------------------------ # Lineage-aware image generation # ------------------------------------------------------------------ def _build_lineage_fields( self, module_name: str, ref_entity_ids: Optional[List[str]] = None, prompt_type: Optional[str] = None, ) -> Dict[str, Any]: """Build lineage fields for an image asset.""" info = get_module_info(module_name) ref_image_ids: List[str] = [] if ref_entity_ids: for eid in ref_entity_ids: ref_img = ( self._db.query(ImageAsset) .filter( ImageAsset.project_id == self._project_id, ImageAsset.entity_id == eid, ImageAsset.asset_type == "reference", ImageAsset.is_primary == 1, ) .first() ) if not ref_img: ref_img = ( self._db.query(ImageAsset) .filter( ImageAsset.project_id == self._project_id, ImageAsset.entity_id == eid, ImageAsset.asset_type == "reference", ) .order_by(ImageAsset.created_at.desc()) .first() ) if ref_img: ref_image_ids.append(ref_img.id) return { "prompt_type": prompt_type, "code_version": info["version"], "prompt_file_version": info.get("prompt_dependency"), "reference_image_ids": json.dumps(ref_image_ids), }