#!/usr/bin/env python3 """수정본 참조 선별 시험 — 요구한 참조만 붙여서 다시 만들어 본다. 배경(2026-08-19 사용자 지시): 수정본을 만들면 원본의 자세·구도가 변질되고 사람이 화면 가운데 차렷 자세로 서는 일이 잦다. 이번 실행 실측으로 수정본 139장면 중 82장면(59%)이 "더 나빠졌다"로 퇴짜를 맞았다. 원인 후보(사용자 지적): 지금 코드는 지적이 요구하지 않아도 **참조를 전부 붙인다**(`multiroll_select.py:765` — 조건 없음). 인물 정본에는 "얼굴·머리· 체형을 정확히 맞춰라"가 적혀 있어, 사람을 다시 그리라는 지시나 다름없다. 실물 예: S56sh8 의 지적 4건 중 참조가 필요한 것은 이전 샷 하나뿐인데 인물 정본 3장이 함께 붙었다. 이 도구가 하는 일: ① 지적마다 **어느 참조가 필요한지 LLM 에게 묻는다**(글자 대조 아님). ② 요구된 참조만 붙인다. 아무도 요구하지 않으면 원본 한 장만 붙인다. ③ 지적이 **없는 것을 새로 넣으라**고 하면 그 사실을 지시문에 명시하고 해당 참조를 붙인다(사용자 지시). ④ 같은 지시문·같은 그림으로 **nb2 와 grok 두 모델**에 각각 보낸다. ⑤ 원본 / 기존 수정본 / 새 수정본 둘을 나란히 놓은 갤러리를 만든다. 프로덕션 무접촉: 읽기만 한다. records/체크포인트/DB 를 쓰지 않고 산출은 artifact/ 전용이다. 프로덕션 코드도 고치지 않는다 — 시험 결과를 보고 사용자가 정한 뒤에 고친다. 사용: .venv/bin/python fix_ref_gate_pilot.py # 선별 + 생성 + 갤러리 .venv/bin/python fix_ref_gate_pilot.py --html # 갤러리만 """ from __future__ import annotations import argparse import base64 import html as H import json import re import sys import time import urllib.request from datetime import datetime, timezone from pathlib import Path ROOT = Path(__file__).resolve().parent PROJ = "5bddbdfc-2681-42a6-9837-43f35f60049d" EPI = "f5372927-bbec-405d-ad2c-100d587f5373" RECIPE = ROOT.parent / "projects" / PROJ / "images" / EPI / "scene" / "recipe" REF_DIR = ROOT.parent / "projects" / PROJ / "images" / EPI / "reference" PACK = (ROOT.parent / "prompts" / "_base" / "multiroll_judge" / "12.202608141305") OUT = ROOT.parent / "artifact" / "20260819_fix_ref_gate" TAGS = ["S56sh8", "S56sh4", "S55sh8", "S55sh1", "S48sh9", "S43sh1"] GATE_MODEL = "gemini-3.1-pro-preview" NB2_MODEL = "gemini-3.1-flash-image-preview" GROK_MODEL = "x-ai/grok-imagine-image-2.0" # 지적마다 어느 참조가 필요한지 묻는 문안. 판단은 전부 모델 몫이다 — # 글자를 뒤져 맞추지 않는다. GATE_SYS = """You decide which reference photographs an image-editing job actually needs. You are given (a) a list of CORRECTIONS to apply to one photograph and (b) a list of REFERENCE photographs that could be attached alongside it. Attaching a reference is not free. A reference that a correction does not need still tells the editing model to look at it, and the model then redraws parts of the photograph that nobody asked it to touch — people get restaged, poses reset, wardrobes swapped. So attach a reference ONLY when a correction cannot be carried out without seeing it. For each correction decide: - needs_ref_indices: which references must be seen to carry it out. Empty when the correction is self-contained (erase something, move something, blur something, change an angle) — those need no reference at all. A reference is needed when the correction says to match, restore or copy something whose appearance is only knowable from that reference. - adds_missing_entity: true when the correction requires putting into the photograph a person or object that is NOT currently there at all. - missing_entity_name: what must be added, when the above is true. Be strict. When in doubt, attach nothing.""" GATE_TAIL = """ OUTPUT FORMAT: reply with ONLY a JSON object, no prose, no code fence: {"corrections": [{"index": 1, "needs_ref_indices": [], "reason_ko": "", "adds_missing_entity": false, "missing_entity_name": ""}]}""" # 없는 것을 새로 넣어야 할 때만 지시문에 덧붙이는 절 (사용자 지시). MISSING_HEAD = "ADD WHAT IS MISSING:" MISSING_TAIL = ( "These are not in the photograph at all — put them in exactly as the " "corrections describe, matching the attached reference where one is " "given. Everything already in the photograph stays untouched.") def env(key: str) -> str: for line in (ROOT / ".env").read_text(encoding="utf-8").splitlines(): if line.startswith(key + "="): return line.split("=", 1)[1].strip() return "" def lenient_json(text: str) -> dict: t = re.sub(r"^```(?:json)?\s*|\s*```$", "", (text or "").strip()) m = re.search(r"\{.*\}", t, re.S) if not m: raise ValueError(f"JSON 없음: {t[:200]!r}") return json.loads(m.group(0)) def post_json(url: str, body: dict, headers: dict, timeout: int = 600): req = urllib.request.Request( url, data=json.dumps(body).encode(), headers=headers, method="POST") with urllib.request.urlopen(req, timeout=timeout) as r: return json.loads(r.read().decode()) # ── 참조 그림 되찾기 ────────────────────────────────────────────────── # records.json 은 인물 정본을 경로가 아니라 `` 로 적어 뒀다. # 그 숫자는 파일 크기다 — 크기로 실제 파일을 되찾는다(추측 없음). def _size_index() -> dict: idx: dict[int, list[Path]] = {} for d in (REF_DIR, RECIPE): if not d.is_dir(): continue for p in d.rglob("*.png"): idx.setdefault(p.stat().st_size, []).append(p) return idx def resolve_refs(refs: list, size_idx: dict) -> list: """[{label, path}] → [(label, Path)] — 못 찾으면 그 항목만 뺀다.""" out = [] for r in refs: label = r.get("label") or "" raw = str(r.get("path") or "") m = re.match(r"$", raw) if m: cands = size_idx.get(int(m.group(1)) or -1) or [] if not cands: print(f" ★참조 못 찾음(크기 {m.group(1)}) — 뺀다") continue out.append((label, cands[0])) else: p = Path(raw) if p.is_file(): out.append((label, p)) else: print(f" ★참조 파일 없음 — 뺀다: {raw[:70]}") return out def short_label(label: str) -> str: """긴 참조 문안에서 사람이 알아볼 짧은 이름만 뽑는다.""" head = label.split("—")[0].strip() if "—" in label else label name = "" if "—" in label: rest = label.split("—", 1)[1].strip() name = rest.split(":")[0].strip()[:24] return f"{head[:34]}{(' / ' + name) if name else ''}" # ── ① 참조 선별 ────────────────────────────────────────────────────── def gate(issues: list, refs: list) -> dict: lines = ["CORRECTIONS:"] for i, it in enumerate(issues, 1): lines.append(f"{i}. {it.get('fix_en') or it.get('issue_ko') or ''}") lines.append("\nAVAILABLE REFERENCES:") for i, (label, _p) in enumerate(refs, 1): lines.append(f"{i}. {label}") body = { "systemInstruction": {"parts": [{"text": GATE_SYS + GATE_TAIL}]}, "contents": [{"role": "user", "parts": [{"text": "\n".join(lines)}]}], "generationConfig": {"response_mime_type": "application/json", "maxOutputTokens": 4096}, } d = post_json( f"https://generativelanguage.googleapis.com/v1beta/models/" f"{GATE_MODEL}:generateContent?key={env('GEMINI_API_KEY')}", body, {"Content-Type": "application/json"}) cands = d.get("candidates") or [] text = "".join(p.get("text", "") for p in ((cands[0].get("content") or {}).get("parts") or [])) return lenient_json(text) # ── ② 지시문 조립 ──────────────────────────────────────────────────── def build_prompt(issues: list, gated: dict) -> str: fix_head = (PACK / "fix_head.md").read_text("utf-8").strip() fix_tail = (PACK / "fix_tail.md").read_text("utf-8").strip() corrections = "CORRECTIONS:\n" + "\n".join( f"- {i.get('fix_en')}" for i in issues) parts = [fix_head, corrections] missing = [c.get("missing_entity_name") or "" for c in gated.get("corrections", []) if c.get("adds_missing_entity")] missing = [m for m in missing if m.strip()] if missing: parts.append(MISSING_HEAD + "\n" + "\n".join(f"- {m}" for m in missing) + "\n" + MISSING_TAIL) parts.append(fix_tail) return "\n\n".join(parts) # ── ③ 생성 ─────────────────────────────────────────────────────────── def gen(model_key: str, prompt: str, labeled: list) -> tuple: sys.path.insert(0, str(ROOT)) if model_key == "nb2": from app.modules.llm.gemini_image_client import GeminiImageClient cli = GeminiImageClient(model=NB2_MODEL) else: from app.modules.llm.grok_image_client import GrokImageClient cli = GrokImageClient(model=GROK_MODEL) return cli.generate_image(prompt, labeled_references=labeled) # ── 갤러리 ─────────────────────────────────────────────────────────── def esc(x): return H.escape(str(x if x is not None else "")) def build_html(state: dict) -> None: rows = [] for tag in TAGS: st = state.get(tag) if not st: continue gl = st.get("gate") or {} gate_rows = [] for i, it in enumerate(st.get("issues", []), 1): c = next((x for x in gl.get("corrections", []) if x.get("index") == i), {}) need = c.get("needs_ref_indices") or [] gate_rows.append( f"{i}{esc((it.get('severity') or '')[:8])}" f"{esc(it.get('issue_ko'))}" f"{'붙임 ' + ', '.join(map(str, need)) if need else '없음'}" f"{' · 새로 넣기: ' + esc(c.get('missing_entity_name')) if c.get('adds_missing_entity') else ''}" f"{esc(c.get('reason_ko'))}") refs_before = st.get("refs_all", []) kept = st.get("refs_kept", []) ref_list = "".join( f"
  • " f"{'붙임' if lb in kept else '뺌'} — {esc(short_label(lb))}
    • " for lb in refs_before) imgs = "".join( f"
    " f"
    {esc(cap)}
    " for cap, fn in st.get("images", []) if fn) rows.append( f"

    {tag}

    " f"

    지적 {len(st.get('issues', []))}건 · " f"기존 참조 {len(refs_before)}장 → 이번 {len(kept)}장

    " f"
    {imgs}
    " f"

    참조 선별

      {ref_list}
    " f"

    지적별 판단

    " "" + "".join(gate_rows) + "
    #무게지적참조이유
    " f"
    이번에 보낸 수정 지시문" f"
    {esc(st.get('prompt'))}
    ") tot_before = sum(len(state[t].get("refs_all", [])) for t in state) tot_after = sum(len(state[t].get("refs_kept", [])) for t in state) html = ( '\n' "수정본 참조 선별 시험 — nb2 vs grok\n\n" "

    수정본 참조 선별 시험 — nb2 vs grok

    \n" f"

    {len(state)}장면 · 붙인 참조 " f"{tot_before} → {tot_after}장 · 같은 지시문·같은 그림으로 두 모델에 각각 보냄

    \n" + "\n".join(rows)) OUT.mkdir(parents=True, exist_ok=True) (OUT / "index.html").write_text(html, encoding="utf-8") print(f"갤러리: {OUT / 'index.html'}") def main() -> None: ap = argparse.ArgumentParser() ap.add_argument("--html", action="store_true") args = ap.parse_args() OUT.mkdir(parents=True, exist_ok=True) (OUT / "img").mkdir(exist_ok=True) state_path = OUT / "state.json" state = (json.loads(state_path.read_text("utf-8")) if state_path.is_file() else {}) if args.html: build_html(state) return records = json.loads((RECIPE / "records.json").read_text("utf-8")) size_idx = _size_index() import shutil for tag in TAGS: rec = records.get(tag) if not rec: print(f"{tag}: 기록 없음 — 건너뜀") continue issues = (rec.get("critique") or {}).get("issues") or [] if not issues: print(f"{tag}: 지적 없음 — 건너뜀") continue print(f"\n=== {tag} · 지적 {len(issues)}건 ===") refs = resolve_refs(rec.get("refs") or [], size_idx) print(f" 되찾은 참조 {len(refs)}장") gl = gate(issues, refs) need_idx = sorted({n for c in gl.get("corrections", []) for n in (c.get("needs_ref_indices") or [])}) kept = [refs[i - 1] for i in need_idx if 1 <= i <= len(refs)] print(f" → 붙일 참조 {len(kept)}장 (기존 {len(refs)}장)") for c in gl.get("corrections", []): if c.get("adds_missing_entity"): print(f" ★새로 넣어야 함: {c.get('missing_entity_name')}") prompt = build_prompt(issues, gl) orig = RECIPE / f"{tag}_sel.png" fix_label = (PACK / "fix_label.md").read_text("utf-8").strip() ref_label = (PACK / "fix_ref_label.md").read_text("utf-8").strip() labeled = [(fix_label, orig.read_bytes())] labeled += [(f"{ref_label} {lb}", p.read_bytes()) for lb, p in kept] images = [] for cap, src in (("원본(선정본)", orig), ("기존 수정본", RECIPE / f"{tag}_fix.png")): if src.is_file(): fn = f"{tag}_{'sel' if 'sel' in src.name else 'oldfix'}.png" shutil.copy(src, OUT / "img" / fn) images.append((cap, fn)) for mk, cap in (("nb2", "새 수정본 — nb2"), ("grok", "새 수정본 — grok i2")): fn = f"{tag}_new_{mk}.png" dst = OUT / "img" / fn if dst.is_file(): images.append((cap, fn)) continue t0 = time.monotonic() try: png, _ms = gen(mk, prompt, labeled) dst.write_bytes(png) images.append((cap, fn)) print(f" {mk} ok {round(time.monotonic()-t0,1)}s") except Exception as exc: # noqa: BLE001 — 실측 기록 print(f" {mk} 실패 {type(exc).__name__}: {exc}"[:220]) images.append((cap + " (실패)", "")) state[tag] = { "issues": issues, "gate": gl, "refs_all": [lb for lb, _ in refs], "refs_kept": [lb for lb, _ in kept], "prompt": prompt, "images": images, "ts": datetime.now(timezone.utc).isoformat(), } state_path.write_text( json.dumps(state, ensure_ascii=False, indent=1), "utf-8") build_html(state) if __name__ == "__main__": main()