"""레인2 spatial view cluster — 스키마·검증·DAG 순서 (설계 v2 Stage C).

군집 기준(합의 결정 4): 같은 고정 구조의 겹치는 면/개구부/레벨/접근축
공유 — 샷 번호·인물·시간 유사도 금지. LLM 은 군집 구조 JSON 만 저작,
코드는 완전성·DAG 무결성 fail-closed 검증과 결정론 topo 순서만 담당.
generation_order 는 LLM 저작이 아니라 parent DAG 에서 코드가 유도한다.
"""
from __future__ import annotations

from typing import Any, Dict, List

MAX_CLUSTERS = 6
_ALLOWED_CLUSTER_IDS = tuple(f"CL{i}" for i in range(1, MAX_CLUSTERS + 1))

_MODULE = "outdoor_view_cluster"

PROMPT_VERSION_MAP = {
    "1": "1.202607150600",
    # v2 (2026-07-15 육안 피드백): INTERIOR SPEC=구조물 규모 SOT 계약 —
    # 플레이트 외견만 보고 'compact' 축소 서술하는 결함 교정.
    "2": "2.202607151000",
    # v3 (재리뷰 BLOCKING): 증상 단어 금지 목록 제거 — 의미 계약만
    # ("framing 에서 크기 추론 금지, INTERIOR SPEC 을 따르라").
    "3": "3.202607151300",
}


def resolve_prompt_version(version: str) -> str:
    if version not in PROMPT_VERSION_MAP:
        raise ValueError(f"outdoor_view_cluster 프롬프트 버전 없음: {version}")
    return PROMPT_VERSION_MAP[version]


def build_cluster_schema(
    shot_keys: List[str], max_clusters: int = MAX_CLUSTERS
) -> Dict[str, Any]:
    """cluster plan 스키마 — cluster_id/member/parent enum 잠금."""
    ids = [f"CL{i}" for i in range(1, max_clusters + 1)]
    cluster = {
        "type": "object",
        "properties": {
            "cluster_id": {"enum": ids},
            "member_shot_keys": {
                "type": "array",
                "items": {"enum": list(shot_keys)},
                "minItems": 1,
            },
            "spatial_scope_en": {"type": "string", "minLength": 8},
            "stable_visible_features_en": {
                "type": "array",
                "items": {"type": "string", "minLength": 3},
                "minItems": 1,
            },
            "view_axis_family_en": {"type": "string", "minLength": 5},
            "overlap_evidence_en": {"type": "string", "minLength": 8},
            "parent_cluster_id": {"enum": ids + [None]},
            "bg_prompt_en": {"type": "string"},
            "rationale_ko": {"type": "string", "minLength": 5},
        },
        "required": [
            "cluster_id", "member_shot_keys", "spatial_scope_en",
            "stable_visible_features_en", "view_axis_family_en",
            "overlap_evidence_en", "parent_cluster_id", "bg_prompt_en",
            "rationale_ko",
        ],
        "additionalProperties": False,
    }
    return {
        "type": "object",
        "properties": {
            "clusters": {"type": "array", "items": cluster,
                         "minItems": 1, "maxItems": max_clusters},
        },
        "required": ["clusters"],
        "additionalProperties": False,
    }


