"""분석 Phase StepRunner 서브클래스 — Steps 1-10.

각 단계는 StepRunner._execute()를 구현하여 LLM 호출 수행.
입력은 이전 단계의 체크포인트에서 로드, 결과는 체크포인트에 저장.
"""

import json
import logging
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from typing import Any, Dict, List, Optional

from app.core.step_runner import StepRunner
from app.modules.llm.llm_client import call_structured, call_text

logger = logging.getLogger(__name__)


class _AnalysisStepMixin:
    """분석 단계 공통 — fulltext 로드, 이전 단계 결과 로드, Opik metadata."""

    def _opik_meta(self, extra_tags: list = None) -> Dict:
        """Opik metadata — build_opik_metadata()가 있으면 사용, 없으면 빈 dict."""
        if hasattr(self, "build_opik_metadata"):
            return self.build_opik_metadata(extra_tags)
        return {}

    def _load_fulltext(self) -> str:
        from app.models.project import Episode
        from sqlalchemy.orm import undefer
        ep = self.db.query(Episode).options(undefer(Episode.fulltext)).filter(
            Episode.id == self.episode_id
        ).first()
        if not ep or not ep.fulltext:
            from app.core.errors import AppError
            raise AppError(code="step.no_fulltext", message="시나리오 텍스트가 없습니다.", status_code=400)
        return ep.fulltext

    def _load_prev_checkpoint(self, step_id: str) -> Optional[Dict]:
        from pathlib import Path
        from app.core.config import settings
        cp = Path(settings.projects_dir) / self.project_id / "checkpoints" / "episodes" / self.episode_id / step_id / "manifest.json"
        if cp.exists():
            return json.loads(cp.read_text(encoding="utf-8"))
        return None

    def _load_prompt(self, module: str, name: str, **kwargs) -> str:
        from app.modules.prompt_loader import load_prompt
        return load_prompt(module, name, db=self.db, **kwargs)

    def _load_schema(self, module: str, name: str) -> Dict:
        from app.modules.prompt_loader import load_schema
        return load_schema(module, name, db=self.db)


# ── Step 1: entity_style ──

class EntityStyleStep(_AnalysisStepMixin, StepRunner):
    def _execute(self, mode="resume") -> Dict[str, Any]:
        fulltext = self._load_fulltext()

        from app.modules.pipeline.entity_extractor_v3 import STEP1_SCHEMA, _load_prompt, _load_system

        system = _load_system()
        step1_prompt = _load_prompt("turn0_style", prior_block="", fulltext=fulltext)
        step1_prompt += "\n\n" + _load_prompt("turn1", prior_block="", fulltext=fulltext)

        result = call_structured(
            step="entity_style",
            system_prompt=system,
            user_prompt=step1_prompt,
            response_schema=STEP1_SCHEMA,
            project_config=self.project_config,
            schema_name="entity_step1",
            opik_metadata=self._opik_meta(),
        )

        return {
            "completed_count": 1,
            "applicable_count": 1,
            "failed_count": 0,
            "data": result,
        }


# ── Step 2: entity_review ──

class EntityReviewStep(_AnalysisStepMixin, StepRunner):
    def _execute(self, mode="resume") -> Dict[str, Any]:
        fulltext = self._load_fulltext()
        prev = self._load_prev_checkpoint("entity_style")
        if not prev or not prev.get("data"):
            from app.core.errors import AppError
            raise AppError(code="step.no_input", message="entity_style 결과 없음", status_code=400)

        step1_data = prev["data"]

        from app.modules.pipeline.entity_extractor_v3 import _load_schema
        review_schema = _load_schema("turn1_review_schema.json")
        from app.modules.pipeline.ref_image_pipeline import _load_lvm_prompt

        def _fmt(items):
            return ", ".join(f"{e['name']}({e['appearances']}회)" for e in items)

        review_prompt = _load_lvm_prompt(
            "entity_list_review",
            screenplay_summary=fulltext,
            characters=_fmt(step1_data.get("characters", [])),
            locations=_fmt(step1_data.get("locations", [])),
            props=_fmt(step1_data.get("props", [])),
        )

        try:
            result = call_structured(
                step="entity_review",
                system_prompt="시나리오 분석 전문가. 추출된 요소 목록의 정확성을 평가한다.",
                user_prompt=review_prompt,
                response_schema=review_schema,
                project_config=self.project_config,
                schema_name="entity_review",
                opik_metadata=self._opik_meta(),
            )
        except Exception as exc:
            logger.warning("Entity review failed, using empty: %s", exc)
            result = {"entities": []}

        # 필터링 적용
        entities_review = result.get("entities", [])
        low_importance = {
            e["name"] for e in entities_review
            if int(e.get("importance", 50)) < 10 and int(e.get("appearances", 0)) < 2
        }

        filtered = {
            "characters": [e["name"] for e in step1_data.get("characters", []) if e["name"] not in low_importance],
            "locations": [e["name"] for e in step1_data.get("locations", []) if e["name"] not in low_importance],
            "props": [e["name"] for e in step1_data.get("props", []) if e["name"] not in low_importance],
            "review": entities_review,
        }

        return {
            "completed_count": 1,
            "applicable_count": 1,
            "failed_count": 0,
            "data": filtered,
        }


