"""중요한 샷을 다시 만든다 — 재생성 실험 (2026-08-06 사용자 지시). `opus_pilot.py` 가 남긴 이중 선정 결과를 입력으로 받는다. **대상 선별을 critique 가 아니라 선정 결과에서 뽑는 이유.** 처음엔 critique 의 `severity=critical` 을 재생성 트리거로 썼는데 7샷 전부에서 critical 0 이 나와 한 건도 발동하지 않았다. 수정을 억제하는 문구를 강하게 쓴 결과 critique 가 지나치게 보수적이 된 것이다. 반면 **선정 단계에서는 두 심판이 후보마다 하드 위반을 이미 적어 놓았다** — 선정된 후보에 위반이 남아 있다는 것은 "최선을 골랐는데도 결함이 있다"는 뜻이고, 그게 다시 찍을 근거다. 한 모델의 심각도 판단보다 두 모델이 각자 본 위반이 견고하다. 흐름 ① 대상 선별 선정본에 하드 위반이 남은 샷 (위반 수 내림차순) ② 교정절 Opus 가 저작한다. 브리프 + 선정본 + **두 심판의 위반 목록**을 주고, 이번에는 무엇이 참이어야 하는지를 긍정형으로 쓰게 한다. (사용자 지시 — 수정 요소는 Opus 가 지시한다) ③ 재생성 원 롤 프롬프트 + 교정절 + 원 참조로 다시 찍는다. ★결함 이미지는 참조로 주지 않는다 — 주면 그 기하를 물려받는다. ④ 재판정 [원본 vs 재생성본] 블라인드 정순+역순을 **두 심판 모두**에게. 4표 중 2표 이상이면 새것 채택. 개악이면 원본을 지킨다. 사용 .venv/bin/python opus_remake.py <선정결과.json> [--top N] """ from __future__ import annotations import argparse import json import re import shutil import sys import threading import traceback from concurrent.futures import ThreadPoolExecutor, as_completed from pathlib import Path from typing import Any, Dict, List sys.path.insert(0, str(Path(__file__).resolve().parent)) from opus_pilot import ( # noqa: E402 JUDGES_DEFAULT, JUDGE_LABEL, RECIPE, REJUDGE_SCHEMA, REJUDGE_SYS, _ask, build_size_index, resolve_refs, ) from app.modules.pipeline.multiroll_gemini import ( # noqa: E402 make_nb2_gen_fn, png_part, ref_parts, ) from app.modules.pipeline.multiroll_select import ( # noqa: E402 _compose_critique_prompt, ) PROJ = "e716bafb-24bb-42b7-aea0-fdb383844ee8" EPI = "d6a9aa85-b75e-400c-980c-4ee7e876a15b" DIRECTOR = "claude-opus" # 수정 지시는 Opus 전담 (사용자 확정) REMAKE_SYS = """\ A film still was chosen as the best of several attempts, and two independent reviewers still found faults in it. You are writing the note that goes to the photographer for the reshoot. You will see the original brief, the chosen still, and the reviewers' findings. Some findings will be wrong or trivial; you are the director, not a clerk. Decide which faults are actually worth reshooting for, and say so. Write `regen_clause_en` as instructions appended to the original brief before it is shot again: - Phrase everything positively — state what must be true in the new frame, not what was wrong in the old one. The photographer will not see the old frame, so "remove the second steering wheel" means nothing; "the cabin has exactly one steering wheel, directly in front of the driver" does. - Be specific about counts, seating, who faces where, and what the camera sees, because those are the faults that survive a retouch. - Where a prop's quantity, grade or denomination is what makes the moment read — money that is supposed to tempt someone, a wound that is supposed to be fatal, a crowd that is supposed to be a mob — say what it has to be, and say it in the terms of the era and place the brief establishes. Do not leave that to chance: given only a category, the image model reaches for the most ordinary member of it, and the most ordinary banknote will not justify a greedy smile. Read the amount off the story, not off a rule. - When you forbid something from appearing (text on a surface, an object, a person), also say what occupies that space instead. A bare prohibition tends to be ignored; a positive description of the same area is followed. - Do not restate what the brief already says. Add only what the brief left ambiguous enough to be got wrong. - Do not ask for a different camera angle or a different moment than the brief specifies. Set `worth_remaking` false when the surviving faults are cosmetic or when the brief itself is what forced the fault (for example, a brief that forbids any body part in frame will keep producing a phone with no hand — that is not something a reshoot can fix). In that case leave `regen_clause_en` empty and say why in `reason_ko`; also name what would have to change upstream in `upstream_fix_ko`. Answer in Korean for `reason_ko` and `upstream_fix_ko`; English for `regen_clause_en`.""" REMAKE_SCHEMA = { "type": "object", "properties": { "worth_remaking": {"type": "boolean"}, "faults_taken": {"type": "array", "items": {"type": "string"}}, "faults_dismissed": {"type": "array", "items": {"type": "string"}}, "reason_ko": {"type": "string"}, "regen_clause_en": {"type": "string"}, "upstream_fix_ko": {"type": "string"}, }, "required": ["worth_remaking", "faults_taken", "faults_dismissed", "reason_ko", "regen_clause_en", "upstream_fix_ko"], "additionalProperties": False, } def stem_key(s: str): m = re.findall(r"\d+", s) return (int(m[0]), int(m[1])) if len(m) >= 2 else (9999, 9999) def run_one(stem: str, prior: Dict[str, Any], rec: Dict[str, Any], size_idx, outdir: Path, gen_fn) -> Dict[str, Any]: sel = prior["select"] winner = sel["winner"] orig = RECIPE / f"{stem}_{winner.lower()}.png" faults = sel.get("violations", {}).get(winner, []) roll_prompts = rec.get("roll_prompts") or {} brief = _compose_critique_prompt( rec.get("prompt", ""), roll_prompts, winner, shared_prompt=None) refs, _ = resolve_refs( (rec.get("roll_refs") or {}).get(winner) or rec.get("refs"), size_idx) # ② 교정절 저작 — Opus parts: List[Dict[str, Any]] = [ {"type": "text", "text": "THE ORIGINAL BRIEF:\n" + brief}] parts += ref_parts(refs) parts.append({"type": "text", "text": "The chosen still:"}) parts.append(png_part(orig)) parts.append({"type": "text", "text": "REVIEWERS' FINDINGS ON THIS STILL:\n" + "\n".join(f"- {f}" for f in faults)}) plan = _ask(f"remake_plan_{stem}", REMAKE_SYS, parts, REMAKE_SCHEMA, model=DIRECTOR) out: Dict[str, Any] = { "stem": stem, "selected_roll": winner, "orig": str(orig), "faults_in": faults, "plan": plan, } if not (plan.get("worth_remaking") and plan.get("regen_clause_en", "").strip()): out["remade"] = False return out # ③ 재생성 — 결함 이미지는 주지 않는다 regen_prompt = ((roll_prompts.get(winner) or rec.get("prompt", "")).strip() + "\n\n" + plan["regen_clause_en"].strip()) out["regen_prompt"] = regen_prompt new = outdir / f"{stem}_remake.png" try: gen_fn(f"remake_{stem}", regen_prompt, refs, new) except Exception as exc: # noqa: BLE001 out["remade"] = False out["error"] = repr(exc) return out # ④ 재판정 — 두 심판 모두, 정순+역순 def pair(a: Path, b: Path, tag: str, model: str): pp: List[Dict[str, Any]] = [ {"type": "text", "text": "THE BRIEF:\n" + brief}] pp += ref_parts(refs) pp.append({"type": "text", "text": "Candidate A:"}) pp.append(png_part(a)) pp.append({"type": "text", "text": "Candidate B:"}) pp.append(png_part(b)) return _ask(tag, REJUDGE_SYS, pp, REJUDGE_SCHEMA, model=model) votes, detail = 0, {} for m in JUDGES_DEFAULT: mk = m.replace("-", "") f = pair(orig, new, f"remake_rj_{stem}_{mk}", m) r = pair(new, orig, f"remake_rj_rev_{stem}_{mk}", m) v = int(f["winner"] == "B") + int(r["winner"] == "A") votes += v detail[m] = {"forward": f, "reverse": r, "new_votes": v} total = 2 * len(JUDGES_DEFAULT) new_won = votes * 2 >= total # 4표 중 2표 이상 = 새것 채택 out.update({ "remade": True, "new": str(new), "rejudge": {"detail": detail, "new_votes": votes, "total": total, "new_won": bool(new_won)}, }) final = new if new_won else orig shutil.copy(final, outdir / f"{stem}_final.png") out["final"] = str(outdir / f"{stem}_final.png") out["final_is_new"] = bool(new_won) return out def main() -> None: ap = argparse.ArgumentParser() ap.add_argument("prior") ap.add_argument("out") ap.add_argument("--top", type=int, default=0, help="위반 수 상위 N개만 (0=위반 있는 전부)") ap.add_argument("--stems", default="", help="특정 샷만, 쉼표 구분") ap.add_argument("--workers", type=int, default=3) args = ap.parse_args() prior = json.loads(Path(args.prior).read_text("utf-8")) records = json.loads((RECIPE / "records.json").read_text("utf-8")) # ① 대상 선별 — 선정본에 하드 위반이 남은 샷 cand = [] for stem, v in prior.items(): sel = v.get("select") if not sel or stem not in records: continue n = len(sel.get("violations", {}).get(sel["winner"], [])) if n: cand.append((n, stem)) cand.sort(key=lambda x: (-x[0], stem_key(x[1]))) if args.top: cand = cand[:args.top] stems = [s for _, s in cand] if args.stems: want = {s.strip() for s in args.stems.split(",") if s.strip()} stems = [s for s in stems if s in want] or sorted( want & set(prior) & set(records), key=stem_key) out = Path(args.out) outdir = out.parent / "remake" outdir.mkdir(parents=True, exist_ok=True) size_idx = build_size_index() gen_fn = make_nb2_gen_fn(project_id=PROJ, episode_id=EPI, operation_type="opus_remake") print(f"대상 {len(stems)}샷 (선정본에 하드 위반 잔존) · " f"지시={JUDGE_LABEL[DIRECTOR]} · " f"재판정={' + '.join(JUDGE_LABEL[m] for m in JUDGES_DEFAULT)}", flush=True) results: Dict[str, Any] = {} lock, done = threading.Lock(), [0] with ThreadPoolExecutor(max_workers=args.workers) as ex: futs = {ex.submit(run_one, s, prior[s], records[s], size_idx, outdir, gen_fn): s for s in stems} for f in as_completed(futs): s = futs[f] try: results[s] = f.result() except Exception as exc: # noqa: BLE001 results[s] = {"stem": s, "error": repr(exc), "trace": traceback.format_exc()[-800:]} with lock: done[0] += 1 ok = sum(1 for v in results.values() if v.get("final_is_new")) print(f" {done[0]}/{len(stems)} 새것 채택 {ok}", flush=True) out.write_text(json.dumps(results, ensure_ascii=False, indent=1), "utf-8") out.write_text(json.dumps(results, ensure_ascii=False, indent=1), "utf-8") made = [k for k, v in results.items() if v.get("remade")] took = [k for k, v in results.items() if v.get("final_is_new")] kept = [k for k, v in results.items() if v.get("remade") and not v.get("final_is_new")] skip = [k for k, v in results.items() if not v.get("remade") and "error" not in v] print(f"\n완료 — 대상 {len(stems)}") print(f" 다시 찍음 : {len(made)}") print(f" ★새것 채택 : {len(took)} {took}") print(f" 개악 → 원본 유지 : {len(kept)} {kept}") print(f" 다시 찍지 않음 : {len(skip)} {skip}") if __name__ == "__main__": main()