"""참조 이미지 생성 서비스 — ImageService에서 분리된 reference/composite 전용. Phase 3b.2 (2026-04-18): ImageService의 참조/합성 관련 7개 메서드를 이관. - generate_base_references (Phase 1 진입점, Phase 3b.4 신규) - generate_composites (Phase 2+3 진입점, Phase 3b.4 신규) - generate_reference_images_only (orchestrator 위임) - generate_single_entity_image (단일 엔티티 참조 재생성) - generate_composite_image (단일 composite 재생성) - regenerate_composites_for_entity (엔티티 기준 composite 전체 재생성) - _validate_reference (validator wrapper) Phase 3b.5b (2026-04-18): ImageService wrapper 전부 제거. 호출자(api/v1/images.py, core/steps/image_steps.py)는 본 서비스를 직접 인스턴스화한다. F24.4.1 (2026-04-24): generate_reference_images_only 본체를 `ReferencePipelineOrchestrator`로 literal lift. facade는 orchestrator.run()에 얇게 위임한다. """ from __future__ import annotations import json import logging from pathlib import Path from typing import Any, Dict, List, Optional from sqlalchemy.orm import Session as OrmSession from app.logging.activity_logger import ActivityLogger from app.modules.image_validator import ImageValidator from app.services.reference_composite_service import ReferenceCompositeService from app.services.reference_entity_service import ReferenceEntityService from app.services.reference_pipeline_orchestrator import ReferencePipelineOrchestrator logger = logging.getLogger(__name__) class ReferenceImageService: """참조/합성 이미지 생성 전담 서비스 (Phase 3b.2).""" 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._entity_svc = ReferenceEntityService( db=db, project_id=project_id, actor_id=actor_id, activity_logger=self._logger, ) self._composite_svc = ReferenceCompositeService( db=db, project_id=project_id, actor_id=actor_id, activity_logger=self._logger, ) self._pipeline_svc = ReferencePipelineOrchestrator( db=db, project_id=project_id, actor_id=actor_id, activity_logger=self._logger, ) # ------------------------------------------------------------------ # Public — Phase 1 / Phase 2+3 entry points (Phase 3b.4) # ------------------------------------------------------------------ def generate_base_references( self, episode_id: str, *, mode: str = "resume", ip: Optional[str] = None, ) -> Dict[str, Any]: """엔티티별 base reference 이미지만 생성 (Phase 1). RefImageGenStep이 본 API로 호출. `skip_composite=True`를 강제하여 Phase 1(reference)까지만 수행한다. """ return self.generate_reference_images_only( episode_id, ip=ip, mode=mode, skip_composite=True, ) def generate_composites( self, episode_id: str, *, mode: str = "resume", ip: Optional[str] = None, ) -> Dict[str, Any]: """composite(character + outlook) 이미지 생성 (Phase 2 + 3). Phase 1(base reference)는 resume 모드 덕에 이미 완료된 entity는 건너뛰며 실제로 Phase 2(outfit 단독) + Phase 3(composite 전신) 단계만 수행한다. CompositeImageGenStep이 본 API로 호출. `skip_composite=False` 기본값. """ return self.generate_reference_images_only( episode_id, ip=ip, mode=mode, skip_composite=False, ) # ------------------------------------------------------------------ # Public — orchestrator entry (facade → ReferencePipelineOrchestrator) # ------------------------------------------------------------------ def generate_reference_images_only( self, episode_id: str, ip: Optional[str] = None, mode: str = "resume", *, skip_composite: bool = False, ) -> Dict[str, Any]: """엔티티 참조 이미지 생성 — orchestrator 위임 (F24.4.1). Phase 3b.4: 신규 호출자는 `generate_base_references` / `generate_composites` public API를 사용할 것. 본 함수는 기존 호출자(api/v1/images.py 및 backend/scripts/experiment_*.py) 호환을 위해 `skip_composite` 플래그 시그니처로 유지된다. """ return self._pipeline_svc.run( episode_id, ip=ip, mode=mode, skip_composite=skip_composite, ) # ------------------------------------------------------------------ # Public — single entity / composite regeneration # ------------------------------------------------------------------ def generate_single_entity_image( self, entity_id: str, custom_prompt: Optional[str] = None, ip: Optional[str] = None, ) -> Dict[str, Any]: return self._entity_svc.generate_single_entity_image( entity_id, custom_prompt=custom_prompt, ip=ip, ) def generate_composite_image( self, character_id: str, outlook_id: str, ip: Optional[str] = None, ) -> Dict[str, Any]: return self._composite_svc.generate_composite_image(character_id, outlook_id, ip=ip) def regenerate_composites_for_entity( self, entity_id: str, ip: Optional[str] = None, ) -> List[Dict[str, Any]]: return self._composite_svc.regenerate_composites_for_entity(entity_id, ip=ip) # ------------------------------------------------------------------ # Public — validator wrapper # ------------------------------------------------------------------ def _validate_reference( self, validator: ImageValidator, file_path: str, entity_info: Dict[str, Any], status: str, review_notes: str, ) -> tuple: try: fp = Path(file_path) if not fp.exists(): return None, None, status, review_notes image_bytes = fp.read_bytes() result = validator.validate_reference_image(image_bytes, entity_info) score = result.get("score", 0) result_json = json.dumps(result, ensure_ascii=False) if score < 60: issues = result.get("issues", []) issue_text = "; ".join(issues) if issues else "Low validation score" status = "needs_fix" review_notes = f"[auto-validation] score={score}: {issue_text}" logger.info("Reference image validated: score=%d, passed=%s", score, result.get("passed")) return score, result_json, status, review_notes except Exception as exc: logger.warning("Reference image validation failed: %s", exc) return None, None, status, review_notes