"""ImageComposerService — T2I Composer + composer prompt 설정 서비스 (W5 F23 Phase 3). ImageService facade(image_service.py)에서 T2I composer 관련 3개 공개 메서드를 literal lift. facade는 얇은 delegate만 유지. 이관된 메서드: - compose_prompts — 씬 스틸에 T2I cinematic/closeup 프롬프트 생성 - get_composer_prompt / update_composer_prompt — 프로젝트 단위 composer 오버라이드 """ import json import logging import uuid from datetime import datetime, timezone 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 EntityCanon, ProjectSettings, SceneStill from app.modules.t2i_prompt_composer import T2IPromptComposer from app.services.image_service_helpers import get_latest_world_guide, 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()) class ImageComposerService: """T2I Composer + composer prompt 설정 서비스.""" def __init__( self, db: OrmSession, project_id: str, actor_id: str, activity_logger: ActivityLogger, ) -> None: self._db = db self._project_id = project_id self._actor_id = actor_id self._logger = activity_logger 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("id") or 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, "short_id": entity.short_id or "", "description": entity.description or "", "stable_traits": entity.stable_traits or "{}", }) return visible_entities def compose_prompts( self, still_id: str, ip: Optional[str] = None, ) -> Dict[str, Any]: """Run T2I composer on a still — save cinematic+closeup prompts.""" 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, ) # 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 composer = T2IPromptComposer( project_llm_config=load_project_llm_config(self._db, self._project_id), 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 = get_latest_world_guide(self._db, self._project_id, 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, }