"""Shot description 검증·재작성 단계.

shot_extract 직후 실행. 각 shot의 description이 '한 찰나' 원칙을 지키는지
LLM으로 검증하고, 시간 연결어/연속 동작이 섞인 경우 재작성한다. 원본은
`original_description` 필드로 백업.

체크포인트 구조는 shot_extract와 동일해서 다운스트림 step들은 로더 경로만
`shot_validator`로 교체하면 validated 버전을 읽는다.
"""
import json as _json
import logging
import re
from concurrent.futures import ThreadPoolExecutor, as_completed
from pathlib import Path as _Path
from typing import Any, Dict, List, Optional

from app.core.config import settings
from app.core.entity_protection import _load_cp, _parse_traits
from app.core.errors import AppError
from app.core.step_runner import StepRunner
from app.models.project import EntityCanon
from app.modules.llm.llm_client import call_structured
from app.modules.prompt_loader import load_prompt, load_schema

logger = logging.getLogger(__name__)

MAX_WORKERS = 4

# character_ids 형식 — short_id base (C##) 만 허용. composite (C##O##) 는 base 로 collapse.
_CHAR_SHORT_ID_RE = re.compile(r"^C\d+$")
# composite full-pattern — partial 매치 ('C1O' 등) 차단 위해 anchored.
_COMPOSITE_CHAR_ID_RE = re.compile(r"^(C\d+)O\d+$")


def _filter_character_ids(
    raw: List[Any],
    pool_keys: set[str],
) -> tuple[List[str], List[Any]]:
    """character_ids LLM 출력 strict 검증.

    각 entry 를 다음 룰로 평가:
      1. None / 빈 문자열 — drop
      2. composite C##O## → base C## 로 collapse 후 재검증
      3. ^C\\d+$ 정규식 미일치 — drop (silent corruption 차단; '42', 'None', 'C-A' 등)
      4. character_pool 이 비어있지 않은데 그 안에 없음 — drop (stale/hallucinated id)
      5. 통과 entry 는 base form 으로 dedupe 보존 (입력 순서)

    Returns: (accepted base ids, rejected raw entries — log 용).
    """
    accepted: List[str] = []
    rejected: List[Any] = []
    seen: set[str] = set()
    for entry in raw:
        if entry is None or entry == "":
            rejected.append(entry)
            continue
        s = str(entry)
        # composite full-pattern → base. 'C1O' 같은 partial trailing-O 는 base
        # 발견 안 됨 → 다음 regex 검증에서 fail.
        # fullmatch — '$' 가 line-end 도 매치하는 회피 차단 (예: 'C01\n' rejection).
        composite_match = _COMPOSITE_CHAR_ID_RE.fullmatch(s)
        base = composite_match.group(1) if composite_match else s
        if not _CHAR_SHORT_ID_RE.fullmatch(base):
            rejected.append(entry)
            continue
        # pool 이 있으면 candidate set 검증 (없으면 첫 run 의 RC-F 부분 비활성 — 통과)
        if pool_keys and base not in pool_keys:
            rejected.append(entry)
            continue
        if base not in seen:
            seen.add(base)
            accepted.append(base)
    return accepted, rejected


def assert_no_failed_scenes(
    shot_validator_cp: Optional[Dict[str, Any]],
    project_config: Optional[Dict[str, Any]],
    *,
    consumer_step: str,
) -> None:
    """G4.6 RO-15 — shot_validator manifest 의 validator_status='failed' 씬 fail-fast.

    **계층 정책** (G4.6 Phase 3 fix iter 3):
      1차 (gate): step_runner.check_gate 가 dep partial + manifest
        allow_partial_downstream=False 차단. project_config 의
        partial_override_config_key (=allow_failed_validator) 가 ``is True``
        (strict bool) 인 경우 통과.
      2차 (consumer guard, 본 helper): force / resume / 비표준 dispatch case 에
        gate 통과 후 도달 시 second line of defense. 동일 키 + 동일 strict 정책.

    **guard 적용 consumer** (description / characters / shot text 직접 사용 —
    failed 씬의 stale 값을 LLM input 으로 사용하면 잘못된 결과):
      - shot_selection_step / shot_dependency_step / shot_director_step
      - shot_dependency_t2i_step / scene_context_loader
      - entity_all_character (entity_steps) / scene_camera_flow_step
      - shot_staging_step / background_master_plan_step

    **guard 미적용 consumer** (cascade dependencies — gate 차단으로 dispatch
    자체 안 됨, 또는 selected_shot_indices 만 사용해서 stale shot text 무관):
      - background_classify / background_planner / background_chain_planning
      - background_prompt / floor_plan_prompt
    """
    if not shot_validator_cp:
        return
    scenes = (shot_validator_cp.get("data") or {}).get("scenes") or []
    failed = sorted(
        sc.get("scene_index")
        for sc in scenes
        if isinstance(sc, dict)
        and sc.get("validator_status") == "failed"
        and sc.get("scene_index") is not None
    )
    if not failed:
        return
    cfg = project_config or {}
    # strict — `is True` 만 허용 (gate-level check_gate 와 동일 정책).
    if cfg.get("allow_failed_validator") is True:
        logger.warning(
            "%s: upstream shot_validator failed for scenes %s; "
            "allow_failed_validator=True — explicit operator override.",
            consumer_step, failed,
        )
        return
    raise AppError(
        code="step.upstream_validator_failed",
        message=(
            f"{consumer_step} blocked: shot_validator failed for scenes {failed}. "
            f"Re-run shot_validator first or set allow_failed_validator=true as explicit operator override."
        ),
        status_code=400,
    )


