"""기획서 분석 step — PDF를 LLM에 직접 전달하여 구조화된 섹션 추출. PDF 파일이 있으면 multimodal (base64)로 직접 분석. 없으면 DB의 planning_doc_text로 fallback. """ import base64 import logging from pathlib import Path from typing import Dict from app.core.config import settings from app.core.step_runner import StepRunner logger = logging.getLogger(__name__) _ANALYSIS_SCHEMA = { "type": "object", "properties": { "characters": { "type": "array", "items": { "type": "object", "properties": { "name": {"type": "string"}, "description": {"type": "string"}, "visual_traits": {"type": "string"}, "age_gender": {"type": "string"}, "role": {"type": "string"}, }, "required": ["name", "description"], "additionalProperties": False, }, }, # ★장소 칸 (2026-09-18). 종전엔 인물만 구조로 받고 장소는 # `world_setting` 문자열에 뭉개져, 장소 추출·상세 단계가 기획서를 # **아예 못 봤다**. 인물과 같은 모양으로 받는다. "locations": { "type": "array", "items": { "type": "object", "properties": { "name": {"type": "string"}, "description": {"type": "string"}, "visual_traits": {"type": "string"}, }, "required": ["name", "description"], "additionalProperties": False, }, }, "world_setting": {"type": "string"}, "tone_mood": {"type": "string"}, "story_arc": {"type": "string"}, "visual_concepts": {"type": "string"}, "key_relationships": { "type": "array", "items": { "type": "object", "properties": { "characters": {"type": "string"}, "relationship": {"type": "string"}, }, "required": ["characters", "relationship"], "additionalProperties": False, }, }, "available_sections": { "type": "array", "items": {"type": "string"}, }, }, "required": ["characters", "locations", "world_setting", "tone_mood", "story_arc", "visual_concepts", "key_relationships", "available_sections"], "additionalProperties": False, } _SYSTEM_PROMPT = """당신은 영화/드라마 기획서 분석 전문가입니다. 주어진 기획서(PDF 또는 텍스트)에서 구조화된 정보를 추출하세요. 이미지, 표, 레이아웃 등 모든 시각적 정보도 참고하세요. ## 추출 규칙 - 기획서에 해당 정보가 **있으면** 상세히 추출 - 기획서에 해당 정보가 **없으면** 빈 문자열("") 또는 빈 배열([]) - available_sections: 실제로 의미 있는 내용이 추출된 섹션 이름만 나열 ## 인물(characters) - 기획서에 인물 설명이 있으면 추출 - visual_traits: 외형 특징 (키, 체형, 머리, 특징적 외모 등) — 있으면 추출, 없으면 "" - age_gender: 나이/성별 — 있으면 추출 - role: 역할 (주인공, 조연, 악역 등) ## 장소(locations) - 기획서에 장소·공간 설명이 있으면 추출 (없으면 빈 배열) - visual_traits: 그 공간의 외형 특징 (구조, 재질, 색감, 조명, 놓인 것들 등) ## 세계관(world_setting) - 시대, 장소, 세계 규칙 등 종합 설명 ## 톤/분위기(tone_mood) - 작품의 톤, 분위기, 비주얼 방향성 ## 줄거리(story_arc) - 전체 스토리 요약 (스포일러 포함 OK) ## 비주얼 컨셉(visual_concepts) - 아트 디렉션, 색감, 촬영 스타일, 컨셉아트 설명 등 ## 인물 관계(key_relationships) - 주요 인물 간 관계 (가족, 연인, 적대 등)""" _EMPTY_RESULT = { "characters": [], "locations": [], "world_setting": "", "tone_mood": "", "story_arc": "", "visual_concepts": "", "key_relationships": [], "available_sections": [], } def compute_available_sections(result: Dict) -> list: """어떤 절이 **실제로 쓸 만하게** 왔는지 — 분석 결과 하나만 보고 정한다. ★이 규칙이 두 곳에 있었다 (스텝 하나 · 업로드 서비스 하나). 2026-09-18 에 장소 칸을 더하면서 스텝만 고쳤더니, **정상 업로드 경로(서비스)가 장소를 지웠다**(Codex BLOCK). 규칙을 한 자리로 모은다 — 새 절을 더할 때 고칠 곳도 여기 하나다. """ out = [] for key in ("characters", "locations", "key_relationships"): if result.get(key): out.append(key) for key in ("world_setting", "tone_mood", "story_arc", "visual_concepts"): if (result.get(key) or "").strip(): out.append(key) # 종전 순서 보존 — available_sections 는 기록에 그대로 남는다 order = ("characters", "locations", "world_setting", "tone_mood", "story_arc", "visual_concepts", "key_relationships") return [k for k in order if k in out] class PlanningDocAnalysisStep(StepRunner): """기획서를 LLM으로 분석 — PDF multimodal 우선, 텍스트 fallback. 기획서 업로드 시점에 ``planning_doc_analysis_service`` 가 project-level checkpoint 를 미리 만들어둠. 본 step 은 episode 분석 cascade 호환을 위해 유지되지만, project-level 결과가 있으면 그대로 mirror 하고 LLM 재호출은 하지 않음. """ def _execute(self, mode="resume") -> Dict: from app.models.catalog import ProjectRegistry from app.services.planning_doc_analysis_service import ( load_project_checkpoint, ) # 1) project-level checkpoint 우선 (single source of truth). # db=self.db 전달 — reader 의 source_hash 계산이 같은 session 으로 # ProjectRegistry 를 조회하도록 일관성 유지. project_data = load_project_checkpoint(self.project_id, db=self.db) if project_data is not None: logger.info( "planning_doc_analysis: project-level checkpoint mirror " "(project=%s, sections=%s)", self.project_id[:8], project_data.get("available_sections", []), ) return project_data project = self.db.query(ProjectRegistry).filter( ProjectRegistry.id == self.project_id ).first() # PDF 파일 확인 pdf_path = Path(settings.projects_dir) / self.project_id / "assets" / "planning_doc.pdf" has_pdf = pdf_path.exists() planning_text = project.planning_doc_text if project else None if not has_pdf and (not planning_text or len(planning_text.strip()) < 100): logger.info("planning_doc_analysis: 기획서 없음 — 빈 결과") return _EMPTY_RESULT from app.modules.llm.llm_client import call_structured def _build_pdf_prompt(): pdf_bytes = pdf_path.read_bytes() pdf_b64 = base64.b64encode(pdf_bytes).decode("utf-8") return [ {"type": "text", "text": "이 기획서 PDF를 분석하여 구조화된 정보를 추출하세요."}, {"type": "image_url", "image_url": { "url": f"data:application/pdf;base64,{pdf_b64}", }}, ], len(pdf_bytes) def _build_text_prompt(): return ( "다음 기획서를 분석하여 구조화된 정보를 추출하세요.\n\n" f"## 기획서 전문\n\n{planning_text}" ), len(planning_text or "") # PDF multimodal 우선 → 실패 시 텍스트 fallback result = None if has_pdf: user_prompt, size = _build_pdf_prompt() logger.info("planning_doc_analysis: PDF multimodal (%d bytes)", size) try: result = call_structured( step="planning_doc_analysis", system_prompt=_SYSTEM_PROMPT, user_prompt=user_prompt, response_schema=_ANALYSIS_SCHEMA, project_config=self.project_config, temperature=0.2, ) except Exception as exc: logger.warning("PDF multimodal failed, falling back to text: %s", exc) if result is None and planning_text and len(planning_text.strip()) >= 100: user_prompt, size = _build_text_prompt() logger.info("planning_doc_analysis: text fallback (%d chars)", size) result = call_structured( step="planning_doc_analysis", system_prompt=_SYSTEM_PROMPT, user_prompt=user_prompt, response_schema=_ANALYSIS_SCHEMA, project_config=self.project_config, temperature=0.2, ) if result is None: logger.warning("planning_doc_analysis: both PDF and text failed") return _EMPTY_RESULT # available_sections 는 **공용 함수 하나**가 정한다 (위 참조). computed_available = compute_available_sections(result) result["available_sections"] = computed_available logger.info( "planning_doc_analysis: %d characters, %d locations, " "%d relationships, sections=%s, mode=%s", len(result.get("characters", [])), len(result.get("locations", [])), len(result.get("key_relationships", [])), computed_available, "pdf" if has_pdf else "text", ) return result