"""ReferenceCompositeService — 합성(face+outlook) 이미지 생성 (W5 F24 Phase 3). ReferenceImageService facade에서 generate_composite_image + regenerate_composites_for_entity를 literal lift. facade는 얇은 delegate만 유지. """ 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.i18n.loader import t from app.logging.activity_logger import ActivityLogger from app.models.project import CharacterOutlook, EntityCanon, EntityEpisodeLink, ImageAsset from app.modules.llm.gemini_image_client import GeminiImageClient, ModerationError from app.modules.llm.gemini_key_pool import key_count as gemini_key_count from app.services.image_capture.annotate import annotate_generated_asset from app.services.image_service_helpers import image_to_dict logger = logging.getLogger(__name__) class ReferenceCompositeService: """합성 이미지 생성 + 엔티티 기준 재생성 서비스.""" 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 generate_composite_image( self, character_id: str, outlook_id: str, ip: Optional[str] = None, ) -> Dict[str, Any]: """합성 이미지 1장 생성 (얼굴 + 아웃룩 → 전신).""" 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) char_ent = self._db.query(EntityCanon).filter(EntityCanon.id == character_id, EntityCanon.project_id == self._project_id).first() outlook_ent = self._db.query(EntityCanon).filter(EntityCanon.id == outlook_id, EntityCanon.project_id == self._project_id).first() if not char_ent or not outlook_ent: raise AppError(code="entity.not_found", message=t("entity.not_found"), status_code=404) if outlook_ent.short_id == "O00": raise AppError( code="image.null_outlook", message=f"{char_ent.name}은 Null Outlook(O00) — 합성 불필요 (캐릭터 참조이미지 직접 사용)", status_code=400, ) # HEAD 동작 보존: 단일 경로는 `[outlook_id:` 레거시 prompt도 face로 허용 # (pipeline_gate.py 는 `outlook_id:{outlook_id}` 키를 character composite로 해석). # 배치 경로에서만 엄격 필터 유지 — 단일 경로는 legacy migration 호환. face_asset = ( self._db.query(ImageAsset) .filter( ImageAsset.entity_id == character_id, ImageAsset.asset_type == "reference", ImageAsset.is_primary == 1, ~ImageAsset.prompt_used.like("[composite:%"), ) .first() ) if not face_asset or not Path(face_asset.file_path).exists(): raise AppError(code="image.face_missing", message="얼굴 참조 이미지가 없습니다. 인물 이미지를 먼저 생성하세요.", status_code=400) outfit_asset = self._db.query(ImageAsset).filter( ImageAsset.entity_id == outlook_id, ImageAsset.asset_type == "reference", ImageAsset.is_primary == 1, ).first() if not outfit_asset or not Path(outfit_asset.file_path).exists(): raise AppError(code="image.outfit_missing", message="아웃룩 참조 이미지가 없습니다. 아웃룩 이미지를 먼저 생성하세요.", status_code=400) face_bytes = Path(face_asset.file_path).read_bytes() outfit_bytes = Path(outfit_asset.file_path).read_bytes() link = self._db.query(EntityEpisodeLink).filter(EntityEpisodeLink.canon_id == character_id).first() episode_id = link.episode_id if link else "shared" reference_dir = Path(settings.projects_dir) / self._project_id / "images" / episode_id / "reference" from app.modules.pipeline.ref_image_pipeline import generate_and_validate_reference, _load_ref_image_prompt gemini_client = GeminiImageClient(model=settings.gemini_image_model) gemini_client.set_context(project_id=self._project_id, episode_id=episode_id, operation_type="composite_single") composite_prompt = _load_ref_image_prompt("composite") or \ "Full body shot, standing pose, plain neutral background. Dress the character in the outfit shown in the reference images." labeled_refs = [ ("Reference face image", face_bytes), (f"Outfit piece: {outlook_ent.name}", outfit_bytes), ] try: pipe_result = generate_and_validate_reference( gemini_client=gemini_client, entity_name=f"{char_ent.name} ({outlook_ent.name})", entity_description=f"{char_ent.name} wearing {outlook_ent.name}", entity_type="composite", t2i_prompt=composite_prompt, output_dir=reference_dir, extra_references=labeled_refs, # Phase 4 iter 7 review I3 — composite manual path 도 trace_meta. trace_meta={ "project_id": self._project_id, "episode_id": episode_id, "operation_type": "composite_single", "entity_id": character_id, }, ) except ModerationError as exc: raise AppError(code="image.generation_blocked", message=f"Image generation blocked: {exc.block_reason}", status_code=400) composite_key = f"composite:{character_id}:{outlook_id}" from sqlalchemy import text as sql_text self._db.execute(sql_text( "UPDATE image_asset SET is_primary = 0 " "WHERE project_id = :pid AND prompt_used LIKE :key AND is_primary = 1" ), {"pid": self._project_id, "key": f"[{composite_key}]%"}) asset = ImageAsset( id=str(uuid.uuid4()), project_id=self._project_id, asset_type="reference", entity_id=character_id, episode_id=episode_id, file_path=to_relative_image_path(pipe_result["file_path"]), prompt_used=f"[{composite_key}] {char_ent.name}+{outlook_ent.name}", generation_model=pipe_result.get("generation_model", settings.gemini_image_model), status="generated", review_notes=json.dumps(pipe_result.get("validation", {}), ensure_ascii=False) if pipe_result.get("validation") else None, validation_score=pipe_result.get("validation", {}).get("score") if pipe_result.get("validation") else None, is_primary=1, created_at=datetime.now(timezone.utc).isoformat(), ) self._db.add(asset) # persist-all Wave 1 — 합성 lineage: 얼굴(face) + 아웃룩(outfit) → 합성. annotate_generated_asset( asset, pipeline_role="reference_composite", stage="reference_composite", input_image_ids=[a.id for a in (face_asset, outfit_asset) if a is not None], ) self._db.commit() return image_to_dict(asset) def regenerate_composites_for_entity( self, entity_id: str, ip: Optional[str] = None, ) -> List[Dict[str, Any]]: """엔티티(인물 또는 아웃룩)의 이미지가 변경될 때, 관련 합성 이미지를 모두 재생성.""" entity = self._db.query(EntityCanon).filter(EntityCanon.id == entity_id, EntityCanon.project_id == self._project_id).first() if not entity: return [] results: List[Dict[str, Any]] = [] if entity.entity_type == "character": combos = self._db.query(CharacterOutlook).filter( CharacterOutlook.character_id == entity_id, CharacterOutlook.project_id == self._project_id, ).all() for co in combos: o_ent = self._db.query(EntityCanon).filter(EntityCanon.id == co.outlook_id).first() if o_ent and o_ent.short_id == "O00": continue try: result = self.generate_composite_image(entity_id, co.outlook_id, ip=ip) results.append(result) except Exception as exc: logger.warning("Composite regen failed for %s+%s: %s", entity_id, co.outlook_id, exc) elif entity.entity_type == "outlook": if entity.short_id == "O00": return results combos = self._db.query(CharacterOutlook).filter( CharacterOutlook.outlook_id == entity_id, CharacterOutlook.project_id == self._project_id, ).all() for co in combos: try: result = self.generate_composite_image(co.character_id, entity_id, ip=ip) results.append(result) except Exception as exc: logger.warning("Composite regen failed for %s+%s: %s", co.character_id, entity_id, exc) return results