"""s21 사전 취합 — 옥탑방 유닛(빌라 포함) 관련 **실외** 샷의 생성 스틸 전수 취합. - 대상: recon(bg_rooftop_unit)의 is_indoor=False 샷 전부 (L03 옥상/현관 + L10 골목/계단). - SOT: DB scene_still(scene_index, shot_index) → image_asset(still_id, pipeline_role='scene_still', status='generated') file_path. - 산출: out/shots_outdoor/<샷키>_.png 복사 + plans/outdoor_shots_collect.json (다음 remake 실험 입력 매핑) + shots_outdoor.html (8897 갤러리). - 읽기 전용(DB write 0), 커밋 금지(scratchpad). 사용: .venv/bin/python s21_shot_collect.py """ import html as html_mod import json import shutil import sys 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" PAGE = Path(__file__).parent / "shots_outdoor.html" def fetch_stills(eid, pairs): conn = psycopg2.connect(host="localhost", user="theroad", password="theroad_dev_2026", dbname="theroad") cur = conn.cursor() cur.execute( """ SELECT ss.scene_index, ss.shot_index, ia.file_path, ia.id, ia.created_at FROM image_asset ia JOIN scene_still ss ON ia.still_id = ss.id WHERE ss.episode_id = %s AND ia.pipeline_role = 'scene_still' AND ia.status = 'generated' AND (ss.scene_index, ss.shot_index) IN %s ORDER BY ss.scene_index, ss.shot_index, ia.created_at """, (eid, tuple(pairs)), ) rows = cur.fetchall() conn.close() return rows def main(): recon = F.load_recon() eid = "0d30302a-51c8-4ac9-98e2-6cd9180512bf" shots = {k: s for k, s in recon["shots"].items() if not s["is_indoor"]} pairs = sorted((s["scene_index"], s["shot_index"]) for s in shots.values()) key_by_pair = {(s["scene_index"], s["shot_index"]): k for k, s in shots.items()} rows = fetch_stills(eid, pairs) OUT.mkdir(parents=True, exist_ok=True) collect = {} missing = [] for sc, sh, fp, aid, _ in rows: key = key_by_pair[(sc, sh)] src = ROOT / fp if not src.exists(): missing.append(fp) continue n = len(collect.get(key, [])) + 1 dst = OUT / f"{key}_{n:02d}.png" if not dst.exists(): shutil.copy2(src, dst) collect.setdefault(key, []).append( {"file": dst.name, "asset_id": aid, "src": fp}) F.save_plan("outdoor_shots_collect", { "episode_id": eid, "group_id": recon.get("group_id"), "shots": {k: {"scene_index": shots[k]["scene_index"], "shot_index": shots[k]["shot_index"], "loc_ids": shots[k]["loc_ids"], "scene_heading": shots[k].get("scene_heading"), "description": shots[k].get("description"), "images": collect.get(k, [])} for k in sorted(shots)}, }) def _esc(t): return html_mod.escape(str(t or "")) secs = [] for k in sorted(shots, key=lambda x: (shots[x]["scene_index"], shots[x]["shot_index"])): s = shots[k] imgs = collect.get(k, []) cells = "".join( f"

" f"

{im['file']}

" for im in imgs) secs.append( f"

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

" f"

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

" f"

{cells or '(생성 스틸 없음)'}

") PAGE.write_text(f""" 실외 샷 스틸 취합 (옥탑 유닛 그룹)

옥탑방 유닛(빌라 포함) 실외 샷 — 생성 스틸 전수 취합 ({len(shots)}샷, 총 {sum(len(v) for v in collect.values())}장)

SOT=DB scene_still×image_asset(pipeline_role=scene_still, status=generated). 다음 remake 실험 입력. 매핑=plans/outdoor_shots_collect.json

{''.join(secs)} """, encoding="utf-8") total = sum(len(v) for v in collect.values()) print(f"수집: {len(shots)}샷 {total}장 -> {OUT}") if missing: print(f"!! 파일 누락 {len(missing)}건:", missing[:3]) print(f"page -> {PAGE}") if __name__ == "__main__": main()