# ── Step 3: entity_detail_batch ──

class EntityDetailBatchStep(_AnalysisStepMixin, StepRunner):
    def _execute(self, mode="resume") -> Dict[str, Any]:
        fulltext = self._load_fulltext()
        prev = self._load_prev_checkpoint("entity_review")
        if not prev or not prev.get("data"):
            from app.core.errors import AppError
            raise AppError(code="step.no_input", message="entity_review 결과 없음", status_code=400)

        filtered = prev["data"]
        entity_queue = []
        for name in filtered.get("characters", []):
            entity_queue.append((name, "character"))
        for name in filtered.get("locations", []):
            entity_queue.append((name, "location"))
        for name in filtered.get("props", []):
            entity_queue.append((name, "prop"))

        from app.modules.pipeline.entity_extractor_v3 import _load_prompt, _load_schema
        detail_schema = _load_schema("turn1_7_detail_batch_schema.json")
        entity_list_text = "\n".join(f"- {name} ({etype})" for name, etype in entity_queue)
        detail_prompt = _load_prompt("turn1_7_detail_batch", entity_list=entity_list_text, fulltext=fulltext)

        gpt_details = {}
        try:
            batch_result = call_structured(
                step="entity_detail_batch",
                system_prompt="시나리오 분석 전문가. 요소별 시각적 상세 정보를 최대한 많이 추출한다.",
                user_prompt=detail_prompt,
                response_schema=detail_schema,
                project_config=self.project_config,
                schema_name="entity_detail_batch",
                opik_metadata=self._opik_meta(),
            )
            for ent in batch_result.get("entities", []):
                gpt_details[ent["name"]] = {
                    "description": ent.get("description", ""),
                    "visual_traits": ent.get("visual_traits", []),
                }

            # 누락 재시도
            missing = [(n, t) for n, t in entity_queue if n not in gpt_details]
            if missing:
                missing_text = "\n".join(f"- {n} ({t})" for n, t in missing)
                retry_prompt = _load_prompt("turn1_7_detail_batch", entity_list=missing_text, fulltext=fulltext)
                try:
                    retry_result = call_structured(
                        step="entity_detail_batch",
                        system_prompt="시나리오 분석 전문가. 요소별 시각적 상세 정보를 최대한 많이 추출한다.",
                        user_prompt=retry_prompt,
                        response_schema=detail_schema,
                        project_config=self.project_config,
                        schema_name="entity_detail_retry",
                        opik_metadata=self._opik_meta(["retry"]),
                    )
                    for ent in retry_result.get("entities", []):
                        gpt_details[ent["name"]] = {
                            "description": ent.get("description", ""),
                            "visual_traits": ent.get("visual_traits", []),
                        }
                except Exception as retry_exc:
                    logger.warning("Entity detail retry failed: %s — 1차 결과만 사용", retry_exc)
        except Exception as exc:
            logger.warning("Entity detail batch failed: %s", exc)

        return {
            "completed_count": len(gpt_details),
            "applicable_count": len(entity_queue),
            "failed_count": len(entity_queue) - len(gpt_details),
            "data": {"entity_queue": entity_queue, "gpt_details": gpt_details},
        }


# ── Step 4: entity_t2i ──

