"""웹북 패키지 생성 모듈 — 시나리오 에피소드를 웹북 에피소드로 분할.""" import json from typing import Any, Dict, List, Optional from app.modules.llm.llm_client import call_structured from app.modules.prompt_loader import load_prompt WEBBOOK_PACKAGE_MAX_OUTPUT_TOKENS = 60000 WEBBOOK_TARGET_EPISODE_CHARS = 6500 LANGUAGE_NAMES = { "ko": ("ko", "Korean"), "en": ("en", "English"), } _MODULE = "prototype_prompts" def _load_prompt(filename: str) -> str: stem = filename.replace(".md", "") if filename.endswith(".md") else filename return load_prompt(_MODULE, stem) def _make_webbook_package_schema( web_episode_count: int, sections_per_episode: int ) -> Dict[str, Any]: section_schema: Dict[str, Any] = { "type": "object", "additionalProperties": False, "properties": { "still_id": {"type": "string"}, "section_title": {"type": "string"}, "image_caption": {"type": "string"}, "opening_paragraphs": { "type": "array", "minItems": 1, "maxItems": 3, "items": {"type": "string"}, }, "closing_paragraphs": { "type": "array", "minItems": 1, "maxItems": 3, "items": {"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", "opening_paragraphs", "closing_paragraphs", "paragraphs", "image_after_paragraph", ], } episode_schema: Dict[str, Any] = { "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 _compact_entities(entities: List[Dict[str, Any]]) -> List[Dict[str, Any]]: compact = [] for item in entities: compact.append( { "name": item.get("name", ""), "entity_type": item.get("entity_type", ""), "description": item.get("description", ""), "stable_traits": item.get("stable_traits", "{}"), } ) return compact def _compact_stills(stills: List[Dict[str, Any]]) -> List[Dict[str, Any]]: compact = [] for s in stills: compact.append( { "still_index": s.get("still_index", 0), "id": s.get("id", ""), "screenplay_scene_heading": s.get("screenplay_scene_heading", ""), "beat_title": s.get("beat_title", ""), "still_frame_prompt": s.get("still_frame_prompt", ""), } ) return compact def _episode_character_count(episode: Dict[str, Any]) -> 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", [])) total += sum( len(text) for text in section.get("opening_paragraphs", []) ) total += sum( len(text) for text in section.get("closing_paragraphs", []) ) return total def _validate_still_coverage( payload: Dict[str, Any], available_stills: List[Dict[str, Any]] ) -> List[str]: """Validate still_ids in the package exist in the available stills list.""" available_ids = {s.get("id", "") for s in available_stills} warnings: List[str] = [] for episode in payload.get("episodes", []): for section in episode.get("sections", []): sid = section.get("still_id", "") if sid and sid not in available_ids: warnings.append( f"still_id '{sid}' in episode {episode.get('episode_number')} " f"not found in available stills" ) return warnings def _validate_webbook_package( payload: Dict[str, Any], 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: {', '.join(too_short)}" ) class WebbookGenerator: """웹북 패키지 생성기.""" def __init__(self, project_llm_config: Optional[Dict] = None) -> None: self._project_config = project_llm_config def generate( self, fulltext: str, entities: List[Dict[str, Any]], stills: List[Dict[str, Any]], world_guide: Dict[str, Any], language: str, web_episode_count: int = 4, sections_per_episode: int = 10, ) -> Dict[str, Any]: """Generate webbook package for an episode. Takes screenplay fulltext, entities, scene stills, world guide, and language. Calls LLM via LiteLLM Router for structured JSON response. Returns webbook package dict with episodes, sections, text blocks including opening_paragraphs, image_caption, and closing_paragraphs per section. """ lang_code, lang_name = LANGUAGE_NAMES.get(language, ("ko", "Korean")) system_prompt = _load_prompt("webbook_package_system.md").format( source_language_code=lang_code, source_language_name=lang_name, web_episode_count=web_episode_count, sections_per_episode=sections_per_episode, ) user_prompt = _load_prompt("webbook_package_user.md").format( source_file="episode", 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=fulltext, entities_json=json.dumps( _compact_entities(entities), ensure_ascii=False, indent=2 ), scene_stills_json=json.dumps( _compact_stills(stills), ensure_ascii=False, indent=2 ), ) schema = _make_webbook_package_schema( web_episode_count, sections_per_episode ) last_error: Exception | None = None for attempt in range(1, 4): payload = call_structured( step="webbook_gen", system_prompt=system_prompt, user_prompt=user_prompt, response_schema=schema, project_config=self._project_config, schema_name="prototype_webbook_package", max_tokens=WEBBOOK_PACKAGE_MAX_OUTPUT_TOKENS, ) # Validate uniqueness of still_ids still_ids = [ section["still_id"] for episode in payload["episodes"] for section in episode["sections"] ] if len(still_ids) != len(set(still_ids)): # 중복 still_id는 경고만 — GPT가 가끔 중복 생성하므로 치명적 에러로 처리하지 않음 pass try: _validate_webbook_package( payload, min_episode_chars=WEBBOOK_TARGET_EPISODE_CHARS ) payload["generation_metadata"] = { "prompt_version": "v5", "web_episode_count": web_episode_count, "sections_per_episode": sections_per_episode, "language": lang_code, } coverage_warnings = _validate_still_coverage( payload, stills ) if coverage_warnings: payload["generation_metadata"][ "still_coverage_warnings" ] = coverage_warnings return payload except RuntimeError as exc: last_error = exc if attempt < 3: user_prompt += ( "\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}" )