"""씬 스틸 추출 모듈 — LLM 기반 스틸 이미지 후보 추출.""" import json import re from pathlib import Path from typing import Any, Dict, List, Optional from app.modules.llm.base import BaseLLMClient PROMPTS_BASE = Path(__file__).resolve().parent.parent.parent.parent / "prompts" / "_base" STILL_PROMPT_DIR = PROMPTS_BASE / "scene_stills" / "v2" SUPPORTED_LANGUAGE_MAP = { "ko": "Korean", "ja": "Japanese", "en": "English", } HEADING_PREFIXES = ("INT.", "EXT.", "INT/EXT.", "I/E.") HEADING_MAX_LEN = 180 # ── JSON Schemas ──────────────────────────────────────────────────── 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": { "angle": {"type": "string"}, # 예: "위에서 내려다보는 구도", "정면", "측면" }, "required": ["angle"], } LIGHTING_SCHEMA: Dict[str, object] = { "type": "object", "additionalProperties": False, "properties": { "mood": {"type": "string"}, # 예: "차가운 형광등", "따뜻한 석양", "어두운 실내" }, "required": ["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", ], } _ENTITY_ITEM_SCHEMA: Dict[str, object] = { "type": "object", "additionalProperties": False, "properties": { "name": {"type": "string"}, "aliases": {"type": "array", "items": {"type": "string"}}, "description": {"type": "string"}, "continuity_reason": {"type": "string"}, "visual_anchor_traits": {"type": "array", "items": {"type": "string"}}, "variant_axes": {"type": "array", "items": {"type": "string"}}, "importance": { "type": "string", "enum": ["major", "supporting", "minor", "unknown"], }, "continuity_priority": { "type": "string", "enum": ["critical", "high", "medium", "low"], }, "reference_image_priority": { "type": "string", "enum": ["required", "helpful", "not_needed"], }, "evidence": {"type": "array", "items": {"type": "string"}}, }, "required": [ "name", "aliases", "description", "continuity_reason", "visual_anchor_traits", "variant_axes", "importance", "continuity_priority", "reference_image_priority", "evidence", ], } _LOCATION_SCHEMA: Dict[str, object] = { "type": "object", "additionalProperties": False, "properties": { **_ENTITY_ITEM_SCHEMA["properties"], "kind": { "type": "string", "enum": ["interior", "exterior", "mixed", "unknown"], }, }, "required": [*_ENTITY_ITEM_SCHEMA["required"], "kind"], } _PROP_SCHEMA: Dict[str, object] = { "type": "object", "additionalProperties": False, "properties": { **{k: v for k, v in _ENTITY_ITEM_SCHEMA["properties"].items() if k != "importance"}, "significance": { "type": "string", "enum": ["key", "recurring", "minor", "unknown"], }, }, "required": [ r if r != "importance" else "significance" for r in _ENTITY_ITEM_SCHEMA["required"] ], } RAW_SCHEMA: Dict[str, object] = { "type": "object", "additionalProperties": False, "properties": { "source_file": {"type": "string"}, "characters": {"type": "array", "items": _ENTITY_ITEM_SCHEMA}, "locations": {"type": "array", "items": _LOCATION_SCHEMA}, "props": {"type": "array", "items": _PROP_SCHEMA}, "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", ], } def _load_prompt(filename: str) -> str: path = STILL_PROMPT_DIR / filename return path.read_text(encoding="utf-8").strip() def detect_scene_headings(fulltext: str) -> List[Dict[str, Any]]: """Detect screenplay scene headings from fulltext.""" headings: List[Dict[str, Any]] = [] for line in fulltext.splitlines(): stripped = line.strip() upper = stripped.upper() for prefix in HEADING_PREFIXES: if upper.startswith(prefix) and len(stripped) <= HEADING_MAX_LEN: headings.append({ "index": len(headings) + 1, "heading": stripped, }) break return headings def _format_heading_catalog(headings: List[Dict[str, Any]]) -> str: lines = [] for h in headings: lines.append(f"{h['index']}. {h['heading']}") return "\n".join(lines) def _build_stills_list(raw: Dict[str, Any]) -> List[Dict[str, Any]]: """Convert raw LLM output to a list of scene still dicts.""" stills: List[Dict[str, Any]] = [] for idx, s in enumerate(raw.get("scene_stills", []), start=1): stills.append({ "still_index": idx, "screenplay_scene_heading": s.get("screenplay_scene_heading", ""), "beat_title": s.get("beat_title", ""), "still_frame_prompt": s.get("still_frame_prompt_raw", ""), "camera": s.get("camera", {}), "lighting": s.get("lighting", {}), "visible_entities": s.get("visible_entities", []), }) return stills def extract_scene_stills( llm_client: BaseLLMClient, fulltext: str, language: str = "ko", source_file: str = "episode", prior_memory: Optional[str] = None, ) -> List[Dict[str, Any]]: """Run scene still extraction on screenplay fulltext. Returns a list of scene still candidate dicts. """ lang_name = SUPPORTED_LANGUAGE_MAP.get(language, "Korean") headings = detect_scene_headings(fulltext) heading_catalog = _format_heading_catalog(headings) system_prompt = _load_prompt("system.md").format( source_language_name=lang_name, source_language_code=language, ) memory_block = prior_memory or "(없음 / None)" user_prompt = _load_prompt("user.md").format( source_file=source_file, source_language_name=lang_name, source_language_code=language, series_memory_block=memory_block, heading_catalog=heading_catalog, screenplay_text=fulltext, ) raw = llm_client.generate_structured( system_prompt=system_prompt, user_prompt=user_prompt, response_schema=RAW_SCHEMA, schema_name="scene_stills", max_tokens=32000, ) return _build_stills_list(raw)