"""ReferencePipelineOrchestrator — Phase 1/2/3 참조 이미지 생성 orchestration. Phase 3b.2 ~ F24.3: reference_image_service.py의 ``generate_reference_images_only`` (666 LOC) 본체를 literal lift. facade ``ReferenceImageService``는 본 서비스의 ``run(...)``에 얇게 위임한다. 이후 F24.4.2~4에서 Phase1/2/3 서비스로 분해될 예정 (현재는 단일 run 메서드). """ from __future__ import annotations import hashlib 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.entity_protection import ( collect_reference_required_canon_ids, compute_variant_pole_ids, should_skip_low_freq, ) from app.core.errors import AppError from app.i18n.loader import t from app.logging.activity_logger import ActivityLogger from app.models.project import ( CharacterOutlook, Episode, EntityCanon, EntityEpisodeLink, ImageAsset, RelationFact, RelationParticipant, SceneStill, WorldGuide, ) from app.modules.entity_dependency import ( build_visual_dependency_graph, topological_sort_entities, ) from app.modules.llm.gemini_image_client import GeminiImageClient from app.modules.llm.gemini_key_pool import key_count as gemini_key_count from app.modules.llm.openai_client import OpenAIClient from app.modules.progress_tracker import ProgressTracker from app.modules.world_guide_generator import WorldGuideGenerator from app.services.reference_phase1_service import ReferencePhase1Service from app.services.reference_phase2_service import ReferencePhase2Service from app.services.reference_phase3_service import ReferencePhase3Service from app.services.reference_pipeline_context import ReferencePipelineContext logger = logging.getLogger(__name__) def _now() -> str: return datetime.now(timezone.utc).isoformat() def _new_id() -> str: return str(uuid.uuid4()) class ReferencePipelineOrchestrator: """Phase 1/2/3 참조 이미지 생성 orchestrator (F24.4).""" 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 self._phase1_svc = ReferencePhase1Service( db=db, project_id=project_id, actor_id=actor_id, activity_logger=activity_logger, ) self._phase2_svc = ReferencePhase2Service( db=db, project_id=project_id, actor_id=actor_id, activity_logger=activity_logger, ) self._phase3_svc = ReferencePhase3Service( db=db, project_id=project_id, actor_id=actor_id, activity_logger=activity_logger, ) # ------------------------------------------------------------------ # Internal helpers # ------------------------------------------------------------------ def _get_episode(self, episode_id: str) -> Episode: ep = ( self._db.query(Episode) .filter(Episode.id == episode_id, Episode.project_id == self._project_id) .first() ) if not ep: raise AppError( code="episode.not_found", message=t("episode.not_found"), status_code=404, ) return ep # ------------------------------------------------------------------ # Public — 참조 이미지 orchestration # ------------------------------------------------------------------ def run( self, episode_id: str, ip: Optional[str] = None, mode: str = "resume", *, skip_composite: bool = False, ) -> Dict[str, Any]: """엔티티 참조 이미지 생성 orchestrator — Phase 1 (base ref) + Phase 2/3 (composite). Phase 3b.4: 신규 호출자는 `generate_base_references` / `generate_composites` public API를 사용할 것. 본 함수는 기존 호출자(api/v1/images.py 및 backend/scripts/experiment_*.py) 호환을 위해 `skip_composite` 플래그 시그니처로 유지된다. mode: "resume" 만 valid. "full" 모드는 feedback_never_delete_images 규칙 위반 (ImageAsset 일괄 DELETE + cp.clear) 으로 영구 차단. 강제 재생성은 RefImageGenStep mode='force' 사용. skip_composite (Phase 3.1 step 경계 재정의): True → Phase 1(base reference)까지만 수행, Phase 2(character+outlook composite) 생략. RefImageGenStep이 이 플래그로 호출하여 자기 책임(ref)만 수행. CompositeImageGenStep은 기본값(False)로 호출하여 resume 모드에서 이미 완료된 reference는 skip되고 composite만 생성됨. """ # feedback_never_delete_images: image step 재실행은 절대 일괄 DELETE 금지. # mode='full' 의 destructive 동작 (ImageAsset DELETE + ref_cp.clear) 영구 차단. # early validation — db/episode 의존성 lookup 전에 fail-fast. if mode not in ("resume",): raise ValueError( f"ReferencePipelineOrchestrator.run: mode={mode!r} unsupported. " "Only 'resume' is allowed (feedback_never_delete_images). " "Use RefImageGenStep mode='force' for non-destructive regeneration." ) episode = self._get_episode(episode_id) if not episode.fulltext: raise AppError(code="analysis.no_text", message=t("analysis.no_text"), status_code=400) if episode.status != "analyzed": raise AppError( code="image.analysis_not_complete", message=f"분석이 완료되지 않았습니다 (현재: {episode.status}). 분석을 먼저 완료하세요.", status_code=400, ) entity_count = self._db.query(EntityCanon).filter( EntityCanon.project_id == self._project_id, ).count() if entity_count == 0: raise AppError( code="image.no_entities", message="추출된 요소가 없습니다. 분석을 먼저 실행하세요.", status_code=400, ) 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) language = episode.language or "ko" project_dir = Path(settings.projects_dir) / self._project_id reference_dir = project_dir / "images" / episode_id / "reference" # ★보류(shelved)를 뺀 **정본 하나**를 쓴다 (2026-09-04). 링크를 직접 # 읽으면 보류 제외가 한 곳만 붙고 나머지는 샌다 — 실측: 이 화 링크를 # 읽는 자리 21곳 중 거르는 곳이 1곳이었다. from app.core.entity_identity import active_episode_canon_ids canon_ids = active_episode_canon_ids( self._db, self._project_id, episode_id) entities_orm = self._db.query(EntityCanon).filter( EntityCanon.id.in_(canon_ids) ).all() if canon_ids else [] entities = [ {"id": e.id, "name": e.name, "entity_type": e.entity_type, "short_id": e.short_id or "", # W21B-W7 W-B: prop anchor overlay 조인 키 "description": e.description or "", "stable_traits": e.stable_traits or "{}", "t2i_prompt": e.t2i_prompt or ""} for e in entities_orm ] # ★★사람이 확인한 C/P 조사 사진 — canonical ref 생성의 실제 입력이자 # **재개 신선도의 입력** (Codex BLOCK 2026-09-02 ×2). RefImageGenStep 과 # API 가 둘 다 이 run 을 지나므로 여기 한 곳에서, already_done 을 정하기 # **앞에서** 읽는다. v2_chunk 인데 중앙 CP 가 없으면 provider 앞에서 선다. from app.modules.pipeline.grounding_canonical_ref_inputs import ( EMPTY_INPUT_DIGEST, GROUNDING_INPUT_DIGEST_KEY, digests_by_final_id, is_fresh, project_grounding_mode, stored_digest_of_asset, verified_policy_references, ) _grounding_mode = project_grounding_mode(self._db, self._project_id) grounding_references = verified_policy_references( self._db, project_id=self._project_id, episode_id=episode_id, grounding_mode=_grounding_mode) _digest_by_fid = digests_by_final_id(grounding_references) # entity id → 지금 입력 지문 (입력 없음도 명시 값) want_digest: Dict[str, str] = { e["id"]: _digest_by_fid.get(e.get("short_id") or "", EMPTY_INPUT_DIGEST) for e in entities } stale_ids: set = set() # ★한쪽이라도 옛 지문이면 되쓰지 않는다 ref_image_map: Dict[str, bytes] = {} already_done: set = set() if mode == "resume": for entity in entities: existing = ( self._db.query(ImageAsset) .filter( ImageAsset.project_id == self._project_id, ImageAsset.entity_id == entity["id"], ImageAsset.asset_type == "reference", ImageAsset.is_primary == 1, ) .order_by(ImageAsset.created_at.desc()) .first() ) if existing: fp = Path(existing.file_path) if not fp.exists(): logger.warning("Ref image file missing, will regenerate: %s → %s", entity["name"], existing.file_path) elif not is_fresh(stored_digest_of_asset(existing), want_digest[entity["id"]]): # ★사람이 확인한 사진(또는 그 철회)이 이 자산에 안 실려 있다 — # 되쓰지 않는다. 옛 primary 는 **새 생성이 성공한 뒤에만** # 내려간다(Phase1 persist 순서 그대로 · Codex C). stale_ids.add(entity["id"]) logger.info("Ref image stale (grounding input changed), will regenerate: %s", entity["name"]) else: already_done.add(entity["id"]) ref_image_map[entity["id"]] = fp.read_bytes() openai_client = OpenAIClient() stills_orm = self._db.query(SceneStill).filter( SceneStill.project_id == self._project_id, SceneStill.episode_id == episode_id, SceneStill.is_selected == True, # noqa: E712 SceneStill.still_index >= 0, SceneStill.status != "stale", ).all() stills = [{"still_frame_prompt": s.still_frame_prompt or ""} for s in stills_orm] _ft = episode.fulltext or "" wg_hash = hashlib.md5(f"{_ft}:{len(entities)}:{len(stills)}".encode()).hexdigest() existing_wg = self._db.query(WorldGuide).filter( WorldGuide.project_id == self._project_id, WorldGuide.episode_id == episode_id, ).order_by(WorldGuide.created_at.desc()).first() if existing_wg and existing_wg.source_hash == wg_hash: world_guide = json.loads(existing_wg.guide_json) else: wg_gen = WorldGuideGenerator(llm_client=openai_client) world_guide = wg_gen.generate( fulltext=episode.fulltext, language=language, source_file=episode.source_filename or "episode", entities=entities, stills=stills, ) wg_record = WorldGuide( id=_new_id(), project_id=self._project_id, episode_id=episode_id, guide_json=json.dumps(world_guide, ensure_ascii=False), source_hash=wg_hash, created_at=_now(), ) self._db.add(wg_record) self._db.commit() gemini_client = GeminiImageClient(model=settings.gemini_image_model) from app.modules.image_checkpoint import ImageCheckpointManager cp_dir = project_dir / "checkpoints" / "images" / episode_id ref_cp = ImageCheckpointManager(cp_dir, "reference") # mode='full' 분기 제거됨 (run() 첫 줄 ValueError 가드) — 단일 resume 경로. for cp_id in ref_cp.get_completed_ids(): cp_data = ref_cp._data.get("completed", {}).get(cp_id, {}) cp_path = cp_data.get("file_path", "") if cp_path and Path(cp_path).exists(): # ★자산 쪽이 stale 이면 CP 가 되살리지 못한다; CP 행의 지문이 옛 # 것이면 자산이 신선해도 되쓰지 않는다 — **둘 다** 같아야 한다. if cp_id in stale_ids: continue if not is_fresh(cp_data.get(GROUNDING_INPUT_DIGEST_KEY), want_digest.get(cp_id, EMPTY_INPUT_DIGEST)): already_done.discard(cp_id) stale_ids.add(cp_id) logger.info("Ref checkpoint stale (grounding input changed): %s", cp_id[:8]) continue already_done.add(cp_id) else: logger.warning("Checkpoint ref file missing, will regenerate: %s", cp_id[:8]) relations_orm = self._db.query(RelationFact).filter(RelationFact.project_id == self._project_id).all() rel_ids = [r.id for r in relations_orm] participants_orm = self._db.query(RelationParticipant).filter( RelationParticipant.relation_id.in_(rel_ids) ).all() if rel_ids else [] deps = build_visual_dependency_graph( entities, [{"id": r.id, "relation_family": r.relation_family} for r in relations_orm], [{"relation_id": p.relation_id, "canon_id": p.canon_id} for p in participants_orm], ) batches = topological_sort_entities(entities, deps) # ── 저빈도 요소 참조이미지 스킵 (G4.6 RC-C: 보호 cascade 추가) ── # ★여기는 **찾아보는 표**다 — 대상 목록이 아니다. 대상은 위 # `active_episode_canon_ids` 로 이미 걸러졌고, 이 표에 보류 행이 # 섞여 있어도 :372 가 그 id 로 물어보지 않는다. 그래서 여기는 # 보류를 안 거른다 (2026-09-04 확인). _t2i_count_map: Dict[str, int] = {} for _lnk in self._db.query(EntityEpisodeLink).filter( EntityEpisodeLink.project_id == self._project_id, EntityEpisodeLink.episode_id == episode_id, ).all(): _t2i_count_map[_lnk.canon_id] = _lnk.t2i_appearance_count or 0 _reverse_dep_ids: set = set() for _dep_set in deps.values(): _reverse_dep_ids.update(_dep_set) # Phase 1 — reference 생성 보호 = scene_detail.required_refs # + **선택된 샷 VE 의 캐릭터** (2026-09-19 — 샷 이미지가 실제로 참조를 # 붙이는 기준. 사유는 `collect_reference_required_canon_ids` 주석). # broad 4-source union (scene_director.present / shot_validator / # shot_director.visible / scene_detail.visible_entities) 은 여전히 # reference 보호에서 제외 — Phase 0 audit 가 차집합 안전성 검증 완료. _required_reason_by_uuid = collect_reference_required_canon_ids( self._db, self._project_id, episode_id, ) _required_canon_uuids: set = set(_required_reason_by_uuid) # variant_self 는 character-character + identity/transformation 만 (RO-1 # 좁힘 — possession prop / location-location 은 variant_self 미적용). _variant_pole_uuids = compute_variant_pole_ids( entities, [{"id": r.id, "relation_family": r.relation_family} for r in relations_orm], [{"relation_id": p.relation_id, "canon_id": p.canon_id} for p in participants_orm], ) _low_freq_skip_ids: set = set() _skip_decisions: list = [] from app.modules.pipeline.grounding_entity_contract import ( canonical_ref_owner_types, ) _canonical_owner_types = set(canonical_ref_owner_types()) for e in entities: eid = e["id"] etype = e.get("entity_type", "") if etype not in _canonical_owner_types: continue # ★계약의 정책 갈래만 — sidecar 갈래는 대상 아님 count = _t2i_count_map.get(eid, 0) is_base_for_variant = eid in _reverse_dep_ids is_variant_self = eid in _variant_pole_uuids required = eid in _required_canon_uuids skipped = should_skip_low_freq( e, count, is_base_for_variant, is_variant_self, required, ) if skipped: _low_freq_skip_ids.add(eid) _reason = "no required_ref + low t2i count + not variant" elif required: _reason = f"protected: {_required_reason_by_uuid[eid]}" elif count > 1: _reason = f"protected: t2i_count={count}" elif is_base_for_variant: _reason = "protected: base_for_variant" elif is_variant_self: _reason = "protected: variant_self" else: _reason = "protected" _skip_decisions.append({ "canon_id": eid, "name": e.get("name"), "entity_type": etype, "skipped": skipped, "reason": _reason, "t2i_count": count, }) logger.info( "Low-freq decision: %s (%s, t2i_count=%d, base_for_variant=%s, " "variant_self=%s, required=%s, skipped=%s)", e.get("name"), etype, count, is_base_for_variant, is_variant_self, required, skipped, ) from app.core.low_freq_skip import save_low_freq_skip_report save_low_freq_skip_report( self._project_id, episode_id, _skip_decisions, ) entity_by_id = {e["id"]: e for e in entities} generated_count = 0 failed_count = 0 low_freq_skipped = len(_low_freq_skip_ids) skipped_count = len(already_done) + low_freq_skipped max_concurrent = settings.max_concurrent_image_gen total_to_gen = len(entities) - skipped_count progress = ProgressTracker(self._db, episode_id, "reference_image_generation", self._project_id) # O00 (Null Outlook) 캐릭터 ID 세트 — 전신 이미지 생성용 _o00_outlook_ids = { e.id for e in self._db.query(EntityCanon).filter( EntityCanon.project_id == self._project_id, EntityCanon.short_id == "O00", EntityCanon.entity_type == "outlook", ).all() } _o00_char_ids = set() if _o00_outlook_ids: # ★이 화의 배정만 — 한 화의 O00 이 다른 화 판단을 오염시켰다. from app.core.entity_identity import episode_outlook_rows for co in episode_outlook_rows(self._db, self._project_id, episode_id): if co.outlook_id in _o00_outlook_ids: _o00_char_ids.add(co.character_id) # ★몸이 곧 신원인 인물 — 판정은 `outlook_phase1` 이 하고, 배치와 # **수동 재생성**이 같은 helper 로 읽는다(둘이 갈리면 손으로 다시 # 만든 참조가 흉상으로 돌아간다). from app.core.body_identity import body_identity_entity_ids _body_identity_char_ids = body_identity_entity_ids( self._db, self._project_id, episode_id) if _body_identity_char_ids: logger.info("몸이 곧 신원인 인물 %d명", len(_body_identity_char_ids)) # ── Phase 1: base reference (ReferencePhase1Service로 위임, F24.4.2) ── ctx = ReferencePipelineContext( episode_id=episode_id, episode=episode, mode=mode, skip_composite=skip_composite, ip=ip, entities=entities, entity_by_id=entity_by_id, deps=deps, batches=batches, reference_dir=reference_dir, gemini_client=gemini_client, max_concurrent=max_concurrent, progress=progress, ref_cp=ref_cp, ref_image_map=ref_image_map, already_done=already_done, low_freq_skip_ids=_low_freq_skip_ids, o00_char_ids=_o00_char_ids, body_identity_char_ids=_body_identity_char_ids, grounding_references=grounding_references, generated_count=0, failed_count=0, low_freq_skipped=low_freq_skipped, skipped_count=skipped_count, ) early_return = self._phase1_svc.run(ctx) if early_return is not None: return early_return # Phase 1 결과를 local로 sync (Phase 2/3은 F24.4.3/4에서 ctx 전환 예정) generated_count = ctx.generated_count failed_count = ctx.failed_count # ── Phase 1/2 경계: skip_composite 조기 종료 ── if skip_composite: progress.complete() return { "generated": generated_count, "skipped": skipped_count, "low_freq_skipped": low_freq_skipped, "failed": failed_count, "total": len(entities), "outlook_generated": 0, "outlook_skipped": 0, "skipped_composite_phase": True, } # ── Phase 2: outfit 단독 (ReferencePhase2Service, F24.4.3) ── self._phase2_svc.run(ctx) outlook_generated = ctx.outlook_generated outlook_skipped = ctx.outlook_skipped # ── Phase 3: composite (ReferencePhase3Service, F24.4.4) ── self._phase3_svc.run(ctx) composite_generated = ctx.composite_generated composite_skipped = ctx.composite_skipped progress.complete() self._logger.log( actor_id=self._actor_id, action="episode.generate_reference_images", resource_type="episode", resource_id=episode_id, project_id=self._project_id, detail={ "generated": generated_count, "skipped": skipped_count, "low_freq_skipped": low_freq_skipped, "outlook_generated": outlook_generated, "outlook_skipped": outlook_skipped, "composite_generated": composite_generated, "composite_skipped": composite_skipped, }, ip_address=ip, ) return { "reference_count": generated_count, "skipped": skipped_count, "low_freq_skipped": low_freq_skipped, "outlook_generated": outlook_generated, "outlook_skipped": outlook_skipped, }