class EntityT2iStep(_AnalysisStepMixin, StepRunner):
    def _execute(self, mode="resume") -> Dict[str, Any]:
        prev = self._load_prev_checkpoint("entity_detail_batch")
        if not prev or not prev.get("data"):
            from app.core.errors import AppError
            raise AppError(code="step.no_input", message="entity_detail_batch 결과 없음", status_code=400)

        entity_queue = prev["data"]["entity_queue"]
        gpt_details = prev["data"]["gpt_details"]

        # resume: 이미 완료된 것 스킵
        cp = self.load_checkpoint()
        done = cp.get("data", {}).get("completed", {}) if cp and mode == "resume" else {}

        remaining = [(i, n, t) for i, (n, t) in enumerate(entity_queue) if n not in done]
        total = len(entity_queue)

        from app.modules.pipeline.entity_extractor_v3 import ENTITY_DETAIL_SCHEMA, _load_system, _load_prompt

        system_prompt = _load_system()

        def _gen_t2i(idx, ename, etype):
            gpt_detail = gpt_details.get(ename, {})
            extra = ""
            if gpt_detail:
                desc = gpt_detail.get("description", "")
                traits = ", ".join(gpt_detail.get("visual_traits", []))
                extra = f"\n\n[시나리오 기반 상세 정보]\n설명: {desc}\n시각적 특징: {traits}"

            turn_msg = _load_prompt("turn_entity_detail", entity_name=ename, entity_type=etype) + extra

            for attempt in range(3):
                try:
                    detail = call_structured(
                        step="entity_t2i",
                        system_prompt=system_prompt,
                        user_prompt=turn_msg,
                        response_schema=ENTITY_DETAIL_SCHEMA,
                        project_config=self.project_config,
                        schema_name="entity_t2i",
                        opik_metadata=self._opik_meta(),
                    )
                    # description / visual_traits 는 source detail
                    # (entity_detail_batch / gpt_detail) 에서 forward — LLM 의
                    # unsourced trait 이 final output 에 도입되는 path 봉쇄.
                    # gpt_detail 이 빈 dict 라도 LLM detail 로 silent fallback
                    # 안 함. (entity_steps.py 와 동일 정책)
                    src = gpt_detail or {}
                    return (idx, ename, etype, {
                        "name": ename,
                        "description": src.get("description", ""),
                        "visual_traits": src.get("visual_traits", []) if isinstance(src.get("visual_traits"), list) else [],
                        "t2i_prompt": detail.get("t2i_prompt", ""),
                    })
                except Exception as exc:
                    if attempt < 2:
                        time.sleep(2 * (attempt + 1))
            src = gpt_detail or {}
            return (idx, ename, etype, {
                "name": ename,
                "description": src.get("description", ""),
                "visual_traits": src.get("visual_traits", []) if isinstance(src.get("visual_traits"), list) else [],
                "t2i_prompt": "",
            })

        if remaining:
            max_workers = min(10, len(remaining))
            results = []
            with ThreadPoolExecutor(max_workers=max_workers) as executor:
                futures = {}
                for i, ename, etype in remaining:
                    if futures:
                        time.sleep(1)
                    futures[executor.submit(_gen_t2i, i, ename, etype)] = i

                for future in as_completed(futures):
                    result = future.result()
                    _, ename, etype, data = result
                    data["entity_type"] = etype
                    done[ename] = data
                    results.append(result)
                    self.update_progress(len(done), total)
                    # 증분 체크포인트 (crash resume용)
                    self.save_checkpoint({
                        "status": "running",
                        "data": {"completed": done, "entity_queue": entity_queue},
                    })

        # 분류
        characters, locations, props = [], [], []
        for name, data in done.items():
            etype = data.get("entity_type", "character")
            if etype == "character":
                characters.append(data)
            elif etype == "location":
                locations.append(data)
            else:
                props.append(data)

        return {
            "completed_count": len(done),
            "applicable_count": total,
            "failed_count": total - len(done),
            "data": {
                "completed": done,
                "characters": characters,
                "locations": locations,
                "props": props,
            },
        }


# ── Step 5: scene_segmentation ──

class SceneSegmentationStep(_AnalysisStepMixin, StepRunner):
    def _execute(self, mode="resume") -> Dict[str, Any]:
        fulltext = self._load_fulltext()
        from app.core.config import settings
        from app.modules.pipeline.scene_extractor_v2 import _segment_by_llm, _segment_by_heading

        segments = _segment_by_llm(fulltext, project_config=self.project_config)

        if not segments:
            logger.info("LLM segmentation empty — trying regex fallback")
            segments = _segment_by_heading(fulltext)

        if not segments:
            segments = [{"scene_index": 1, "heading": "전체", "start_char": 0, "end_char": len(fulltext), "length": len(fulltext)}]

        return {
            "completed_count": len(segments),
            "applicable_count": len(segments),
            "failed_count": 0,
            "data": {"segments": segments},
        }


# ── Step 6: scene_split ──

class SceneSplitStep(_AnalysisStepMixin, StepRunner):
    def check_applicability(self) -> bool:
        prev = self._load_prev_checkpoint("scene_segmentation")
        if not prev:
            return False
        segments = prev.get("data", {}).get("segments", [])
        from app.models.project import ProjectSettings
        ps = self.db.query(ProjectSettings).filter(ProjectSettings.project_id == self.project_id).first()
        threshold = (ps.scene_split_threshold if ps and ps.scene_split_threshold else 600)
        return any(s.get("length", 0) > threshold for s in segments)

    def _execute(self, mode="resume") -> Dict[str, Any]:
        fulltext = self._load_fulltext()
        prev = self._load_prev_checkpoint("scene_segmentation")
        segments = prev["data"]["segments"]

        from app.models.project import ProjectSettings
        from app.modules.pipeline.scene_extractor_v2 import _split_large_scene_by_llm

        ps = self.db.query(ProjectSettings).filter(ProjectSettings.project_id == self.project_id).first()
        threshold = (ps.scene_split_threshold if ps and ps.scene_split_threshold else 600)
        final_segments = []
        split_count = 0

        for seg in segments:
            if seg.get("length", 0) > threshold:
                scene_text = seg.get("text") or fulltext[seg.get("start_char", 0):seg.get("end_char", len(fulltext))]
                splits = _split_large_scene_by_llm(
                    scene_text=scene_text,
                    parent_seg=seg,
                    max_chars=threshold,
                    project_config=self.project_config,
                )
                if splits and len(splits) > 1:
                    final_segments.extend(splits)
                    split_count += 1
                else:
                    final_segments.append(seg)
            else:
                final_segments.append(seg)

        # scene_index 재번호
        for i, seg in enumerate(final_segments):
            seg["scene_index"] = i + 1

        return {
            "completed_count": len(final_segments),
            "applicable_count": len(segments),
            "failed_count": 0,
            "data": {"segments": final_segments, "splits": split_count},
        }


