"""plate_multiroll — 배경 플레이트에 공통 생성 파이프 적용 (2026-07-13, opt-in). s40 확정: 플레이트도 스틸과 동일한 공통 파이프를 통과한다 — nb2 N롤 → Gemini 단독 판정(judge_still: 프롬프트 충실+참조 일관성 축) → 선정 → 결함 검사(CRITIQUE) → 수정 프롬프트 → nb2 i2i(선정 원본 단독) → 수정본이 최종 PNG(후속 체인 앵커 — 실내 체이닝의 '직전 선정본' 계약). `plate_multiroll_enabled` ON 시 background_render.render_one_background 가 gpt-image-2 단롤 대신 이 렌더러로 위임한다(OFF=기존 경로 byte-identical). 롤/수정 중간 파일은 out_path 옆 `{stem}_a.png`… 로 남아 재개(파일 단위 skip)와 리뷰 소스가 된다. 참조 우선순위·경로 조립은 호출자(기존 계약) 그대로 — 여기서는 라벨만 부여한다. """ from __future__ import annotations import logging import shutil from pathlib import Path from typing import Any, Callable, Dict, List, Optional logger = logging.getLogger(__name__) # 플레이트 참조 공통 라벨 — 종류(fp/prior/anchor) 구분 정보가 이 레벨에 없어 # 범용 공간 진실 라벨 1종 (실험 s37 FP/ROOT/PREV 세분 라벨의 보수적 통합) PLATE_REF_LABEL = ( "SPATIAL REFERENCE — the same place this background belongs to:" " its layout, materials and fixed features are truth; keep every" " shared fixed feature identical where the views overlap. Never" " copy its camera framing." ) def render_plate_multiroll( *, prompt: str, ref_paths: List[Path], out_path: Path, bg_id: str, info: Dict[str, Any], project_id: str = "", episode_id: Optional[str] = None, project_config: Optional[Dict[str, Any]] = None, sanitizer: Any = None, roll_count: Optional[int] = None, critique_enabled: Optional[bool] = None, force: bool = False, gen_fn: Optional[Callable] = None, judge_fn: Optional[Callable] = None, critique_fn: Optional[Callable] = None, # 재판정 대역 주입구 (2026-08-06) — 기본값 None 이면 기존과 동일하게 # 플래그를 보고 실제 판정 함수를 짓는다. fix_rejudge_fn: Optional[Callable] = None, ) -> Dict[str, Any]: """단일 플레이트 multiroll 렌더 — render_one_background 의 info 계약 유지. gen/judge/critique callable 은 테스트 주입용 — None 이면 production 어댑터(multiroll_gemini)로 구성. record 는 `{out_stem}_record.json` sidecar 로 phase 단위 durable persist(재개·critique 소급·지문 대조 SOT, Codex 1차 리뷰 BLOCKING-3). force=True 는 지문 일치와 무관하게 재생성. """ import json as _json from app.core.config import settings from app.modules.pipeline.multiroll_select import ( build_critique_schema, build_judge_schema, roll_labels, run_multiroll_select, ) if roll_count is None: roll_count = int(settings.still_recipe_roll_count) if critique_enabled is None: critique_enabled = bool(settings.still_recipe_critique_enabled) from app.modules.pipeline.multiroll_gemini import ( make_gemini_critique_fn, make_gemini_judge_fn, make_nb2_gen_fn, resolve_judge_texts, ) judge_texts = resolve_judge_texts(roll_count, judge_name="judge_still") if gen_fn is None: gen_fn = make_nb2_gen_fn( project_id=project_id, episode_id=episode_id, operation_type="plate_multiroll_roll", sanitizer=sanitizer, ) if judge_fn is None: judge_fn = make_gemini_judge_fn( judge_sys=judge_texts["judge_sys"], judge_schema=build_judge_schema(roll_labels(roll_count)), project_config=project_config, step_tag="plate_multiroll_judge", ) if critique_enabled and critique_fn is None: critique_fn = make_gemini_critique_fn( critique_sys=judge_texts["critique_sys"], critique_schema=build_critique_schema(), project_config=project_config, step_tag="plate_multiroll_critique", ) # E2E10 fix②: i2i 수정본 무판정 확정 → 2후보 블라인드 재판정 — # 공통 생성 파이프 전체 적용(플레이트 포함). OFF=미전달 byte-identical. # ★2026-08-06: 재판정 기본값이 ON 으로 올라가면서, 주입 이음매가 없으면 # 단위 테스트가 실제 판정 API 를 때린다(실측 — 가짜 이미지 바이트로 # provider 400). 호출측이 대역을 넣을 수 있게 파라미터를 연다. fix_rejudge_header = None if fix_rejudge_fn is not None: pass elif critique_enabled and bool( getattr(settings, "multiroll_fix_rejudge_enabled", False) ): from app.modules.pipeline.multiroll_gemini import ( load_fix_rejudge_header, ) _rj_texts = resolve_judge_texts(2, judge_name="judge_still") # Codex HIGH-4: provenance 비노출 중립 헤더(팩 v3) — 기본 헤더는 # repair prompt 생성 수정본에 거짓 fix_rejudge_header = load_fix_rejudge_header() fix_rejudge_fn = make_gemini_judge_fn( judge_sys=_rj_texts["judge_sys"], judge_schema=build_judge_schema(roll_labels(2)), project_config=project_config, step_tag="plate_multiroll_fix_rejudge", prompt_header=fix_rejudge_header, ) # E2E11 fix③: GPT 구도 critique — Gemini 결함과 합산 수정 (OFF=미전달) composition_critique_fn = None if critique_enabled and bool( getattr(settings, "multiroll_gpt_composition_enabled", False) ): from app.modules.pipeline.multiroll_gemini import ( make_gpt_composition_critique_fn, ) composition_critique_fn = make_gpt_composition_critique_fn( critique_schema=build_critique_schema(), project_config=project_config, step_tag="plate_multiroll_gpt_composition", ) labeled_refs = [(PLATE_REF_LABEL, p) for p in ref_paths] out_stem = out_path.parent / out_path.stem record_path = out_stem.parent / f"{out_stem.name}_record.json" record_in: Optional[Dict[str, Any]] = None if record_path.exists(): try: record_in = _json.loads(record_path.read_text(encoding="utf-8")) except Exception: # noqa: BLE001 logger.warning( "plate_multiroll %s: record sidecar 파싱 실패 — 재생성", bg_id ) def _persist_record(rec: Dict[str, Any]) -> None: tmp = record_path.with_suffix(".tmp") tmp.write_text( _json.dumps(rec, ensure_ascii=False, indent=1), encoding="utf-8" ) tmp.replace(record_path) # Codex 배치 리뷰 HIGH-3: judge 팩=산출 실질 입력 — literal 이 아니라 # 실사용 selector 해석값을 지문에 넣는다 (v2 전환이 reuse 를 뚫도록) from app.modules.pipeline.multiroll_gemini import ( resolve_judge_pack_version, resolve_select_judge_model, resolve_select_judge_model_physical, ) # 선정 판정 모델(alias+물리)도 산출 실질 입력 — 2026-08-06 Claude Opus # 전환으로 어느 롤이 뽑히는지가 바뀐다. extra_fingerprint = { "image_model": settings.gemini_image_model, "judge_pack": resolve_judge_pack_version(), "judge_model": resolve_select_judge_model(), "select_judge_model_physical": resolve_select_judge_model_physical(), } if fix_rejudge_fn is not None: # Codex BLOCKING-2: 재판정 물리 judge 모델·중립 헤더도 산출 실질 # 입력 (정책 버전은 run_multiroll_select 가 자체 기여) extra_fingerprint["fix_rejudge_judge_model_physical"] = str( resolve_select_judge_model_physical() ) if fix_rejudge_header is not None: extra_fingerprint["fix_rejudge_judge_header"] = ( fix_rejudge_header) if composition_critique_fn is not None: from app.modules.pipeline.multiroll_gemini import ( GPT_COMPOSITION_MODEL, GPT_COMPOSITION_PACK_VERSION, resolve_judge_pack_version as _judge_pack_resolve, ) extra_fingerprint["gpt_composition_pack"] = _judge_pack_resolve( GPT_COMPOSITION_PACK_VERSION) # Codex HIGH-2: alias 는 물리 모델 교체를 감지 못함 — 물리 모델 # 병행 스탬프 (스틸 배선과 대칭) extra_fingerprint["gpt_composition_model"] = GPT_COMPOSITION_MODEL extra_fingerprint["gpt_composition_model_physical"] = str( getattr(settings, "openai_model", "") ) try: sel_path, record = run_multiroll_select( tag=f"plate_{bg_id}", prompt=prompt, labeled_refs=labeled_refs, out_stem=out_stem, gen_fn=gen_fn, judge_fn=judge_fn, critique_fn=critique_fn, fix_gen_fn=gen_fn if critique_enabled else None, roll_count=roll_count, critique_enabled=critique_enabled, fix_head=judge_texts["fix_head"], fix_tail=judge_texts["fix_tail"], fix_label=judge_texts["fix_label"], record=record_in, extra_fingerprint=extra_fingerprint, persist_record_fn=_persist_record, force=force, fix_rejudge_fn=fix_rejudge_fn, composition_critique_fn=composition_critique_fn, ) shutil.copy(sel_path, out_path) info["attempts"] = roll_count info["status"] = "ok" info["png_path"] = str(out_path) info["multiroll"] = { "selected": record.get("selected"), "totals": record.get("totals"), "ranking": record.get("ranking"), "fix_applied": bool(record.get("fix_prompt")), "issues": (record.get("critique") or {}).get("issues"), } if record.get("fix_rejudge") is not None: # E2E10 fix② 감사 — 재판정 승자(수정본 개악=원본 유지 가시화) info["multiroll"]["fix_rejudge_won"] = ( record["fix_rejudge"].get("fix_won") ) return info except Exception as exc: # noqa: BLE001 — 노드 단위 실패 격리 (기존 계약) logger.exception("plate_multiroll %s: 실패", bg_id) info["status"] = "failed" info["final_block_reason"] = str(exc)[:200] return info