"""plate_select — 같은 location 복수 플레이트의 샷별 VLM 선택 판정.

E2E6 육안 피드백 ⑤ (2026-07-16): 한 location 에 서브공간별 플레이트가
여러 장일 때 상류 배정(shot_ids)이 샷의 실제 서브공간과 어긋나는 실측
(S18sh1 안방 침대 샷 ↔ 거실 플레이트). 후보가 2장 이상이면 VLM(gemini
-pro)이 샷 텍스트·장소 서술 근거로 서브공간을 선택한다 — 불명확하면
현행 배정 유지(계약 명시, 판정=전부 LLM/VLM·하드코딩 0).

bg group id 는 "L{loc}B{n}" 계약 형식 — location 묶음은 이 식별자
구조 파싱(의미 판단 아님)으로만 한다.
"""
from __future__ import annotations

import logging
import re
from pathlib import Path
from typing import Any, Callable, Dict, List, Optional, Tuple

from app.modules.prompt_loader import load_prompt, load_schema

logger = logging.getLogger(__name__)

_MODULE = "plate_select"

PROMPT_VERSION_MAP = {
    "1": "1.202607161540",
}

# R1 (2026-07-16 Codex 설계 리뷰 BLOCKING-1): 플레이트 권위 선행 고정 계약
# 버전 — 콘티 생성 전에 plate_select 를 1회 수행·영속하고, 스틸(A/B 포함)은
# 그 기록을 재사용한다. 판정 시점·소비 계약이 바뀌면 bump (conti/scene
# config_hash 에 스탬프).
PLATE_AUTHORITY_VERSION = 1

# bg group id 계약 형식: L<loc>B<n> (예: L04B02)
_BG_ID_RE = re.compile(r"^(L\d+)B\d+$")

_LABELS = [chr(ord("A") + i) for i in range(8)]  # 후보 최대 8


def resolve_prompt_version(selector: str) -> str:
    try:
        return PROMPT_VERSION_MAP[selector]
    except KeyError:
        raise ValueError(
            f"unknown plate_select prompt version selector "
            f"{selector!r} (known: {sorted(PROMPT_VERSION_MAP)})"
        )


def location_of_bg_id(bg_id: str) -> Optional[str]:
    """bg group id → location 코드 (계약 형식 벗어나면 None)."""
    m = _BG_ID_RE.match(str(bg_id or ""))
    return m.group(1) if m else None


def plate_candidates_by_location(
    bg_groups: Dict[str, Dict[str, Any]],
) -> Dict[str, List[Tuple[str, Path]]]:
    """location 코드 → [(bg_id, plate_path)] (status ok + png 실재만).

    입력=background_render CP data.groups shape. 정렬=bg_id (결정론).
    """
    out: Dict[str, List[Tuple[str, Path]]] = {}
    for bg_id in sorted(bg_groups or {}):
        g = bg_groups[bg_id]
        if not isinstance(g, dict) or g.get("status") != "ok":
            continue
        loc = location_of_bg_id(bg_id)
        if not loc:
            continue
        png = g.get("png_path") or ""
        p = Path(png) if png else None
        # 재리뷰 NARROW-3: 디렉토리/소실 파일을 후보로 넣지 않는다
        # (이후 fingerprint read 예외 차단) — 파일 실재만
        if p is None or not p.is_file():
            continue
        out.setdefault(loc, []).append((bg_id, p))
    return out


def load_bg_groups_and_assignment(
    projects_dir: str, project_id: str, episode_id: str,
) -> Tuple[Dict[str, Dict[str, Any]], Dict[str, str]]:
    """(bg_groups 병합, shot_key("si_shi")→bg_id) — resolve_shot_plate_map
    과 동일 소스·선순위(background_render 우선, chain fallback)."""
    import json as _json

    from app.modules.pipeline.shot_conti_light import _SHOT_ID_RE

    groups: Dict[str, Dict[str, Any]] = {}
    assign: Dict[str, str] = {}
    for step in ("background_render", "background_chain_render"):
        cp = (
            Path(projects_dir) / project_id / "checkpoints" / "episodes"
            / episode_id / step / "manifest.json"
        )
        if not cp.exists():
            continue
        try:
            data = _json.loads(
                cp.read_text(encoding="utf-8")).get("data", {})
        except Exception as exc:  # noqa: BLE001
            logger.warning("plate_select: %s 로드 실패: %s", step, exc)
            continue
        for bg_id, g in (data.get("groups") or {}).items():
            if not isinstance(g, dict) or g.get("status") != "ok":
                continue
            if bg_id not in groups:
                groups[bg_id] = g
            for sid in g.get("shot_ids") or []:
                m = _SHOT_ID_RE.match(str(sid))
                key = (f"{int(m.group(1))}_{int(m.group(2))}"
                       if m else str(sid))
                if key not in assign:
                    assign[key] = bg_id
    return groups, assign