# ── Step 7: scene_dependency ──

class SceneDependencyStep(_AnalysisStepMixin, StepRunner):
    def _execute(self, mode="resume") -> Dict[str, Any]:
        fulltext = self._load_fulltext()
        prev = self._load_prev_checkpoint("scene_split")
        if not prev:
            prev = self._load_prev_checkpoint("scene_segmentation")
        if not prev or not prev.get("data"):
            from app.core.errors import AppError
            raise AppError(code="step.no_input", message="세그먼트 결과 없음", status_code=400)
        segments = prev["data"]["segments"]

        from app.modules.pipeline.scene_dependency_extractor import extract_scene_dependencies
        from app.core.config import settings

        deps = extract_scene_dependencies(
            segments=segments,
            fulltext=fulltext,
            checkpoint_dir=None,
            project_llm_config=self.project_config,
        )

        return {
            "completed_count": len(deps),
            "applicable_count": len(segments),
            "failed_count": 0,
            "data": {"dependencies": deps},
        }


# ── Step 8: outlook_extraction ──

class OutlookExtractionStep(_AnalysisStepMixin, StepRunner):
    def _execute(self, mode="resume") -> Dict[str, Any]:
        fulltext = self._load_fulltext()

        seg_prev = self._load_prev_checkpoint("scene_split")
        if not seg_prev:
            seg_prev = self._load_prev_checkpoint("scene_segmentation")
        segments = seg_prev["data"]["segments"]

        entity_prev = self._load_prev_checkpoint("entity_t2i")
        characters = [d["name"] for d in entity_prev["data"].get("characters", [])] if entity_prev else []

        # short_id 매핑 (DB에서)
        from app.modules.short_id import build_short_id_info
        sid_info = build_short_id_info(self.db, self.project_id, self.episode_id)
        char_sid_list = [
            {"id": sid, "name": info["name"]}
            for sid, info in sorted(sid_info.items())
            if info["type"] == "character"
        ]

        # visual_world_rules 로드
        style_prev = self._load_prev_checkpoint("entity_style")
        visual_rules = style_prev["data"].get("visual_world_rules", []) if style_prev else []

        # scene_director 결과 — 씬별 물리적 존재 인물
        director_prev = self._load_prev_checkpoint("scene_director")
        scene_present_chars = None
        if director_prev and director_prev.get("data"):
            scene_present_chars = director_prev["data"].get("scene_present_characters", {})

        from app.modules.pipeline.outlook_extractor import extract_all_outlooks

        result = extract_all_outlooks(
            segments=segments,
            fulltext=fulltext,
            characters=characters,
            char_sid_list=char_sid_list,
            project_llm_config=self.project_config,
            visual_world_rules=visual_rules,
            scene_present_characters=scene_present_chars,
        )

        outlooks = result.get("outlooks", [])
        assignments = result.get("scene_assignments", [])

        return {
            "completed_count": len(outlooks),
            "applicable_count": 1,
            "failed_count": 0,
            "data": result,
        }


# ── Step 9: scene_detail ──

