"""ReferenceEntityService — 단일 엔티티 참조 이미지 생성 (W5 F24 Phase 2). ReferenceImageService facade에서 generate_single_entity_image를 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, 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 ( Episode, 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_service_helpers import ( auto_set_primary, build_lineage_fields, get_latest_world_guide, image_to_dict, ) logger = logging.getLogger(__name__) class ReferenceEntityService: """단일 엔티티 참조 이미지 생성 서비스.""" 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_single_entity_image( self, entity_id: str, custom_prompt: Optional[str] = None, ip: Optional[str] = None, ) -> Dict[str, Any]: """Generate a single reference image for an entity.""" 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, ) entity = ( self._db.query(EntityCanon) .filter(EntityCanon.id == entity_id, EntityCanon.project_id == self._project_id) .first() ) if not entity: raise AppError( code="entity.not_found", message=t("entity.not_found"), status_code=404, ) link = ( self._db.query(EntityEpisodeLink) .filter(EntityEpisodeLink.canon_id == entity_id) .first() ) episode_id = link.episode_id if link else "shared" project_dir = Path(settings.projects_dir) / self._project_id reference_dir = project_dir / "images" / episode_id / "reference" entity_data = { "id": entity.id, "name": entity.name, "entity_type": entity.entity_type, "description": entity.description or "", "stable_traits": entity.stable_traits or "{}", } world_guide = get_latest_world_guide(self._db, self._project_id, episode_id) language = "ko" if episode_id != "shared": ep = self._db.query(Episode).filter(Episode.id == episode_id).first() if ep: language = ep.language or "ko" if custom_prompt: entity_data["description"] = custom_prompt from app.modules.pipeline.ref_image_pipeline import generate_and_validate_reference gemini_client = GeminiImageClient(model=settings.gemini_image_model) gemini_client.set_context( project_id=self._project_id, episode_id=episode_id, operation_type="single_ref", step_name="generate_single_entity_image", ) # W21B-W7 W-B (2026-06-12): printed_prop anchor overlay — phase1 배치 # 경로(reference_phase1_service._gen_ref)와 동일 계약. custom_prompt 는 # 운영자 명시 입력이라 anchor 보다 우선 (overlay 미적용). canon 비변경. _single_t2i = custom_prompt or entity.t2i_prompt or entity_data["description"] if not custom_prompt and entity.entity_type == "prop" and episode_id != "shared": from app.core.steps.visual_continuity_anchor_step import ( load_printed_prop_anchor_context, ) _vca_ctx = load_printed_prop_anchor_context(self._project_id, episode_id) _anchor = (_vca_ctx.get("by_prop") or {}).get(entity.short_id or "") if _vca_ctx else None if _anchor: from app.modules.pipeline.visual_continuity_anchor_plan import ( build_prop_ref_prompt_overlay, ) _single_t2i = build_prop_ref_prompt_overlay(_anchor, _single_t2i) # ★실측(2026-06-12): prop 은 ref_image_pipeline 외부 템플릿이 # entity_description 으로 조립 (t2i 는 fallback) — desc 도 overlay. entity_data["description"] = build_prop_ref_prompt_overlay( _anchor, entity_data["description"], include_framing=False) logger.info( "visual_continuity_anchor: prop %s 단건 ref prompt overlay 적용", entity.short_id, ) try: result = generate_and_validate_reference( gemini_client=gemini_client, entity_name=entity.name, entity_description=entity_data["description"], entity_type=entity.entity_type, t2i_prompt=_single_t2i, output_dir=reference_dir, # Phase 4 iter 7 review I3 — single ref manual path 도 trace_meta # 명시. set_context 만 의존하면 entity_id metadata 누락. trace_meta={ "project_id": self._project_id, "episode_id": episode_id, "operation_type": "single_ref", "entity_id": entity_id, }, ) except ModerationError as exc: raise AppError( code="image.generation_blocked", message=f"Image generation blocked: {exc.block_reason}", status_code=400, ) lineage = build_lineage_fields(self._db, self._project_id, "ref_image_pipeline") asset_id = str(uuid.uuid4()) asset = ImageAsset( id=asset_id, project_id=self._project_id, asset_type="reference", entity_id=entity_id, still_id=None, episode_id=episode_id, file_path=to_relative_image_path(result["file_path"]), prompt_used=result.get("prompt_used", ""), generation_model=result.get("generation_model", settings.gemini_image_model), width=None, height=None, status="generated", review_notes=json.dumps(result.get("validation", {}), ensure_ascii=False) if result.get("validation") else None, validation_score=result.get("validation", {}).get("score") if result.get("validation") else None, sanitization_strategy=result.get("sanitization_info", {}).get("strategy") if result.get("sanitization_info") else None, original_prompt=result.get("original_prompt"), sanitization_note=result.get("sanitization_info", {}).get("note") if result.get("sanitization_info") else None, code_version=lineage["code_version"], prompt_file_version=lineage["prompt_file_version"], reference_image_ids=lineage["reference_image_ids"], created_at=datetime.now(timezone.utc).isoformat(), ) self._db.add(asset) auto_set_primary(self._db, self._project_id, asset) self._db.commit() self._logger.log( actor_id=self._actor_id, action="image.generate_single_entity", resource_type="image", resource_id=asset_id, project_id=self._project_id, detail={"entity_id": entity_id}, ip_address=ip, ) return image_to_dict(asset)