def load_bg_assignment_any_status(
    projects_dir: str, project_id: str, episode_id: str,
) -> Dict[str, str]:
    """shot_key → bg_id 배정 기록 — **status 불문** (seed-bg B2 계약).

    load_bg_groups_and_assignment 는 status=ok 그룹만 통과시켜 producer
    실패(failed/rejected) 그룹의 배정 기록이 사라진다 — 그러면 '실패한
    생산'이 '구조적 무생산'으로 오인되어 seed fallback 이 fail-open 된다
    (Codex seed-bg 리뷰 BLOCKING-2). 이 로더는 생산 기록의 존재 자체를
    보존한다: 어떤 상태든 배정 기록이 있으면 seed 하강 금지 대상.
    """
    import json as _json

    from app.modules.pipeline.shot_conti_light import _SHOT_ID_RE

    assign: Dict[str, str] = {}
    for step in ("background_render", "background_chain_render"):
        cp = (
            Path(projects_dir) / project_id / "checkpoints" / "episodes"
            / episode_id / step / "manifest.json"
        )
        if not cp.exists():
            continue
        try:
            data = _json.loads(
                cp.read_text(encoding="utf-8")).get("data", {})
        except Exception as exc:  # noqa: BLE001
            logger.warning(
                "plate_select(any-status): %s 로드 실패: %s", step, exc)
            continue
        for bg_id, g in (data.get("groups") or {}).items():
            if not isinstance(g, dict):
                continue
            for sid in g.get("shot_ids") or []:
                m = _SHOT_ID_RE.match(str(sid))
                key = (f"{int(m.group(1))}_{int(m.group(2))}"
                       if m else str(sid))
                if key not in assign:
                    assign[key] = bg_id
    return assign