class SceneDetailStep(_AnalysisStepMixin, StepRunner):
    def _execute(self, mode="resume") -> Dict[str, Any]:
        fulltext = self._load_fulltext()

        seg_prev = self._load_prev_checkpoint("scene_split")
        if not seg_prev:
            seg_prev = self._load_prev_checkpoint("scene_segmentation")
        segments = seg_prev["data"]["segments"]

        dep_prev = self._load_prev_checkpoint("scene_dependency")
        deps = dep_prev["data"]["dependencies"] if dep_prev else {}

        outlook_prev = self._load_prev_checkpoint("outlook_extraction")
        outlook_assignments = {}
        if outlook_prev and outlook_prev.get("data"):
            for sa in outlook_prev["data"].get("scene_assignments", []):
                outlook_assignments[sa["scene_index"]] = sa.get("characters", [])

        entity_prev = self._load_prev_checkpoint("entity_t2i")
        style_prev = self._load_prev_checkpoint("entity_style")
        entities = entity_prev["data"] if entity_prev else {}
        style = style_prev["data"] if style_prev else {}

        from app.modules.pipeline.scene_extractor_v2 import extract_scenes_multiturn
        from app.core.config import settings

        visual_rules = style.get("visual_world_rules", [])

        # cinematography 결과 로드 — 씬별 촬영 기법
        cine_prev = self._load_prev_checkpoint("scene_cinematography")
        scene_shot_map = {}
        if cine_prev and cine_prev.get("data"):
            from sqlalchemy import text as sql_text
            # shot_type의 llm_description 로드
            shot_desc_map = {}
            shot_rows = self.db.execute(sql_text(
                "SELECT name, llm_description FROM shot_type WHERE is_active = true"
            )).fetchall()
            for r in shot_rows:
                shot_desc_map[r[0]] = r[1]

            for s in cine_prev["data"].get("scenes", []):
                si = s["scene_index"]
                shot1_desc = shot_desc_map.get(s.get("shot_1"), "")
                shot2_desc = shot_desc_map.get(s.get("shot_2"), "")
                scene_shot_map[si] = {
                    "shot_1": {"name": s.get("shot_1", ""), "description": shot1_desc, "focus": s.get("shot_1_focus", "")},
                    "shot_2": {"name": s.get("shot_2", ""), "description": shot2_desc, "focus": s.get("shot_2_focus", "")},
                }

        # scene_director의 씬별 엔티티 ID 맵
        director_prev = self._load_prev_checkpoint("scene_director")
        scene_present_entities_map = None
        if director_prev and director_prev.get("data"):
            scene_present_entities_map = director_prev["data"].get("scene_present_entities", {})
            # int/str 키 통일
            if scene_present_entities_map:
                unified = {}
                for k, v in scene_present_entities_map.items():
                    unified[int(k) if isinstance(k, str) and k.isdigit() else k] = v
                scene_present_entities_map = unified

        # entities에 id + short_id 추가 (DB에서) — episode 스코핑
        from sqlalchemy import text as sql_text
        db_ents = self.db.execute(sql_text(
            "SELECT ec.id, ec.name, ec.entity_type, ec.short_id FROM entity_canon ec "
            "JOIN entity_episode_link eel ON ec.id = eel.canon_id "
            "WHERE ec.project_id = :p AND eel.episode_id = :e"
        ), {"p": self.project_id, "e": self.episode_id}).fetchall()
        name_to_id = {(r[1], r[2]): r[0] for r in db_ents}
        name_to_short = {(r[1], r[2]): r[3] for r in db_ents if r[3]}
        for etype in ["characters", "locations", "props"]:
            singular = etype.rstrip("s")
            for e in entities.get(etype, []):
                if "id" not in e:
                    e["id"] = name_to_id.get((e["name"], singular), "")
                if "short_id" not in e:
                    e["short_id"] = name_to_short.get((e["name"], singular), "")

        # outlook entities에 short_id 추가 (C01O02 매핑용)
        outlook_entities = []
        ol_ents = [r for r in db_ents if r[2] == "outlook"]
        for r in ol_ents:
            outlook_entities.append({"name": r[1], "short_id": r[3] or "", "id": r[0]})

        scene_result = extract_scenes_multiturn(
            fulltext=fulltext,
            entities={"characters": entities.get("characters", []),
                       "locations": entities.get("locations", []),
                       "props": entities.get("props", []),
                       "outlooks": outlook_entities},
            style=style,
            on_scene_progress=lambda done, total: self.update_progress(done, total),
            split_threshold=600,
            outlook_assignments=outlook_assignments,
            scene_dependencies=deps,
            pre_segments=segments,
            scene_llm=settings.scene_detail_llm,
            project_llm_config=self.project_config,
            visual_world_rules=visual_rules,
            scene_shot_map=scene_shot_map,
            scene_present_entities_map=scene_present_entities_map,
            opik_metadata=self._opik_meta(),
        )

        total = scene_result.get("total_scenes", 0)

        return {
            "completed_count": total,
            "applicable_count": total,
            "failed_count": 0,
            "data": scene_result,
        }


# ── Step 10: scene_verify ──

class SceneVerifyStep(_AnalysisStepMixin, StepRunner):
    def check_applicability(self) -> bool:
        prev = self._load_prev_checkpoint("scene_detail")
        if not prev:
            return False
        scenes = prev.get("data", {}).get("scenes", [])
        return any(
            len([v for v in s.get("visible_entities", []) if v.get("entity_type") == "character"]) >= 2
            for s in scenes
        )

    def _execute(self, mode="resume") -> Dict[str, Any]:
        fulltext = self._load_fulltext()

        prev = self._load_prev_checkpoint("scene_detail")
        scenes = prev["data"]["scenes"]

        seg_prev = self._load_prev_checkpoint("scene_split")
        if not seg_prev:
            seg_prev = self._load_prev_checkpoint("scene_segmentation")
        segments = seg_prev["data"]["segments"]

        from app.modules.pipeline.scene_validator import validate_visible_entities

        verified = validate_visible_entities(
            scenes=scenes,
            fulltext=fulltext,
            segments=segments,
            project_config=self.project_config,
        )

        return {
            "completed_count": len(verified),
            "applicable_count": len(scenes),
            "failed_count": 0,
            "data": {"scenes": verified},
        }


# ── Auxiliary: project_summary ──