def validate_cluster_plan(
    plan: Any, expected_shot_keys: List[str]
) -> List[str]:
    """결정론 완전성·DAG 무결성 검증 — 위반 리스트 반환 (fail-closed).

    스키마가 정상 경로를 잠그더라도 public 소비자(topo/배경 생성)가 이
    validator 를 직접 신뢰하므로 자체 완결 (Stage B N3 교훈): root/
    cluster 타입, 멤버 완전성, root 유일, 순환, bg_prompt 규칙까지.
    """
    if not isinstance(plan, dict):
        return ["plan 이 객체 아님"]
    clusters = plan.get("clusters")
    if not isinstance(clusters, list) or not clusters:
        return ["clusters 가 비어 있거나 배열 아님"]

    violations: List[str] = []
    # graph 순회는 입력 dict 가 아니라 이 정규화 맵만 소비 — malformed
    # 타입(unhashable 등)이 순회에 재진입할 수 없다 (재리뷰 NARROW_1).
    parent_by_cid: Dict[str, Any] = {}
    seen_members: List[str] = []
    roots: List[str] = []
    _REQUIRED_TEXT = (
        "spatial_scope_en", "view_axis_family_en",
        "overlap_evidence_en", "rationale_ko",
    )
    _INVALID = "__invalid__"
    for i, c in enumerate(clusters):
        if not isinstance(c, dict):
            violations.append(f"[{i}] cluster 가 객체 아님")
            continue
        raw_cid = c.get("cluster_id")
        cid_valid = (isinstance(raw_cid, str)
                     and raw_cid in _ALLOWED_CLUSTER_IDS)
        if not cid_valid:
            violations.append(f"[{i}] cluster_id {raw_cid!r} 허용 밖")
        elif raw_cid in parent_by_cid:
            violations.append(f"cluster_id {raw_cid} 중복")
            cid_valid = False
        label = raw_cid if cid_valid else f"[{i}]"
        members = c.get("member_shot_keys")
        if not isinstance(members, list) or not members:
            violations.append(f"{label} member_shot_keys 가 비어 있거나 "
                              "배열 아님")
            members = []
        for m in members:
            if not isinstance(m, str):
                violations.append(f"{label} member {m!r} 문자열 아님")
                continue
            if m in seen_members:
                violations.append(f"샷 {m} 가 복수 cluster 에 중복 소속")
            seen_members.append(m)
            if m not in expected_shot_keys:
                violations.append(f"대상 밖 샷 {m}")
        # required text 4종 — 타입·필수성 (Codex Stage C BLOCKING)
        for field in _REQUIRED_TEXT:
            val = c.get(field)
            if not isinstance(val, str) or not val.strip():
                violations.append(
                    f"{label} {field} 이 비어 있거나 문자열 아님")
        feats = c.get("stable_visible_features_en")
        if not isinstance(feats, list) or not feats:
            violations.append(f"{label} stable_visible_features 비어 있음")
        else:
            for f in feats:
                if not isinstance(f, str) or not f.strip():
                    violations.append(
                        f"{label} features 원소 {f!r} 문자열 아님")
        # parent — required 키: 누락은 명시적 null 과 동치 아님 (재리뷰)
        if "parent_cluster_id" not in c:
            violations.append(
                f"{label} parent_cluster_id 키 누락 — root 는 명시적 null")
            parent_norm: Any = _INVALID
        else:
            parent = c["parent_cluster_id"]
            if parent is None:
                parent_norm = None
            elif isinstance(parent, str):
                parent_norm = parent
            else:
                violations.append(
                    f"{label} parent_cluster_id {parent!r} 타입 위반 "
                    "(null 또는 문자열)")
                parent_norm = _INVALID
        if parent_norm is None:
            roots.append(label)
        # bg — required string: 키 누락/None/비문자 = 위반, root 는
        # 명시적 빈 문자열만 허용 (재리뷰)
        bg_val = c.get("bg_prompt_en")
        bg_known = isinstance(bg_val, str)
        if not bg_known:
            violations.append(
                f"{label} bg_prompt_en 이 문자열 아님/누락 — root 도 명시적 "
                "빈 문자열이어야 함")
        bg = bg_val.strip() if bg_known else ""
        if parent_norm is None and bg_known and bg:
            violations.append(
                f"root {label} 에 bg_prompt_en 금지 — root 배경=확정 플레이트")
        if parent_norm is not None and parent_norm != _INVALID and (
                bg_known and not bg):
            violations.append(f"{label} bg_prompt_en 누락 (비-root 필수)")
        if cid_valid:
            parent_by_cid[raw_cid] = parent_norm

    missing = [k for k in expected_shot_keys if k not in seen_members]
    if missing:
        violations.append(f"멤버 누락 샷: {missing}")
    if len(roots) != 1:
        violations.append(f"root(parent null) 는 정확히 1개여야 함: {roots}")

    # parent 유효·자기참조·순환 — 정규화 맵만 소비
    for cid, parent in parent_by_cid.items():
        if parent is None or parent == _INVALID:
            continue
        if parent == cid:
            violations.append(f"{cid} 자기 참조 parent")
            continue
        if parent not in parent_by_cid:
            violations.append(f"{cid} 의 parent {parent!r} 미정의")
            continue
        hops, cur = 0, parent
        while (cur is not None and cur != _INVALID
               and hops <= len(parent_by_cid)):
            cur = parent_by_cid.get(cur)
            hops += 1
        if hops > len(parent_by_cid):
            violations.append(f"{cid} 에서 parent 순환 감지")
    return violations


