"""최종 256샷 갤러리 + Gemini·Qwen 이중 재평가 (#64).
각 샷에 대해:
· 후보 롤(a/b/c…) 과 프로덕션 선택 — records.json 에서
· 수정사항 — 결함 검사 내용·수정 적용 여부·수정 재판정 결과
· **재평가**: Gemini(gemini-pro, Router 구조화 호출)와 Qwen(3.8 Max,
DashScope 우회)이 같은 계약(opus_pilot 의 SELECT_SYS + select_schema,
같은 브리프·같은 참조·같은 후보 제시)으로 후보를 다시 고른다.
· 지목 25건 태그 — 2026-08-07 육안 판정 문서의 샷 목록과 대조
돈 주의: 판정 호출만(이미지 생성 없음). 재실행하면 이미 판정된 샷은
건너뛴다(결과 json 에 있으면 skip — 증분 저장).
사용:
.venv/bin/python build_final_eval_gallery.py # 판정 + 갤러리
.venv/bin/python build_final_eval_gallery.py --html # 갤러리만 재생성
.venv/bin/python build_final_eval_gallery.py --limit 6 # 시험 소량
"""
from __future__ import annotations
import argparse
import html as _html
import json
import sys
import threading
import traceback
from concurrent.futures import ThreadPoolExecutor, as_completed
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent))
from opus_pilot import ( # noqa: E402
EPI, PROJ, RECIPE, ROOT, SELECT_SYS, build_size_index, select_schema,
)
from qwen_vlm_pilot import ask_qwen, build_parts # noqa: E402
OUT_DIR = ROOT / "artifact" / "20260809_final_eval_gallery"
RESULTS = OUT_DIR / "results.json"
INDEX = OUT_DIR / "index.html"
WORKERS = 6
# 2026-08-07 육안 판정 25건이 걸린 샷(문서에서 뽑은 태그 23개 —
# 일부 지목은 같은 샷을 두 번 다룬다)
FLAGGED = {
"S1sh7", "S2sh5", "S13sh3", "S13sh4", "S15sh3", "S15sh5", "S18sh1",
"S18sh5", "S42sh4", "S57sh3", "S61sh12", "S62sh4", "S73sh4", "S74sh8",
"S76sh5", "S79sh1", "S88sh3", "S88sh6", "S93sh5", "S93sh8", "S97sh5",
"S98sh6", "S104sh7",
}
# 이번 JIT 재실행에서 다시 만들어진 샷
REGEN_TODAY = {"S1sh5", "S1sh7", "S96sh1", "S96sh4", "S98sh2", "S98sh5",
"S98sh6"}
def tag_sort_key(tag: str):
s, sh = tag[1:].split("sh")
return (int(s), int(sh))
def load_records() -> dict:
return json.loads((RECIPE / "records.json").read_text("utf-8"))
def current_tags() -> set:
"""지금 선정된 스틸의 태그만 — records 에는 이전 선정분(스테일)도
남아 있어 그대로 돌면 스테일 판정에 돈이 샌다."""
import subprocess
out = subprocess.run(
["psql", "-h", "localhost", "-U", "theroad", "-d", "theroad",
"-t", "-A", "-c",
"SELECT scene_index||'sh'||shot_index FROM scene_still "
f"WHERE episode_id='{EPI}' AND is_selected AND still_index>=0 "
"AND status!='stale';"],
capture_output=True, text=True,
env={"PGPASSWORD": "theroad_dev_2026", "PATH": "/usr/bin:/bin:"
"/usr/local/bin:/opt/homebrew/bin"},
)
tags = {f"S{line.strip()}" for line in out.stdout.splitlines()
if line.strip()}
if not tags:
raise RuntimeError(f"선정 스틸 조회 실패: {out.stderr[:200]}")
return tags
def judge_gemini(stem: str, parts: list, labels: list) -> dict:
from app.modules.llm.llm_client import call_structured
return call_structured(
"gallery_reselect",
SELECT_SYS,
parts,
select_schema(labels),
project_config={"gallery_reselect": {"model": "gemini-pro"}},
schema_name="reselect",
opik_metadata={"operation_type": "gallery_reselect", "shot_tag": stem,
"project_id": PROJ, "episode_id": EPI},
enable_fallback=False,
)
def run_judges(limit: int = 0) -> None:
records = load_records()
size_idx = build_size_index()
done: dict = {}
if RESULTS.exists():
done = json.loads(RESULTS.read_text("utf-8"))
live = current_tags()
tags = sorted(
(t for t, r in records.items()
if isinstance(r, dict) and "::" not in t and r.get("selected")
and t in live),
key=tag_sort_key)
jobs = [t for t in tags
if done.get(t, {}).get("gemini") is None
or done.get(t, {}).get("qwen") is None]
if limit:
jobs = jobs[:limit]
print(f"판정 대상 {len(jobs)} / 전체 {len(tags)} (이미 완료 "
f"{len(tags) - len(jobs)})", flush=True)
lock = threading.Lock()
count = [0]
def one(stem: str):
rec = records[stem]
parts, labels = build_parts(stem, rec, size_idx)
entry = done.get(stem) or {}
if not labels:
return stem, {"skipped": "후보 롤 파일 없음"}
if entry.get("gemini") is None:
try:
entry["gemini"] = judge_gemini(stem, parts, labels)
except Exception as exc: # noqa: BLE001
entry["gemini_error"] = repr(exc)[:300]
if entry.get("qwen") is None:
try:
payload, meta = ask_qwen(SELECT_SYS, parts,
select_schema(labels))
entry["qwen"] = payload
entry["qwen_meta"] = {
"attempts": len(meta.get("attempts", []))}
except Exception as exc: # noqa: BLE001
entry["qwen_error"] = repr(exc)[:300]
entry["labels"] = labels
return stem, entry
OUT_DIR.mkdir(parents=True, exist_ok=True)
with ThreadPoolExecutor(max_workers=WORKERS) as ex:
futs = {ex.submit(one, t): t for t in jobs}
for f in as_completed(futs):
stem = futs[f]
try:
k, v = f.result()
done[k] = v
except Exception: # noqa: BLE001
done[stem] = {"error": traceback.format_exc()[-400:]}
with lock:
count[0] += 1
if count[0] % 5 == 0 or count[0] == len(jobs):
RESULTS.write_text(
json.dumps(done, ensure_ascii=False, indent=1),
"utf-8")
g_ok = sum(1 for v in done.values() if v.get("gemini"))
q_ok = sum(1 for v in done.values() if v.get("qwen"))
print(f" {count[0]}/{len(jobs)} G성공 {g_ok} "
f"Q성공 {q_ok}", flush=True)
RESULTS.write_text(json.dumps(done, ensure_ascii=False, indent=1),
"utf-8")
# ── 갤러리 ──────────────────────────────────────────────────────────
def esc(s) -> str:
return _html.escape(str(s if s is not None else ""))
def img_url(name: str) -> str:
return f"/projects/{PROJ}/images/{EPI}/scene/recipe/{name}"
def render(records: dict, results: dict) -> str:
rows = []
live = current_tags()
tags = sorted(
(t for t, r in records.items()
if isinstance(r, dict) and "::" not in t and r.get("selected")
and t in live),
key=tag_sort_key)
n = {"g_agree": 0, "q_agree": 0, "gq_agree": 0, "g_ok": 0, "q_ok": 0,
"g_fail_flag": 0, "q_fail_flag": 0, "fixed": 0}
for tag in tags:
rec = records[tag]
res = results.get(tag) or {}
sel = str(rec.get("selected", "")).upper()
g = res.get("gemini") or {}
q = res.get("qwen") or {}
g_win = str(g.get("winner", "")).upper()
q_win = str(q.get("winner", "")).upper()
if g_win:
n["g_ok"] += 1
n["g_agree"] += int(g_win == sel)
n["g_fail_flag"] += int(bool(g.get("all_candidates_fail")))
if q_win:
n["q_ok"] += 1
n["q_agree"] += int(q_win == sel)
n["q_fail_flag"] += int(bool(q.get("all_candidates_fail")))
if g_win and q_win:
n["gq_agree"] += int(g_win == q_win)
crit = rec.get("critique") or {}
issues = [i.get("issue_ko", "") for i in (crit.get("issues") or [])]
fixed = "fix_prompt" in rec
n["fixed"] += int(fixed)
rj = rec.get("fix_rejudge") or {}
labels = res.get("labels") or sorted(
p.stem.split("_")[-1].upper()
for p in RECIPE.glob(f"{tag}_[abc].png"))
cls = []
disagree = (g_win and g_win != sel) or (q_win and q_win != sel)
if disagree:
cls.append("disagree")
if tag in FLAGGED:
cls.append("flagged")
if tag in REGEN_TODAY:
cls.append("regen")
def reading_block(name, payload, err):
if err:
return (f"
{name} 오류: "
f"{esc(err)}
")
if not payload:
return f"
{name} 결과 없음
"
win = str(payload.get("winner", "")).upper()
mark = "일치" if win == sel else f"불일치(프로덕션 {sel})"
fail = (" · 전 후보 실패 판정"
if payload.get("all_candidates_fail") else "")
body = [f"
{name} → {esc(win)} "
f""
f"{mark}{fail} {esc(payload.get('why', ''))}"]
for r in payload.get("readings", []):
body.append(
f"{esc(r.get('label'))} — "
f"{esc(r.get('score'))}점, 위반 "
f"{len(r.get('hard_violations') or [])}건"
f"
방향: {esc(r.get('direction'))}
"
f"
구조: {esc(r.get('built_space'))}
"
f"
대상: {esc(r.get('entities'))}
"
f"
위반: {esc('; '.join(r.get('hard_violations') or []) or '없음')}
"
f"")
body.append("
")
return "".join(body)
thumbs = []
for lab in labels:
badges = []
if lab == sel:
badges.append("선택")
if lab == g_win:
badges.append("G")
if lab == q_win:
badges.append("Q")
thumbs.append(
f""
f"{lab} {''.join(badges)}")
final_imgs = (
f""
f"최종(_sel{'·수정 반영' if fixed else ''})"
f"")
fix_html = ""
if issues or fixed:
rj_txt = ""
if rj:
rj_txt = (" · 수정 재판정: "
+ ("수정본 승" if rj.get("winner") == "B"
else f"원본 승({esc(rj.get('winner'))})"))
fix_html = (
"
수정사항 — 결함 "
f"{len(issues)}건{', 수정 적용' if fixed else ', 수정 없음'}"
f"{rj_txt}
"
+ "".join(f"
{esc(i)}
" for i in issues)
+ "
")
tags_html = "".join(
[("지목 25건" if tag in FLAGGED else ""),
("오늘 재생성"
if tag in REGEN_TODAY else "")])
rows.append(
f""
f"