"""s21 참조 추적 — 취합된 실외 샷 스틸 40장에 실제 투입된 참조 이미지 표시. - 입력: plans/outdoor_shots_collect.json (s21_shot_collect 산출). - SOT: image_asset.generation_call_id → llm_call_log.reference_image_ids (실첨부 역할 라벨) + image_asset.input_image_ids (해석된 UUID lineage). 두 리스트 길이 일치(40/40 사전 검증)일 때만 인덱스 페어링 — 라벨 의미 파싱 0 (기존 Phase D overlay 패턴 재사용). - 산출: out/shots_outdoor/refs/ 참조 이미지 복사(dedup) + plans/outdoor_shots_refs.json + shots_outdoor.html 재빌드 (이후 페이지 SOT=이 스크립트, s21_shot_collect 의 기본 페이지를 대체). - 읽기 전용(DB write 0), LLM/이미지 생성 0, 커밋 금지(scratchpad). 사용: backend/.venv/bin/python s21_shot_refs.py """ import html as html_mod import json import shutil import sys from datetime import datetime, timedelta from pathlib import Path sys.path.insert(0, str(Path(__file__).parent)) import forest_lib as F # noqa: E402 import psycopg2 # noqa: E402 ROOT = Path("/Users/manta/Documents/Projects/TheRoad-I1") OUT = F.OUT / "shots_outdoor" REFS_DIR = OUT / "refs" PAGE = Path(__file__).parent / "shots_outdoor.html" def _kst(iso): if not iso: return "" dt = datetime.fromisoformat(iso) return (dt + timedelta(hours=9)).strftime("%m-%d %H:%M") def fetch(collect): aids = [im["asset_id"] for s in collect["shots"].values() for im in s["images"]] conn = psycopg2.connect(host="localhost", user="theroad", password="theroad_dev_2026", dbname="theroad") cur = conn.cursor() cur.execute( """ SELECT ia.id, ia.created_at, ia.input_image_ids, cl.reference_image_ids, cl.metadata_json FROM image_asset ia LEFT JOIN llm_call_log cl ON cl.id = ia.generation_call_id WHERE ia.id = ANY(%s) """, (aids,), ) stills = {} for aid, created, iids, labels, meta in cur.fetchall(): ii = json.loads(iids or "[]") lb = json.loads(labels or "[]") md = json.loads(meta or "{}") stills[aid] = { "created_at": created, "input_image_ids": ii, "ref_labels": lb, "labels_paired": len(ii) == len(lb), "beat_title": md.get("beat_title"), } ref_ids = sorted({r for s in stills.values() for r in s["input_image_ids"]}) refs = {} if ref_ids: cur.execute( """ SELECT id, file_path, pipeline_role, asset_type, entity_id FROM image_asset WHERE id = ANY(%s) """, (ref_ids,), ) for rid, fp, role, atype, ent in cur.fetchall(): refs[rid] = {"file_path": fp, "pipeline_role": role, "asset_type": atype, "entity_id": ent} ents = sorted({r["entity_id"] for r in refs.values() if r["entity_id"]}) if ents: cur.execute("SELECT id, short_id, name FROM entity_canon" " WHERE id = ANY(%s)", (ents,)) emap = {i: {"short_id": s, "name": n} for i, s, n in cur.fetchall()} for r in refs.values(): if r["entity_id"] in emap: r["entity"] = emap[r["entity_id"]] conn.close() return stills, refs def copy_refs(refs): REFS_DIR.mkdir(parents=True, exist_ok=True) unresolved = [] for rid, r in refs.items(): src = ROOT / r["file_path"] if not src.exists(): unresolved.append(rid) continue dst = REFS_DIR / f"{r['pipeline_role']}_{rid[:8]}.png" if not dst.exists(): shutil.copy2(src, dst) r["file"] = f"refs/{dst.name}" return unresolved def build_page(collect, stills, refs): def _esc(t): return html_mod.escape(str(t or "")) shots = collect["shots"] secs = [] for k in sorted(shots, key=lambda x: (shots[x]["scene_index"], shots[x]["shot_index"])): s = shots[k] imgs = list(s["images"]) imgs.sort(key=lambda im: stills.get(im["asset_id"], {}) .get("created_at") or "", reverse=True) imgs = imgs[:1] # 샷당 최신 1장만 cells = [] for im in imgs: st = stills.get(im["asset_id"], {}) badge = "" beat = st.get("beat_title") beat_html = (f"

{_esc(beat)}

" if beat else "") ref_cells = [] for i, rid in enumerate(st.get("input_image_ids", [])): r = refs.get(rid) lbls = st.get("ref_labels", []) lbl = (lbls[i] if st.get("labels_paired") and i < len(lbls) else "") if not r or "file" not in r: ref_cells.append( f"

미해석" f" {_esc(rid[:8])}

" f"

{_esc(lbl)}

") continue ent = r.get("entity") ent_txt = (f"{ent['short_id']} {ent['name']}" if ent else "") ref_cells.append( f"
" f"" f"

{_esc(r['pipeline_role'])}" f"{(' · ' + _esc(ent_txt)) if ent_txt else ''}

" f"

{_esc(lbl)}

") refs_html = (f"
{''.join(ref_cells)}
" if ref_cells else "

(참조 없음 — 프롬프트만)

") cells.append( f"
" f"" f"

{badge}{_esc(im['file'])}" f" · {_esc(_kst(st.get('created_at')))}

" f"{beat_html}" f"

투입 참조 {len(ref_cells)}장

" f"{refs_html}
") secs.append( f"

{_esc(k)} — {_esc(s.get('scene_heading'))}" f" (loc={','.join(s['loc_ids'])})

" f"

{_esc(s.get('description'))}

" f"
{''.join(cells) or '(생성 스틸 없음)'}
") n_stills = sum(1 for s in shots.values() if s["images"]) PAGE.write_text(f""" 실외 샷 최신 스틸 + 투입 참조 (옥탑 유닛 그룹)

옥탑방 유닛(빌라 포함) 실외 샷 — 샷별 최신 스틸 1장 + 투입 참조 ({len(shots)}샷, 표시 {n_stills}장)

스틸=DB scene_still×image_asset(pipeline_role=scene_still, status=generated) 중 샷별 created_at 최신 1장. 투입 참조=image_asset.input_image_ids(UUID lineage) + llm_call_log.reference_image_ids(실첨부 역할 라벨, 인덱스 페어링 — 길이 일치 검증 통과분만). 매핑=plans/outdoor_shots_refs.json

{''.join(secs)} """, encoding="utf-8") def main(): collect = json.loads( (Path(__file__).parent / "plans" / "outdoor_shots_collect.json") .read_text(encoding="utf-8")) stills, refs = fetch(collect) unresolved = copy_refs(refs) F.save_plan("outdoor_shots_refs", { "stills": stills, "refs": refs, "unresolved_files": unresolved}) build_page(collect, stills, refs) n_pairs = sum(1 for s in stills.values() if s["labels_paired"]) print(f"스틸 {len(stills)}장 (라벨 페어링 {n_pairs}/{len(stills)}), " f"고유 참조 {len(refs)}장 -> {REFS_DIR}") if unresolved: print(f"!! 참조 파일 누락 {len(unresolved)}건:", unresolved) print(f"page -> {PAGE}") if __name__ == "__main__": main()