def generation_order(plan: Dict[str, Any]) -> List[str]:
    """DAG topo 순서 — root 먼저, parent 가 항상 앞, 동레벨=id 정렬."""
    clusters = plan.get("clusters") or []
    by_id = {c["cluster_id"]: c for c in clusters}
    order: List[str] = []
    remaining = sorted(by_id)
    while remaining:
        progressed = False
        for cid in list(remaining):
            parent = by_id[cid].get("parent_cluster_id")
            if parent is None or parent in order:
                order.append(cid)
                remaining.remove(cid)
                progressed = True
        if not progressed:  # 순환 — validator 가 사전 차단하지만 방어
            raise ValueError(f"cluster DAG 순환: {remaining}")
    return order


def _shot_key(sh: Dict[str, Any]) -> str:
    return f"S{sh['scene_index']}_Shot{sh['shot_index']}"


def run_view_cluster_plan_group(
    *,
    spec: Dict[str, Any],
    structure_shots: List[Dict[str, Any]],
    scene_texts: Dict[int, str],
    plate_images: List[tuple],
    interior_block: str = "",
    prompt_version: str = "3",
    call_structured_fn=None,
    project_config: Dict[str, Any] | None = None,
    opik_metadata: Dict[str, Any] | None = None,
    max_attempts: int = 3,
) -> Dict[str, Any]:
    """그룹 1개 view cluster plan — 위반 힌트 재시도.

    반환 {"plan": <검증 통과 plan>, "attempts": n}.
    소진 시 AppError(step.contract_violation.outdoor_view_cluster).
    """
    from app.core.errors import AppError

    # upstream 결손 fail-fast (Codex Stage C HIGH) — 씬 원문·확정 플레이트
    # 없이 조용히 성공하면 "root 배경=확정 플레이트/씬 전문" acceptance 가
    # caller 실수·재개 파일 결손 시 무력화된다.
    missing: List[str] = []
    if not structure_shots:
        missing.append("structure_shots 비어 있음")
    for sh in structure_shots:
        si = sh.get("scene_index")
        text = scene_texts.get(si)
        if not (isinstance(text, str) and text.strip()):
            missing.append(f"scene {si} 원문 결손")
    if not plate_images:
        missing.append("확정 플레이트(plate_images) 0장")
    if missing:
        raise AppError(
            code="step.missing_input.outdoor_view_cluster",
            message="view cluster 입력 결손: " + "; ".join(sorted(set(missing))),
            status_code=422,
        )

    if call_structured_fn is None:
        from app.modules.llm.llm_client import call_structured

        call_structured_fn = call_structured

    from app.modules.pipeline.multiroll_gemini import png_part
    from app.modules.pipeline.outdoor_shot_grounding import (
        build_legend_block,
        build_shot_block,
    )
    from app.modules.prompt_loader import load_prompt

    resolved = resolve_prompt_version(prompt_version)
    system = load_prompt(_MODULE, "system", version=resolved)
    template = load_prompt(_MODULE, "user_template", version=resolved)
    shot_keys = [_shot_key(sh) for sh in structure_shots]
    schema = build_cluster_schema(shot_keys)

    filled = template
    for key, val in {
        "legend_block": build_legend_block(spec),
        "zones_block": "\n".join(
            f"- {z}" for z in spec.get("zone_labels_en", []) or []
        ),
        "interior_block": interior_block or "(실내 스펙 없음)",
        "shots_block": "\n\n".join(
            build_shot_block(sh) for sh in structure_shots
        ),
        # 씬 원문 전문 — 절대 자르지 않는다 (CLAUDE.md 절대 규칙)
        "scene_texts_block": "\n\n".join(
            f"[Scene {si}]\n{scene_texts[si]}" for si in sorted(scene_texts)
        ),
    }.items():
        filled = filled.replace("{" + key + "}", val)
    base_parts: List[Dict[str, Any]] = []
    for label, png in plate_images:
        base_parts.append({
            "type": "text",
            "text": f"CONFIRMED STRUCTURE PLATE — {label}:",
        })
        base_parts.append(png_part(png))
    base_parts.append({"type": "text", "text": filled})

    attempts = 0
    parts = base_parts
    last: List[str] = []
    while attempts < max_attempts:
        attempts += 1
        result = call_structured_fn(
            _MODULE, system, parts, schema,
            project_config=project_config,
            schema_name=_MODULE,
            opik_metadata=opik_metadata,
        )
        violations = validate_cluster_plan(result or {}, shot_keys)
        if not violations:
            return {"plan": result, "attempts": attempts}
        last = violations
        hint = "\n".join(
            ["", "", "[재시도 — 직전 응답이 아래 계약을 위반했습니다. 전부",
             " 고쳐서 전체 결과를 다시 출력하세요:]"]
            + [f"  - {v}" for v in violations]
        )
        parts = base_parts + [{"type": "text", "text": hint}]

    raise AppError(
        code="step.contract_violation.outdoor_view_cluster",
        message=f"view cluster 계약 위반 (attempts={max_attempts}): "
                + "; ".join(last[:8]),
        status_code=422,
    )


