"""Shot별 촬영 기법 2개 선정 — beat 연결된 shot에 카메라 앵글/구도/포커스 추가. Usage: cd backend .venv/bin/python experiments/shot_cinematography_test.py \ --project-id 84e9e90b-... --episode-id 78bc78d1-... \ --shots experiments/shot_results_gemini-pro_v3.json \ --model gemini-pro --threads 4 """ import argparse import json import sys import time from concurrent.futures import ThreadPoolExecutor, as_completed from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) from app.modules.llm.llm_client import call_structured BUNDLE_MAX_SHOTS = 15 # 한 번들에 최대 15 shots SYSTEM = """너는 촬영 감독(DP)이다. 각 Shot(스틸컷)에 가장 효과적인 촬영 기법 2가지를 선택한다. ## 선택 원칙 1. 전체 에피소드의 **감정 곡선**을 고려: 긴장 고조 → 폭발 → 감정 정리 2. 연속된 Shot에서 **같은 기법 반복 최소화**: 변화와 리듬감 유지 3. Shot의 **핵심 감정/상황**에 맞는 기법 선택 4. 2가지 기법은 서로 **다른 관점**이어야 함 (예: 인물 중심 + 환경 중심, 클로즈 + 와이드) 5. 각 기법은 하나의 독립적 이미지(컷)가 된다 — 같은 장면을 ���른 구도로 촬영 6. focus에는 카메라가 초점을 맞추는 대상을 구체적으로 명시 (인물이면 누구, 사물이면 무엇)""" USER_TEMPLATE = """\ 아래 Shot 목록 각각에 대해, 촬영 기법 목록에서 가장 적합한 2가지를 선택하세요. [사용 가능한 촬영 기법] {shot_types_block} [Shot 목록] {shots_block} 각 Shot에 대해 technique_1, technique_2를 선택하세요. 각 항목에는 name(촬영 기법명), reason(선택 이유 한 줄), focus(카메라 초점 대상)를 포함하세요. """ SCHEMA = { "type": "object", "properties": { "shots": { "type": "array", "items": { "type": "object", "properties": { "scene_index": {"type": "integer"}, "shot_index": {"type": "integer"}, "technique_1": { "type": "object", "properties": { "name": {"type": "string"}, "reason": {"type": "string"}, "focus": {"type": "string"}, }, "required": ["name", "reason", "focus"], "additionalProperties": False, }, "technique_2": { "type": "object", "properties": { "name": {"type": "string"}, "reason": {"type": "string"}, "focus": {"type": "string"}, }, "required": ["name", "reason", "focus"], "additionalProperties": False, }, }, "required": ["scene_index", "shot_index", "technique_1", "technique_2"], "additionalProperties": False, }, }, }, "required": ["shots"], "additionalProperties": False, } def load_shot_types(): """DB에서 촬영 기법 목록 로드.""" from sqlalchemy import create_engine, text engine = create_engine("postgresql://theroad:theroad_dev_2026@localhost:5432/theroad") with engine.connect() as conn: rows = conn.execute(text( "SELECT name, category, description, llm_description " "FROM shot_type WHERE is_active = true ORDER BY sort_order" )).fetchall() block = "\n".join( f"- {r[0]} [{r[1]}]: {r[2]} | Camera: {r[3][:100]}" for r in rows ) return block def load_shots(shots_path: str): return json.loads(Path(shots_path).read_text()) def _build_bundles(shots_data): """shot을 BUNDLE_MAX_SHOTS개씩 묶기.""" all_shots = [] for s in shots_data: for sh in s.get("shots", []): all_shots.append({ "scene_index": s["scene_index"], "scene_heading": s.get("scene_heading", ""), "shot_index": sh["shot_index"], "description": sh["description"], "characters": sh.get("characters", []), "based_on_beat": sh.get("based_on_beat", 0), }) bundles = [] for i in range(0, len(all_shots), BUNDLE_MAX_SHOTS): bundles.append(all_shots[i:i + BUNDLE_MAX_SHOTS]) return bundles def _call_one_bundle(bundle, call_idx, model, shot_types_block): shots_lines = [] for sh in bundle: chars = ", ".join(sh["characters"]) beat = f"beat:{sh['based_on_beat']}" if sh["based_on_beat"] else "원문" shots_lines.append( f"Scene {sh['scene_index']} Shot {sh['shot_index']} ({beat}): " f"{sh['description']} [{chars}]" ) shots_block = "\n".join(shots_lines) user_prompt = USER_TEMPLATE.format( shot_types_block=shot_types_block, shots_block=shots_block, ) scene_range = f"S{bundle[0]['scene_index']}-S{bundle[-1]['scene_index']}" call_start = time.time() try: result = call_structured( step="shot_cinematography_test", system_prompt=SYSTEM, user_prompt=user_prompt, response_schema=SCHEMA, project_config={"default_model": model}, schema_name=f"shot_cine_{call_idx}", ) elapsed = time.time() - call_start shots_result = result.get("shots", []) print(f" call {call_idx} ({scene_range}, {len(bundle)} shots, {elapsed:.1f}s): " f"{len(shots_result)} results", flush=True) return call_idx, shots_result, None except Exception as exc: elapsed = time.time() - call_start print(f" call {call_idx} ({scene_range}) FAILED ({elapsed:.1f}s): {exc}", flush=True) return call_idx, [], str(exc) def extract_cinematography(shots_data, model, shot_types_block, max_threads=4): bundles = _build_bundles(shots_data) print(f" {sum(len(b) for b in bundles)} shots in {len(bundles)} bundles", flush=True) total_start = time.time() results_by_idx = {} with ThreadPoolExecutor(max_workers=max_threads) as pool: futures = { pool.submit(_call_one_bundle, b, i + 1, model, shot_types_block): i + 1 for i, b in enumerate(bundles) } for fut in as_completed(futures): idx, shots, err = fut.result() results_by_idx[idx] = shots all_results = [] for i in sorted(results_by_idx): all_results.extend(results_by_idx[i]) total_elapsed = time.time() - total_start print(f"\n Total: {len(all_results)} shots, {len(bundles)} calls, {total_elapsed:.1f}s", flush=True) return all_results def merge_results(shots_data, cine_results): """원본 shot 데이터에 촬영 기법 정보를 병합.""" cine_map = {} for c in cine_results: key = (c["scene_index"], c["shot_index"]) cine_map[key] = c merged = [] for s in shots_data: scene_out = { "scene_index": s["scene_index"], "scene_heading": s.get("scene_heading", ""), "shots": [], } for sh in s.get("shots", []): key = (s["scene_index"], sh["shot_index"]) cine = cine_map.get(key, {}) shot_out = {**sh} if cine: shot_out["technique_1"] = cine.get("technique_1", {}) shot_out["technique_2"] = cine.get("technique_2", {}) scene_out["shots"].append(shot_out) merged.append(scene_out) return merged def main(): parser = argparse.ArgumentParser() parser.add_argument("--project-id", required=True) parser.add_argument("--episode-id", required=True) parser.add_argument("--shots", required=True) parser.add_argument("--model", default="gemini-pro") parser.add_argument("--threads", type=int, default=4) parser.add_argument("--output", default=None) args = parser.parse_args() print("Loading shot types from DB...") shot_types_block = load_shot_types() print(f" {shot_types_block.count(chr(10)) + 1} techniques loaded") print(f"Loading shots from {args.shots}...") shots_data = load_shots(args.shots) total_shots = sum(len(s.get("shots", [])) for s in shots_data) print(f" {total_shots} shots") print(f"\nExtracting cinematography with model={args.model}, threads={args.threads}...") cine_results = extract_cinematography(shots_data, args.model, shot_types_block, args.threads) print("\nMerging results...") merged = merge_results(shots_data, cine_results) out_path = args.output or f"experiments/shot_cine_results_{args.model}.json" Path(out_path).parent.mkdir(parents=True, exist_ok=True) Path(out_path).write_text(json.dumps(merged, ensure_ascii=False, indent=2)) print(f"Saved to {out_path}") # 요약 print("\n=== Sample (first 5 shots) ===") count = 0 for s in merged: for sh in s.get("shots", []): if count >= 5: break t1 = sh.get("technique_1", {}) t2 = sh.get("technique_2", {}) print(f" S{s['scene_index']:02d}-Shot{sh['shot_index']:02d}: {sh['description'][:60]}") print(f" T1: {t1.get('name','')} — focus: {t1.get('focus','')}") print(f" T2: {t2.get('name','')} — focus: {t2.get('focus','')}") count += 1 if count >= 5: break if __name__ == "__main__": main()