class ProjectSummaryStep(_AnalysisStepMixin, StepRunner):
    def _execute(self, mode="resume") -> Dict[str, Any]:
        from app.models.project import Episode
        from app.models.catalog import ProjectRegistry

        ep = self.db.query(Episode).filter(
            Episode.id == self.episode_id, Episode.project_id == self.project_id
        ).first()
        if not ep:
            return {"completed_count": 0, "applicable_count": 1, "failed_count": 1}

        # entity_style 체크포인트에서 episode_summary 가져오기
        entity_cp = self._load_prev_checkpoint("entity_style")
        summary = ""
        if entity_cp and entity_cp.get("status") == "completed":
            data = entity_cp.get("data", {})
            summary = data.get("episode_summary", "")

        if not summary:
            # fulltext 앞부분으로 요약 생성
            ft = self._load_fulltext()
            excerpt = ft
            summary = call_text(
                step="project_summary",
                system_prompt="시나리오 요약 전문가. 핵심 인물, 갈등, 세계관을 간결히 요약한다.",
                user_prompt=f"아래 시나리오를 5문장 이내로 요약하세요.\n\n{excerpt}",
                project_config=self.project_config,
                opik_metadata=self._opik_meta(),
                temperature=0.3,
            )

        # episode.summary 업데이트
        ep.summary = summary
        self.db.commit()

        # 프로젝트 요약도 업데이트
        proj = self.db.query(ProjectRegistry).filter(ProjectRegistry.id == self.project_id).first()
        if proj:
            proj.description = summary[:500] if len(summary) > 500 else summary
            self.db.commit()

        logger.info("Project summary generated: %d chars", len(summary))
        return {"completed_count": 1, "applicable_count": 1, "failed_count": 0,
                "data": {"summary": summary}}


# ── Auxiliary: outlook_dedup ──

class OutlookDedupStep(_AnalysisStepMixin, StepRunner):
    def _execute(self, mode="resume") -> Dict[str, Any]:
        from app.modules.pipeline.outlook_dedup import dedup_scene_markers, find_duplicate_outlooks, apply_outlook_merge

        # outlook_extraction 체크포인트 로드
        ol_cp = self._load_prev_checkpoint("outlook_extraction")
        if not ol_cp or ol_cp.get("status") != "completed":
            return {"completed_count": 0, "applicable_count": 0, "failed_count": 0,
                    "data": {"message": "outlook_extraction not completed"}}

        outlooks = ol_cp.get("data", {}).get("outlooks", [])
        if len(outlooks) < 2:
            return {"completed_count": 0, "applicable_count": 0, "failed_count": 0,
                    "data": {"message": "too few outlooks to dedup"}}

        # GPT로 중복 판별 — 반환값은 List[{keep, remove, reason}]
        duplicates = find_duplicate_outlooks(outlooks, project_config=self.project_config)
        merge_count = len(duplicates) if isinstance(duplicates, list) else 0

        if merge_count > 0:
            apply_outlook_merge(self.db, self.project_id, duplicates)
            logger.info("Outlook dedup: merged %d groups", merge_count)
        else:
            logger.info("Outlook dedup: no duplicates found")

        return {"completed_count": 1, "applicable_count": 1, "failed_count": 0,
                "data": {"merge_groups": merge_count, "duplicates": duplicates}}


# ── Step Registry ──