BG_PACK_MODULE = "cluster_bg"

BG_PACK_VERSION_MAP = {
    "1": "1.202607150610",
    # v2 (2026-07-15 육안 피드백): 씨드 시점 설명(SEED VIEWPOINT)+단일
    # 구조물 dedup 절(CL4 두 옥상 병치 실측 교정)+SCALE SPEC(규모 SOT)
    # +root plate 교정 생성 head.
    "2": "2.202607151010",
    # v3 (재리뷰 BLOCKING+H2): dedup 절 일반화(target_structure —
    # 'rooftop' 하드코딩 제거, 주변 구조물 과제약 완화), root_head 를
    # 무조건 재촬영 계약으로(SCALE 언급은 scale_clause 존재 시만 —
    # dangling SCALE SPEC 참조 제거).
    "3": "3.202607151310",
}


def resolve_bg_pack_version(version: str) -> str:
    if version not in BG_PACK_VERSION_MAP:
        raise ValueError(f"cluster_bg 팩 버전 없음: {version}")
    return BG_PACK_VERSION_MAP[version]


# selector 별 dedup 절 stem — 명시적 과거 selector 실행 계약 유지
# (재리뷰 NARROW_1: 무조건 target_structure 로드가 v1/v2 를
# FileNotFoundError 로 깨뜨림). v1 은 dedup 절 자체가 없던 조립.
_DEDUP_STEM_BY_PACK = {
    BG_PACK_VERSION_MAP["2"]: "single_structure",
    BG_PACK_VERSION_MAP["3"]: "target_structure",
}


def build_cluster_bg_prompt(
    *,
    cluster: Dict[str, Any],
    place_text: str,
    world_anchor: str,
    seed_view_en: str = "",
    scale_note_en: str = "",
    prompt_version: str = "3",
) -> str:
    """비-root cluster 배경 프롬프트 조립 (head → seed(+시점) → target
    dedup → scale → parent 계약 → LOCATION → features → no_people).
    seed_view_en default="" 는 v1 호출 시그니처 호환(f5441f2a 이전에는
    이 인자가 없었음) — v2/v3 경로는 분기 안 non-empty 가드가 잠근다.
    root 는 배경=확정 플레이트이므로 ValueError.

    seed_view_en = 첨부 씨드가 보여주는 뷰(root cluster 의
    spatial_scope_en) — 씨드 정체 설명 없이는 생성 모델이 씨드 속
    구조물을 인접 복제할 수 있다 (CL4 두 옥상 병치 실측).
    scale_note_en = 실내 구성 규모 앵커(같은 building 실내 loc 서술).
    """
    from app.modules.prompt_loader import load_prompt

    if cluster.get("parent_cluster_id") is None:
        raise ValueError("root cluster 배경은 확정 플레이트 — 생성 금지")
    bg = (cluster.get("bg_prompt_en") or "").strip()
    if not bg:
        raise ValueError(f"{cluster.get('cluster_id')} bg_prompt_en 없음")

    resolved = resolve_bg_pack_version(prompt_version)
    feats = ", ".join(cluster.get("stable_visible_features_en") or [])
    if resolved == BG_PACK_VERSION_MAP["1"]:
        # v1 원 조립 계약 — seed_view/dedup/scale 절 없던 버전
        return "\n\n".join([
            f"A photographic background plate{world_anchor}. {bg}",
            load_prompt(BG_PACK_MODULE, "seed_clause",
                        version=resolved).strip(),
            load_prompt(BG_PACK_MODULE, "parent_clause",
                        version=resolved).strip(),
            f"THE LOCATION: {place_text}",
            f"KEEP CONSISTENT (fixed features of this view family): {feats}",
            load_prompt(BG_PACK_MODULE, "no_people",
                        version=resolved).strip(),
        ])

    if not (seed_view_en or "").strip():
        raise ValueError("seed_view_en 결손 — 씨드 시점 설명 필수")
    parts = [
        f"A photographic background plate{world_anchor}. {bg}",
        load_prompt(BG_PACK_MODULE, "seed_clause", version=resolved).strip(),
        f"SEED VIEWPOINT — the attached plate shows: {seed_view_en}",
        load_prompt(BG_PACK_MODULE, _DEDUP_STEM_BY_PACK[resolved],
                    version=resolved).strip(),
    ]
    if (scale_note_en or "").strip():
        parts.append(
            load_prompt(BG_PACK_MODULE, "scale_clause",
                        version=resolved).strip()
            + f"\nINTERIOR: {scale_note_en}"
        )
    parts += [
        load_prompt(BG_PACK_MODULE, "parent_clause",
                    version=resolved).strip(),
        f"THE LOCATION: {place_text}",
        f"KEEP CONSISTENT (fixed features of this view family): {feats}",
        load_prompt(BG_PACK_MODULE, "no_people", version=resolved).strip(),
    ]
    return "\n\n".join(parts)