def select_plate_for_shot(
    *,
    shot_desc: str,
    place_text: str,
    assigned_bg_id: str,
    candidates: List[Tuple[str, Path]],
    call_structured_fn: Callable[..., Dict[str, Any]],
    project_config: Optional[Dict[str, Any]] = None,
    prompt_version: str = "1",
    opik_metadata: Optional[Dict[str, Any]] = None,
) -> Tuple[str, Path, Dict[str, Any]]:
    """샷의 플레이트 VLM 선택 — (chosen_bg_id, path, record).

    후보 1장이거나 배정이 후보에 없으면 판정 없이 현행 유지(no-op,
    record.skipped 사유 기록). 판정 실패/계약 밖 choice = 현행 유지
    (fail-open — 배정 SOT 보존이 안전측, 기록 남김).
    """
    from app.modules.pipeline.multiroll_gemini import png_part

    by_id = {bid: p for bid, p in candidates}
    if assigned_bg_id not in by_id:
        return assigned_bg_id, Path(""), {
            "skipped": f"배정 {assigned_bg_id} 후보 밖 — 판정 생략"}
    if len(candidates) < 2:
        return assigned_bg_id, by_id[assigned_bg_id], {
            "skipped": "후보 1장 — 판정 불요"}
    if len(candidates) > len(_LABELS):
        # 라벨 초과 — 판정 생략(현행 유지)이 안전측. 조용히 자르지 않는다.
        return assigned_bg_id, by_id[assigned_bg_id], {
            "skipped": f"후보 {len(candidates)}장 > {len(_LABELS)} — 생략"}

    resolved = resolve_prompt_version(prompt_version)
    judge_sys = load_prompt(_MODULE, "judge_system", version=resolved)
    judge_schema = load_schema(_MODULE, "judge_schema", None,
                               version=resolved)
    labels = list(_LABELS[: len(candidates)])
    label_by_id = {bid: lab for (bid, _p), lab in zip(candidates, labels)}
    parts: List[Dict[str, Any]] = [{
        "type": "text",
        "text": (
            f"SHOT TEXT (authoritative, Korean): {shot_desc}\n"
            f"LOCATION: {place_text}"
        ),
    }]
    for (bid, p), lab in zip(candidates, labels):
        mark = (" — CURRENTLY ASSIGNED"
                if bid == assigned_bg_id else "")
        parts.append({"type": "text", "text": f"Candidate {lab}{mark}:"})
        parts.append(png_part(p))
    pc = {**(project_config or {}), f"{_MODULE}_judge": {"model": "gemini-pro"}}
    try:
        res = call_structured_fn(
            f"{_MODULE}_judge", judge_sys, parts, judge_schema,
            project_config=pc, schema_name=f"{_MODULE}_judge",
            opik_metadata=opik_metadata,
        )
    except Exception as exc:  # noqa: BLE001
        logger.warning("plate_select 판정 실패 — 배정 유지: %s", exc)
        return assigned_bg_id, by_id[assigned_bg_id], {
            "error": str(exc), "kept": assigned_bg_id}
    choice = (res.get("choice") or "").strip()
    id_by_label = {lab: bid for bid, lab in label_by_id.items()}
    chosen = id_by_label.get(choice)
    record = {
        "candidates": {lab: bid for bid, lab in label_by_id.items()},
        "assigned": assigned_bg_id,
        "choice": choice,
        "confident": bool(res.get("confident")),
        "reason_ko": res.get("reason_ko") or "",
    }
    if chosen is None:
        logger.warning(
            "plate_select: 계약 밖 choice %r — 배정 유지", choice)
        record["kept"] = assigned_bg_id
        return assigned_bg_id, by_id[assigned_bg_id], record
    # Codex 배치 리뷰 BLOCKING-1: 팩 계약("불명확=배정 유지")을 코드도
    # 강제 — confident 아닌 전환은 배정 유지 fail-open (기록 남김).
    if chosen != assigned_bg_id and res.get("confident") is not True:
        logger.info(
            "plate_select: low-confidence 전환 차단 %s→%s — 배정 유지",
            assigned_bg_id, chosen,
        )
        record["kept"] = assigned_bg_id
        record["blocked_low_confidence"] = chosen
        return assigned_bg_id, by_id[assigned_bg_id], record
    record["chosen"] = chosen
    return chosen, by_id[chosen], record


def plate_authority_fingerprint(
    *,
    shot_desc: str,
    place_text: str,
    assigned_bg_id: str,
    candidates: List[Tuple[str, Path]],
    judge_model: str,
    prompt_version: str = "1",
) -> str:
    """권위 판정의 결정 입력 지문 — 후보 플레이트 **bytes** 포함.

    플레이트 재생성·후보 구성·배정·샷 텍스트·팩·모델이 바뀌면 재판정,
    일치하면 판정 0콜 재사용 (Codex R1: 선택 기록 재사용 계약).
    """
    import json as _json

    from app.modules.pipeline.multiroll_select import (
        compute_input_fingerprint,
    )

    payload = _json.dumps(
        {
            "shot_desc": shot_desc,
            "place_text": place_text,
            "assigned": assigned_bg_id,
            "candidate_ids": [bid for bid, _ in candidates],
            "pack": resolve_prompt_version(prompt_version),
            "judge_model": judge_model,
            "authority_version": PLATE_AUTHORITY_VERSION,
        },
        sort_keys=True, ensure_ascii=False,
    )
    return compute_input_fingerprint(
        prompt=payload,
        labeled_refs=[(bid, p) for bid, p in candidates],
        roll_count=1,
        critique_enabled=False,
    )