class SceneDirectorStep(_AnalysisStepMixin, StepRunner):
    """씬 감독 — 전체 씬 일괄 분석, ID 기반 엔티티 물리적 존재 판별."""

    def _execute(self, mode="resume") -> Dict[str, Any]:
        import json as _json
        fulltext = self._load_fulltext()

        seg_prev = self._load_prev_checkpoint("scene_split")
        if not seg_prev:
            seg_prev = self._load_prev_checkpoint("scene_segmentation")
        segments = seg_prev["data"]["segments"]

        entity_prev = self._load_prev_checkpoint("entity_t2i")
        entities = entity_prev["data"] if entity_prev else {}

        style_prev = self._load_prev_checkpoint("entity_style")
        visual_rules = style_prev["data"].get("visual_world_rules", []) if style_prev else []

        from app.modules.prompt_loader import load_prompt, load_schema

        system_prompt = load_prompt("scene_director", "system", db=self.db)
        schema = load_schema("scene_director", "analyze_schema", db=self.db)

        rules_text = "\n".join(f"- {r}" for r in visual_rules) if visual_rules else "없음"

        # DB short_id 기반 엔티티 목록 (C01/L01/P01)
        from app.modules.short_id import build_short_id_info, build_short_id_map

        sid_info = build_short_id_info(self.db, self.project_id, self.episode_id)
        short_to_uuid = {sid: info["uuid"] for sid, info in sid_info.items()}

        entity_list_items = []
        for sid in sorted(sid_info.keys()):
            info = sid_info[sid]
            if info["type"] in ("character", "location", "prop"):
                desc = info["description"]
                entity_list_items.append(
                    _json.dumps({"id": sid, "name": info["name"], "type": info["type"], "description": desc}, ensure_ascii=False)
                )

        entity_list_block = "\n".join(f"  {item}" for item in entity_list_items)

        # schema에 동적 enum 주입 — LLM이 DB short_id 외의 값을 반환 불가
        valid_ids = [sid for sid, info in sid_info.items() if info["type"] in ("character", "location", "prop")]
        schema = _json.loads(_json.dumps(schema))  # deep copy
        items_spec = schema["properties"]["scenes"]["items"]["properties"]["present_entity_ids"]["items"]
        items_spec["enum"] = valid_ids
        # not_present의 id도 enum 제약
        np_spec = schema["properties"]["scenes"]["items"]["properties"].get("not_present")
        if np_spec:
            np_items = np_spec.get("items", {}).get("properties", {}).get("id")
            if np_items:
                np_items["enum"] = valid_ids

        # 전체 씬 JSON
        scenes_json = []
        for seg in segments:
            scenes_json.append({
                "scene_index": seg["scene_index"],
                "heading": seg.get("heading", ""),
                "text": seg.get("text") or fulltext[seg.get("start_char", 0):seg.get("end_char", len(fulltext))],
            })
        scenes_str = _json.dumps(scenes_json, ensure_ascii=False, indent=1)

        user_prompt = (
            "아래는 전체 시나리오의 씬 목록입니다.\n"
            "각 씬에서 아래 엔티티 목록의 항목 중 **해당 장소에 물리적으로 존재하는 것**의 id를 선택하세요.\n\n"
            "인물: 자기 물리적 몸으로 존재하는 인물만. 대사를 하더라도 빙의/원격접속 중이면 제외.\n"
            "배경: 이 씬의 촬영 장소와 일치하는 배경.\n"
            "소품: 이 씬에서 카메라에 보이는 소품.\n\n"
            "중요: 앞쪽 씬의 맥락을 반드시 참고하세요.\n"
            "- 앞쪽 씬에서 인물A가 인물B의 몸에 접속/빙의/라이드했다면, "
            "이후 씬에서 인물A가 대사를 하더라도 인물A의 물리적 몸은 접속 장소에 있고 "
            "현장에는 인물B의 몸만 있습니다.\n"
            "- 씬 순서대로 읽으며 접속/빙의 상태를 추적하세요.\n\n"
            "결과에는 엔티티 목록에서 제공된 **id**(C01, L01, P01 등)를 그대로 사용하세요. 이름이 아닌 id로 반환.\n"
            "description을 참고하여 씬 텍스트의 표현과 매칭하세요 (이름이 정확히 일치하지 않을 수 있음).\n\n"
            f"[시각적 세계관 규칙]\n{rules_text}\n\n"
            f"[엔티티 목록 (인물+배경+소품)]\n{entity_list_block}\n\n"
            f"[전체 씬 목록]\n{scenes_str}"
        )

        logger.info("Scene director: analyzing %d scenes, %d entities (short-ID mapped)", len(segments), len(entity_list_items))

        result = call_structured(
            step="scene_director",
            system_prompt=system_prompt,
            user_prompt=user_prompt,
            response_schema=schema,
            project_config=self.project_config,
            schema_name="scene_director_batch",
            opik_metadata=self._opik_meta(),
        )

        results = result.get("scenes", [])

        # short_id → UUID/info 매핑 (sid_info에서)
        id_to_info = {info["uuid"]: {"name": info["name"], "type": info["type"]} for info in sid_info.values()}

        # 이름 기반 fallback 매핑 (LLM이 short ID 무시 시)
        name_to_uuid = {}
        for info in sid_info.values():
            name_upper = info["name"].upper().replace(" ", "_")
            name_to_uuid[info["name"]] = info["uuid"]
            name_to_uuid[name_upper] = info["uuid"]
            prefix = {"character": "CHAR_", "location": "BG_", "prop": "PROP_"}.get(info["type"], "")
            if prefix:
                name_to_uuid[f"{prefix}{name_upper}"] = info["uuid"]

        def _resolve_entity_id(raw_id: str) -> Optional[str]:
            """short ID → UUID, 실패 시 이름 매칭 fallback."""
            if raw_id in short_to_uuid:
                return short_to_uuid[raw_id]
            # LLM이 자체 ID 생성한 경우 — 이름으로 매칭 시도
            if raw_id in name_to_uuid:
                return name_to_uuid[raw_id]
            # CHAR_DONGBYUK → 부분 매칭
            clean = raw_id.replace("CHAR_", "").replace("BG_", "").replace("PROP_", "")
            if clean in name_to_uuid:
                return name_to_uuid[clean]
            return None

        # 씬별 PRESENT 엔티티 분류 (인물/배경/소품 분리)
        scene_present_entities = {}  # scene_index → {characters: [uuid], locations: [uuid], props: [uuid]}
        scene_present_characters = {}  # 호환용 — outlook_extraction에 이름 전달
        fallback_count = 0
        for r in results:
            si = int(r["scene_index"])
            chars, locs, props = [], [], []
            char_names = []
            for raw_id in r.get("present_entity_ids", []):
                real_uuid = _resolve_entity_id(raw_id)
                if not real_uuid:
                    logger.warning("Scene %d: unknown entity ID '%s' from LLM, skipping", si, raw_id)
                    continue
                if raw_id not in short_to_uuid:
                    fallback_count += 1
                info = id_to_info.get(real_uuid, {})
                etype = info.get("type", "")
                if etype == "character":
                    chars.append(real_uuid)
                    char_names.append(info.get("name", ""))
                elif etype == "location":
                    locs.append(real_uuid)
                elif etype == "prop":
                    props.append(real_uuid)
            scene_present_entities[si] = {"characters": chars, "locations": locs, "props": props}
            scene_present_characters[si] = char_names

        if fallback_count:
            logger.warning("Scene director: %d IDs resolved via name fallback (LLM ignored short IDs)", fallback_count)

        # data.scenes는 LLM 원본 그대로 보관 (short ID)
        # UUID는 scene_present_entities에만 존재 (내부 필터링용)

        return {
            "completed_count": len(results),
            "applicable_count": len(segments),
            "failed_count": len(segments) - len(results),
            "data": {
                "scenes": results,
                "scene_present_entities": scene_present_entities,
                "scene_present_characters": scene_present_characters,  # outlook_extraction 호환
            },
        }


