"""Phase 0 reference-necessity audit CLI (fail-closed GATE). 사용: cd backend && python -m scripts.reference_necessity_audit checkpoint + DB 를 읽어 usage matrix / safety_diff 를 계산하고 `projects/{pid}/checkpoints/episodes/{eid}/entity_reference_usage_report.json` 에 저장한다. GATE FAIL 시 exit code 1. GATE 는 pure 함수(checkpoint presence/status/selected_map) + CLI 의 DB step_run completed·non-stale 검사 + SceneStill t2i ID cross-check 를 모두 통과해야 PASS. """ from __future__ import annotations import json import re import sys from pathlib import Path _ID_RE = re.compile(r"\b([CPLO]\d{2,3})(?:O\d{2,3})?\b") _REQUIRED_STEPS = ( "shot_selection", "shot_director", "scene_director", "shot_validator", "scene_detail", ) def main(project_id: str, episode_id: str) -> int: from app.core.config import settings from app.core.database import SessionLocal # app.db.session 없음 — app.core.database 가 정규 경로 from app.models.project import EntityCanon, EntityEpisodeLink, ImageAsset, SceneStill from app.services.reference_necessity_audit import compute_reference_usage_report checkpoints_dir = ( Path(settings.projects_dir) / project_id / "checkpoints" / "episodes" / episode_id ) if not checkpoints_dir.exists(): print(f"checkpoint 디렉토리 없음: {checkpoints_dir}", file=sys.stderr) return 2 db = SessionLocal() try: links = db.query(EntityEpisodeLink).filter( EntityEpisodeLink.project_id == project_id, EntityEpisodeLink.episode_id == episode_id, ).all() canon_ids = [lnk.canon_id for lnk in links] canons = db.query(EntityCanon).filter( EntityCanon.id.in_(canon_ids) ).all() if canon_ids else [] entity_catalog = { c.short_id.split("O")[0]: {"name": c.name, "entity_type": c.entity_type} for c in canons if c.short_id } canon_by_id = {c.id: c for c in canons} generated_ref_counts: dict = {} ref_assets = db.query(ImageAsset).filter( ImageAsset.project_id == project_id, ImageAsset.episode_id == episode_id, ImageAsset.asset_type == "reference", ).all() for a in ref_assets: c = canon_by_id.get(a.entity_id) if c and c.short_id: sid = c.short_id.split("O")[0] generated_ref_counts[sid] = generated_ref_counts.get(sid, 0) + 1 # cross-check: DB 의 selected non-stale SceneStill t2i_prompt 에서 # short_id 추출 → checkpoint scene_detail 의 prompt_id_count 와 비교. db_selected_ids: dict = {} sel_stills = db.query(SceneStill).filter( SceneStill.project_id == project_id, SceneStill.episode_id == episode_id, SceneStill.is_selected == True, # noqa: E712 SceneStill.still_index >= 0, SceneStill.status != "stale", ).all() for st in sel_stills: try: variations = json.loads(st.t2i_variations_json or "[]") except (json.JSONDecodeError, TypeError): variations = [] texts = [v.get("t2i_prompt", "") for v in variations] if not texts: texts = [st.t2i_prompt_cinematic or ""] for t in texts: for m in _ID_RE.finditer(t): b = m.group(1).split("O")[0] db_selected_ids[b] = db_selected_ids.get(b, 0) + 1 # DB step_run — 필수 5 step 전부 status='completed' 여야 GATE PASS. # step_run 은 ORM 모델 없음 — raw SQL (step_runner.py:510 패턴). from sqlalchemy import text as _sql_text step_run_status: dict = {} for sid in _REQUIRED_STEPS: row = db.execute(_sql_text( "SELECT status FROM step_run WHERE project_id = :pid " "AND episode_id = :eid AND step_id = :sid" ), {"pid": project_id, "eid": episode_id, "sid": sid}).fetchone() step_run_status[sid] = row[0] if row else "missing" finally: db.close() report = compute_reference_usage_report( checkpoints_dir=checkpoints_dir, entity_catalog=entity_catalog, generated_ref_counts=generated_ref_counts, ) report["project_id"] = project_id report["episode_id"] = episode_id # cross-check 결과 첨부 (checkpoint prompt_id_count vs DB selected stills) cp_prompt_ids = { r["short_id"]: r["prompt_id_count"] for r in report["entities"] if r["prompt_id_count"] > 0 } divergence = sorted( set(cp_prompt_ids) ^ set(db_selected_ids) ) report["cross_check"] = { "db_selected_still_id_count": db_selected_ids, "checkpoint_prompt_id_count": cp_prompt_ids, "id_set_divergence": divergence, } if divergence: report["gate"]["fail_reasons"].append( f"cross-check divergence: checkpoint t2i ID 와 DB selected " f"SceneStill ID 불일치 {divergence}" ) report["gate"]["passed"] = False # DB step_run GATE — 필수 5 step 전부 status='completed' 여야 PASS. # missing / partial / failed / pending / running / stale 전부 FAIL. report["db_step_run_status"] = step_run_status _bad = {s: st for s, st in step_run_status.items() if st != "completed"} if _bad: for s, st in sorted(_bad.items()): report["gate"]["fail_reasons"].append(f"step_run {s}: {st}") report["gate"]["passed"] = False out = checkpoints_dir / "entity_reference_usage_report.json" out.write_text( json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8", ) print(f"\n=== reference-necessity audit: {project_id} / {episode_id} ===") print("input_status:", report["gate"]["input_status"]) print(f"{'short_id':<10}{'type':<12}{'visible':<9}{'promptID':<10}" f"{'reqRef':<8}{'genRef':<8}name") for r in report["entities"]: print(f"{r['short_id']:<10}{(r['entity_type'] or ''):<12}" f"{r['visible_shot_count']:<9}{r['prompt_id_count']:<10}" f"{r['required_ref_count']:<8}{r['generated_refs']:<8}{r['name'] or ''}") sd = report["safety_diff"] print(f"\naudit_set ({len(sd['audit_set'])}): {sd['audit_set']}") print(f"audit_set_with_t2i_usage (diagnostic): " f"{sd['audit_set_with_t2i_usage']}") print(f"gate_blocking_ids ({len(sd['gate_blocking_ids'])}): " f"{sd['gate_blocking_ids']}") _w = sd["warnings"] print(f"warnings: out_of_scope={_w['out_of_scope_prompt_id_ids']} " f"prop={_w['prompt_id_only_prop_ids']} " f"expected_text_only={_w['expected_text_only_prompt_id_ids']}") print(f"cross_check divergence: {report['cross_check']['id_set_divergence']}") print(f"\nGATE: {'PASS' if report['gate']['passed'] else 'FAIL'}") for fr in report["gate"]["fail_reasons"]: print(f" - {fr}") print(f"report 저장: {out}") return 0 if report["gate"]["passed"] else 1 if __name__ == "__main__": if len(sys.argv) != 3: print("usage: python -m scripts.reference_necessity_audit " " ", file=sys.stderr) sys.exit(2) sys.exit(main(sys.argv[1], sys.argv[2]))