class ShotValidatorStep(StepRunner):
    """각 shot description을 '한 찰나' 원칙에 맞게 검증·재작성."""

    def _config_hash(self) -> str:
        """프롬프트 팩 실효 버전을 지문에 싣는다 (2026-08-07).

        이 스텝은 팩을 **버전 미지정**으로 로드해 항상 최신을 쓴다. 그런데
        지문에는 팩이 없어서, 팩을 고쳐도 완료된 체크포인트가 그대로
        재사용됐다 — 이번에 실측한 "만들어 놓고 안 나간다" 결함과 같은
        형태다. 팩이 바뀌면 재실행되게 한다.

        ★재실행 비용이 크다: 이 스텝의 산출(샷 description)은 하류
        거의 전부의 입력이다. 팩을 올릴 때는 재실행 범위를 함께 판단할 것.
        """
        import hashlib
        import json as _json

        from app.core.step_runner import compute_config_hash

        # 팩 **내용**으로 지문을 만든다. 버전 문자열이 아니라 실제로 나가는
        # 바이트다 — provenance 조회는 caller 의 db 전달 여부에 따라 빈 값을
        # 돌려줄 수 있고(실측), 그러면 "팩 무관"으로 조용히 떨어진다.
        text = load_prompt("shot_validator", "system", db=self.db) or ""
        payload = {
            "prompt_sha": hashlib.sha256(
                text.encode("utf-8")).hexdigest()[:16],
            "project_config_hash": compute_config_hash(self.project_config),
        }
        return hashlib.sha256(
            _json.dumps(payload, sort_keys=True).encode("utf-8")
        ).hexdigest()

    def _load_prev_checkpoint(self, step_id: str) -> Optional[Dict[str, Any]]:
        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 _execute(self, mode: str = "resume") -> Dict[str, Any]:
        shot_cp = self._load_prev_checkpoint("shot_extract")
        if not shot_cp or not shot_cp.get("data", {}).get("scenes"):
            raise AppError(
                code="step.no_shots",
                message="shot_extract 결과가 없습니다. 먼저 shot_extract를 실행하세요.",
                status_code=400,
            )

        # scene_save의 실제 구조: data.segments (각 segment에 text 필드)
        save_cp = self._load_prev_checkpoint("scene_save")
        scene_texts: Dict[int, str] = {}
        if save_cp:
            save_data = save_cp.get("data", {})
            segs = save_data.get("segments") or save_data.get("scenes") or []
            for s in segs:
                si = s.get("scene_index")
                if si is not None:
                    scene_texts[si] = s.get("text", "") or s.get("segment", "")

        system = load_prompt("shot_validator", "system", db=self.db)
        schema = load_schema("shot_validator", "validator_schema", db=self.db)

        scenes_input = shot_cp["data"]["scenes"]
        # character pool 한 번만 로드 — entity_blocks 빌드 + character_ids 검증 양쪽에 사용.
        character_pool = self._load_character_pool()
        # shot_cp 재활용 (cache) — universal-pool 분기에서 shot_extract 재로드 회피.
        entity_blocks = self._build_entity_blocks(character_pool, shot_cp=shot_cp)
        # scene_index None 방어 — 데이터 손실 방지
        missing = [i for i, sc in enumerate(scenes_input) if sc.get("scene_index") is None]
        if missing:
            raise AppError(
                code="step.invalid_data",
                message=f"shot_extract 결과에 scene_index 없는 씬 {len(missing)}개 — upstream 데이터 손상",
                status_code=422,
            )

        results: List[Dict[str, Any]] = []
        changed_total = 0
        shots_total = 0
        failed_scenes = 0

        with ThreadPoolExecutor(max_workers=MAX_WORKERS) as pool:
            futures = {
                pool.submit(
                    self._validate_scene,
                    sc,
                    scene_texts.get(sc.get("scene_index"), ""),
                    entity_blocks.get(sc.get("scene_index"), ""),
                    system,
                    schema,
                    character_pool,
                ): sc.get("scene_index")
                for sc in scenes_input
            }
            for fut in as_completed(futures):
                si = futures[fut]
                try:
                    out_scene, changed = fut.result()
                except Exception as exc:
                    logger.warning("shot_validator: scene %s failed — %s", si, exc)
                    failed_scenes += 1
                    # 실패 시 원본 유지 + validator_status="failed" 마킹.
                    # 다운스트림이 partial state 를 인지하고 fail-fast / explicit
                    # operator override 정책을 결정할 수 있도록 명시 마커 추가
                    # (silent carry 차단).
                    orig = next((x for x in scenes_input if x.get("scene_index") == si), None)
                    if orig is not None:
                        marked = dict(orig)
                        marked["validator_status"] = "failed"
                        marked["validator_failure_reason"] = str(exc)[:200]
                        marked["shots"] = [
                            {**sh, "validator_status": "failed_carry_original"}
                            for sh in (orig.get("shots") or [])
                        ]
                        results.append(marked)
                    else:
                        logger.error(
                            "shot_validator: scene %s not found in original input — skipping",
                            si,
                        )
                    continue
                results.append(out_scene)
                changed_total += changed
                shots_total += len(out_scene.get("shots", []))

        results.sort(key=lambda s: s.get("scene_index", 0))
        logger.info(
            "shot_validator: %d scenes, %d shots total, %d revised, %d failed",
            len(results),
            shots_total,
            changed_total,
            failed_scenes,
        )
        return {
            "completed_count": len(results) - failed_scenes,
            "applicable_count": len(scenes_input),
            "failed_count": failed_scenes,
            # ★`_config_hash()` 를 정의했으면 저장도 그 값으로 해야 한다.
            #  체크포인트 저장은 data 에 `config_hash` 가 없으면
            #  `compute_config_hash(project_config)` 로 떨어지는데(step_runner
            #  :388), drift 검사는 step-local `_config_hash()` 로 한다
            #  (:1533). 짝을 안 맞추면 방금 성공한 실행이 다음 순간
            #  "contract drift" 로 막힌다 — 실측으로 전체 재실행이 여기서
            #  멈췄다. `shot_ref_classify_step` 이 같은 패턴을 쓴다.
            "config_hash": self._config_hash(),
            "data": {
                "scenes": results,
                "total_shots": shots_total,
                "changed_shots": changed_total,
            },
        }

    def _validate_scene(
        self,
        original_scene: Dict[str, Any],
        scene_text: str,
        entity_block: str,
        system: str,
        schema: Dict[str, Any],
        character_pool: Optional[Dict[str, Any]] = None,
    ) -> tuple[Dict[str, Any], int]:
        """단일 씬의 shots 검증. (결과 씬, 변경된 shot 수) 반환.

        entity_block: 사전 빌드된 [Entity map for this scene] 문자열 (또는 빈 문자열).
        character_pool: {short_id: ...} — character_ids 검증 시 candidate set.
        """
        pool_keys = set(character_pool.keys()) if character_pool else set()
        shots = original_scene.get("shots", [])
        si = original_scene.get("scene_index")

        if not shots:
            return original_scene, 0

        shots_block = "\n".join(
            f"Shot {sh.get('shot_index', '?')}: {sh.get('description', '')}"
            for sh in shots
        )
        user_prompt = (
            f"[씬 {si}]\n"
            f"{scene_text}\n\n"
            f"{entity_block}"
            f"[Shots 검증 대상]\n{shots_block}\n\n"
            f"각 shot의 description이 '한 찰나' 원칙에 맞는지 검증하고, "
            f"위반 시 재작성하세요. character_ids 는 entity map 의 short_id 사용."
        )

        # 시도 전략 (동적, 사후 retry chain — censorship 사전 대응 X):
        #   1) primary      — gemini-pro (manifest 기본)
        #   2) retry_same   — 동일 모델 재시도 (일시 오류/네트워크 에러용).
        #                     deterministic provider block (예외 message 의
        #                     content_filter/safety/recitation/empty response) 감지 시 skip.
        #   3) fallback_gpt — gpt 모델로 project_config override (cross-provider fallback).
        gpt_config = {
            **self.project_config,
            "shot_validator": {
                **self.project_config.get("shot_validator", {}),
                "model": "gpt",
            },
        }
        attempts = [
            ("primary", self.project_config),
            ("retry_same", self.project_config),
            ("fallback_gpt", gpt_config),
        ]
        res = None
        last_exc: Optional[Exception] = None
        skip_retry_same = False
        for attempt_label, cfg in attempts:
            if attempt_label == "retry_same" and skip_retry_same:
                logger.info(
                    "shot_validator: scene %s skipping retry_same (deterministic failure detected)",
                    si,
                )
                continue
            try:
                candidate = call_structured(
                    step="shot_validator",
                    system_prompt=system,
                    user_prompt=user_prompt,
                    response_schema=schema,
                    project_config=cfg,
                    schema_name=f"shot_validator_s{si}_{attempt_label}",
                    opik_metadata=self.build_opik_metadata(extra_tags=[attempt_label], extra_metadata={"scene_index": si}),
                )
                if not candidate or not candidate.get("shots"):
                    raise ValueError("empty response (no shots)")
                res = candidate
                if attempt_label != "primary":
                    logger.info(
                        "shot_validator: scene %s succeeded on %s",
                        si, attempt_label,
                    )
                break
            except Exception as exc:
                last_exc = exc
                msg = str(exc).lower()
                # content_filter / safety / recitation는 동일 모델 재시도해도 같은 결과
                # → retry_same 스킵, 바로 fallback_gpt로
                if any(k in msg for k in ("content_filter", "content filter", "safety", "recitation", "empty response")):
                    if attempt_label == "primary":
                        skip_retry_same = True
                logger.warning(
                    "shot_validator: scene %s %s failed (%s)",
                    si, attempt_label, exc,
                )
        if res is None:
            assert last_exc is not None
            raise last_exc

        revisions = {r.get("shot_index"): r for r in res.get("shots", [])}
        new_shots: List[Dict[str, Any]] = []
        changed = 0

        for sh in shots:
            rev = revisions.get(sh.get("shot_index"))
            new_sh = dict(sh)
            if rev:
                if rev.get("changed"):
                    revised = rev.get("revised_description", "").strip()
                    if revised and revised != sh.get("description", ""):
                        new_sh["original_description"] = sh.get("description", "")
                        new_sh["description"] = revised
                        new_sh["validator_reason"] = rev.get("reason", "")
                        changed += 1
                # character_ids / characters merge — changed=false 여도 LLM 매핑은 적용.
                # 다운스트림 (shot_dependency / scene_detail) 가 두 표현을 모두 사용.
                rev_char_ids = rev.get("character_ids")
                if isinstance(rev_char_ids, list):
                    accepted, rejected = _filter_character_ids(rev_char_ids, pool_keys)
                    new_sh["character_ids"] = accepted
                    if rejected:
                        logger.warning(
                            "shot_validator: scene %s shot %s — dropping malformed/out-of-pool "
                            "character_ids %s (kept %s).",
                            si, sh.get("shot_index"), rejected, accepted,
                        )
                rev_chars = rev.get("characters")
                if isinstance(rev_chars, list):
                    # None 만 drop (str 강제 보존 — 다운스트림 lookup 의 raw name 유지).
                    new_sh["characters"] = [str(x) for x in rev_chars if x is not None]
            # post-validate: characters / character_ids 모두 비어있는 경우 marker
            # 추가는 보류 (Phase 5 carry). 이전 도입한 'no_character_mapping' 은
            # 배경/소품 shot 에도 false-positive — visible-human-action 분류 없이
            # marker 가 의미 약함. Phase 5 RO-26 contract validator 가 정밀 분류 후
            # 마킹 책임. 현재는 raw shot 그대로 carry.
            new_shots.append(new_sh)

        return {**original_scene, "shots": new_shots}, changed

    def _build_entity_blocks_for_all_scenes(self) -> Dict[int, str]:
        """Test-only wrapper — pool 을 자체 로드해서 빌드.

        production path 에서는 _execute() 가 pool 을 한 번 로드 후
        _build_entity_blocks(pool, shot_cp=...) 직접 호출.
        """
        return self._build_entity_blocks(self._load_character_pool())

    def _build_entity_blocks(
        self,
        pool: Dict[str, tuple[str, List[str]]],
        *,
        shot_cp: Optional[Dict[str, Any]] = None,
    ) -> Dict[int, str]:
        """[Entity map for this scene] 사전 빌드 — character (C##) 만.

        Source 우선순위 (multi-source — DAG order 7.25 가 entity_merge/scene_director
        보다 빨라 first-run 에서 RC-F 가 부분 비활성화 — force re-run 시 활성화):

        1. **Primary character pool**: entity_merge cp 의 characters → {short_id, name, [stable_traits]}.
           부재 시 EntityCanon DB fallback. 둘 다 부재면 RC-F 비활성 → WARNING + {}.
        2. **씬별 좁힘 (refinement)**: scene_director.present_entity_ids 의 C## 만.
           부재 시 모든 씬에 universal pool block 적용.

        pool 인자는 caller 가 _load_character_pool() 결과를 한번만 로드해서 전달
        (ThreadPoolExecutor 병렬 path 에서도 동일 pool 재사용).
        """
        if not pool:
            logger.warning(
                "shot_validator: entity character pool unavailable "
                "(entity_merge + EntityCanon both empty) — RC-F entity mapping "
                "inactive this run; re-run after entity pipeline produces characters."
            )
            return {}

        # ── Optional refinement: scene_director ──
        director_cp = _load_cp(self.project_id, self.episode_id, "scene_director")
        if not director_cp:
            logger.info(
                "shot_validator: scene_director cp absent — applying universal "
                "entity pool to all scenes (no per-scene narrowing)."
            )
            cp_for_indices = shot_cp or self._load_prev_checkpoint("shot_extract")
            if not cp_for_indices:
                return {}
            universal = self._format_entity_block(pool, sorted(pool.keys()))
            return {
                sc["scene_index"]: universal
                for sc in (cp_for_indices.get("data") or {}).get("scenes", []) or []
                if sc.get("scene_index") is not None
            }

        # ── scene_director 있음: 씬별 좁힘 ──
        director_scenes = (director_cp.get("data") or {}).get("scenes", []) or []
        blocks: Dict[int, str] = {}
        for sc in director_scenes:
            si = sc.get("scene_index")
            if si is None:
                continue
            # I3 — character (C##) 만. composite C##O## → base C##.
            #   L## / P## / O## 는 character_ids 필드 대상 아님.
            scene_char_ids = sorted({
                bid for bid in (
                    sid.split("O")[0]
                    for sid in (sc.get("present_entity_ids") or [])
                    if sid
                )
                if bid.startswith("C") and bid in pool
            })
            if not scene_char_ids:
                continue
            blocks[si] = self._format_entity_block(pool, scene_char_ids)

        return blocks

    def _load_character_pool(self) -> Dict[str, tuple[str, List[str]]]:
        """character pool — entity_merge cp 우선, 부재 시 EntityCanon DB.

        반환: {short_id: (name, stable_traits[:2])} — character (C##) 만.
        """
        # entity_merge cp 우선
        merge_cp = _load_cp(self.project_id, self.episode_id, "entity_merge")
        if merge_cp:
            chars = (merge_cp.get("data") or {}).get("characters") or []
            pool: Dict[str, tuple[str, List[str]]] = {}
            for c in chars:
                if not isinstance(c, dict):
                    continue
                sid = c.get("short_id")
                if not isinstance(sid, str) or not sid.startswith("C"):
                    continue
                name = c.get("name") or sid
                traits = _parse_traits(c.get("stable_traits"))[:2]
                pool[sid] = (str(name), traits)
            if pool:
                return pool

        # EntityCanon DB fallback
        try:
            rows = self.db.query(
                EntityCanon.short_id,
                EntityCanon.name,
                EntityCanon.stable_traits,
            ).filter(
                EntityCanon.project_id == self.project_id,
                EntityCanon.entity_type == "character",
            ).all()
        except Exception as exc:  # noqa: BLE001 — DB 미초기화 / 세션 오류 모두 fallback
            logger.warning(
                "shot_validator: EntityCanon character query failed (%s) — "
                "RC-F pool empty for this run.", exc,
            )
            return {}
        return {
            r.short_id: (r.name, _parse_traits(r.stable_traits)[:2])
            for r in rows
            if isinstance(r.short_id, str) and r.short_id.startswith("C")
        }

    @staticmethod
    def _format_entity_block(
        pool: Dict[str, tuple[str, List[str]]],
        sorted_ids: List[str],
    ) -> str:
        """[Entity map for this scene] 문자열 포맷 — 표시용 helper."""
        lines = ["[Entity map for this scene]"]
        for bid in sorted_ids:
            name, traits = pool[bid]
            traits_str = f" — {', '.join(traits)}" if traits else ""
            lines.append(f"- {bid}: {name}{traits_str}")
        return "\n".join(lines) + "\n\n"