"""씬 이미지 변형(variant) 선택 및 편집 서비스 — W5 F22 Phase B. SceneImageService에서 variation 관련 로직을 이관. 공개 API는 facade가 유지하고 본 서비스는 구현을 담당. 공개 API (SceneImageService facade가 delegation): - Phase B.1: `select_variant` / `select_original` - Phase B.2: `regenerate_variation` (fal.ai 앵글 i2i) Module-level helper (facade + variation_svc 양쪽에서 공유): - `set_variant_primary(db, project_id, image_id, still_id)` """ from __future__ import annotations import json import logging import uuid 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.core.file_paths import to_relative_image_path from app.services.image_capture.annotate import annotate_generated_asset from app.i18n.loader import t from app.logging.activity_logger import ActivityLogger from app.models.project import ImageAsset, SceneStill from app.modules.llm.gemini_key_pool import key_count as gemini_key_count, get_next_key from app.services.fal_angle_helpers import ( apply_fal_angle as _apply_fal_angle, select_final_best as _select_final_best, ) from app.services.image_service_helpers import ( build_lineage_fields, image_to_dict, load_project_llm_config, ) logger = logging.getLogger(__name__) def _now() -> str: return datetime.now(timezone.utc).isoformat() def _new_id() -> str: return str(uuid.uuid4()) def set_variant_primary( db: OrmSession, project_id: str, image_id: str, still_id: str, ) -> None: """해당 image를 still의 primary로 설정 (siblings는 is_primary=0). scene_image_service / scene_variation_service 양쪽에서 공유되는 helper. commit은 호출자 책임. """ siblings = ( db.query(ImageAsset) .filter( ImageAsset.project_id == project_id, ImageAsset.still_id == still_id, ImageAsset.id != image_id, ) .all() ) for sib in siblings: sib.is_primary = 0 target = db.query(ImageAsset).filter(ImageAsset.id == image_id).first() if target: target.is_primary = 1 def apply_validation_override( var_results_list: List[Dict[str, Any]], best_idx: int, num_saved_assets: int, fal_generated: bool, ) -> "tuple[int, Optional[tuple[int, int]]]": """Validation score 기반 best_idx override. W5 F22 Phase B.22.7 (2026-04-23): scene_image_service.generate_images의 validation_score override 블록(~16 LOC)을 이관. 처리: 1. 현재 선택된 variation의 validation_score (None → 50) 조회 2. score < 40이고 variation 2+개면 override 후보 탐색 3. fal.ai 결과 제외한 original N개 중 가장 높은 score 후보 선택 4. 후보 score가 현재보다 20점 초과 높으면 override (strict >) Args: var_results_list: 전체 variation 결과 (fal.ai 결과 포함 가능) best_idx: Gemini Vision이 선택한 (또는 fallback된) 현재 best 인덱스 num_saved_assets: DB에 저장된 asset 개수 (var_results_list 총 길이와 동일) fal_generated: fal.ai N+1 이미지가 추가되어 있는지 Returns: (new_best_idx, override_scores): - override_scores=None: override 발생 안 함, best_idx 그대로 - override_scores=(old, new): override 발생, caller 로그용 """ 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: orig_results = var_results_list[: num_saved_assets - (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]) return override_idx, ( _vscore(selected_result), _vscore(alternatives[0]), ) return best_idx, None def select_best_variation_idx( var_results_list: List[Dict[str, Any]], beat_title: str, ) -> "tuple[int, bool]": """Gemini Vision으로 variations 중 최종 대표 이미지 인덱스 선택. W5 F22 Phase B.22.5 (2026-04-22): scene_image_service.generate_images의 Gemini Vision final best 선택 블록(~22 LOC)을 이관. 처리: 1. var_results_list 순회, 존재하는 file_path의 bytes만 valid_pairs로 수집 2. len(valid_pairs) > 1이면 _select_final_best 호출 → raw_best 인덱스 3. valid_indices[min(raw, len-1)]로 원래 인덱스 복원 4. 예외 또는 valid_pairs <=1이면 (0, False) 반환 Returns: (best_idx, was_selected): - was_selected=True: Gemini Vision이 실제로 선택을 수행함 (caller가 로그 출력) - was_selected=False: valid_pairs<=1 또는 예외로 선택 스킵 (best_idx=0 fallback) beat_title은 Gemini Vision prompt에 사용. """ best_idx = 0 was_selected = False 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, beat_title) best_idx = valid_indices[min(raw_best, len(valid_indices) - 1)] was_selected = True except Exception as exc: logger.warning("Final best selection failed, using first: %s", exc) best_idx = 0 was_selected = False return best_idx, was_selected class SceneVariationService: """씬 이미지 variant 선택 전담 서비스 (W5 F22 Phase B.1).""" def __init__( self, db: OrmSession, project_id: str, actor_id: str, ) -> None: self._db = db self._project_id = project_id self._actor_id = actor_id self._logger = ActivityLogger(db) 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: set_variant_primary(self._db, self._project_id, 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 _get_project_dir(self) -> Path: """프로젝트 루트 디렉토리 (W5 F22 Phase B.4: scene_image_service에서 복제).""" return Path(settings.projects_dir) / self._project_id def _generate_variation_i2i( 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]]: """단일 variation을 i2i 편집으로 생성 (W5 F22 Phase B.5 이관). facade의 generate_scene_with_variations가 본 메서드를 호출한다. 이전에는 `_generate_variation` internal helper였으나 variation-side 로직으로 분리. """ try: angle_params = json.loads(angle_json) if angle_json else None except json.JSONDecodeError: angle_params = None # persist-all Wave2 (B4): 카메라 다이어그램은 i2i 입력용 diagnostic 중간물. # ★diagram-only narrow scope — edit_angle 내부가 동일 다이어그램을 재생성하지만 # 그 호출은 이 scope 밖이라 i2i_edit 최종 출력은 capture 되지 않는다(중복 0, # Codex 합의: i2i output 은 Wave1 annotate 대상). 결정론 PIL 재계산은 저렴. if var_type in ("angle", "angle+color") and angle_params: try: from app.modules.gemini_i2i_editor import _create_camera_diagram from app.services.image_capture.context import generation_context with generation_context( self._project_id, episode_id, stage="angle_camera_diagram", still_id=still.id, scene_index=getattr(still, "scene_index", None), shot_index=getattr(still, "shot_index", None), ): _create_camera_diagram( int(angle_params.get("horizontal", 0)), int(angle_params.get("vertical", 0)), float(angle_params.get("zoom", 1.0)), ) except Exception as _exc: # 비치명 — 다이어그램 capture 실패가 변형을 막지 않음 logger.warning("B4 camera_diagram capture failed (non-fatal): %s", _exc) 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=to_relative_image_path(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) # persist-all Wave 1 — i2i 변형 lineage: source_image_id 구조FK → input. annotate_generated_asset( asset, pipeline_role="scene_still", stage="scene_image", input_image_ids=[original_id] if original_id else [], ) self._db.flush() return image_to_dict(asset) def generate_variations_only(self, episode_id: str) -> None: """기존 원본 이미지 기반 A/B 변형 추천 + 생성 (W5 F22 Phase B.4 이관). for each still: 1) variation_a_type/b_type 없으면 V2 recommender로 추천 → DB 저장 2) 추천된 variation 중 아직 생성 안 된 것 GeminiI2IEditor로 생성 """ 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, SceneStill.is_selected == True, # noqa: E712 SceneStill.still_index >= 0, SceneStill.status != "stale", ) .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=to_relative_image_path(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) # persist-all Wave 1 — i2i 변형 lineage: source → 변형. annotate_generated_asset( var_asset, pipeline_role="scene_still", stage="scene_image", input_image_ids=[original_asset.id] if original_asset else [], ) 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 recommend_variations( self, still_id: str, ip: Optional[str] = None, ) -> Dict[str, Any]: """LLM에게 A/B 변형 추천 받기 (W5 F22 Phase B.3 이관).""" from app.core.openai_keys import has_openai_key if not has_openai_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 recommender = VariationRecommender( project_llm_config=load_project_llm_config(self._db, self._project_id) ) 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 regenerate_variation( self, image_id: str, angle_json: Optional[str] = None, color_prompt: Optional[str] = None, ip: Optional[str] = None, ) -> Dict[str, Any]: """원본 기반 fal.ai i2i로 변형 재생성 (W5 F22 Phase B.2 이관). 현재 구현: angle_json(horizontal/vertical/zoom) 기반 **앵글만** 편집. color_prompt 파라미터는 reserved — color-only 호출은 image.no_edit_params 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, ) # facade의 get_image와 동일 로직 — facade 참조를 피하기 위해 inline 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, ) # 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: # 의도적: 잘못된 angle_json은 아래 not angle_params 분기의 400(AppError)으로 흡수. 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_ai_enabled or 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) # PNG 메타데이터 삽입 from app.modules.png_metadata import embed_png_metadata embed_png_metadata(file_path, { "source_image": Path(source_img.file_path).name, "angle_horizontal": str(h), "angle_vertical": str(v), "angle_zoom": str(z), "model": "fal-ai/qwen-image-edit-2511-multiple-angles", "created_at": _now(), }) 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=to_relative_image_path(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) # persist-all Wave 1 — fal 앵글 lineage: source → 앵글. annotate_generated_asset( asset, pipeline_role="scene_angle_fal", stage="scene_image", input_image_ids=[source_id] if source_id else [], ) 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": "fal_angle", }, ip_address=ip, ) return image_to_dict(asset) 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, ) set_variant_primary(self._db, self._project_id, 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 image_to_dict(img)