def run_plate_authority(
    *,
    tags: List[str],
    plate_map: Dict[str, Path],
    assign_by_key: Dict[str, str],
    cands_by_loc: Dict[str, List[Tuple[str, Path]]],
    shot_desc_by_tag: Dict[str, str],
    place_text_by_scene: Dict[int, str],
    sidecar: Dict[str, Any],
    call_structured_fn: Callable[..., Dict[str, Any]],
    judge_model: str,
    project_config: Optional[Dict[str, Any]] = None,
    prompt_version: str = "1",
    opik_metadata: Optional[Dict[str, Any]] = None,
    force: bool = False,
) -> Dict[str, Dict[str, Any]]:
    """콘티 생성 **전** 플레이트 권위 판정 (R1 — 선행 고정).

    tags = 콘티 대상 샷(선택·person-visible·no-prev, 맵 플레이트 샷 제외).
    같은 location 후보 2장 이상 + 배정 존재 샷만 판정. 반환
    {tag: {assigned, chosen, plate_path, record, fingerprint}} — caller 가
    plate_map override 및 CP/사이드카 영속. sidecar 는 in-place 갱신
    (지문 일치=판정 0콜 재사용, force=재판정).

    판정된 샷만 entry 를 갖는다 — 후보 1장/배정 없음/플레이트 없음 샷은
    entry 없음(배정 플레이트가 콘티·스틸 공통 소스라 권위 충돌 없음).
    """
    from app.modules.pipeline.shot_ref_classify import parse_tag

    authority: Dict[str, Dict[str, Any]] = {}
    for tag in sorted(tags, key=parse_tag):
        si, shi = parse_tag(tag)
        key = f"{si}_{shi}"
        if plate_map.get(key) is None:
            continue
        assigned = assign_by_key.get(key) or ""
        if not assigned:
            continue
        loc = location_of_bg_id(assigned)
        cands = cands_by_loc.get(loc or "", [])
        if len(cands) < 2:
            continue
        # 배치 리뷰 HIGH-4: 배정이 후보 밖이면 판정 자체가 무의미 —
        # entry 를 만들지 않고 현재 plate_map 유지 (select 의 skipped
        # 반환 Path("") 이 '.' 로 영속되던 결함 차단)
        if assigned not in {bid for bid, _ in cands}:
            logger.warning(
                "plate_authority %s: 배정 %s 후보 밖 — 판정 생략(배정 "
                "유지)", tag, assigned,
            )
            continue
        fingerprint = plate_authority_fingerprint(
            shot_desc=shot_desc_by_tag.get(tag) or "",
            place_text=place_text_by_scene.get(si, ""),
            assigned_bg_id=assigned,
            candidates=cands,
            judge_model=judge_model,
            prompt_version=prompt_version,
        )
        prev = sidecar.get(tag) or {}
        if (
            not force
            and prev.get("fingerprint") == fingerprint
            and (prev.get("plate_path") or "")
            # HIGH-4: exists() 는 디렉토리('.')도 통과 — 파일 실재만
            and Path(prev["plate_path"]).is_file()
        ):
            authority[tag] = {
                "assigned": prev.get("assigned", assigned),
                "chosen": prev.get("chosen", assigned),
                "plate_path": prev["plate_path"],
                "record": prev.get("record") or {},
                "fingerprint": fingerprint,
                "reused": True,
            }
            continue
        chosen_id, chosen_path, record = select_plate_for_shot(
            shot_desc=shot_desc_by_tag.get(tag) or "",
            place_text=place_text_by_scene.get(si, ""),
            assigned_bg_id=assigned,
            candidates=cands,
            call_structured_fn=call_structured_fn,
            project_config=project_config,
            prompt_version=prompt_version,
            opik_metadata=opik_metadata,
        )
        if not str(chosen_path) or not Path(chosen_path).is_file():
            # HIGH-4 방어: 선택 결과가 실파일이 아니면 영속 금지 —
            # 배정(plate_map) 유지
            logger.warning(
                "plate_authority %s: 선택 결과 비파일(%r) — 배정 유지",
                tag, str(chosen_path),
            )
            continue
        entry = {
            "assigned": assigned,
            "chosen": chosen_id,
            "plate_path": str(chosen_path),
            "record": record,
            "fingerprint": fingerprint,
        }
        authority[tag] = entry
        sidecar[tag] = {
            k: entry[k]
            for k in ("assigned", "chosen", "plate_path", "record",
                      "fingerprint")
        }
    return authority