"""ReferencePhase1Service — Phase 1 (base reference) 단독 서비스. F24.4.2 (2026-04-24): `ReferencePipelineOrchestrator.run()`의 Phase 1 루프 (batch 단위 base reference 이미지 생성 + early return)를 literal lift. 반환: - None: Phase 1 성공 or 0건. 다음 Phase로 진행. - Dict: Phase 1 부분 실패. orchestrator가 early return. - raise AppError: Phase 1 전체 실패. """ from __future__ import annotations import json import logging import uuid from concurrent.futures import ThreadPoolExecutor, as_completed 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.errors import AppError from app.core.file_paths import to_relative_image_path from app.services.image_capture.annotate import annotate_generated_asset from app.core.image_call_budget import bind_current_budget from app.logging.activity_logger import ActivityLogger from app.models.project import ImageAsset 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 ReferencePhase1Service: """Phase 1: 엔티티별 base reference 이미지 병렬 생성.""" 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 run(self, ctx: ReferencePipelineContext) -> Optional[Dict[str, Any]]: from app.modules.pipeline.ref_image_pipeline import generate_and_validate_reference from app.modules.pipeline.grounding_canonical_ref_inputs import ( GROUNDING_INPUT_DIGEST_KEY, GROUNDING_INPUTS_KEY, GROUNDING_REFERENCE_LABEL, grounding_input_digest, provenance_of, ) from app.modules.pipeline.grounding_entity_contract import ( canonical_ref_owner_types, ) # ★canonical ref 를 만드는 갈래는 계약 하나에서 — 게이트·readiness 와 같다 _canonical_owner_types = set(canonical_ref_owner_types()) # W21B-W7 W-B (2026-06-12): printed_prop visual-continuity anchor — # prop ref 프롬프트 overlay 컨텍스트 (flag OFF / cp 부재 시 빈 dict = # 기존 경로 byte-identical). episode 당 1회 로드. from app.core.steps.visual_continuity_anchor_step import ( load_printed_prop_anchor_context, ) vca_prop_ctx = load_printed_prop_anchor_context(self._project_id, ctx.episode_id) # local capture (literal lift — orchestrator에서 이름 그대로) batches = ctx.batches entity_by_id = ctx.entity_by_id already_done = ctx.already_done _low_freq_skip_ids = ctx.low_freq_skip_ids deps = ctx.deps ref_image_map = ctx.ref_image_map ref_cp = ctx.ref_cp _o00_char_ids = ctx.o00_char_ids _body_identity_ids = ctx.body_identity_char_ids from app.core.body_identity import reference_entity_type grounding_references = ctx.grounding_references gemini_client = ctx.gemini_client reference_dir = ctx.reference_dir max_concurrent = ctx.max_concurrent progress = ctx.progress entities = ctx.entities episode_id = ctx.episode_id skipped_count = ctx.skipped_count low_freq_skipped = ctx.low_freq_skipped total_to_gen = ctx.total_to_gen for batch_idx, batch in enumerate(batches): batch_entities = [ entity_by_id[eid] for eid in batch if eid in entity_by_id and eid not in already_done and eid not in _low_freq_skip_ids and entity_by_id[eid].get("entity_type") in _canonical_owner_types ] if not batch_entities: continue progress.update( f"참조 이미지 {ctx.generated_count + skipped_count}/{len(entities)}", ctx.generated_count, total_to_gen, ) def _gen_ref(entity): entity_deps = deps.get(entity["id"], set()) is_variant = bool(entity_deps) and entity.get("entity_type") == "character" is_null_outlook = entity["id"] in _o00_char_ids dep_refs = [ (f"Base form — KEEP THIS FACE: {entity_by_id.get(d, {}).get('name', '')}" if is_variant and not is_null_outlook else f"Reference: {entity_by_id.get(d, {}).get('name', '')}", ref_image_map[d]) for d in entity_deps if d in ref_image_map ] t2i = entity.get("t2i_prompt") or entity.get("description") or entity["name"] desc = entity.get("description", "") # W21B-W7 W-B: printed_prop anchor → reference prompt overlay. # canon(EntityCanon.t2i_prompt / entity_t2i cp) 은 변경하지 않는다 # — canon 자체가 provenance. overlay 는 이 생성 호출의 prompt 에만 # 반영 (Codex 가이드 6 / W_A review). ★실측(2026-06-12): prop 은 # ref_image_pipeline 의 외부 템플릿이 entity_description 으로 # 조립되고 t2i_prompt 는 템플릿 부재시 fallback — 둘 다 overlay. if entity.get("entity_type") == "prop" and vca_prop_ctx: _anchor = (vca_prop_ctx.get("by_prop") or {}).get( entity.get("short_id") or "") if _anchor: from app.modules.pipeline.visual_continuity_anchor_plan import ( build_prop_ref_prompt_overlay, ) t2i = build_prop_ref_prompt_overlay(_anchor, t2i) desc = build_prop_ref_prompt_overlay( _anchor, desc, include_framing=False) logger.info( "visual_continuity_anchor: prop %s ref prompt overlay 적용 " "(canon 비변경)", entity.get("short_id"), ) if dep_refs and entity.get("entity_type") == "character" and not is_null_outlook: base_names = [entity_by_id.get(d, {}).get("name", "") for d in entity_deps if d in ref_image_map] variant_instruction = ( f"CRITICAL — This is a VARIANT form of {', '.join(base_names)}. " f"The reference image shows the base form. " f"Keep the EXACT same face, bone structure, skin tone, and identity. " f"Only change what the description below specifies.\n\n" ) desc = variant_instruction + desc t2i = variant_instruction + t2i # ★★사람이 확인한 조사 사진 → 이 canonical ref 의 **실제 입력** # (Codex BLOCK 2026-09-02: 종전엔 ID 로만 올라가 사진 없이 생성됐다). # `dep_refs` 뒤에 **따로** 둔다 — 위 variant 판정이 `dep_refs` # 유무를 보기 때문이다. 결속은 typed id(short_id == 중앙 CP 의 # final_id)로만 — 이름 결속 없음. 호출 수는 그대로다. grounding_refs = [ (GROUNDING_REFERENCE_LABEL, g["bytes"]) for g in (grounding_references.get(entity.get("short_id") or "") or ()) ] # ★전신 기본 형태를 쓰는 기준은 **두 가지**다 (2026-09-18). # 종전에는 「아웃룩이 없나」 하나였고, 그래서 옷을 입는 로봇은 # 사람 경로(얼굴 흉상)로 가 몸이 아웃룩마다 새로 지어졌다. is_body_identity = entity["id"] in _body_identity_ids ref_entity_type = reference_entity_type( entity.get("entity_type", "character"), is_body_identity=is_body_identity, is_null_outlook=is_null_outlook) pipe_result = generate_and_validate_reference( gemini_client=gemini_client, entity_name=entity["name"], entity_description=desc, entity_type=ref_entity_type, t2i_prompt=t2i, output_dir=reference_dir, extra_references=(dep_refs + grounding_refs) or None, style_context="", # Phase 4 iter 7 B2 — observability: PID/EID 명시 forward. trace_meta={ "project_id": self._project_id, "episode_id": episode_id, "operation_type": "ref_image_gen_phase1", "entity_id": entity["id"], }, ) result = { "id": _new_id(), "entity_id": entity["id"], "asset_type": "reference", "still_id": None, "file_path": pipe_result["file_path"], "prompt_used": t2i, "generation_model": pipe_result["generation_model"], "width": None, "height": None, "status": "generated", "review_notes": json.dumps(pipe_result.get("validation", {}), ensure_ascii=False), "validation_score": pipe_result.get("validation", {}).get("score"), "created_at": _now(), # ★출처 — 어느 사진(sha)이 이 참조의 입력이었나. 없으면 []. GROUNDING_INPUTS_KEY: provenance_of( grounding_references.get(entity.get("short_id") or "") or []), # ★입력 지문 — 입력 없음도 명시 값. 자산·CP 행 둘 다에 남는다 GROUNDING_INPUT_DIGEST_KEY: grounding_input_digest( grounding_references.get(entity.get("short_id") or "") or []), } return entity["id"], result with ThreadPoolExecutor(max_workers=min(max_concurrent, len(batch_entities))) as executor: # W20E5 Codex B2 — propagate parent-thread image-call budget # (_gen_ref → ReferencePipeline → GeminiImageClient image gen). _submit_gen_ref = bind_current_budget(_gen_ref) futures = {executor.submit(_submit_gen_ref, e): e for e in batch_entities} for future in as_completed(futures): try: eid, result = future.result() fp = Path(result.get("file_path", "")) if fp.exists(): ref_image_map[eid] = fp.read_bytes() self._db.query(ImageAsset).filter( ImageAsset.project_id == self._project_id, ImageAsset.entity_id == result["entity_id"], ImageAsset.is_primary == 1, ).update({"is_primary": 0}) asset = ImageAsset( id=result["id"], project_id=self._project_id, asset_type="reference", entity_id=result["entity_id"], episode_id=episode_id, file_path=to_relative_image_path(result["file_path"]), prompt_used=result["prompt_used"], generation_model=result["generation_model"], status=result["status"], review_notes=result.get("review_notes"), validation_score=result.get("validation_score"), is_primary=1, created_at=result["created_at"], ) self._db.add(asset) # persist-all Wave 1 — 얼굴(face)=T2I 루트(이미지 입력 없음). annotate_generated_asset( asset, pipeline_role="reference_face", input_image_ids=[], # ★어떤 사진이 실렸나(없으면 [])와 그 지문을 **늘** 남긴다 — # resume 이 이 지문으로 신선도를 본다 (Codex B) pipeline_metadata={ GROUNDING_INPUTS_KEY: result[GROUNDING_INPUTS_KEY], GROUNDING_INPUT_DIGEST_KEY: result[GROUNDING_INPUT_DIGEST_KEY], }, ) self._db.commit() ctx.generated_count += 1 ref_cp.mark_completed(eid, { "asset_id": result["id"], "file_path": result["file_path"], # ★CP 행에도 같은 지문 — 자산과 **둘 다** 맞아야 되쓴다 GROUNDING_INPUT_DIGEST_KEY: result[GROUNDING_INPUT_DIGEST_KEY], }) progress.update( f"참조 이미지 {ctx.generated_count + skipped_count}/{len(entities)}: {entity_by_id.get(eid, {}).get('name', '')}", ctx.generated_count, total_to_gen, ) except Exception as exc: failed_entity = futures[future] eid_failed = failed_entity.get("id", "") logger.warning("Ref image failed (%s): %s", failed_entity.get("name", eid_failed), exc) ref_cp.mark_failed(eid_failed, str(exc)[:500]) ctx.failed_count += 1 if ctx.generated_count > 0 and ctx.failed_count == ctx.generated_count: progress.error("참조 이미지 생성 전부 실패 (API 오류)") raise AppError( code="image.all_ref_failed", message=f"참조 이미지 {ctx.failed_count}개 모두 실패했습니다. API 상태를 확인하세요.", status_code=500, ) if ctx.failed_count > 0: logger.warning( "참조 이미지 %d개 실패 — 합성 이미지 생성 건너뜀. resume으로 참조를 먼저 완성하세요.", ctx.failed_count, ) progress.complete() return { "generated": ctx.generated_count, "skipped": skipped_count, "low_freq_skipped": low_freq_skipped, "failed": ctx.failed_count, "total": len(entities), "message": f"참조 이미지 {ctx.failed_count}개 실패. resume으로 재시도하세요.", } return None