"""Shot 기반 인물 추출 실험 — entity_all_character의 shot 버전. shot 결과 + 원본 씬 텍스트를 3000자 번들로 묶어 인물 추출. scene_count 대신 shot_count (몇 개 샷에 등장하는지). 체이닝 방식: 이전 번들 결과를 다음 번들에 전달. Usage: cd backend .venv/bin/python experiments/entity_from_shots_test.py \ --project-id 84e9e90b-... --episode-id 78bc78d1-... \ --shots experiments/shot_results_gpt.json \ --model gpt-mini --threads 1 """ import argparse import json import sys import time from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) from app.modules.llm.llm_client import call_structured from app.modules.pdf_parser import extract_text_from_pdf BUNDLE_TARGET = 3000 SYSTEM = """시나리오 분석 전문가. 시나리오를 읽고 이미지 생성(T2I)에 필요한 시각적 요소를 리스팅한다. ## 핵심 원칙 - 카메라로 찍을 수 있는 시각적 요소만 리스팅 - 이 단계에서는 이름만 나열 (상세 설명은 다음 단계에서) - 리스팅에만 집중 — 빠짐없이 모든 요소를 나열 ## 예외 — 카메라에 찍히면 포함 회상, 상상, 꿈, 화상통화 등 실제 존재하지 않더라도 **카메라에 찍히는 인물/배경/소품은 리스팅 대상**이다.""" CHARACTER_PROMPT = """시나리오에서 등장하는 인물(캐릭터)을 모두 나열하세요. ## 리스팅 항목 - name: 인물 이름 (이름이 있다면 반드시 사용) - shot_count: 등장하는 샷(스틸컷) 수 ## 포함 기준 - 되도록 최대한 많은 인물을 포함 - 이름이 있거나 시각적으로 구분되는 인물은 모두 포함 - 이름이 없어도 역할명으로 식별 가능하면 포함 - 인간이 아니더라도 지속적으로 스스로 행동할 수 있는 존재라면 무조건 추출 ## 제외 기준 - 엑스트라(행인, 군중, 이름 없는 단역 등 누구든 대체 가능한 인물)만 제외 ## 통합 기준 - 동일 인물은 1개로 통합 (과거/현재 포함) - 이름은 반드시 단수형으로 — 복수형 이름 절대 금지 ## 분리 기준 — 외형이 달라지면 반드시 별도 인물로 분리 - 변신/변형: 인간↔요괴, 인간↔괴물, 본체↔변신체 등 외형이 달라지는 경우 - 나이 변화: 어린 시절↔성인↔노인 등 얼굴이 크게 달라지는 경우 - 이름 구분: "A", "A (변형 상태)" — 괄호 안에 변형 상태를 명시""" SCHEMA = { "type": "object", "properties": { "characters": { "type": "array", "items": { "type": "object", "properties": { "name": {"type": "string"}, "shot_count": {"type": "integer"}, }, "required": ["name", "shot_count"], "additionalProperties": False, }, }, }, "required": ["characters"], "additionalProperties": False, } def load_scenes(project_dir: Path, episode_id: str): cp_path = (project_dir / "checkpoints" / "episodes" / episode_id / "scene_segmentation" / "manifest.json") cp = json.loads(cp_path.read_text()) segments = cp["data"]["segments"] pdf_dir = project_dir / "assets" / "screenplays" pdfs = list(pdf_dir.glob("*.pdf")) fulltext, _ = extract_text_from_pdf(pdfs[0]) scene_texts = [] for seg in segments: start = seg["start_char"] end = seg["end_char"] scene_texts.append({ "idx": seg["scene_index"], "heading": seg.get("heading", ""), "text": fulltext[start:end], "length": end - start, }) return scene_texts def load_shots(shots_path: str): """shot 결과를 scene_index → shots 매핑으로 로드.""" shots_data = json.loads(Path(shots_path).read_text()) shots_by_scene = {} for s in shots_data: idx = s["scene_index"] shots_by_scene[idx] = s.get("shots", []) return shots_data, shots_by_scene def _format_shots_for_scene(shots): if not shots: return " (샷 없음)" lines = [] for sh in shots: chars = ", ".join(sh.get("characters", [])) lines.append(f" Shot {sh['shot_index']}: {sh['description'][:100]} [{chars}]") return "\n".join(lines) def extract_characters(scene_texts, shots_by_scene, model: str): """체이닝 방식으로 인물 추출. 순차 실행 (체이닝이라 병렬 불가).""" all_entities = [] existing_names = set() i = 0 call_count = 0 total_start = time.time() while i < len(scene_texts): # 번들 구성 (BUNDLE_TARGET 이하) bundle = [] bundle_len = 0 while i < len(scene_texts) and bundle_len + scene_texts[i]["length"] <= BUNDLE_TARGET: bundle.append(scene_texts[i]) bundle_len += scene_texts[i]["length"] i += 1 if not bundle and i < len(scene_texts): bundle.append(scene_texts[i]) bundle_len = scene_texts[i]["length"] i += 1 # 번들 내 씬+샷 텍스트 구성 bundle_parts = [] for s in bundle: shots = shots_by_scene.get(s["idx"], []) shot_text = _format_shots_for_scene(shots) bundle_parts.append( f"--- Scene {s['idx']}: {s['heading']} ---\n" f"[샷 목록]\n{shot_text}\n\n" f"[원문]\n{s['text']}" ) bundle_text = "\n\n".join(bundle_parts) # 체이닝 — 이전 결과 포함 prev_list = "" if all_entities: prev_list = ( "지금까지 찾은 인물:\n" + "\n".join(f"- {e['name']} (shot_count: {e['shot_count']})" for e in all_entities) + "\n\n위 목록에 없는 새로운 인물만 추가하세요. " "이미 있는 인물의 shot_count는 이 번들에서의 추가 등장 횟수만 적어주세요.\n\n" ) user_prompt = prev_list + bundle_text + "\n\n" + CHARACTER_PROMPT call_count += 1 call_start = time.time() scene_range = f"{bundle[0]['idx']}~{bundle[-1]['idx']}" try: result = call_structured( step="entity_shot_test", system_prompt=SYSTEM, user_prompt=user_prompt, response_schema=SCHEMA, project_config={"default_model": model}, schema_name=f"entity_shot_{call_count}", ) elapsed = time.time() - call_start new_chars = result.get("characters", []) added = 0 for e in new_chars: name = e.get("name", "") if not name: continue if name in existing_names: # shot_count 누적 for existing in all_entities: if existing["name"] == name: new_sc = e.get("shot_count", 0) if new_sc > 0: existing["shot_count"] += new_sc break else: all_entities.append(e) existing_names.add(name) added += 1 print(f" call {call_count} (scenes {scene_range}, {bundle_len}chars, {elapsed:.1f}s): " f"+{added} new (total {len(all_entities)})", flush=True) except Exception as exc: elapsed = time.time() - call_start print(f" call {call_count} (scenes {scene_range}) FAILED ({elapsed:.1f}s): {exc}", flush=True) total_elapsed = time.time() - total_start print(f"\n Total: {len(all_entities)} characters, {call_count} calls, {total_elapsed:.1f}s", flush=True) return all_entities 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="gpt-mini") parser.add_argument("--output", default=None) args = parser.parse_args() projects_root = Path(__file__).resolve().parent.parent.parent / "projects" project_dir = projects_root / args.project_id print(f"Loading scenes...") scene_texts = load_scenes(project_dir, args.episode_id) print(f" {len(scene_texts)} scenes") print(f"Loading shots from {args.shots}...") shots_data, shots_by_scene = load_shots(args.shots) total_shots = sum(len(v) for v in shots_by_scene.values()) print(f" {total_shots} total shots across {len(shots_by_scene)} scenes") print(f"\nExtracting characters with model={args.model} (chained, sequential)...") results = extract_characters(scene_texts, shots_by_scene, args.model) # shot_count 내림차순 정렬 results.sort(key=lambda x: x.get("shot_count", 0), reverse=True) out_path = args.output or f"experiments/entity_shot_results_{args.model}.json" Path(out_path).parent.mkdir(parents=True, exist_ok=True) Path(out_path).write_text(json.dumps(results, ensure_ascii=False, indent=2)) print(f"\nSaved to {out_path}") print("\n=== Characters (sorted by shot_count) ===") for e in results: print(f" {e['name']:20s} | shots: {e.get('shot_count', 0)}") if __name__ == "__main__": main()