"""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 _storage_episode(self, character_id: str, episode_id: Optional[str]) -> str: """이 합성을 **어느 화 아래에** 적을 것인가. ★★★임의로 고르지 않는다 (Codex BLOCK 2026-09-04). 앞 판은 `EntityEpisodeLink.canon_id == character_id` 의 **첫 행**을 집었다 — 프로젝트 필터도 정렬도 없었다. 공유 인물 `C01` 이 1·2화에 다 있으면 **2화 옷의 합성이 1화 폴더·DB·trace 에 기록된다.** 화 삭제·감사·자산 계보가 통째로 어긋난다. 규칙: 부른 쪽이 화를 주면 → 그 화 안 주고 링크가 **딱 하나**면 → 그 화 (종전 동작과 같다) 0개거나 여럿이면 → `"shared"` — **안 고른다** """ if episode_id: return episode_id from app.models.project import EntityEpisodeLink eps = { r[0] for r in self._db.query(EntityEpisodeLink.episode_id).filter( EntityEpisodeLink.project_id == self._project_id, EntityEpisodeLink.canon_id == character_id, ).all() } if len(eps) == 1: return eps.pop() if eps: logger.warning( "합성 저장 화를 못 정했다 — 인물 %s 가 %d개 화에 걸려 있는데 " "부른 쪽이 화를 안 줬다. `shared` 아래에 적는다 " "(임의로 고르면 계보가 어긋난다)", character_id, len(eps)) return "shared" def generate_composite_image( self, character_id: str, outlook_id: str, ip: Optional[str] = None, episode_id: Optional[str] = None, ) -> Dict[str, Any]: """합성 이미지 1장 생성 (얼굴 + 아웃룩 → 전신). ★``episode_id`` — **어느 화의 배정인가.** 안 주면 위 규칙으로 정하되 임의로 고르지 않는다. """ 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) # ★★판정을 **옷 사진을 찾기 전에** 한다 (2026-09-20 Codex BLOCK). # # 종전에는 derive 여부를 알기도 전에 옷 사진을 **필수로** 읽어, # 없으면 400 으로 막았다. 그런데 몸=신원 인물에게는 애초에 옷 # 사진을 **안 준다**(아래). 쓰지도 않을 것이 없다고 막은 셈이고, # 옷 참조가 아직 없는 로봇은 수동 합성 자체를 못 했다. from app.core.body_identity import body_identity_short_ids _derive = (char_ent.short_id or "") in body_identity_short_ids( self._project_id) outfit_asset = self._db.query(ImageAsset).filter( ImageAsset.entity_id == outlook_id, ImageAsset.asset_type == "reference", ImageAsset.is_primary == 1, ).first() _outfit_ok = bool( outfit_asset and Path(outfit_asset.file_path).exists()) if not _derive and not _outfit_ok: raise AppError(code="image.outfit_missing", message="아웃룩 참조 이미지가 없습니다. 아웃룩 이미지를 먼저 생성하세요.", status_code=400) face_bytes = Path(face_asset.file_path).read_bytes() episode_id = self._storage_episode(character_id, episode_id) 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") # ★★수동 합성도 **배치와 같은 판정**을 읽는다 (Codex BLOCK, 2026-09-19). # 종전에는 사람 경로로 고정돼 있어, 로봇 기본 참조를 전신으로 고쳐 놓아도 # 사용자가 합성 재생성을 누르면 **몸을 다시 새로 그렸다**. # 판정 자체는 위에서 이미 했다(옷 사진 필수 조건과 얽혀 있다). _ctype = "composite_derive" if _derive else "composite" composite_prompt = _load_ref_image_prompt(_ctype) or \ "Full body shot, standing pose, plain neutral background. Dress the character in the outfit shown in the reference images." labeled_refs = [ (("Reference image 1 — the character's COMPLETE BODY" if _derive else "Reference face image"), face_bytes), ] # ★★몸이 곧 신원인 인물에게는 **옷 사진을 안 준다** (2026-09-19 # 사용자 지적 · 배치는 3c01121f 에서 이미 고쳤다). # # 옷 참조는 **사람 모양 마네킹 전신 사진**이다. 참조 두 장이 둘 다 # 몸이면 모델이 둘을 섞는다 — 실측으로 고릴라 비율 로봇이 사람 # 다리에 카고바지를 입고 로봇 팔만 붙은 그림이 됐다. 글로 # 「마네킹은 진열대일 뿐」이라고 적어도 안 먹었고, **두 번째 몸을 # 아예 안 보여주는 것**이 답이었다. # # ★이 수리가 **배치에만** 들어가 있었다 — 수동·엔티티별 재생성은 # 같은 실패를 계속 살 수 있었다(2026-09-20 Codex 감사). if not _derive: labeled_refs.append( (f"Outfit piece: {outlook_ent.name}", Path(outfit_asset.file_path).read_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=_ctype, t2i_prompt=composite_prompt, output_dir=reference_dir, extra_references=labeled_refs, # ★파생 문안의 `{outlook_description}` 슬롯을 채운다 # (2026-09-20 Codex BLOCK). 안 주면 그릴 재료가 없어 # `<인물> wearing <옷 이름>` 한 줄로 떨어진다 — 그래서 # 「커다란 밀짚모자와 거대한 장화」 같은 상세가 최종 # 생성 문안에 **도달하지 못했다**. 배치는 이미 넘긴다 # (`reference_phase3_service.py:178,228`). outlook_description=(outlook_ent.description or ""), # 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) # ★키도 배치와 **같은 모양** — 기본 참조 id 를 붙여야 소비자가 # 「지금 몸으로 그린 합성」을 고를 수 있다. composite_key = f"composite:{character_id}:{outlook_id}:{face_asset.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, # ★기본 참조 id 를 뺀 **(인물, 아웃룩) 접두**로 내린다 — 키에 id 를 # 붙인 뒤로 전체 키로 찾으면 **옛 합성이 안 내려가** 둘이 공존한다. "key": f"[composite:{character_id}:{outlook_id}%"}) 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", # ★파생 갈래는 옷 사진을 **안 붙인다** — 안 쓴 입력을 계보에 # 남기면 「무엇을 보고 그렸나」가 거짓이 된다 (배치와 같은 # 계약: `reference_phase3_service.py:170-174`). input_image_ids=[ a.id for a in ( face_asset, None if _derive else (outfit_asset if _outfit_ok else None), ) 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 [] # ★여기는 **엔티티 단위 재생성**이라 화 문맥이 없다 (호출부가 # episode_id 를 안 준다). 「이 인물의 합성을 다 다시 만든다」가 # 뜻이므로 화마다 다른 옷도 **다 만드는 것이 맞다** — 프로젝트 범위가 # 의도다 (2026-09-04 확인, 화 범위로 좁히지 않는다). 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, episode_id=co.episode_id) 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, episode_id=co.episode_id) results.append(result) except Exception as exc: logger.warning("Composite regen failed for %s+%s: %s", co.character_id, entity_id, exc) return results