class SceneCinematographyStep(_AnalysisStepMixin, StepRunner):
    """촬영 감독 — 전체 씬 일괄로 촬영 기법 2가지 선택."""

    def _execute(self, mode="resume") -> Dict[str, Any]:
        import json as _json
        fulltext = self._load_fulltext()

        seg_prev = self._load_prev_checkpoint("scene_split")
        if not seg_prev:
            seg_prev = self._load_prev_checkpoint("scene_segmentation")
        segments = seg_prev["data"]["segments"]

        from app.modules.prompt_loader import load_prompt, load_schema
        from sqlalchemy import text as sql_text

        system_prompt = load_prompt("scene_cinematography", "system", db=self.db)
        analyze_template = load_prompt("scene_cinematography", "analyze", db=self.db)
        schema = load_schema("scene_cinematography", "analyze_schema", db=self.db)

        # DB에서 활성 샷 타입 로드
        shot_rows = self.db.execute(sql_text(
            "SELECT name, category, description FROM shot_type WHERE is_active = true ORDER BY sort_order"
        )).fetchall()
        shot_types_block = "\n".join(
            f"- {r[0]} [{r[1]}]: {r[2]}" for r in shot_rows
        )

        # 전체 씬 JSON (제목+요약만, 텍스트 전문 대신)
        scenes_block_items = []
        for seg in segments:
            scene_text = seg.get("text") or fulltext[seg.get("start_char", 0):seg.get("end_char", len(fulltext))]
            scenes_block_items.append(f"씬 {seg['scene_index']}: {seg.get('heading', '')}\n{scene_text}")
        scenes_block = "\n\n".join(scenes_block_items)

        user_prompt = analyze_template.format(
            shot_types_block=shot_types_block,
            scenes_block=scenes_block,
        )

        logger.info("Scene cinematography: %d scenes, %d shot types", len(segments), len(shot_rows))

        result = call_structured(
            step="scene_cinematography",
            system_prompt=system_prompt,
            user_prompt=user_prompt,
            response_schema=schema,
            project_config=self.project_config,
            schema_name="scene_cinematography",
            opik_metadata=self._opik_meta(),
        )

        scenes = result.get("scenes", [])

        # DB에 저장 (scene_still.shot_type_1, shot_type_2)
        for s in scenes:
            si = s["scene_index"]
            self.db.execute(sql_text(
                "UPDATE scene_still SET shot_type_1 = :s1, shot_type_2 = :s2 "
                "WHERE project_id = :pid AND episode_id = :eid AND still_index = :si"
            ), {"s1": s.get("shot_1"), "s2": s.get("shot_2"), "pid": self.project_id, "eid": self.episode_id, "si": si})
        self.db.commit()

        return {
            "completed_count": len(scenes),
            "applicable_count": len(segments),
            "failed_count": len(segments) - len(scenes),
            "data": {"scenes": scenes},
        }


ANALYSIS_STEP_CLASSES = {
    "entity_style": EntityStyleStep,
    "entity_review": EntityReviewStep,
    "entity_detail_batch": EntityDetailBatchStep,
    "entity_t2i": EntityT2iStep,
    "scene_segmentation": SceneSegmentationStep,
    "scene_split": SceneSplitStep,
    "scene_director": SceneDirectorStep,
    "scene_cinematography": SceneCinematographyStep,
    "scene_dependency": SceneDependencyStep,
    "outlook_extraction": OutlookExtractionStep,
    "scene_detail": SceneDetailStep,
    "scene_verify": SceneVerifyStep,
    "project_summary": ProjectSummaryStep,
    "outlook_dedup": OutlookDedupStep,
}