#!/usr/bin/env python3 """Extract fulltext scene still candidates from a screenplay PDF.""" from __future__ import annotations import argparse import hashlib import json import re import sqlite3 import sys import time from dataclasses import dataclass from difflib import SequenceMatcher from pathlib import Path from typing import Dict, Iterable, List, Sequence, Tuple import extract_entities as common SCRIPT_DIR = Path(__file__).resolve().parent PROMPTS_DIR = SCRIPT_DIR / "scene_still_prompts" PROMPTS_MANIFEST = PROMPTS_DIR / "manifest.json" HEADING_PREFIXES = ("INT.", "EXT.", "INT/EXT.", "I/E.") HEADING_MAX_LEN = 180 VISIBLE_ENTITY_REF_SCHEMA: Dict[str, object] = { "type": "object", "additionalProperties": False, "properties": { "entity_name": {"type": "string"}, "entity_type": { "type": "string", "enum": ["character", "location", "prop"], }, "role": {"type": "string"}, "prominence": { "type": "string", "enum": ["primary", "secondary", "background", "detail"], }, }, "required": ["entity_name", "entity_type", "role", "prominence"], } CAMERA_SCHEMA: Dict[str, object] = { "type": "object", "additionalProperties": False, "properties": { "shot_size": {"type": "string"}, "angle": {"type": "string"}, "lens": {"type": "string"}, "composition": {"type": "string"}, "focus": {"type": "string"}, }, "required": ["shot_size", "angle", "lens", "composition", "focus"], } LIGHTING_SCHEMA: Dict[str, object] = { "type": "object", "additionalProperties": False, "properties": { "setup": {"type": "string"}, "direction": {"type": "string"}, "quality": {"type": "string"}, "color": {"type": "string"}, "contrast": {"type": "string"}, "motivation": {"type": "string"}, "mood": {"type": "string"}, }, "required": ["setup", "direction", "quality", "color", "contrast", "motivation", "mood"], } SCENE_STILL_SCHEMA: Dict[str, object] = { "type": "object", "additionalProperties": False, "properties": { "heading_catalog_index": {"type": "integer", "minimum": 1}, "screenplay_scene_heading": {"type": "string"}, "page_start": {"type": "integer", "minimum": 1}, "page_end": {"type": "integer", "minimum": 1}, "beat_index_within_heading": {"type": "integer", "minimum": 1}, "still_kind": { "type": "string", "enum": [ "establishing", "group", "dialogue", "action", "detail", "reaction", "reveal", "insert", "climax", "aftermath", "other", ], }, "beat_title": {"type": "string"}, "still_frame_prompt_raw": {"type": "string"}, "visible_entities": { "type": "array", "minItems": 1, "items": VISIBLE_ENTITY_REF_SCHEMA, }, "camera": CAMERA_SCHEMA, "lighting": LIGHTING_SCHEMA, "evidence": {"type": "array", "minItems": 1, "items": {"type": "string"}}, }, "required": [ "heading_catalog_index", "screenplay_scene_heading", "page_start", "page_end", "beat_index_within_heading", "still_kind", "beat_title", "still_frame_prompt_raw", "visible_entities", "camera", "lighting", "evidence", ], } RAW_SCHEMA: Dict[str, object] = { "type": "object", "additionalProperties": False, "properties": { "source_file": {"type": "string"}, "characters": common.CHUNK_SCHEMA["properties"]["characters"], "locations": common.CHUNK_SCHEMA["properties"]["locations"], "props": common.CHUNK_SCHEMA["properties"]["props"], "scene_stills": { "type": "array", "minItems": 1, "items": SCENE_STILL_SCHEMA, }, "summary": {"type": "string"}, "notes": {"type": "array", "items": {"type": "string"}}, }, "required": ["source_file", "characters", "locations", "props", "scene_stills", "summary", "notes"], } ENTITY_SECTIONS: Sequence[Tuple[str, str]] = ( ("characters", "character"), ("locations", "location"), ("props", "prop"), ) PRIORITY_RANK = {"critical": 3, "high": 2, "medium": 1, "low": 0} REFERENCE_RANK = {"required": 2, "helpful": 1, "not_needed": 0} TOKEN_RE = re.compile(r"[\s/,:;()\-]+") KO_CINEMATIC_MAP = { "Wide-angle lens": "광각 렌즈", "Standard lens": "표준 렌즈", "Telephoto lens": "망원 렌즈", "Macro lens": "매크로 렌즈", "Available light": "현장광", "Natural lighting": "자연광 조명", "Natural daylight": "자연광 주광", "Outdoor daylight": "야외 주광", "Practical lighting": "실광 조명", "Spotlight": "스포트라이트", "Top lighting": "탑 라이트", "Side lighting": "사이드 라이트", "Under lighting": "언더라이트", "Front lighting": "전면광", "Backlighting": "역광", "Shallow focus": "얕은 심도", "Wide Shot": "와이드 샷", "Full Shot": "풀 샷", "Wide Shot": "와이드 샷", "Medium Shot": "미디엄 샷", "Medium Close-Up": "미디엄 클로즈업", "Close-Up": "클로즈업", "Extreme Close-Up": "익스트림 클로즈업", "Extreme": "익스트림", "Eye Level": "아이레벨", "High Angle": "하이 앵글", "Slight High Angle": "약한 하이 앵글", "Low Angle": "로우 앵글", "Dutch Angle": "더치 앵글", "Low-key lighting": "로우키 조명", "High-key lighting": "하이키 조명", "Hard": "하드", "Soft": "소프트", "High": "높음", "Medium": "중간", "Low": "낮음", "Deep focus": "딥 포커스", } JA_CINEMATIC_MAP = { "Wide-angle lens": "広角レンズ", "Standard lens": "標準レンズ", "Telephoto lens": "望遠レンズ", "Macro lens": "マクロレンズ", "Available light": "現場光", "Natural lighting": "自然光照明", "Natural daylight": "自然光", "Outdoor daylight": "屋外の自然光", "Practical lighting": "実景照明", "Spotlight": "スポットライト", "Top lighting": "トップライト", "Side lighting": "サイドライト", "Under lighting": "アンダーライト", "Front lighting": "前方光", "Backlighting": "逆光", "Shallow focus": "浅い被写界深度", "Wide Shot": "ワイドショット", "Full Shot": "フルショット", "Wide Shot": "ワイドショット", "Medium Shot": "ミディアムショット", "Medium Close-Up": "ミディアムクローズアップ", "Close-Up": "クローズアップ", "Extreme Close-Up": "エクストリームクローズアップ", "Extreme": "エクストリーム", "Eye Level": "アイレベル", "High Angle": "ハイアングル", "Slight High Angle": "ややハイアングル", "Low Angle": "ローアングル", "Dutch Angle": "ダッチアングル", "Low-key lighting": "ローキー照明", "High-key lighting": "ハイキー照明", "Hard": "ハード", "Soft": "ソフト", "High": "高い", "Medium": "中間", "Low": "低い", "Deep focus": "ディープフォーカス", } @dataclass class HeadingCandidate: page: int heading: str order_index: int @dataclass class PromptBundle: version: str system: str user: str manifest_path: str version_path: str prompt_files: Dict[str, str] prompt_hashes: Dict[str, str] def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description="Extract scene still candidates from a screenplay PDF.") parser.add_argument( "--provider", default="openai", choices=["openai", "gemini"], help="LLM provider to use.", ) parser.add_argument( "--input", default="screenplay/srd part 1 blue revision.pdf", help="Path to the screenplay PDF.", ) parser.add_argument( "--output", default="screenplay/srd_part_1_scene_stills.json", help="Path to write the extracted JSON.", ) parser.add_argument( "--sqlite-output", default=None, help="Optional SQLite file to store extracted scene still data.", ) parser.add_argument( "--episode-key", default=None, help="Stable episode key for IDs and SQLite storage. Defaults to the PDF stem.", ) parser.add_argument( "--model", default=None, help="Model to use. Defaults depend on --provider.", ) parser.add_argument( "--fallback-model", default=None, help="Fallback model to try if the primary model is unavailable.", ) parser.add_argument( "--prompt-version", default=None, help="Prompt version to use. Defaults to the scene prompt manifest current_version.", ) parser.add_argument( "--source-language", default="auto", choices=["auto", "ko", "ja", "en"], help="Force the screenplay language instead of auto detection.", ) parser.add_argument( "--series-memory-json", default=None, help="Optional JSON file containing prior series scene memory.", ) parser.add_argument( "--temperature", type=float, default=0.15, help="Sampling temperature for extraction.", ) return parser.parse_args() def load_prompt_bundle(version_override: str | None) -> PromptBundle: try: manifest = json.loads(PROMPTS_MANIFEST.read_text(encoding="utf-8")) except FileNotFoundError as exc: raise RuntimeError(f"Prompt manifest not found: {PROMPTS_MANIFEST}") from exc version = version_override or manifest["current_version"] version_entry = manifest["versions"].get(version) if not version_entry: raise RuntimeError(f"Unknown scene prompt version: {version}") version_dir = PROMPTS_DIR / version_entry["path"] prompt_files: Dict[str, str] = {} prompt_hashes: Dict[str, str] = {} def read_file(name: str) -> str: path = version_dir / name try: content = path.read_text(encoding="utf-8").strip() except FileNotFoundError as exc: raise RuntimeError(f"Prompt file not found: {path}") from exc prompt_files[name] = str(path) prompt_hashes[name] = hashlib.sha256(content.encode("utf-8")).hexdigest() return content return PromptBundle( version=version, system=read_file("system.md"), user=read_file("user.md"), manifest_path=str(PROMPTS_MANIFEST), version_path=str(version_dir), prompt_files=prompt_files, prompt_hashes=prompt_hashes, ) def sanitize_text(text: str) -> str: text = text.replace("\x00", " ") text = text.replace("\u200b", "") lines = [" ".join(line.split()) for line in text.splitlines()] return "\n".join(line for line in lines if line).strip() def sanitize_page_texts(page_texts: Sequence[str]) -> List[str]: return [sanitize_text(page_text) for page_text in page_texts if sanitize_text(page_text)] def extract_page_number(page_text: str) -> int: first_line = page_text.splitlines()[0].strip() if first_line.startswith("[PAGE ") and first_line.endswith("]"): try: return int(first_line[len("[PAGE ") : -1]) except ValueError: return 0 return 0 def heading_lines(page_text: str) -> Iterable[str]: lines = page_text.splitlines() for line in lines[1:]: stripped = " ".join(line.split()) if not stripped or len(stripped) > HEADING_MAX_LEN: continue upper = stripped.upper() if upper.startswith(HEADING_PREFIXES): yield stripped def detect_scene_headings(page_texts: Sequence[str]) -> List[HeadingCandidate]: headings: List[HeadingCandidate] = [] order_index = 1 for page_text in page_texts: page_number = extract_page_number(page_text) for heading in heading_lines(page_text): headings.append( HeadingCandidate( page=page_number, heading=heading, order_index=order_index, ) ) order_index += 1 return headings def heading_catalog_text(headings: Sequence[HeadingCandidate]) -> str: if not headings: return "No scene headings were detected automatically." return "\n".join( f"{item.order_index}. [p{item.page}] {item.heading}" for item in headings ) def render_prompt(template: str, **values: object) -> str: return template.format(**values) def normalize_name(name: str) -> str: return " ".join(name.lower().split()) def compact_identifier(text: str) -> str: return "".join(ch for ch in text.lower() if ch.isalnum()) def token_set(text: str) -> set[str]: return {token for token in TOKEN_RE.split(text.lower()) if token} def entity_identifiers(item: Dict[str, object]) -> set[str]: identifiers = {normalize_name(str(item["name"])), compact_identifier(str(item["name"]))} for alias in item.get("aliases", []): if isinstance(alias, str) and alias.strip(): identifiers.add(normalize_name(alias)) identifiers.add(compact_identifier(alias)) return {identifier for identifier in identifiers if identifier} def merge_lists_preserve(items: Sequence[str], limit: int = 20) -> List[str]: merged = list(dict.fromkeys(item for item in items if item)) return merged[:limit] def choose_canonical_name(current_name: str, candidate_name: str) -> str: current_compact = compact_identifier(current_name) candidate_compact = compact_identifier(candidate_name) if current_compact == candidate_compact: return candidate_name if len(candidate_name) > len(current_name) else current_name if current_compact in candidate_compact and len(candidate_compact) > len(current_compact): return candidate_name if candidate_compact in current_compact and len(current_compact) >= len(candidate_compact): return current_name return candidate_name if len(candidate_name) > len(current_name) else current_name def similarity_score(entity: Dict[str, object], canon: Dict[str, object]) -> float: entity_ids = entity_identifiers(entity) canon_ids = set(canon.get("identifiers", entity_identifiers(canon))) if entity_ids & canon_ids: return 1.0 best = 0.0 for left in entity_ids: for right in canon_ids: if not left or not right: continue ratio = SequenceMatcher(None, left, right).ratio() if len(left) > 3 and len(right) > 3 and (left in right or right in left): ratio = max(ratio, 0.94) best = max(best, ratio) entity_tokens = token_set(str(entity["name"])) canon_tokens = set(canon.get("name_tokens", token_set(str(canon["name"])))) if entity_tokens and canon_tokens: union = entity_tokens | canon_tokens if union: best = max(best, len(entity_tokens & canon_tokens) / len(union)) entity_traits = {trait.lower() for trait in entity.get("visual_anchor_traits", [])} canon_traits = {trait.lower() for trait in canon.get("visual_anchor_traits", [])} if entity_traits and canon_traits: overlap = len(entity_traits & canon_traits) if overlap >= 2: best = max(best, 0.88) return best def find_canon_match(entity: Dict[str, object], canons: List[Dict[str, object]]) -> Tuple[int | None, float]: best_index = None best_score = 0.0 for index, canon in enumerate(canons): score = similarity_score(entity, canon) if score > best_score: best_score = score best_index = index if best_score >= 0.93: return best_index, best_score return None, best_score def ensure_series_memory_shape(series_memory: Dict[str, object]) -> Dict[str, object]: memory = dict(series_memory or {}) memory.setdefault("counters", {"character": 0, "location": 0, "prop": 0}) for _, entity_type in ENTITY_SECTIONS: key = f"{entity_type}s" memory.setdefault(key, []) return memory def create_memory_entry(entity_type: str, entity: Dict[str, object], entity_id: str, episode_key: str) -> Dict[str, object]: return { "entity_id": entity_id, "entity_type": entity_type, "name": entity["name"], "aliases": merge_lists_preserve(entity.get("aliases", []), limit=12), "identifiers": sorted(entity_identifiers(entity)), "name_tokens": sorted(token_set(str(entity["name"]))), "visual_anchor_traits": merge_lists_preserve(entity.get("visual_anchor_traits", []), limit=16), "variant_axes": merge_lists_preserve(entity.get("variant_axes", []), limit=12), "continuity_priority": entity.get("continuity_priority", "low"), "reference_image_priority": entity.get("reference_image_priority", "not_needed"), "episodes": [episode_key], "appearance_count": 1, } def merge_memory_entry(canon: Dict[str, object], entity: Dict[str, object], episode_key: str) -> None: canon["name"] = choose_canonical_name(canon["name"], entity["name"]) canon["aliases"] = merge_lists_preserve([*canon["aliases"], *entity.get("aliases", [])], limit=12) canon["identifiers"] = sorted(entity_identifiers({"name": canon["name"], "aliases": canon["aliases"]})) canon["name_tokens"] = sorted(token_set(canon["name"])) canon["visual_anchor_traits"] = merge_lists_preserve( [*canon["visual_anchor_traits"], *entity.get("visual_anchor_traits", [])], limit=16, ) canon["variant_axes"] = merge_lists_preserve([*canon["variant_axes"], *entity.get("variant_axes", [])], limit=12) if PRIORITY_RANK[entity.get("continuity_priority", "low")] > PRIORITY_RANK[canon["continuity_priority"]]: canon["continuity_priority"] = entity["continuity_priority"] if REFERENCE_RANK[entity.get("reference_image_priority", "not_needed")] > REFERENCE_RANK[canon["reference_image_priority"]]: canon["reference_image_priority"] = entity["reference_image_priority"] if episode_key not in canon["episodes"]: canon["episodes"].append(episode_key) canon["appearance_count"] += 1 def next_series_id(memory: Dict[str, object], entity_type: str) -> str: memory["counters"][entity_type] = int(memory["counters"].get(entity_type, 0)) + 1 prefix = {"character": "CHAR", "location": "LOC", "prop": "PROP"}[entity_type] return f"{prefix}_{memory['counters'][entity_type]:04d}" def localized_backfill_fields(entity_type: str, name: str, language: common.LanguageInfo) -> Dict[str, object]: if language.code == "ko": description = f"{name}에 대한 간이 시각 앵커." reason = "씬 스틸의 visible_entities에서 역보강된 시각 앵커." elif language.code == "ja": description = f"{name} の簡易ビジュアルアンカー。" reason = "scene_stills.visible_entities から補完した視覚アンカー。" else: description = f"Lightweight visual anchor for {name}." reason = "Backfilled from scene_stills.visible_entities." common_fields = { "name": name, "aliases": [], "description": description, "continuity_reason": reason, "visual_anchor_traits": [], "variant_axes": [], "continuity_priority": "medium", "reference_image_priority": "helpful", "evidence": [], } if entity_type in {"character", "location"}: common_fields["importance"] = "supporting" else: common_fields["significance"] = "recurring" if entity_type == "location": common_fields["kind"] = "unknown" return common_fields def build_entity_lookup(entity_index: Dict[str, List[Dict[str, object]]]) -> Dict[Tuple[str, str], Dict[str, object]]: lookup: Dict[Tuple[str, str], Dict[str, object]] = {} for section, entity_type in ENTITY_SECTIONS: for item in entity_index[section]: names = [str(item["name"]), *[alias for alias in item.get("aliases", []) if isinstance(alias, str)]] for name in names: normalized = normalize_name(name) if normalized: lookup[(entity_type, normalized)] = item return lookup def resolve_visible_entity( reference: Dict[str, object], lookup: Dict[Tuple[str, str], Dict[str, object]], candidates: Sequence[Dict[str, object]], ) -> Dict[str, object] | None: entity_type = str(reference["entity_type"]) normalized = normalize_name(str(reference["entity_name"])) exact = lookup.get((entity_type, normalized)) if exact: return exact stub = { "name": str(reference["entity_name"]), "aliases": [], "visual_anchor_traits": [], } best_index, best_score = find_canon_match(stub, list(candidates)) if best_index is not None and best_score >= 0.9: return list(candidates)[best_index] return None def dedupe_top_level_entities(raw_payload: Dict[str, object]) -> Dict[str, List[Dict[str, object]]]: return { "characters": common.merge_entity_lists(raw_payload["characters"], "characters"), "locations": common.merge_entity_lists(raw_payload["locations"], "locations"), "props": common.merge_entity_lists(raw_payload["props"], "props"), } def assign_series_entity_ids( raw_payload: Dict[str, object], series_memory: Dict[str, object], episode_key: str, language: common.LanguageInfo, ) -> Tuple[Dict[str, List[Dict[str, object]]], Dict[str, object]]: memory = ensure_series_memory_shape(series_memory) deduped = dedupe_top_level_entities(raw_payload) stats = {"matched_existing": 0, "created_new": 0, "backfilled_from_scene_refs": 0} for section, entity_type in ENTITY_SECTIONS: memory_key = f"{entity_type}s" canons = memory[memory_key] for item in deduped[section]: match_index, _ = find_canon_match(item, canons) if match_index is None: entity_id = next_series_id(memory, entity_type) canons.append(create_memory_entry(entity_type, item, entity_id, episode_key)) stats["created_new"] += 1 else: entity_id = canons[match_index]["entity_id"] merge_memory_entry(canons[match_index], item, episode_key) stats["matched_existing"] += 1 item["entity_id"] = entity_id item["entity_type"] = entity_type entity_index = {section: list(items) for section, items in deduped.items()} lookup = build_entity_lookup(entity_index) for scene in raw_payload["scene_stills"]: resolved_visible_entities: List[Dict[str, object]] = [] for reference in scene["visible_entities"]: section = {"character": "characters", "location": "locations", "prop": "props"}[reference["entity_type"]] candidates = entity_index[section] matched = resolve_visible_entity(reference, lookup, candidates) if not matched: backfilled = localized_backfill_fields(str(reference["entity_type"]), str(reference["entity_name"]), language) match_index, _ = find_canon_match(backfilled, memory[f"{reference['entity_type']}s"]) if match_index is None: entity_id = next_series_id(memory, str(reference["entity_type"])) memory[f"{reference['entity_type']}s"].append( create_memory_entry(str(reference["entity_type"]), backfilled, entity_id, episode_key) ) else: entity_id = memory[f"{reference['entity_type']}s"][match_index]["entity_id"] merge_memory_entry(memory[f"{reference['entity_type']}s"][match_index], backfilled, episode_key) backfilled["entity_id"] = entity_id backfilled["entity_type"] = str(reference["entity_type"]) entity_index[section].append(backfilled) for name in [backfilled["name"], *backfilled.get("aliases", [])]: normalized = normalize_name(str(name)) if normalized: lookup[(str(reference["entity_type"]), normalized)] = backfilled matched = backfilled stats["backfilled_from_scene_refs"] += 1 resolved_visible_entities.append( { "entity_id": matched["entity_id"], "entity_name": matched["name"], "entity_type": matched["entity_type"], "role": reference["role"], "prominence": reference["prominence"], } ) scene["visible_entities"] = resolved_visible_entities for section in entity_index: entity_index[section].sort(key=lambda item: str(item["entity_id"])) return entity_index, stats def localized_anchor_prefix(language: common.LanguageInfo) -> Tuple[str, str]: if language.code == "ko": return "보이는 핵심 요소", "정지 화면 설명" if language.code == "ja": return "見える重要要素", "静止画説明" return "Visible anchors", "Still frame" def localize_cinematic_text(value: str, language: common.LanguageInfo) -> str: if language.code == "ko": mapping = KO_CINEMATIC_MAP elif language.code == "ja": mapping = JA_CINEMATIC_MAP else: return value localized = value for english, translated in mapping.items(): localized = localized.replace(english, translated) return localized def localize_cinematic_block(block: Dict[str, object], language: common.LanguageInfo) -> Dict[str, object]: return {key: localize_cinematic_text(str(value), language) for key, value in block.items()} def build_still_frame_prompt( raw_prompt: str, visible_entities: Sequence[Dict[str, object]], language: common.LanguageInfo, ) -> str: anchor_label, frame_label = localized_anchor_prefix(language) anchor_tags = ", ".join(f"[{item['entity_id']}] {item['entity_name']}" for item in visible_entities) return f"{anchor_label}: {anchor_tags}. {frame_label}: {raw_prompt}".strip() def localize_summary( *, source_file: str, character_count: int, location_count: int, prop_count: int, scene_count: int, language: common.LanguageInfo, ) -> str: if language.code == "ko": return ( f"{source_file} 씬 스틸 추출 결과: " f"인물 {character_count}개, 배경/장소 {location_count}개, 중요 소품 {prop_count}개, 씬 스틸 {scene_count}개." ) if language.code == "ja": return ( f"{source_file} のシーン静止画抽出結果: " f"人物 {character_count}件、背景・場所 {location_count}件、重要小道具 {prop_count}件、シーン静止画 {scene_count}件。" ) return ( f"Scene still extraction from {source_file}: " f"{character_count} characters, {location_count} locations, {prop_count} key props, {scene_count} stills." ) def enrich_payload( raw_payload: Dict[str, object], *, source_file: str, source_language: common.LanguageInfo, episode_key: str, series_memory: Dict[str, object], detected_headings: Sequence[HeadingCandidate], ) -> Tuple[Dict[str, object], Dict[str, object]]: entity_index, id_stats = assign_series_entity_ids(raw_payload, series_memory, episode_key, source_language) scene_stills: List[Dict[str, object]] = [] heading_counts: Dict[str, int] = {} for idx, scene in enumerate(raw_payload["scene_stills"], start=1): heading = str(scene["screenplay_scene_heading"]).strip() heading_key = normalize_name(heading) heading_counts[heading_key] = heading_counts.get(heading_key, 0) + 1 visible_entities = list(scene["visible_entities"]) scene_stills.append( { "still_id": f"{episode_key}_still_{idx:03d}", "heading_catalog_index": scene["heading_catalog_index"], "screenplay_scene_heading": heading, "page_start": scene["page_start"], "page_end": scene["page_end"], "beat_index_within_heading": scene["beat_index_within_heading"], "still_kind": scene["still_kind"], "beat_title": scene["beat_title"], "visible_entities": visible_entities, "visible_entity_ids": [item["entity_id"] for item in visible_entities], "still_frame_prompt_raw": scene["still_frame_prompt_raw"], "still_frame_prompt": build_still_frame_prompt( str(scene["still_frame_prompt_raw"]), visible_entities, source_language, ), "camera": localize_cinematic_block(scene["camera"], source_language), "lighting": localize_cinematic_block(scene["lighting"], source_language), "evidence": scene["evidence"], } ) payload = { "source_file": source_file, "source_language": { "code": source_language.code, "name": source_language.name, }, "entity_index": entity_index, "scene_stills": scene_stills, "summary": localize_summary( source_file=source_file, character_count=len(entity_index["characters"]), location_count=len(entity_index["locations"]), prop_count=len(entity_index["props"]), scene_count=len(scene_stills), language=source_language, ), "notes": list(dict.fromkeys(raw_payload.get("notes", []))), "scene_heading_catalog": [ {"order_index": item.order_index, "page": item.page, "heading": item.heading} for item in detected_headings ], "extraction_metrics": { "detected_heading_count": len(detected_headings), "covered_heading_catalog_count": len({item["heading_catalog_index"] for item in scene_stills}), "distinct_output_heading_count": len({normalize_name(item["screenplay_scene_heading"]) for item in scene_stills}), "scene_still_count": len(scene_stills), "id_assignment": id_stats, }, } return payload, series_memory def init_scene_sqlite(db_path: Path) -> None: db_path.parent.mkdir(parents=True, exist_ok=True) with sqlite3.connect(db_path) as conn: conn.executescript( """ PRAGMA foreign_keys = ON; CREATE TABLE IF NOT EXISTS scene_extraction_run ( id TEXT PRIMARY KEY, source_file TEXT NOT NULL, episode_key TEXT NOT NULL, provider TEXT NOT NULL, model TEXT NOT NULL, prompt_version TEXT NOT NULL, source_language_code TEXT NOT NULL, source_language_name TEXT NOT NULL, payload_path TEXT, created_at TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS scene_entity ( id TEXT PRIMARY KEY, run_id TEXT NOT NULL REFERENCES scene_extraction_run(id) ON DELETE CASCADE, entity_id TEXT NOT NULL, entity_type TEXT NOT NULL CHECK (entity_type IN ('character', 'location', 'prop')), name TEXT NOT NULL, aliases_json TEXT NOT NULL, description TEXT NOT NULL, continuity_reason TEXT NOT NULL, visual_anchor_traits_json TEXT NOT NULL, variant_axes_json TEXT NOT NULL, continuity_priority TEXT NOT NULL, reference_image_priority TEXT NOT NULL, kind TEXT, importance TEXT, significance TEXT, evidence_json TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS scene_still ( id TEXT PRIMARY KEY, run_id TEXT NOT NULL REFERENCES scene_extraction_run(id) ON DELETE CASCADE, still_id TEXT NOT NULL, heading_catalog_index INTEGER NOT NULL, screenplay_scene_heading TEXT NOT NULL, page_start INTEGER NOT NULL, page_end INTEGER NOT NULL, beat_index_within_heading INTEGER NOT NULL, still_kind TEXT NOT NULL, beat_title TEXT NOT NULL, still_frame_prompt_raw TEXT NOT NULL, still_frame_prompt TEXT NOT NULL, camera_json TEXT NOT NULL, lighting_json TEXT NOT NULL, evidence_json TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS scene_still_visible_entity ( id TEXT PRIMARY KEY, still_row_id TEXT NOT NULL REFERENCES scene_still(id) ON DELETE CASCADE, entity_id TEXT NOT NULL, entity_type TEXT NOT NULL CHECK (entity_type IN ('character', 'location', 'prop')), entity_name TEXT NOT NULL, role TEXT NOT NULL, prominence TEXT NOT NULL, sort_order INTEGER NOT NULL ); """ ) columns = { row[1] for row in conn.execute("PRAGMA table_info(scene_still)") } if "heading_catalog_index" not in columns: conn.execute( "ALTER TABLE scene_still ADD COLUMN heading_catalog_index INTEGER NOT NULL DEFAULT 0" ) def stable_id(prefix: str, *parts: str) -> str: digest = hashlib.sha1("||".join(parts).encode("utf-8")).hexdigest()[:16] return f"{prefix}_{digest}" def store_payload_in_sqlite( *, db_path: Path, input_path: Path, payload_path: Path | None, payload: Dict[str, object], provider: str, model: str, prompt_version: str, episode_key: str, ) -> None: init_scene_sqlite(db_path) run_id = stable_id("scene_run", str(input_path), provider, model, prompt_version, episode_key) with sqlite3.connect(db_path) as conn: conn.execute( """ INSERT OR REPLACE INTO scene_extraction_run ( id, source_file, episode_key, provider, model, prompt_version, source_language_code, source_language_name, payload_path, created_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, datetime('now')) """, ( run_id, payload["source_file"], episode_key, provider, model, prompt_version, payload["source_language"]["code"], payload["source_language"]["name"], str(payload_path) if payload_path else None, ), ) for section, entity_type in ENTITY_SECTIONS: for item in payload["entity_index"][section]: row_id = stable_id("scene_entity", run_id, str(item["entity_id"])) conn.execute( """ INSERT OR REPLACE INTO scene_entity ( id, run_id, entity_id, entity_type, name, aliases_json, description, continuity_reason, visual_anchor_traits_json, variant_axes_json, continuity_priority, reference_image_priority, kind, importance, significance, evidence_json ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( row_id, run_id, item["entity_id"], entity_type, item["name"], json.dumps(item.get("aliases", []), ensure_ascii=False), item.get("description", ""), item.get("continuity_reason", ""), json.dumps(item.get("visual_anchor_traits", []), ensure_ascii=False), json.dumps(item.get("variant_axes", []), ensure_ascii=False), item.get("continuity_priority", "low"), item.get("reference_image_priority", "not_needed"), item.get("kind"), item.get("importance"), item.get("significance"), json.dumps(item.get("evidence", []), ensure_ascii=False), ), ) for item in payload["scene_stills"]: still_row_id = stable_id("scene_still", run_id, item["still_id"]) conn.execute( """ INSERT OR REPLACE INTO scene_still ( id, run_id, still_id, heading_catalog_index, screenplay_scene_heading, page_start, page_end, beat_index_within_heading, still_kind, beat_title, still_frame_prompt_raw, still_frame_prompt, camera_json, lighting_json, evidence_json ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( still_row_id, run_id, item["still_id"], item["heading_catalog_index"], item["screenplay_scene_heading"], item["page_start"], item["page_end"], item["beat_index_within_heading"], item["still_kind"], item["beat_title"], item["still_frame_prompt_raw"], item["still_frame_prompt"], json.dumps(item["camera"], ensure_ascii=False), json.dumps(item["lighting"], ensure_ascii=False), json.dumps(item["evidence"], ensure_ascii=False), ), ) for index, visible in enumerate(item["visible_entities"], start=1): conn.execute( """ INSERT OR REPLACE INTO scene_still_visible_entity ( id, still_row_id, entity_id, entity_type, entity_name, role, prominence, sort_order ) VALUES (?, ?, ?, ?, ?, ?, ?, ?) """, ( stable_id("scene_vis", still_row_id, visible["entity_id"], str(index)), still_row_id, visible["entity_id"], visible["entity_type"], visible["entity_name"], visible["role"], visible["prominence"], index, ), ) def load_series_memory(memory_path: str | None) -> Dict[str, object]: if not memory_path: return {} path = Path(memory_path).expanduser().resolve() try: return json.loads(path.read_text(encoding="utf-8")) except FileNotFoundError as exc: raise RuntimeError(f"Series memory JSON not found: {path}") from exc def render_series_memory_block(series_memory: Dict[str, object]) -> str: if not series_memory: return "No prior series memory is available for this episode." compact = { "characters": [ { "entity_id": item["entity_id"], "name": item["name"], "aliases": item.get("aliases", []), "visual_anchor_traits": item.get("visual_anchor_traits", []), "episodes": item.get("episodes", []), } for item in series_memory.get("characters", []) ], "locations": [ { "entity_id": item["entity_id"], "name": item["name"], "aliases": item.get("aliases", []), "visual_anchor_traits": item.get("visual_anchor_traits", []), "episodes": item.get("episodes", []), } for item in series_memory.get("locations", []) ], "props": [ { "entity_id": item["entity_id"], "name": item["name"], "aliases": item.get("aliases", []), "visual_anchor_traits": item.get("visual_anchor_traits", []), "episodes": item.get("episodes", []), } for item in series_memory.get("props", []) ], } return json.dumps(compact, ensure_ascii=False, indent=2) def extract_scene_stills_for_pdf( *, input_path: Path, output_path: Path | None, sqlite_output_path: Path | None, episode_key: str, provider: str, model: str, fallback_model: str, prompt_version: str | None, source_language_override: str, series_memory: Dict[str, object], temperature: float, ) -> Tuple[Dict[str, object], Dict[str, object]]: page_texts = sanitize_page_texts(common.extract_pdf_pages(input_path)) if not page_texts: raise RuntimeError("No extractable text found in PDF.") source_language = common.resolve_source_language(page_texts, source_language_override) headings = detect_scene_headings(page_texts) heading_catalog = heading_catalog_text(headings) full_text = "\n\n".join(page_texts) bundle = load_prompt_bundle(prompt_version) requested_model = model print(f"Extracting scene stills from {input_path.name}", flush=True) print(f"Provider: {provider}", flush=True) print(f"Model: {model}", flush=True) print(f"Pages with text: {len(page_texts)}", flush=True) print(f"Detected headings: {len(headings)}", flush=True) print(f"Detected language: {source_language.name} ({source_language.code})", flush=True) print(f"Prompt version: {bundle.version}", flush=True) started_at = time.time() raw_payload, model_used = common.extract_chunk_with_fallback( provider=provider, primary_model=model, fallback_model=fallback_model, instructions=render_prompt( bundle.system, source_language_code=source_language.code, source_language_name=source_language.name, ), user_input=render_prompt( bundle.user, source_file=input_path.name, source_language_code=source_language.code, source_language_name=source_language.name, series_memory_block=render_series_memory_block(series_memory), heading_catalog=heading_catalog, screenplay_text=full_text, ), schema_name="screenplay_scene_stills_fulltext", schema=RAW_SCHEMA, temperature=temperature, ) payload, updated_memory = enrich_payload( raw_payload, source_file=input_path.name, source_language=source_language, episode_key=episode_key, series_memory=ensure_series_memory_shape(series_memory), detected_headings=headings, ) payload["extraction_metadata"] = { "provider": provider, "mode": "fulltext", "source_language": { "code": source_language.code, "name": source_language.name, }, "prompt_version": bundle.version, "prompt_manifest": bundle.manifest_path, "prompt_version_path": bundle.version_path, "prompt_files": bundle.prompt_files, "prompt_hashes": bundle.prompt_hashes, "requested_model": requested_model, "fallback_model": fallback_model, "model_used": model_used, "elapsed_seconds": round(time.time() - started_at, 2), } if output_path is not None: output_path.parent.mkdir(parents=True, exist_ok=True) output_path.write_text( json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8", ) print(f"Saved JSON to: {output_path}", flush=True) if sqlite_output_path is not None: store_payload_in_sqlite( db_path=sqlite_output_path, input_path=input_path, payload_path=output_path, payload=payload, provider=provider, model=model_used, prompt_version=bundle.version, episode_key=episode_key, ) print(f"Saved SQLite to: {sqlite_output_path}", flush=True) return payload, updated_memory def main() -> int: args = parse_args() input_path = Path(args.input).expanduser().resolve() output_path = Path(args.output).expanduser().resolve() sqlite_output_path = ( Path(args.sqlite_output).expanduser().resolve() if args.sqlite_output else None ) if not input_path.exists(): print(f"Input PDF not found: {input_path}", file=sys.stderr) return 1 model, fallback_model = common.resolve_models(args.provider, args.model, args.fallback_model) episode_key = args.episode_key or input_path.stem.replace(" ", "_") series_memory = load_series_memory(args.series_memory_json) try: extract_scene_stills_for_pdf( input_path=input_path, output_path=output_path, sqlite_output_path=sqlite_output_path, episode_key=episode_key, provider=args.provider, model=model, fallback_model=fallback_model, prompt_version=args.prompt_version, source_language_override=args.source_language, series_memory=series_memory, temperature=args.temperature, ) except RuntimeError as exc: print(str(exc), file=sys.stderr) return 1 return 0 if __name__ == "__main__": raise SystemExit(main())