"""
def esc(x) -> str:
return H.escape(str(x if x is not None else ""))
def q(sql: str) -> list[list[str]]:
out = subprocess.run(
["psql", "-h", "localhost", "-U", "theroad", "-d", "theroad",
"-t", "-A", "-F", "\x1f", "-c", sql],
env={"PGPASSWORD": "theroad_dev_2026",
"PATH": "/usr/local/bin:/opt/homebrew/bin:/usr/bin:/bin"},
capture_output=True, text=True, check=True)
return [line.split("\x1f") for line in out.stdout.splitlines() if line]
def tag_key(tag: str):
s, sh = tag[1:].split("sh")
return (int(s), int(sh))
def scan_files() -> dict[str, dict[str, str]]:
files: dict[str, dict[str, str]] = {}
for p in sorted(RECIPE.iterdir()):
m = re.match(
r"^(S\d+sh\d+)(__bgfirst_bg|_confinedfp|_fix|_sel|_cine|_([a-e]))\.png$",
p.name)
if not m:
continue
kind = m.group(3) or m.group(2).lstrip("_")
files.setdefault(m.group(1), {})[kind] = p.name
return files
# 검열 거부 판별 — 프로덕션과 같은 함수를 쓴다(글자 목록을 따로 두면 갈린다).
sys.path.insert(0, str(Path(__file__).resolve().parent))
from app.modules.llm.image_moderation import is_moderation_text # noqa: E402
def build_index(shots, segs, sel_of, files, manifest) -> int:
body = []
total = 0
for si in sorted(shots):
seg = segs.get(si) or {}
heading = (seg.get("heading") or f"S#{si}.").strip()
title = re.sub(r"^S#\d+\.\s*", "", heading)
text = (seg.get("text") or "").strip()
cells = []
for shi, beat, desc, var in shots[si]:
tag = f"S{si}sh{shi}"
f = files.get(tag) or {}
src = f.get("cine") or f.get("sel")
if not src:
continue
total += 1
manifest.append((RECIPE / src, f"images/{tag}_sel.png"))
sel = var or sel_of.get(tag) or ""
sel_txt = f" · 선정 롤 {esc(sel)}" if sel else ""
cine_txt = " · 시네마틱 변환본" if f.get("cine") else " · 원본(변환 미적용)"
cells.append(
f"
"
+ "\n".join(body) + MODAL + "\n\n")
(OUT_DIR / "index.html").write_text(html, "utf-8")
return total
def build_process(files, records, manifest) -> None:
main_recs = {k: v for k, v in records.items()
if "::" not in k and isinstance(v, dict)}
cine = {k.split("::")[0]: v for k, v in records.items()
if k.endswith("::cine") and isinstance(v, dict)}
tags = sorted(set(main_recs) | set(files), key=tag_key)
routes = Counter()
n_fix = n_cine = n_cine_fail = n_cine_declined = 0
rows: list[str] = []
cur_scene = None
for tag in tags:
r = main_recs.get(tag) or {}
f = files.get(tag) or {}
si = tag_key(tag)[0]
if si != cur_scene:
cur_scene = si
rows.append(f"
S{si}
")
gq = r.get("gq") or {}
if not gq:
jf = r.get("judge_flip") or {}
gq = (jf.get("forward_raw") or {}).get("gq") or {}
route = gq.get("route") or ""
if route:
routes[route] += 1
sel = r.get("selected") or "?"
fixed = bool(r.get("repair_mode") or f.get("fix"))
if fixed:
n_fix += 1
cn = cine.get(tag) or {}
cine_ok = bool(cn.get("applied")) and bool(f.get("cine"))
# 2026-08-20: 포기(declined)는 **닫힌 결말**이지 실패가 아니다 —
# 검열 거부가 기준만큼 쌓여 변환을 그만두고 원본을 최종본으로
# 확정한 상태다. 실패로 그리면 아직 고칠 것이 남은 판으로 읽힌다.
# 포기 표식이 붙기 시작한 것은 2026-08-20 이다. 그 전에 버려진 샷은
# applied=false + 검열 거부 원문만 남아 있어, 표식만 보면 「아직 고칠
# 것이 남았다」로 읽힌다. 옛 모양도 사실대로 읽는다 — 판별은 프로덕션이
# 쓰는 것과 **같은 공용 함수**다(따로 적으면 둘이 갈린다).
cine_declined = bool(cn.get("declined")) or (
bool(cn) and not cn.get("applied")
# 포기 셈이 들어오기 전(2026-08-20 이전)에 버려진 것만 이 갈래다.
# 그 뒤 기록에는 셈(moderation_refusals)이 있고, 셈이 있는데
# declined 가 없으면 **아직 기준에 못 미쳐 다시 시도할 샷**이다
# — 그것을 포기로 그리면 결말이 아닌 것을 결말로 읽는다.
and "moderation_refusals" not in cn
and is_moderation_text(str(cn.get("error") or "")))
cine_fail = bool(cn) and not cn.get("applied") and not cine_declined
if cine_ok:
n_cine += 1
if cine_fail:
n_cine_fail += 1
if cine_declined:
n_cine_declined += 1
badges = [f"선정 {esc(sel)}"]
if route:
badges.append(f"판정 {esc(route)}")
if fixed:
badges.append("수정")
if cine_ok:
badges.append("시네마틱 변환")
if cine_fail:
badges.append("변환 실패(원본 fallback)")
if cine_declined:
badges.append(
"변환 포기 — 원본 확정"
"(검열 거부 누적)")
figs = []
def fig(kind: str, label: str, hl: bool = False):
name = f.get(kind)
if not name:
return
jpg = f"procimg/{tag}_{kind}.png"
manifest.append((RECIPE / name, jpg))
style = " style='outline:3px solid #2a7'" if hl else ""
figs.append(
f""
f"{esc(label)}")
fig("bgfirst_bg", "배경(bgfirst)")
fig("confinedfp", "confined 도면")
for roll in ROLL_ORDER:
fig(roll, f"롤 {roll.upper()}", hl=(sel == roll.upper()))
fig("fix", "수정본")
fig("sel", "선정")
fig("cine", "최종 — 시네마틱 변환", hl=cine_ok)
det = []
if gq:
det.append(
f"
"
f"{''.join(det)}")
head = (
"\n"
"마지막 임무 — 전 과정 갤러리\n"
"\n")
summary = (
f"
마지막 임무 — 전 과정 갤러리 ({len(tags)}샷)
"
f"
판정 route: {dict(routes)} · 수정 {n_fix}샷 · 시네마틱 변환 {n_cine}"
+ (f" · 변환 미적용 {n_cine_fail}" if n_cine_fail else "")
+ (f" · 변환 포기(원본 확정) {n_cine_declined}"
if n_cine_declined else "")
+ " · 최종 리포트로 →
\n")
(OUT_DIR / "process.html").write_text(head + summary + "\n".join(rows),
"utf-8")
def main() -> None:
rows = q(
"SELECT scene_index, shot_index, coalesce(beat_title,''),"
" coalesce(shot_description,''), coalesce(selected_variant,'')"
f" FROM scene_still WHERE episode_id='{EPI}' AND is_selected=true"
" ORDER BY scene_index, shot_index;")
shots = defaultdict(list)
for si, shi, beat, desc, var in rows:
shots[int(si)].append((int(shi), beat, desc, var))
segs = {s["scene_index"]: s for s in
(json.loads(CP_SAVE.read_text("utf-8")).get("data") or {})
.get("segments", [])}
records = json.loads((RECIPE / "records.json").read_text("utf-8"))
sel_of = {k: v.get("selected") for k, v in records.items()
if "::" not in k and isinstance(v, dict)}
files = scan_files()
OUT_DIR.mkdir(parents=True, exist_ok=True)
(OUT_DIR / "images").mkdir(exist_ok=True)
(OUT_DIR / "procimg").mkdir(exist_ok=True)
manifest: list[tuple[Path, str]] = []
total = build_index(shots, segs, sel_of, files, manifest)
build_process(files, records, manifest)
# 이미지 복사 목록 — 원본 PNG 그대로(압축 없음)
lst = OUT_DIR / "_convert_list.tsv"
lst.write_text(
"\n".join(f"{src}\t{OUT_DIR / rel}" for src, rel in manifest),
"utf-8")
print(f"완료: index {total}샷 · 복사 대상 {len(manifest)}장 → {lst}")
if __name__ == "__main__":
main()
"
f"