#!/usr/bin/env python3 """Build a multi-episode webbook prototype for one screenplay episode.""" from __future__ import annotations import argparse import base64 import hashlib import json import os import socket import sys import time import urllib.error import urllib.request from dataclasses import dataclass from pathlib import Path from typing import Dict, List, Sequence, Tuple from fpdf import FPDF from PIL import Image import extract_entities as entity_extractor import extract_scene_stills as still_extractor SCRIPT_DIR = Path(__file__).resolve().parent PROTOTYPE_PROMPTS_DIR = SCRIPT_DIR / "prototype_prompts" PROTOTYPE_PROMPTS_MANIFEST = PROTOTYPE_PROMPTS_DIR / "manifest.json" DEFAULT_GEMINI_IMAGE_MODEL = "gemini-3.1-flash-image-preview" GEMINI_API_URL_TEMPLATE = "https://generativelanguage.googleapis.com/v1beta/models/{model}:generateContent?key={api_key}" APPLE_SD_GOTHIC = "/System/Library/Fonts/AppleSDGothicNeo.ttc" SCENE_TARGET_SIZE = (1600, 900) REFERENCE_TARGETS = { "character": (1024, 1365), "location": (1600, 900), "prop": (1024, 1024), } SCENE_ASPECT_RATIO = "16:9" REFERENCE_ASPECT_RATIO = { "character": "3:4", "location": "16:9", "prop": "1:1", } PROMINENCE_RANK = {"primary": 40, "secondary": 28, "background": 16, "detail": 10} DIRECT_RELATION_FAMILIES = {"identity", "kinship", "social", "conflict", "collaboration", "possession", "membership", "control"} WEBBOOK_PAGE_WIDTH_PT = 399.685 WEBBOOK_TOP_MARGIN_PT = 26 WEBBOOK_SIDE_MARGIN_PT = 36 WEBBOOK_IMAGE_WIDTH_PT = 328 WEBBOOK_IMAGE_HEIGHT_PT = WEBBOOK_IMAGE_WIDTH_PT * (SCENE_TARGET_SIZE[1] / SCENE_TARGET_SIZE[0]) WEBBOOK_MIN_PAGE_HEIGHT_PT = 5200 WEBBOOK_MAX_PAGE_HEIGHT_PT = 18000 WEBBOOK_TARGET_EPISODE_CHARS = 6500 WEBBOOK_PACKAGE_MAX_OUTPUT_TOKENS = 60000 SPECIAL_ENTITY_GUARDRAILS = { "DR.NEX": { "ko": [ "인간 체형 위에 근미래 전술 장비와 철제 가면을 쓴 수준으로 유지하라.", "전신 파워아머, 우주복형 장갑복, 초중장갑 SF 슈트로 과장하지 마라.", ], "en": [ "Keep DR.NEX as a human figure in near-future tactical gear with a metal mask.", "Do not exaggerate into a full power-armor suit, spacesuit, or heavy sci-fi exoshell.", ], }, "한치호": { "ko": [ "현대 특수요원에 가까운 전술복과 보호 장비로 표현하라.", "전신 장갑복이나 메카형 실루엣으로 바꾸지 마라.", ], "en": [ "Render as a modern special-operations soldier with tactical clothing and protective gear.", "Do not convert into a full armored suit or mech-like silhouette.", ], }, "은성": { "ko": [ "지휘형 전술 요원으로 보이게 하되, 장비는 인간 중심 현대 전술 장비의 확장선으로 유지하라.", "초대형 메카 조종사나 SF 중장갑 기사처럼 만들지 마라.", ], "en": [ "Keep Eunseong as a command-oriented tactical operator with lightly extrapolated modern gear.", "Do not render as a giant mech pilot or heavily armored sci-fi knight.", ], }, "서현": { "ko": [ "날카로운 여성 전술 요원/연구 요원 이미지로 유지하라.", "판타지 암살자, 초능력 전사, 갑옷형 슈트로 만들지 마라.", ], "en": [ "Keep Seohyeon as a sharp female tactical/research agent.", "Do not turn her into a fantasy assassin, superpowered warrior, or armored suit character.", ], }, "오리엔티스": { "ko": [ "시설과 병력은 현대 한국 기반의 연구기관+전술조직 수준으로 유지하라.", "우주기지, 스타십 내부, 초미래 군사기지처럼 과장하지 마라.", ], "en": [ "Keep Orientis as a Korean near-future research and tactical facility.", "Do not exaggerate into a spaceship interior or ultra-futuristic military base.", ], }, } WORLD_GUIDE_SCHEMA: Dict[str, object] = { "type": "object", "additionalProperties": False, "properties": { "world_time_period": {"type": "string"}, "world_setting_summary": {"type": "string"}, "technology_level": {"type": "string"}, "era_guardrails": {"type": "array", "minItems": 3, "maxItems": 8, "items": {"type": "string"}}, "costume_guardrails": {"type": "array", "minItems": 3, "maxItems": 8, "items": {"type": "string"}}, "location_guardrails": {"type": "array", "minItems": 2, "maxItems": 6, "items": {"type": "string"}}, "prop_guardrails": {"type": "array", "minItems": 2, "maxItems": 6, "items": {"type": "string"}}, "prohibited_visual_misreads": {"type": "array", "minItems": 4, "maxItems": 12, "items": {"type": "string"}}, "continuity_guardrails": {"type": "array", "minItems": 3, "maxItems": 8, "items": {"type": "string"}}, "image_generation_notes": {"type": "array", "minItems": 2, "maxItems": 6, "items": {"type": "string"}}, }, "required": [ "world_time_period", "world_setting_summary", "technology_level", "era_guardrails", "costume_guardrails", "location_guardrails", "prop_guardrails", "prohibited_visual_misreads", "continuity_guardrails", "image_generation_notes", ], } @dataclass class PromptBundle: version: str world_guide_system: str world_guide_user: str webbook_package_system: str webbook_package_user: str entity_reference_ko: str entity_reference_en: str scene_image_ko: str scene_image_en: str manifest_path: str version_path: str prompt_files: Dict[str, str] prompt_hashes: Dict[str, str] def make_webbook_package_schema(web_episode_count: int, sections_per_episode: int) -> Dict[str, object]: section_schema: Dict[str, object] = { "type": "object", "additionalProperties": False, "properties": { "still_id": {"type": "string"}, "section_title": {"type": "string"}, "image_caption": {"type": "string"}, "paragraphs": { "type": "array", "minItems": 3, "maxItems": 4, "items": {"type": "string"}, }, "image_after_paragraph": {"type": "integer", "minimum": 1, "maximum": 4}, }, "required": ["still_id", "section_title", "image_caption", "paragraphs", "image_after_paragraph"], } episode_schema: Dict[str, object] = { "type": "object", "additionalProperties": False, "properties": { "episode_number": {"type": "integer", "minimum": 1, "maximum": web_episode_count}, "title": {"type": "string"}, "subtitle": {"type": "string"}, "opener_paragraphs": { "type": "array", "minItems": 2, "maxItems": 3, "items": {"type": "string"}, }, "sections": { "type": "array", "minItems": sections_per_episode, "maxItems": sections_per_episode, "items": section_schema, }, "closer_paragraphs": { "type": "array", "minItems": 2, "maxItems": 3, "items": {"type": "string"}, }, }, "required": ["episode_number", "title", "subtitle", "opener_paragraphs", "sections", "closer_paragraphs"], } return { "type": "object", "additionalProperties": False, "properties": { "series_title": {"type": "string"}, "adaptation_subtitle": {"type": "string"}, "logline": {"type": "string"}, "episodes": { "type": "array", "minItems": web_episode_count, "maxItems": web_episode_count, "items": episode_schema, }, }, "required": ["series_title", "adaptation_subtitle", "logline", "episodes"], } def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description="Build a webbook prototype for one screenplay PDF.") parser.add_argument( "--input", default="screenplay/srd part 1 blue revision.pdf", help="Input screenplay PDF.", ) parser.add_argument( "--output-dir", default="prototype/webbook_episode_01", help="Directory for analysis, images, manifests, and rendered PDFs.", ) parser.add_argument( "--web-episode-count", type=int, default=4, help="How many webbook episodes to split the screenplay into.", ) parser.add_argument( "--sections-per-episode", type=int, default=10, help="How many illustrated sections to put inside each webbook episode.", ) parser.add_argument( "--prompt-version", default=None, help="Prototype prompt version. Defaults to manifest current_version.", ) parser.add_argument( "--force", action="store_true", help="Re-run every step even if outputs already exist.", ) parser.add_argument( "--only-scene-ids", default="", help="Comma-separated still_ids to regenerate while reusing the existing package.", ) parser.add_argument( "--review-notes-path", default=None, help="Optional JSON file mapping still_id to reviewer notes for targeted regeneration.", ) return parser.parse_args() def write_json(path: Path, payload: Dict[str, object]) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") def load_json(path: Path) -> Dict[str, object]: return json.loads(path.read_text(encoding="utf-8")) def parse_scene_id_list(raw_value: str | None) -> List[str]: if not raw_value: return [] return list(dict.fromkeys(item.strip() for item in raw_value.split(",") if item.strip())) def load_review_notes(path_value: str | None) -> Dict[str, str]: if not path_value: return {} path = Path(path_value).expanduser().resolve() if not path.exists(): raise RuntimeError(f"Review notes JSON not found: {path}") payload = json.loads(path.read_text(encoding="utf-8")) if not isinstance(payload, dict): raise RuntimeError("Review notes JSON must be an object of {still_id: notes}.") notes: Dict[str, str] = {} for scene_id, note in payload.items(): if note is None: continue cleaned = str(note).strip() if cleaned: notes[str(scene_id)] = cleaned return notes def resolved_sections_per_episode(requested: int, scene_count: int, web_episode_count: int) -> int: if requested <= 0: requested = max(1, min(12, scene_count // max(web_episode_count, 1))) max_possible = max(1, scene_count // max(web_episode_count, 1)) if requested > max_possible: return max_possible return requested def normalize_name(name: str) -> str: return " ".join(name.lower().split()) def load_prompt_bundle(version_override: str | None) -> PromptBundle: manifest = json.loads(PROTOTYPE_PROMPTS_MANIFEST.read_text(encoding="utf-8")) version = version_override or manifest["current_version"] version_entry = manifest["versions"].get(version) if not version_entry: raise RuntimeError(f"Unknown prototype prompt version: {version}") version_dir = PROTOTYPE_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 content = path.read_text(encoding="utf-8").strip() prompt_files[name] = str(path) prompt_hashes[name] = hashlib.sha256(content.encode("utf-8")).hexdigest() return content return PromptBundle( version=version, world_guide_system=read_file("world_guide_system.md"), world_guide_user=read_file("world_guide_user.md"), webbook_package_system=read_file("webbook_package_system.md"), webbook_package_user=read_file("webbook_package_user.md"), entity_reference_ko=read_file("entity_reference_ko.md"), entity_reference_en=read_file("entity_reference_en.md"), scene_image_ko=read_file("scene_image_ko.md"), scene_image_en=read_file("scene_image_en.md"), manifest_path=str(PROTOTYPE_PROMPTS_MANIFEST), version_path=str(version_dir), prompt_files=prompt_files, prompt_hashes=prompt_hashes, ) def extract_screenplay_text(pdf_path: Path) -> Tuple[str, still_extractor.common.LanguageInfo]: page_texts = still_extractor.sanitize_page_texts(entity_extractor.extract_pdf_pages(pdf_path)) if not page_texts: raise RuntimeError("No extractable screenplay text found.") language = entity_extractor.resolve_source_language(page_texts, "auto") return "\n\n".join(page_texts), language def compact_entities_payload(entities_payload: Dict[str, object]) -> Dict[str, object]: compact: Dict[str, object] = { "characters": [], "locations": [], "props": [], "relation_facts": entities_payload.get("relation_facts", []), } for section in ("characters", "locations", "props"): for item in entities_payload.get(section, []): compact[section].append( { "name": item["name"], "aliases": item.get("aliases", []), "description": item.get("description", ""), "visual_anchor_traits": item.get("visual_anchor_traits", []), "variant_axes": item.get("variant_axes", []), "continuity_reason": item.get("continuity_reason", ""), } ) return compact def compact_scene_payload(scene_payload: Dict[str, object]) -> List[Dict[str, object]]: compact: List[Dict[str, object]] = [] for still in scene_payload["scene_stills"]: compact.append( { "still_id": still["still_id"], "heading_catalog_index": still["heading_catalog_index"], "screenplay_scene_heading": still["screenplay_scene_heading"], "beat_title": still["beat_title"], "page_start": still["page_start"], "page_end": still["page_end"], "still_kind": still["still_kind"], "still_frame_prompt_raw": still["still_frame_prompt_raw"], "still_frame_prompt": still["still_frame_prompt"], "camera": still["camera"], "lighting": still["lighting"], "visible_entities": [ { "entity_id": item["entity_id"], "entity_name": item["entity_name"], "entity_type": item["entity_type"], "role": item["role"], "prominence": item["prominence"], } for item in still["visible_entities"] ], "evidence": still["evidence"], } ) return compact def ensure_analysis(input_path: Path, analysis_dir: Path, force: bool) -> Tuple[Path, Path]: entities_path = analysis_dir / "entities.json" stills_path = analysis_dir / "scene_stills.json" analysis_dir.mkdir(parents=True, exist_ok=True) cache_entity = SCRIPT_DIR / "srd_part_1_entities_v5.json" cache_stills = SCRIPT_DIR / "scene_still_results_full_v2" / "openai_fulltext" / "episodes" / "episode_01.json" can_seed_from_episode1_cache = input_path.name == "srd part 1 blue revision.pdf" if not entities_path.exists() and cache_entity.exists() and not force and can_seed_from_episode1_cache: entities_path.write_text(cache_entity.read_text(encoding="utf-8"), encoding="utf-8") if not stills_path.exists() and cache_stills.exists() and not force and can_seed_from_episode1_cache: stills_path.write_text(cache_stills.read_text(encoding="utf-8"), encoding="utf-8") if force or not entities_path.exists(): payload = entity_extractor.extract_entities_for_pdf( input_path=input_path, output_path=entities_path, sqlite_output_path=None, episode_key="prototype_episode_01", provider="openai", model="gpt-5.4", fallback_model="gpt-5", prompt_version=None, source_language_override="auto", mode="fulltext", series_memory={}, chunk_chars=18000, temperature=0.15, ) write_json(entities_path, payload) if force or not stills_path.exists(): payload, _ = still_extractor.extract_scene_stills_for_pdf( input_path=input_path, output_path=stills_path, sqlite_output_path=None, episode_key="prototype_episode_01", provider="openai", model="gpt-5.4", fallback_model="gpt-5", prompt_version=None, source_language_override="auto", series_memory={}, temperature=0.15, ) write_json(stills_path, payload) return entities_path, stills_path def call_openai_structured( *, model: str, instructions: str, user_input: str, schema_name: str, schema: Dict[str, object], max_output_tokens: int | None = None, ) -> Dict[str, object]: return entity_extractor.call_openai_structured( model=model, instructions=instructions, user_input=user_input, schema_name=schema_name, schema=schema, temperature=0.15, max_output_tokens=max_output_tokens, ) def generate_world_guide( *, bundle: PromptBundle, screenplay_text: str, language: still_extractor.common.LanguageInfo, source_file: str, entities_payload: Dict[str, object], scene_payload: Dict[str, object], ) -> Dict[str, object]: instructions = bundle.world_guide_system.format( source_language_code=language.code, source_language_name=language.name, ) user_input = bundle.world_guide_user.format( source_file=source_file, screenplay_text=screenplay_text, entities_json=json.dumps(compact_entities_payload(entities_payload), ensure_ascii=False, indent=2), scene_stills_json=json.dumps(compact_scene_payload(scene_payload), ensure_ascii=False, indent=2), ) return call_openai_structured( model="gpt-5.4", instructions=instructions, user_input=user_input, schema_name="prototype_world_guide", schema=WORLD_GUIDE_SCHEMA, ) def generate_webbook_package( *, bundle: PromptBundle, screenplay_text: str, language: still_extractor.common.LanguageInfo, source_file: str, entities_payload: Dict[str, object], scene_payload: Dict[str, object], world_guide: Dict[str, object], web_episode_count: int, sections_per_episode: int, ) -> Dict[str, object]: schema = make_webbook_package_schema(web_episode_count, sections_per_episode) instructions = bundle.webbook_package_system.format( source_language_code=language.code, source_language_name=language.name, web_episode_count=web_episode_count, sections_per_episode=sections_per_episode, ) user_input = bundle.webbook_package_user.format( source_file=source_file, web_episode_count=web_episode_count, sections_per_episode=sections_per_episode, world_guide_json=json.dumps(world_guide, ensure_ascii=False, indent=2), screenplay_text=screenplay_text, entities_json=json.dumps(compact_entities_payload(entities_payload), ensure_ascii=False, indent=2), scene_stills_json=json.dumps(compact_scene_payload(scene_payload), ensure_ascii=False, indent=2), ) last_error: Exception | None = None for attempt in range(1, 4): payload = call_openai_structured( model="gpt-5.4", instructions=instructions, user_input=user_input, schema_name="prototype_webbook_package", schema=schema, max_output_tokens=WEBBOOK_PACKAGE_MAX_OUTPUT_TOKENS, ) still_ids = [section["still_id"] for episode in payload["episodes"] for section in episode["sections"]] if len(still_ids) != len(set(still_ids)): last_error = RuntimeError("Webbook package reused the same still_id in multiple sections.") else: try: validate_webbook_package(payload, min_episode_chars=WEBBOOK_TARGET_EPISODE_CHARS) return payload except RuntimeError as exc: last_error = exc if attempt < 3: user_input += ( "\n\nRegeneration correction:\n" "- The last output was too short or structurally weak.\n" "- Make each episode materially longer.\n" "- Keep the same strict schema.\n" "- Preserve chronology and still uniqueness.\n" ) raise RuntimeError(f"Failed to generate sufficient webbook package: {last_error}") def episode_character_count(episode: Dict[str, object]) -> int: total = 0 for part in ("opener_paragraphs", "closer_paragraphs"): total += sum(len(text) for text in episode.get(part, [])) for section in episode.get("sections", []): total += len(section.get("section_title", "")) total += len(section.get("image_caption", "")) total += sum(len(text) for text in section.get("paragraphs", [])) return total def validate_webbook_package(payload: Dict[str, object], min_episode_chars: int) -> None: episodes = payload.get("episodes", []) if not isinstance(episodes, list) or not episodes: raise RuntimeError("Webbook package did not include episodes.") too_short: List[str] = [] for episode in episodes: char_count = episode_character_count(episode) if char_count < min_episode_chars: too_short.append(f"{episode.get('episode_number')}:{char_count}") if too_short: raise RuntimeError(f"Webbook episodes were too short for sample-style output: {', '.join(too_short)}") def build_entity_lookup(scene_payload: Dict[str, object], entities_payload: Dict[str, object]) -> Dict[str, Dict[str, object]]: lookup: Dict[str, Dict[str, object]] = {} for section in ("characters", "locations", "props"): for item in scene_payload["entity_index"][section]: lookup[item["entity_id"]] = dict(item) extra_by_name: Dict[Tuple[str, str], Dict[str, object]] = {} for section, entity_type in (("characters", "character"), ("locations", "location"), ("props", "prop")): for item in entities_payload.get(section, []): extra_by_name[(entity_type, normalize_name(item["name"]))] = item for entity in lookup.values(): extra = extra_by_name.get((entity["entity_type"], normalize_name(entity["name"]))) if extra: entity["description"] = extra.get("description", "") entity["visual_anchor_traits"] = extra.get("visual_anchor_traits", []) entity["continuity_reason"] = extra.get("continuity_reason", "") return lookup def build_still_lookup(scene_payload: Dict[str, object]) -> Dict[str, Dict[str, object]]: return {item["still_id"]: item for item in scene_payload["scene_stills"]} def visible_entity_priority( visible_entity: Dict[str, object], entity_lookup: Dict[str, Dict[str, object]], relation_facts: Sequence[Dict[str, object]], visible_name_set: set[str], ) -> int: score = PROMINENCE_RANK[visible_entity["prominence"]] entity_record = entity_lookup.get(visible_entity["entity_id"], {}) score += {"critical": 8, "high": 5, "medium": 3, "low": 1}.get(entity_record.get("continuity_priority", "medium"), 3) if visible_entity["entity_type"] == "location": score += 4 if visible_entity["entity_type"] == "prop": score += 2 normalized_name = normalize_name(visible_entity["entity_name"]) for relation in relation_facts: participants = relation.get("participants", []) participant_names = {normalize_name(item["entity_name"]) for item in participants} if normalized_name in participant_names and len(participant_names & visible_name_set) >= 2: score += 6 if relation.get("relation_family") in DIRECT_RELATION_FAMILIES else 3 return score def build_relation_context(still: Dict[str, object], relation_facts: Sequence[Dict[str, object]]) -> List[str]: visible_names = {normalize_name(item["entity_name"]) for item in still["visible_entities"]} lines: List[str] = [] for relation in relation_facts: participants = relation.get("participants", []) participant_names = {normalize_name(item["entity_name"]) for item in participants} if len(participant_names & visible_names) < 2: continue named = ", ".join(item["entity_name"] for item in participants if normalize_name(item["entity_name"]) in visible_names) lines.append(f"{named}: {relation.get('relation_type')} / {relation.get('continuity_reason')}") if len(lines) >= 3: break return lines def should_include_previous_scene(current_still: Dict[str, object], previous_still: Dict[str, object] | None) -> bool: if not previous_still: return False current_ids = set(current_still["visible_entity_ids"]) previous_ids = set(previous_still["visible_entity_ids"]) current_locations = {item["entity_id"] for item in current_still["visible_entities"] if item["entity_type"] == "location"} previous_locations = {item["entity_id"] for item in previous_still["visible_entities"] if item["entity_type"] == "location"} if len(current_ids & previous_ids) >= 2: return True if current_locations & previous_locations and current_ids & previous_ids: return True if abs(int(current_still["heading_catalog_index"]) - int(previous_still["heading_catalog_index"])) <= 2 and current_ids & previous_ids: return True return False def infer_mime_type(path: Path) -> str: if path.suffix.lower() in {".jpg", ".jpeg"}: return "image/jpeg" return "image/png" def encode_image_part(path: Path) -> Dict[str, object]: return { "inlineData": { "mimeType": infer_mime_type(path), "data": base64.b64encode(path.read_bytes()).decode("ascii"), } } def call_gemini_image_generate( *, model: str, prompt: str, reference_paths: Sequence[Path], aspect_ratio: str, ) -> Tuple[bytes, Dict[str, object]]: api_key = os.environ.get("GEMINI_API_KEY") if not api_key: raise RuntimeError("GEMINI_API_KEY is not set.") parts: List[Dict[str, object]] = [encode_image_part(path) for path in reference_paths] parts.append({"text": prompt}) body = { "contents": [{"parts": parts}], "generationConfig": { "responseModalities": ["TEXT", "IMAGE"], "imageConfig": {"aspectRatio": aspect_ratio}, }, } url = GEMINI_API_URL_TEMPLATE.format(model=model, api_key=api_key) request = urllib.request.Request( url, data=json.dumps(body).encode("utf-8"), headers={"Content-Type": "application/json"}, method="POST", ) last_error: Exception | None = None for attempt in range(1, 4): try: with urllib.request.urlopen(request, timeout=600) as response: payload = json.loads(response.read().decode("utf-8")) break except urllib.error.HTTPError as exc: error_text = exc.read().decode("utf-8", errors="replace") last_error = RuntimeError(f"Gemini image API error {exc.code}: {error_text}") if exc.code in {429, 500, 502, 503, 504} and attempt < 3: time.sleep(2 * attempt) continue raise RuntimeError(f"Gemini image API error {exc.code}: {error_text}") from exc except (urllib.error.URLError, socket.timeout) as exc: last_error = exc if attempt < 3: time.sleep(2 * attempt) continue raise RuntimeError(f"Gemini image API request failed after retries: {exc}") from exc else: raise RuntimeError(f"Gemini image API request failed: {last_error}") candidates = payload.get("candidates", []) for candidate in candidates: content = candidate.get("content", {}) for part in content.get("parts", []): inline_data = part.get("inlineData") if isinstance(inline_data, dict) and inline_data.get("data"): return base64.b64decode(inline_data["data"]), payload raise RuntimeError(f"Gemini image API returned no image parts: {payload}") def normalize_and_save_image(image_bytes: bytes, output_path: Path, target_size: Tuple[int, int]) -> None: output_path.parent.mkdir(parents=True, exist_ok=True) tmp_path = output_path.parent / f".{output_path.stem}.tmp" tmp_path.write_bytes(image_bytes) image = Image.open(tmp_path).convert("RGB") src_w, src_h = image.size target_w, target_h = target_size src_ratio = src_w / src_h target_ratio = target_w / target_h if src_ratio > target_ratio: new_h = target_h new_w = int(new_h * src_ratio) else: new_w = target_w new_h = int(new_w / src_ratio) image = image.resize((new_w, new_h), Image.LANCZOS) left = max(0, (new_w - target_w) // 2) top = max(0, (new_h - target_h) // 2) image = image.crop((left, top, left + target_w, top + target_h)) image.save(output_path, format="PNG") tmp_path.unlink(missing_ok=True) def relation_priority_summary(still: Dict[str, object], relation_facts: Sequence[Dict[str, object]]) -> List[str]: return build_relation_context(still, relation_facts) def build_world_guide_block(world_guide: Dict[str, object]) -> str: summary = world_guide["world_setting_summary"] if len(summary) > 220: summary = summary[:217].rstrip() + "..." lines = [ f"시대/배경: {world_guide['world_time_period']}", f"세계 요약: {summary}", f"기술 수준: {world_guide['technology_level']}", "핵심 시대 규칙:", *[f"- {item}" for item in world_guide["era_guardrails"][:3]], "핵심 복장 규칙:", *[f"- {item}" for item in world_guide["costume_guardrails"][:3]], "금지 오독:", *[f"- {item}" for item in world_guide["prohibited_visual_misreads"][:4]], "연속성 규칙:", *[f"- {item}" for item in world_guide["continuity_guardrails"][:3]], ] return "\n".join(lines) def format_bulleted_block(items: Sequence[str], empty_line: str) -> str: if not items: return f"- {empty_line}" return "\n".join(f"- {item}" for item in items) def entity_type_label(entity_type: str, language_code: str) -> str: labels = { "ko": {"character": "인물", "location": "배경/장소", "prop": "중요 소품"}, "en": {"character": "character", "location": "location", "prop": "prop"}, } return labels["ko" if language_code == "ko" else "en"][entity_type] def entity_type_reference_rules(entity_type: str, language_code: str) -> List[str]: if language_code == "ko": mapping = { "character": [ "중립적인 기본 자세로 보여 주고, 얼굴과 체형 식별이 우선이다.", "장면성 포즈나 포박 상태보다 기본 외형과 기본 복장 톤을 우선한다.", "현대 한국/근미래 한국 기준의 현실적 기본 복장을 사용하라.", ], "location": [ "공간 구조가 한눈에 들어오는 establishing 구도로 표현하라.", "군중이나 주연 인물은 넣지 않거나 최소화하라.", "공간 재질, 조명 구조, 출입 동선, 핵심 설비를 분명히 보여라.", ], "prop": [ "사람 손이나 몸에서 떼어 낸 단독 기준 이미지로 보여라.", "형태, 재질, 장착 구조, 조작부가 선명해야 한다.", "장비의 크기감이 읽히도록 깔끔하게 배치하라.", ], } else: mapping = { "character": [ "Use a neutral baseline pose and prioritize face and body recognition.", "Prefer stable appearance over scene-specific action or restraint state.", "Use realistic present-day / near-future Korean baseline clothing.", ], "location": [ "Use an establishing composition that clearly explains the space.", "Avoid crowds and named characters unless absolutely necessary.", "Show material, lighting structure, circulation path, and key built-in devices.", ], "prop": [ "Show it as an isolated reference without hands or a human body.", "Make shape, material, attachment structure, and controls clearly readable.", "Present it cleanly so scale and silhouette are easy to understand.", ], } return mapping[entity_type] def special_entity_guardrail_lines(names: Sequence[str], language_code: str) -> List[str]: locale = "ko" if language_code == "ko" else "en" lines: List[str] = [] for name in names: entry = SPECIAL_ENTITY_GUARDRAILS.get(name) if not entry: continue lines.extend(entry[locale]) return lines def reference_description(entity: Dict[str, object], language_code: str) -> str: if entity["name"] == "DR.NEX": if language_code == "ko": return "철제 가면을 쓴 인간 리더. 과장된 전신 갑옷이 아니라 인간 체형 위의 근미래 전술 장비와 금속 마스크가 핵심이다." return "A human leader with a metal mask. The key is near-future tactical gear on a human body, not exaggerated full-body armor." if entity["entity_type"] == "prop": if language_code == "ko": return f"{entity['name']}의 기준 형태, 재질, 구조를 보여 주는 중립 설명." return f"Neutral description focused on the stable shape, material, and structure of {entity['name']}." if entity["entity_type"] == "location": if language_code == "ko": return entity.get("description", entity["name"]) + " 사람보다 공간 구조가 우선이다." return entity.get("description", entity["name"]) + " Prioritize the space itself over people." return entity.get("description", entity["name"]) def localized_entity_reference_prompt( bundle: PromptBundle, entity: Dict[str, object], world_guide: Dict[str, object], language_code: str, ) -> str: description = reference_description(entity, language_code) world_block = build_world_guide_block(world_guide) traits_block = format_bulleted_block(entity.get("visual_anchor_traits", [])[:6], entity["name"]) rule_items = entity_type_reference_rules(entity["entity_type"], language_code) + special_entity_guardrail_lines([entity["name"]], language_code) rules_block = format_bulleted_block(rule_items, "기본 식별 규칙을 유지하라.") template = bundle.entity_reference_ko if language_code == "ko" else bundle.entity_reference_en return template.format( entity_id=entity["entity_id"], entity_name=entity["name"], entity_type_label=entity_type_label(entity["entity_type"], language_code), description=description, traits_block=traits_block, world_guide_block=world_block, entity_type_rules=rules_block, aspect_ratio=REFERENCE_ASPECT_RATIO[entity["entity_type"]], ) def build_scene_generation_prompt( *, bundle: PromptBundle, still: Dict[str, object], ordered_visible_entities: Sequence[Dict[str, object]], include_previous_scene: bool, previous_still: Dict[str, object] | None, world_guide: Dict[str, object], relation_lines: Sequence[str], language_code: str, manual_fix_note: str = "", ) -> str: world_block = build_world_guide_block(world_guide) visible_names = [item["entity_name"] for item in ordered_visible_entities] special_guardrails = special_entity_guardrail_lines(visible_names, language_code) if language_code == "ko": reference_lines = [ f"- 참조 이미지 {index}: [{visible['entity_id']}] {visible['entity_name']} ({entity_type_label(visible['entity_type'], 'ko')}) 기준 이미지. " "정체성과 형태를 고정하고, 장면 상태는 아래 지시를 따른다." for index, visible in enumerate(ordered_visible_entities, start=1) ] if include_previous_scene and previous_still is not None: reference_lines.append( f"- 참조 이미지 {len(ordered_visible_entities) + 1}: 바로 앞의 연속 장면 이미지. " f"이전 장면 제목은 '{previous_still['beat_title']}' 이다. 의상 상태와 위치 연속성을 이어라." ) else: reference_lines = [ f"- Reference image {index}: [{visible['entity_id']}] {visible['entity_name']} ({entity_type_label(visible['entity_type'], 'en')}) anchor. " "Preserve identity and shape; use this scene description for actual wardrobe and state." for index, visible in enumerate(ordered_visible_entities, start=1) ] if include_previous_scene and previous_still is not None: reference_lines.append( f"- Reference image {len(ordered_visible_entities) + 1}: previous related scene image from '{previous_still['beat_title']}'. " "Carry wardrobe and spatial continuity forward." ) template = bundle.scene_image_ko if language_code == "ko" else bundle.scene_image_en prompt = template.format( world_guide_block=world_block, beat_title=still["beat_title"], scene_heading=still["screenplay_scene_heading"], still_frame_prompt_raw=still["still_frame_prompt_raw"], still_kind=still["still_kind"], reference_block="\n".join(reference_lines + [f"- {item}" for item in special_guardrails]), relation_block=format_bulleted_block(relation_lines, "직접 관계 없음." if language_code == "ko" else "No direct relation note."), evidence_block=format_bulleted_block(still["evidence"][:3], "근거 없음." if language_code == "ko" else "No evidence line."), camera_block=format_bulleted_block([f"{key}: {value}" for key, value in still["camera"].items()], "카메라 지시 없음." if language_code == "ko" else "No camera note."), lighting_block=format_bulleted_block([f"{key}: {value}" for key, value in still["lighting"].items()], "조명 지시 없음." if language_code == "ko" else "No lighting note."), aspect_ratio=SCENE_ASPECT_RATIO, ) cleaned_note = manual_fix_note.strip() if cleaned_note: heading = "[수정 우선 메모]" if language_code == "ko" else "[Manual fix notes]" intro = ( "- 아래 검수 메모를 이번 재생성에서 우선 반영하라." if language_code == "ko" else "- Prioritize the following reviewer notes in this regeneration." ) note_lines = [f"- {line.strip()}" for line in cleaned_note.splitlines() if line.strip()] prompt = "\n\n".join([prompt, heading, "\n".join([intro] + note_lines)]) return prompt def episode_file_stub(episode_number: int, episode_title: str) -> str: cleaned = "".join(char if char.isalnum() else "_" for char in episode_title).strip("_").lower() if not cleaned: cleaned = f"episode_{episode_number:02d}" return f"web_episode_{episode_number:02d}_{cleaned}" def new_webbook_pdf(page_height: float) -> FPDF: pdf = FPDF(unit="pt", format=(WEBBOOK_PAGE_WIDTH_PT, page_height)) pdf.set_auto_page_break(auto=False) pdf.add_font("AppleSDGothic", "", APPLE_SD_GOTHIC) pdf.set_margins(WEBBOOK_SIDE_MARGIN_PT, WEBBOOK_TOP_MARGIN_PT, WEBBOOK_SIDE_MARGIN_PT) pdf.add_page() return pdf def draw_webbook_episode( pdf: FPDF, *, package_title: str, episode: Dict[str, object], section_images: Dict[str, Path], ) -> None: content_w = WEBBOOK_PAGE_WIDTH_PT - (WEBBOOK_SIDE_MARGIN_PT * 2) image_x = (WEBBOOK_PAGE_WIDTH_PT - WEBBOOK_IMAGE_WIDTH_PT) / 2 pdf.set_xy(WEBBOOK_SIDE_MARGIN_PT, WEBBOOK_TOP_MARGIN_PT) pdf.set_font("AppleSDGothic", size=8.5) pdf.set_text_color(120, 120, 120) pdf.multi_cell(content_w, 12, f"{package_title} WEBBOOK", align="L") pdf.ln(8) pdf.set_font("AppleSDGothic", size=10) pdf.set_text_color(130, 130, 130) pdf.multi_cell(content_w, 14, f"EPISODE {episode['episode_number']}", align="C") pdf.ln(6) pdf.set_text_color(18, 18, 18) pdf.set_font("AppleSDGothic", size=22) pdf.multi_cell(content_w, 28, episode["title"], align="C") pdf.ln(2) pdf.set_font("AppleSDGothic", size=11) pdf.set_text_color(90, 90, 90) pdf.multi_cell(content_w, 16, episode["subtitle"], align="C") pdf.ln(16) pdf.set_font("AppleSDGothic", size=10.8) pdf.set_text_color(35, 35, 35) for paragraph in episode["opener_paragraphs"]: pdf.multi_cell(content_w, 17, paragraph, align="J") pdf.ln(8) for section in episode["sections"]: image_path = section_images[section["still_id"]] image_after = int(section["image_after_paragraph"]) for paragraph_index, paragraph in enumerate(section["paragraphs"], start=1): pdf.multi_cell(content_w, 17, paragraph, align="J") pdf.ln(8) if paragraph_index == image_after: pdf.image(str(image_path), x=image_x, w=WEBBOOK_IMAGE_WIDTH_PT, h=WEBBOOK_IMAGE_HEIGHT_PT) pdf.ln(WEBBOOK_IMAGE_HEIGHT_PT + 14) pdf.ln(8) for paragraph in episode.get("closer_paragraphs", []): pdf.multi_cell(content_w, 17, paragraph, align="J") pdf.ln(8) def estimate_episode_page_height( *, package_title: str, episode: Dict[str, object], section_images: Dict[str, Path], ) -> float: pdf = new_webbook_pdf(WEBBOOK_MAX_PAGE_HEIGHT_PT) with pdf.offset_rendering() as dummy: draw_webbook_episode( dummy, package_title=package_title, episode=episode, section_images=section_images, ) estimated = dummy.get_y() + 72 return max(WEBBOOK_MIN_PAGE_HEIGHT_PT, min(WEBBOOK_MAX_PAGE_HEIGHT_PT, estimated)) def render_episode_pdf( *, package_title: str, episode: Dict[str, object], section_images: Dict[str, Path], output_path: Path, ) -> None: page_height = estimate_episode_page_height( package_title=package_title, episode=episode, section_images=section_images, ) pdf = new_webbook_pdf(page_height) draw_webbook_episode( pdf, package_title=package_title, episode=episode, section_images=section_images, ) output_path.parent.mkdir(parents=True, exist_ok=True) pdf.output(str(output_path)) def main() -> int: args = parse_args() input_path = Path(args.input).expanduser().resolve() if not input_path.exists(): print(f"Input screenplay not found: {input_path}", file=sys.stderr) return 1 requested_scene_ids = parse_scene_id_list(args.only_scene_ids) review_notes = load_review_notes(args.review_notes_path) output_dir = Path(args.output_dir).expanduser().resolve() analysis_dir = output_dir / "analysis" reference_dir = output_dir / "reference_images" scene_dir = output_dir / "scene_images" rendered_dir = output_dir / "rendered" metadata_dir = output_dir / "metadata" bundle = load_prompt_bundle(args.prompt_version) screenplay_text, language = extract_screenplay_text(input_path) entities_path, stills_path = ensure_analysis(input_path, analysis_dir, args.force) entities_payload = load_json(entities_path) scene_payload = load_json(stills_path) scene_count = len(scene_payload.get("scene_stills", [])) sections_per_episode = resolved_sections_per_episode(args.sections_per_episode, scene_count, args.web_episode_count) world_guide_path = metadata_dir / "world_guide.json" if args.force or not world_guide_path.exists(): world_guide = generate_world_guide( bundle=bundle, screenplay_text=screenplay_text, language=language, source_file=input_path.name, entities_payload=entities_payload, scene_payload=scene_payload, ) write_json(world_guide_path, world_guide) else: world_guide = load_json(world_guide_path) package_path = metadata_dir / "webbook_package.json" if args.force or not package_path.exists(): webbook_package = generate_webbook_package( bundle=bundle, screenplay_text=screenplay_text, language=language, source_file=input_path.name, entities_payload=entities_payload, scene_payload=scene_payload, world_guide=world_guide, web_episode_count=args.web_episode_count, sections_per_episode=sections_per_episode, ) webbook_package["generation_metadata"] = { "model": "gpt-5.4", "prompt_version": bundle.version, "prompt_manifest": bundle.manifest_path, "prompt_files": bundle.prompt_files, "prompt_hashes": bundle.prompt_hashes, "world_guide_model": "gpt-5.4", "web_episode_count": args.web_episode_count, "sections_per_episode": sections_per_episode, } write_json(package_path, webbook_package) else: webbook_package = load_json(package_path) entity_lookup = build_entity_lookup(scene_payload, entities_payload) still_lookup = build_still_lookup(scene_payload) relation_facts = entities_payload.get("relation_facts", []) selected_still_ids = [ section["still_id"] for episode in webbook_package["episodes"] for section in episode["sections"] ] selected_still_ids = list(dict.fromkeys(selected_still_ids)) invalid_scene_ids = [scene_id for scene_id in requested_scene_ids if scene_id not in selected_still_ids] if invalid_scene_ids: print( f"Unknown still_id(s) for this package: {', '.join(invalid_scene_ids)}", file=sys.stderr, ) return 1 targeted_scene_ids = set(requested_scene_ids) if targeted_scene_ids: print( "[scene-select] targeted regeneration for " + ", ".join(requested_scene_ids), flush=True, ) used_entity_ids: List[str] = [] for still_id in selected_still_ids: used_entity_ids.extend(still_lookup[still_id]["visible_entity_ids"]) used_entity_ids = list(dict.fromkeys(used_entity_ids)) manifest_path = metadata_dir / "image_generation_manifest.json" if manifest_path.exists() and not args.force: image_manifest = load_json(manifest_path) else: image_manifest = {} image_manifest.update( { "model": DEFAULT_GEMINI_IMAGE_MODEL, "prompt_version": bundle.version, "prompt_manifest": bundle.manifest_path, "image_prompt_files": { "entity_reference_ko.md": bundle.prompt_files["entity_reference_ko.md"], "entity_reference_en.md": bundle.prompt_files["entity_reference_en.md"], "scene_image_ko.md": bundle.prompt_files["scene_image_ko.md"], "scene_image_en.md": bundle.prompt_files["scene_image_en.md"], }, "image_prompt_hashes": { "entity_reference_ko.md": bundle.prompt_hashes["entity_reference_ko.md"], "entity_reference_en.md": bundle.prompt_hashes["entity_reference_en.md"], "scene_image_ko.md": bundle.prompt_hashes["scene_image_ko.md"], "scene_image_en.md": bundle.prompt_hashes["scene_image_en.md"], }, "world_guide": world_guide, } ) image_manifest.setdefault("entities", {}) image_manifest.setdefault("scenes", {}) reference_dir.mkdir(parents=True, exist_ok=True) for entity_id in used_entity_ids: entity = entity_lookup[entity_id] out_path = reference_dir / f"{entity_id}.png" if args.force or not out_path.exists(): prompt = localized_entity_reference_prompt(bundle, entity, world_guide, language.code) print(f"[entity-ref] {entity_id} {entity['name']}", flush=True) image_bytes, payload = call_gemini_image_generate( model=DEFAULT_GEMINI_IMAGE_MODEL, prompt=prompt, reference_paths=[], aspect_ratio=REFERENCE_ASPECT_RATIO[entity["entity_type"]], ) normalize_and_save_image(image_bytes, out_path, REFERENCE_TARGETS[entity["entity_type"]]) image_manifest["entities"][entity_id] = { "path": str(out_path), "prompt": prompt, "aspect_ratio": REFERENCE_ASPECT_RATIO[entity["entity_type"]], "target_size": REFERENCE_TARGETS[entity["entity_type"]], "response_model_version": payload.get("modelVersion"), } else: image_manifest["entities"].setdefault(entity_id, {"path": str(out_path)}) scene_dir.mkdir(parents=True, exist_ok=True) section_images: Dict[str, Path] = {} previous_generated_still: Dict[str, object] | None = None previous_generated_path: Path | None = None for episode in webbook_package["episodes"]: for section in episode["sections"]: still = still_lookup[section["still_id"]] visible_name_set = {normalize_name(item["entity_name"]) for item in still["visible_entities"]} ordered_visible_entities = sorted( still["visible_entities"], key=lambda item: visible_entity_priority(item, entity_lookup, relation_facts, visible_name_set), reverse=True, ) reference_paths = [reference_dir / f"{item['entity_id']}.png" for item in ordered_visible_entities] include_previous = should_include_previous_scene(still, previous_generated_still) and previous_generated_path is not None if include_previous and previous_generated_path is not None: reference_paths.append(previous_generated_path) relation_lines = relation_priority_summary(still, relation_facts) manual_fix_note = review_notes.get(section["still_id"], "") scene_prompt = build_scene_generation_prompt( bundle=bundle, still=still, ordered_visible_entities=ordered_visible_entities, include_previous_scene=include_previous, previous_still=previous_generated_still, world_guide=world_guide, relation_lines=relation_lines, language_code=language.code, manual_fix_note=manual_fix_note, ) out_path = scene_dir / f"{section['still_id']}.png" should_generate_scene = ( args.force or not out_path.exists() or (bool(targeted_scene_ids) and section["still_id"] in targeted_scene_ids) ) if should_generate_scene: print(f"[scene-img] {section['still_id']} {still['beat_title']}", flush=True) image_bytes, payload = call_gemini_image_generate( model=DEFAULT_GEMINI_IMAGE_MODEL, prompt=scene_prompt, reference_paths=reference_paths, aspect_ratio=SCENE_ASPECT_RATIO, ) normalize_and_save_image(image_bytes, out_path, SCENE_TARGET_SIZE) scene_manifest_entry = { "path": str(out_path), "prompt": scene_prompt, "reference_paths": [str(path) for path in reference_paths], "target_size": SCENE_TARGET_SIZE, "aspect_ratio": SCENE_ASPECT_RATIO, "response_model_version": payload.get("modelVersion"), } if manual_fix_note: scene_manifest_entry["manual_fix_note"] = manual_fix_note image_manifest["scenes"][section["still_id"]] = scene_manifest_entry else: image_manifest["scenes"].setdefault(section["still_id"], {"path": str(out_path)}) section_images[section["still_id"]] = out_path previous_generated_still = still previous_generated_path = out_path if out_path.exists() else None write_json(manifest_path, image_manifest) rendered_dir.mkdir(parents=True, exist_ok=True) rendered_files: List[str] = [] for episode in webbook_package["episodes"]: stub = episode_file_stub(episode["episode_number"], episode["title"]) pdf_path = rendered_dir / f"{stub}.pdf" render_episode_pdf( package_title=webbook_package["series_title"], episode=episode, section_images=section_images, output_path=pdf_path, ) rendered_files.append(str(pdf_path)) summary = { "source_file": input_path.name, "world_guide_path": str(world_guide_path), "webbook_package_path": str(package_path), "image_manifest_path": str(manifest_path), "rendered_episode_pdfs": rendered_files, "selected_still_count": len(selected_still_ids), "reference_entity_count": len(used_entity_ids), "web_episode_count": len(webbook_package["episodes"]), "sections_per_episode": len(webbook_package["episodes"][0]["sections"]) if webbook_package["episodes"] else 0, "scene_still_catalog_count": scene_count, } write_json(metadata_dir / "prototype_summary.json", summary) print(f"World guide JSON: {world_guide_path}") print(f"Webbook package JSON: {package_path}") print(f"Image manifest JSON: {manifest_path}") print("Rendered PDFs:") for path in rendered_files: print(f" - {path}") return 0 if __name__ == "__main__": raise SystemExit(main())