def build_root_plate_prompt(
    *,
    cluster: Dict[str, Any],
    place_text: str,
    world_anchor: str,
    scale_note_en: str = "",
    prompt_version: str = "3",
) -> str:
    """root plate 교정 생성 프롬프트 — 기존 확정 플레이트를 씨드로 같은
    뷰를 재촬영. root_head 는 무조건 동일성 계약이고 SCALE 언급은
    scale_note 가 있을 때만 삽입 (재리뷰 H2 — dangling SCALE SPEC 참조가
    실내 스펙 없는 그룹에서 임의 규모·내부 발명 유도). 비-root 전달 시
    ValueError."""
    from app.modules.prompt_loader import load_prompt

    if cluster.get("parent_cluster_id") is not None:
        raise ValueError("root plate 프롬프트는 root cluster 전용")

    resolved = resolve_bg_pack_version(prompt_version)
    if resolved == BG_PACK_VERSION_MAP["1"]:
        # v1 팩에는 root_head 가 없다 — 원래 존재하지 않던 경로이므로
        # FileNotFound 대신 명시적 fail-fast (재리뷰 NARROW_1)
        raise ValueError(
            "root plate 프롬프트는 cluster_bg 팩 v2 이상 (v1 미지원)")
    feats = ", ".join(cluster.get("stable_visible_features_en") or [])
    parts = [
        f"A photographic background plate{world_anchor}.",
        load_prompt(BG_PACK_MODULE, "root_head", version=resolved).strip(),
        f"THE VIEW: {cluster.get('spatial_scope_en') or ''}",
    ]
    if (scale_note_en or "").strip():
        parts.append(
            load_prompt(BG_PACK_MODULE, "scale_clause",
                        version=resolved).strip()
            + f"\nINTERIOR: {scale_note_en}"
        )
    parts += [
        load_prompt(BG_PACK_MODULE, _DEDUP_STEM_BY_PACK[resolved],
                    version=resolved).strip(),
        f"THE LOCATION: {place_text}",
        f"KEEP CONSISTENT (fixed features): {feats}",
        load_prompt(BG_PACK_MODULE, "no_people", version=resolved).strip(),
    ]
    return "\n\n".join(parts)


def build_cluster_bg_refs(
    *,
    root_plate: Any,
    parent_bg: Any = None,
    root_label: str,
    parent_label: str,
) -> List[tuple]:
    """cluster 배경 생성 라벨드 참조 — [root 씨드(+부모 배경)].

    맵/콘티/photo canon 은 구조상 진입 불가 (레인2 acceptance).
    root seed 결손은 조용히 통과 금지 (Codex Stage C HIGH) — 씨드 없는
    배경 생성은 'root→cluster 체인' acceptance 무력화.
    """
    if root_plate is None:
        raise ValueError("root_plate 결손 — root 씨드 없이 배경 생성 금지")
    if not (root_label or "").strip():
        raise ValueError("root_label 비어 있음")
    if parent_bg is not None and not (parent_label or "").strip():
        raise ValueError("parent_bg 첨부인데 parent_label 비어 있음")
    refs: List[tuple] = [(root_label, root_plate)]
    if parent_bg is not None:
        refs.append((parent_label, parent_bg))
    return refs