#!/usr/bin/env python3 """scene_detail 산출을 잰다 — **입력 근거를 대조한 뒤에** 위반으로 센다. ★`repeat_scene_detail.py` 의 결함(2026-08-25 실측): regex 적중을 곧바로 위반으로 셌다. 그 탓에 - S3sh5 의 `goes out`·`gone dark` 를 조명 왜곡으로 셌는데, 그 샷의 입력 description 이 이미 「형광등 불빛이 **꺼져** 일순간 어두워진」이었다. scene_detail 은 충실했고 왜곡은 상류(shot_validator)가 만든 것이다. - `while leaving` 을 시간 연결어 위반으로 셌는데, 입력 fixed_element 원문에 있던 말이었다. 입력이 정당화하는 것을 빼고 나면 R4·R5 는 **전 판 0건**이고 남는 것은 R1(촬영 용어 누출)뿐이다. 그래서 이 도구는 R1 을 primary 로 놓는다. usage: measure_scene_detail.py # 이미 쌓인 기록만 다시 읽는다 (무료) measure_scene_detail.py --run N # N 회 강제 재실행 후 잰다 (유료) """ import json import re import sys import time import requests sys.path.insert(0, "/Users/manta/Documents/Projects/TheRoad-I1/scratchpad") from _opik_env import opik_target # noqa: E402 ★cwd 를 backend 로 고정 from tools.opik_prompt_audit.audit.fetch import fetch_spans # noqa: E402 B = "http://localhost:8000" ST = json.loads(open("/Users/manta/Documents/Projects/TheRoad-I1/" "scratchpad/minimal_e2e_state.json").read()) P, E = ST["project_id"], ST["episode_id"] # ── primary: 촬영 용어가 t2i_prompt 로 샌 것 (입력 근거와 무관하게 위반) ── R1 = re.compile(r"\b(close-?up|wide shot|medium shot|long shot|wide view|" r"establishing shot|full shot)\b", re.I) # ── 실수치: 카드 camera_reference 가 준 값을 옮긴 것도 t2i 로는 위반 ── R3 = re.compile(r"\b\d+\s?(?:cm|mm|m|meters?|inch\w*|feet|ft)\b", re.I) # ── 조건부: 입력이 그 상태를 말했으면 위반이 아니다 ── R4 = re.compile(r"(goes? out|went out|gone dark|black-?out|power cut|" r"switched? off|turned? off|unlit|extinguish\w*|" r"pitch dark|darkness)", re.I) R4_IN = re.compile(r"꺼[져진]|나간다|정전|goes? out|gone dark|blackout|unlit") R5 = re.compile(r"\b(and then|while \w+ing|after \w+ing|before \w+ing|" r"as (?:he|she|they|it) \w+s?)\b", re.I) SENT = re.compile(r"[.!?]+(?:\s|$)") def _parts(span): inp = span.get("input") msgs = inp if isinstance(inp, list) else (inp or {}).get("messages") or [] syslen, user = 0, "" for m in msgs: c = m.get("content") if isinstance(c, list): c = "\n".join(p.get("text", "") for p in c if isinstance(p, dict)) c = c or "" if m.get("role") == "system": syslen = len(c) elif m.get("role") == "user": user = c mm = re.search(r'"shot_key":\{"scene_index":(\d+),"shot_index":(\d+)\}', user) shot = f"S{mm.group(1)}sh{mm.group(2)}" if mm else "?" dm = re.search(r"\[분석 대상 Shot\](.{0,600}?)(?:\n\[|\Z)", user, re.S) return syslen, shot, (dm.group(1) if dm else ""), user def rows_since(since, until="2026-12-31T00:00:00"): base, ws, proj = opik_target() out = [] for x in fetch_spans(base, ws, proj, since, until): if "op:scene_detail" not in (x.get("tags") or []): continue syslen, shot, shot_desc, user = _parts(x) fixed = user[user.find("[교차 샷 고정 요소]"):][:2500] \ if "[교차 샷 고정 요소]" in user else "" try: c = json.loads(x["output"]["choices"][0]["message"]["content"]) except Exception: continue for v in (c.get("t2i_variations") or []): p = v.get("t2i_prompt") or "" sents = [s for s in SENT.split(p) if s.strip()] out.append({ "ts": x.get("start_time", ""), "sys": syslen, "shot": shot, "len": len(p), "r1": bool(R1.search(p)), "r3": bool(R3.search(p)), # ★입력이 그 상태를 말했으면 위반이 아니다 "r4": bool(R4.search(p)) and not R4_IN.search(shot_desc), "r5": bool(R5.search(p)) and "while leaving" not in fixed, "r4_raw": bool(R4.search(p)), "r5_raw": bool(R5.search(p)), "nsent": len(sents), "longest": max((len(s) for s in sents), default=0), "commas": p.count(",") + p.count(";"), }) return out def report(rows, title=""): if not rows: print(" (없음)") return n = len(rows) print(f"\n── {title} · {n}컷 ──") print(f" ★R1 촬영용어 누출 : {sum(r['r1'] for r in rows)}/{n} (primary)") print(f" R3 실수치 : {sum(r['r3'] for r in rows)}/{n}" f" ※카드 camera_reference 유래인지 확인할 것") print(f" R4 조명 왜곡 : {sum(r['r4'] for r in rows)}/{n}" f" (근거 대조 전 {sum(r['r4_raw'] for r in rows)})") print(f" R5 시간 연결어 : {sum(r['r5'] for r in rows)}/{n}" f" (근거 대조 전 {sum(r['r5_raw'] for r in rows)})") ln = sorted(r["len"] for r in rows) print(f" 길이 중앙 {ln[n//2]:,} · p90 {ln[int(n*0.9)-1 if n>1 else 0]:,}" f" · 최대 {ln[-1]:,} · 900 미만 {sum(1 for x in ln if x<900)}/{n}") print(f" 문장 평균 {sum(r['nsent'] for r in rows)/n:.1f}개 · " f"최장 문장 중앙 {sorted(r['longest'] for r in rows)[n//2]:,}자 · " f"쉼표 평균 {sum(r['commas'] for r in rows)/n:.1f}") def run_once(s): import datetime as dt t0 = dt.datetime.now(dt.timezone.utc) - dt.timedelta(seconds=30) while True: r = s.post(f"{B}/api/v1/projects/{P}/episodes/{E}/steps/scene_detail", params={"mode": "force"}, timeout=120) if r.status_code == 200: break print(f" (시작 대기 {r.status_code})", flush=True) time.sleep(15) for _ in range(120): time.sleep(10) st = {x["step_id"]: x for x in s.get( f"{B}/api/v1/projects/{P}/episodes/{E}/steps", timeout=60).json()["steps"]}["scene_detail"] if st["status"] in ("completed", "failed", "skipped"): return t0.strftime("%Y-%m-%dT%H:%M:%S"), st["status"] return t0.strftime("%Y-%m-%dT%H:%M:%S"), "timeout" if __name__ == "__main__": if "--run" in sys.argv: n = int(sys.argv[sys.argv.index("--run") + 1]) s = requests.Session() s.post(f"{B}/api/v1/auth/login", json={"username": "admin", "password": "admin123"}, timeout=30).raise_for_status() for i in range(1, n + 1): since, status = run_once(s) report(rows_since(since), f"{i}회차 ({status})") else: since = sys.argv[1] if len(sys.argv) > 1 else "2026-08-24T16:00:00" rows = rows_since(since) by = {} for r in rows: by.setdefault(r["sys"], []).append(r) for sysl in sorted(by, reverse=True): report(by[sysl], f"system {sysl:,}자")