"""이미지 생성 서비스 — 공용 유틸리티 + list/get/review/primary/upload/compose_prompts/validate. 레퍼런스/씬 이미지 생성 파이프라인은 Phase 3b.2 / 3b.3 / 3b.5 / 3b.5b 완료 후 ReferenceImageService / SceneImageService로 완전히 이관되었다. 본 모듈은 공용 helper shim과 엔드포인트용 보조 메서드만 유지한다. """ 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.i18n.loader import t from app.logging.activity_logger import ActivityLogger from app.models.project import ( Episode, EntityCanon, GenerationTrace, ImageAsset, ProjectSettings, SceneStill, ) from app.modules.generation_tracker import GenerationTracker from app.services.image_composer_service import ImageComposerService from app.services.image_review_service import ImageReviewService from app.services.image_upload_service import ImageUploadService logger = logging.getLogger(__name__) def _now() -> str: return datetime.now(timezone.utc).isoformat() def _new_id() -> str: return str(uuid.uuid4()) 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 self._review_svc = ImageReviewService( db=db, project_id=project_id, actor_id=actor_id, activity_logger=self._logger, ) self._composer_svc = ImageComposerService( db=db, project_id=project_id, actor_id=actor_id, activity_logger=self._logger, ) self._upload_svc = ImageUploadService( db=db, project_id=project_id, actor_id=actor_id, activity_logger=self._logger, ) 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_style_context(self, episode_id: str = None) -> str: """프로젝트 세계관을 T2I 프롬프트 앞에 붙일 컨텍스트로 변환.""" # 1. ProjectSettings에서 로드 ps = self._db.query(ProjectSettings).filter( ProjectSettings.project_id == self._project_id).first() if ps and ps.style_rules_json: 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','')}." ) # 2. Fallback: visual_world_rules 체크포인트에서 직접 로드 if episode_id: from pathlib import Path cp = ( Path(settings.projects_dir) / self._project_id / "checkpoints" / "episodes" / episode_id / "visual_world_rules" / "manifest.json" ) if cp.exists(): vwr = json.loads(cp.read_text(encoding="utf-8")).get("data", {}) era = vwr.get("era", "") region = vwr.get("region", "") if era or region: return f"Photorealistic cinematic still. Setting: {era}, {region}." return "Photorealistic cinematic still." 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]: """Phase 3b.2 shim → image_service_helpers.image_to_dict.""" from app.services.image_service_helpers import image_to_dict return image_to_dict(img) @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): # 의도적: angle_applied 파싱 실패 시 파생 필드(angle_*)만 생략. 원본 응답은 정상 유지. 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: return self._review_svc.get_image(image_id) def update_review( self, image_id: str, status: str, notes: str, ip: Optional[str] = None, ) -> ImageAsset: return self._review_svc.update_review(image_id, status, notes, ip=ip) def regenerate_image( self, image_id: str, ip: Optional[str] = None, ) -> None: return self._review_svc.regenerate_image(image_id, ip=ip) def regenerate_needs_fix( self, episode_id: str, ip: Optional[str] = None, ) -> int: return self._review_svc.regenerate_needs_fix(episode_id, ip=ip) # ------------------------------------------------------------------ # Primary image management # ------------------------------------------------------------------ def set_primary_image( self, image_id: str, ip: Optional[str] = None, ) -> Dict[str, Any]: return self._review_svc.set_primary_image(image_id, ip=ip) # ------------------------------------------------------------------ # Single image generation (delegates to ImageUploadService) # ------------------------------------------------------------------ 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]: return self._upload_svc.upload_custom_image( file_bytes=file_bytes, filename=filename, entity_id=entity_id, still_id=still_id, episode_id=episode_id, ip=ip, ) 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, } # ------------------------------------------------------------------ # 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 (delegates to ImageReviewService) # ------------------------------------------------------------------ def validate_image( self, image_id: str, ip: Optional[str] = None, ) -> Dict[str, Any]: return self._review_svc.validate_image(image_id, ip=ip) def get_validation(self, image_id: str) -> Dict[str, Any]: return self._review_svc.get_validation(image_id) # ------------------------------------------------------------------ # T2I Prompt Composer (delegates to ImageComposerService) # ------------------------------------------------------------------ def compose_prompts( self, still_id: str, ip: Optional[str] = None, ) -> Dict[str, Any]: return self._composer_svc.compose_prompts(still_id, ip=ip) def get_composer_prompt(self) -> Dict[str, Any]: return self._composer_svc.get_composer_prompt() def update_composer_prompt( self, system_prompt: Optional[str] = None, user_prompt: Optional[str] = None, ip: Optional[str] = None, ) -> Dict[str, Any]: return self._composer_svc.update_composer_prompt( system_prompt=system_prompt, user_prompt=user_prompt, ip=ip, )