"""FloorPlanPromptStep — Phase 7 Step 3.

각 group plan의 floor_plans[]에 대해 LLM 1회씩 t2i prompt 생성.
도면 간에는 depends_on_fp DAG가 있으므로 level 병렬 가능 (대부분 single level).
"""
from __future__ import annotations

import json
import logging
from concurrent.futures import ThreadPoolExecutor, as_completed
from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple

from app.core.step_runner import StepRunner
from app.modules.pipeline.floor_plan_prompt import (
    PROMPT_VERSION_MAP,
    FloorPlanPromptError,
    build_fp_user_prompt,
    derive_fp_space_roles,
    resolve_prompt_version,
    run_floor_plan_prompt,
)

logger = logging.getLogger(__name__)
SCHEMA_VERSION = 2
# Phase 8 v2 prompt: numbered_elements + camera_recommendations + cultural cues.
# D6 v4: ``camera_recommendations[].bg_id`` pattern → BG_ID_RE (`^L\d{2,3}B\d{2,3}$`).
# 옛 lowercase pattern (`^[a-z0-9][a-z0-9_]*$`) 가 D6 ID 구조적 reject → LLM
# garbage hallucinate 결함 fix. user prompt 에 ``Valid bg_ids:`` block + bg
# detail 의 ``state_label_raw`` (D6 raw intent 필드) 사용.
# C8 W3a v5: 콘티 example·구체 location ID example·region-specific 가구 example
# 추상화 (prompt-pack content-only, code behavior 무변경).
# W19A: 이 constant 는 **default selector ("5") 가 가리키는 v5 pack** 의
# 디렉토리명을 backward-compat 으로 유지한다. 실제 _config_hash 에 들어가는
# prompt_version 값은 settings.floor_plan_prompt_version 을
# ``resolve_prompt_version`` 으로 풀어서 얻고, default 일 때 본 상수와 동일하다
# → 기존 v5 checkpoint hash 보존. v6 opt-in 시에만 새로운 hash 발생.
PROMPT_VERSION = PROMPT_VERSION_MAP["5"]


def _resolved_prompt_version() -> str:
    """settings.floor_plan_prompt_version selector 를 on-disk pack 이름으로 풀이.

    default selector "5" → ``PROMPT_VERSION_MAP["5"]`` ("5.202605201406") 그대로,
    opt-in "6" → ``PROMPT_VERSION_MAP["6"]``. import 시점에 settings 를 잡지
    않고 호출 시점 evaluation 으로 둬서 test/runtime override (monkeypatch /
    .env) 가 안정적으로 반영되게 한다.
    """
    from app.core.config import settings
    return resolve_prompt_version(settings.floor_plan_prompt_version)


class FloorPlanPromptStep(StepRunner):
    def _config_hash(self) -> str:
        import hashlib
        import json as _json

        from app.core.config import settings

        payload = {
            "background_mode": settings.background_mode,
            "schema_version": SCHEMA_VERSION,
            # W19A: selector default "5" → ``PROMPT_VERSION_MAP["5"]`` 와 동일
            # → 기존 v5 hash 와 byte-identical. opt-in "6" 만 새로운 hash.
            # selector 자체는 hash 에 넣지 않고 resolved pack name 만 사용해서
            # mapping table 이 안정 한 default 값에 영향을 안 주게 한다.
            "prompt_version": _resolved_prompt_version(),
        }
        return hashlib.sha256(
            _json.dumps(payload, sort_keys=True).encode("utf-8")
        ).hexdigest()[:16]

    def _load_prev_checkpoint(self, step_id: str) -> Optional[Dict[str, Any]]:
        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():
            try:
                return json.loads(cp.read_text(encoding="utf-8"))
            except Exception as exc:
                logger.warning(
                    "floor_plan_prompt: %s parse failed: %s", step_id, exc
                )
        return None

    def _execute(self, mode: str = "resume") -> Dict[str, Any]:
        from app.core.config import settings

        if settings.background_mode not in {"on", "floor_plan_anchored"}:
            return self._not_applicable()

        # W19A: selector ("5" default | "6" opt-in). resolve_prompt_version 이
        # unknown value 를 ValueError 로 reject 하므로 settings 에 잘못된 값이
        # 있으면 step 진입 직후 fail-fast (silent shift 방지). default 경로의
        # 결과 hash/load 는 모두 v5 와 동일하게 흐른다.
        fp_prompt_selector = settings.floor_plan_prompt_version
        # 검증을 위해 resolved version 도 1회 계산. 결과는 _config_hash 가 동일
        # helper 로 다시 계산하므로 여기서는 lookup 단계만 시도 (drift / typo
        # 조기 차단).
        resolve_prompt_version(fp_prompt_selector)

        plans_cp = self._load_prev_checkpoint("background_master_plan")
        scene_save_cp = self._load_prev_checkpoint("scene_save")
        shot_validator_cp = self._load_prev_checkpoint("shot_validator")
        shot_selection_cp = self._load_prev_checkpoint("shot_selection")
        rules_cp = self._load_prev_checkpoint("visual_world_rules")
        world_guide_cp = self._load_prev_checkpoint("world_guide")
        scene_director_cp = self._load_prev_checkpoint("scene_director")

        # W21B-wave-4 (Codex 합의): background_master_plan 과 동일한 scene → primary_location
        # fallback. shot 에 location_id 가 비어 있는 프로젝트에서는 이게 없으면
        # _select_scenes_for_group 이 모든 shot 을 skip → Scene segments 가 비어
        # floor_plan_prompt 모델이 씬 텍스트 0 으로 평면도를 설계한다 (모든 FP 영향).
        # 구조 필드(primary_location) exact join 만 사용 — regex / substring /
        # 시나리오 특화 없음.
        scene_primary: Dict[int, str] = {}
        if scene_director_cp:
            for sc in (scene_director_cp.get("data", {}) or {}).get("scenes", []) or []:
                si = sc.get("scene_index")
                primary = sc.get("primary_location", "") or ""
                if si is not None and primary:
                    scene_primary[int(si)] = primary

        plans_map = ((plans_cp or {}).get("data", {}) or {}).get("plans", {}) or {}
        # Phase 8.1: world_guide(풍부한 cultural cues) + visual_world_rules 통합.
        # Phase 7까지 사용했던 ``data.rules_text`` 키는 visual_world_rules cp에
        # 존재하지 않아 LLM에 빈 문자열이 들어갔다.
        from app.services.visual_context_helper import build_visual_context_block
        rules_text = build_visual_context_block(
            world_guide_cp=world_guide_cp,
            visual_world_rules_cp=rules_cp,
        )
        scene_segments = (
            ((scene_save_cp or {}).get("data", {}) or {}).get("segments", []) or []
        )

        # Reuse the scene-selection helper from BackgroundMasterPlanStep so that
        # group→scene mapping stays consistent with Step 2's master plan.
        from app.core.steps.background_master_plan_step import (
            _select_scenes_for_group,
        )
        from app.modules.llm.llm_client import call_structured

        # ★★같은 도면을 **여러 그룹이 같이 쓸 수 있다** (2026-09-19 실측:
        #  `fp_container_interior` 를 두 그룹이 썼다). 종전에는 그룹 안에서만
        #  배경을 모아 도면 이름으로 저장해서, **뒤 그룹이 앞 그룹을 덮어썼고**
        #  앞 그룹 배경의 카메라 추천이 통째로 사라졌다 — 세 단계 뒤
        #  `floor_plan_overlay_payload` 에서 「카메라 추천 없음」으로 죽었다.
        #  도면 하나당 **모든 그룹의 배경을 합쳐** 한 번만 묻는다.
        _by_fp: Dict[str, Dict[str, Any]] = {}
        for gid, entry in plans_map.items():
            if entry.get("status") != "ok":
                continue
            plan = entry.get("plan") or {}
            for fp in plan.get("floor_plans") or []:
                applied_bgs = [
                    bg
                    for bg in (plan.get("backgrounds") or [])
                    if fp["fp_id"] in (bg.get("depends_on_fp") or [])
                ]
                slot = _by_fp.setdefault(fp["fp_id"], {
                    "fp_id": fp["fp_id"], "fp_spec": dict(fp),
                    "applied_bgs": [], "applied_shots": [],
                    "group_id": gid, "group_ids": [], "fp_specs": [],
                })
                slot["group_ids"].append(gid)
                slot["fp_specs"].append((gid, fp))
                _seen_bg = {b.get("bg_id") for b in slot["applied_bgs"]}
                for bg in applied_bgs:
                    if bg.get("bg_id") not in _seen_bg:
                        slot["applied_bgs"].append(bg)
                        _seen_bg.add(bg.get("bg_id"))
                    slot["applied_shots"].extend(bg.get("applies_to_shots") or [])

        fp_jobs: List[Dict[str, Any]] = []
        shared_fp_diag: List[Dict[str, Any]] = []
        for slot in _by_fp.values():
            if len(slot["group_ids"]) > 1:
                # ★★**첫 그룹 것만 쓰면 안 된다** (Codex BLOCK). fp_id 는 그룹
                #  안에서만 유일한 **로컬 이름**이라, 같은 이름이 서로 다른
                #  장소를 가리킬 수 있다(실측: L166 「container_home」 과
                #  L170 「container_living_space」 가 둘 다 fp_container_interior).
                #  정의를 합치고, 신원이 갈리면 진단으로 남겨 사람이 본다.
                specs = slot["fp_specs"]
                ids = {(fp.get("loc_id"), fp.get("space_key_hint"),
                        fp.get("sub_location")) for _g, fp in specs}
                scopes, deps = [], []
                for _g, fp in specs:
                    sc = (fp.get("scope") or "").strip()
                    if sc and sc not in scopes:
                        scopes.append(sc)
                    for d in (fp.get("depends_on_fp") or []):
                        if d not in deps:
                            deps.append(d)
                merged = dict(specs[0][1])
                merged["scope"] = " / ".join(scopes)
                merged["depends_on_fp"] = deps
                slot["fp_spec"] = merged
                diag = {
                    "fp_id": slot["fp_id"],
                    "group_ids": list(slot["group_ids"]),
                    "identities": sorted(str(i) for i in ids),
                    "same_identity": len(ids) == 1,
                }
                shared_fp_diag.append(diag)
                (logger.info if diag["same_identity"] else logger.warning)(
                    "floor_plan_prompt: 도면 %s 를 그룹 %s 가 같이 쓴다 "
                    "(신원 %s · 배경 %d개를 합쳐 묻는다)",
                    slot["fp_id"], slot["group_ids"],
                    "같음" if diag["same_identity"] else f"다름 {diag['identities']}",
                    len(slot["applied_bgs"]))
            slot["applied_shots"] = list(dict.fromkeys(slot["applied_shots"]))
            slot["members"] = [
                {"loc_id": loc, "label": ""}
                for loc in {bg.get("loc_id") for bg in slot["applied_bgs"]
                            if bg.get("loc_id")}
            ]
            slot["all_group_ids"] = list(dict.fromkeys(slot["group_ids"]))
            slot.pop("group_ids", None)
            slot.pop("fp_specs", None)
            fp_jobs.append(slot)

        if not fp_jobs:
            # D6 T5a: empty path 도 hash stamp (preflight 가 D6/legacy 분기).
            # T5-fix B1: shot_binding_hash 도 stamp — applied_shots 가 user prompt
            # `{shots_block}` 으로 inject 되므로 binding 변경 시 입력 변함.
            plans_data_empty = (plans_cp or {}).get("data", {}) or {}
            return {
                "applicable_count": 1,
                "completed_count": 1,
                "failed_count": 0,
                "schema_version": SCHEMA_VERSION,
                "config_hash": self._config_hash(),
                "data": {
                    "floor_plans": {},
                    "consumed_bg_catalog_hash": plans_data_empty.get("bg_catalog_hash", "") or "",
                    "consumed_shot_binding_hash": plans_data_empty.get("shot_binding_hash", "") or "",
                },
            }

        # 제작자 정정 채널 (wave3 4번째 주입 사이트 — W21B scale-guideline):
        # v10 Rule 11 이 정정을 scale evidence source 로 인용한다. 정정 0건 =
        # "" = 프롬프트 byte-identical. wave3 계약 동일: 정정 내용은
        # config_hash 미포함 — completed cp 는 force/redo 로만 반영.
        from app.core.creator_corrections import project_corrections_block
        corrections_block = project_corrections_block(self.project_id)

        # group별 scenes 캐시 (반복 호출 절약)
        group_scene_cache: Dict[str, Tuple[List[Dict[str, Any]], List[str]]] = {}
        for gid, entry in plans_map.items():
            if entry.get("status") != "ok":
                continue
            plan = entry.get("plan") or {}
            group_member_locs = {
                bg.get("loc_id") for bg in (plan.get("backgrounds") or [])
            }
            synthetic_group = {
                "group_id": gid,
                "members": [
                    {"loc_id": loc, "label": ""}
                    for loc in group_member_locs
                    if loc
                ],
            }
            scenes, _shots = _select_scenes_for_group(
                synthetic_group,
                scene_segments,
                shot_validator_cp,
                shot_selection_cp,
                scene_primary=scene_primary,
            )
            group_scene_cache[gid] = (scenes, _shots)

        # 도면간 DAG는 거의 다 root이지만 generic helper로 안전하게 처리
        from app.modules.pipeline._dag_levels import compute_dag_levels

        items = {
            j["fp_id"]: {
                "parent_id": (j["fp_spec"].get("depends_on_fp") or [""])[0]
            }
            for j in fp_jobs
        }
        order = [j["fp_id"] for j in fp_jobs]
        renderable = set(order)
        levels = compute_dag_levels(
            order, items, renderable, parent_field="parent_id"
        )

        results: Dict[str, Any] = {}
        failed = 0
        opik = self.build_opik_metadata()

        def _process(job: Dict[str, Any]) -> Tuple[str, Dict[str, Any]]:
            # ★씬 원문도 **공유한 모든 그룹** 것을 합친다 — 한 그룹 것만
            #  실으면 다른 그룹 배경의 근거가 프롬프트에 안 들어간다.
            scenes = []
            _seen_scene = set()
            for _gid in (job.get("all_group_ids") or [job["group_id"]]):
                for sc in (group_scene_cache.get(_gid, ([], []))[0] or []):
                    key = sc.get("scene_index", id(sc))
                    if key not in _seen_scene:
                        _seen_scene.add(key)
                        scenes.append(sc)
            # W21B-wave-5: relay the LLM-assigned surface roles of this fp's
            # backgrounds into the prompt (code does not pick a primary type).
            # Only selector "8" injects them; older packs ignore the token.
            space_roles = derive_fp_space_roles(job["applied_bgs"])
            up = build_fp_user_prompt(
                fp_spec=job["fp_spec"],
                applied_backgrounds=job["applied_bgs"],
                applied_shots=job["applied_shots"],
                scene_segments=scenes,
                visual_world_rules=rules_text,
                prompt_version=fp_prompt_selector,
                space_roles=space_roles,
            )
            # Phase 8 v2: master_plan에서 이 fp_id를 depends_on_fp로 가진 모든
            # bg_id를 사전 추출 → run_floor_plan_prompt가 camera_recommendations
            # bg_id 부분집합 검증에 사용.
            expected_bg_ids = {
                bg["bg_id"]
                for bg in job["applied_bgs"]
                if bg.get("bg_id")
            }
            try:
                out = run_floor_plan_prompt(
                    user_prompt=up,
                    expected_fp_id=job["fp_id"],
                    call_structured_fn=call_structured,
                    project_config=self.project_config,
                    opik_metadata=opik,
                    expected_bg_ids=expected_bg_ids,
                    prompt_version=fp_prompt_selector,
                    creator_corrections_block=corrections_block,
                )
                return job["fp_id"], {
                    "status": "ok",
                    "t2i_prompt": out["t2i_prompt"],
                    "key_elements": out.get("key_elements", []),
                    # Phase 8 v2: persist new fields verbatim for downstream
                    # (background_prompt) consumption. Default to [] when LLM
                    # returns null/missing so consumers always see a list shape.
                    "numbered_elements": out.get("numbered_elements") or [],
                    "camera_recommendations": out.get("camera_recommendations") or [],
                    # W21B scale-guideline (v10+): 평면 도면이 못 싣는 수직
                    # 규모 메타데이터를 하류(볼륨 변환/렌더 프롬프트)로 전달.
                    # v6~v9 pack 출력에는 없으므로 빈 문자열로 통일.
                    "scale_guideline": out.get("scale_guideline") or "",
                    "applied_shots": job["applied_shots"],
                    "group_id": job["group_id"],
                    "depends_on_fp": job["fp_spec"].get("depends_on_fp") or [],
                }
            except FloorPlanPromptError as exc:
                logger.error(
                    "floor_plan_prompt: fp %s failed: %s", job["fp_id"], exc
                )
                return job["fp_id"], {
                    "status": "failed",
                    "error": str(exc)[:200],
                    "group_id": job["group_id"],
                }

        jobs_by_id = {j["fp_id"]: j for j in fp_jobs}
        from app.modules.pipeline._workers import resolve_workers
        max_workers = resolve_workers(default=4, cap=8)
        for level in levels:
            if not level:
                continue
            level_workers = max(1, min(max_workers, len(level)))
            with ThreadPoolExecutor(max_workers=level_workers) as pool:
                futures = {
                    pool.submit(_process, jobs_by_id[fid]): fid for fid in level
                }
                for fut in as_completed(futures):
                    fid = futures[fut]
                    try:
                        _fid, res = fut.result()
                    except Exception as exc:
                        logger.error(
                            "floor_plan_prompt: fp %s thread raised: %s",
                            fid,
                            exc,
                        )
                        res = {"status": "failed", "error": str(exc)[:200]}
                    results[fid] = res
                    if res.get("status") != "ok":
                        failed += 1

        # input order 보존
        ordered = {fid: results[fid] for fid in order if fid in results}

        # D6 T5a + T5-fix B1: master_plan cp 의 두 hash 모두 stamp.
        # floor_plan_prompt 가 catalog 의 fp 정보 (sub_location 등) + shot_background_map
        # (applied_shots → user prompt `{shots_block}` inject) 둘 다 의존.
        # binding 변경 시 prompt 입력이 변하므로 shot_binding_hash 도 추적 의무.
        plans_data = (plans_cp or {}).get("data", {}) or {}
        consumed_catalog_hash = plans_data.get("bg_catalog_hash", "") or ""
        consumed_binding_hash = plans_data.get("shot_binding_hash", "") or ""

        return {
            "applicable_count": 1,
            "completed_count": 1 if failed == 0 else 0,
            "failed_count": failed,
            "schema_version": SCHEMA_VERSION,
            "config_hash": self._config_hash(),
            "data": {
                "floor_plans": ordered,
                # ★여러 그룹이 같이 쓴 도면 — 신원이 갈린 것은 사람이 봐야 한다.
                #  로그로만 남기면 체크포인트를 읽는 쪽이 못 본다.
                "shared_floor_plans": shared_fp_diag,
                "consumed_bg_catalog_hash": consumed_catalog_hash,
                "consumed_shot_binding_hash": consumed_binding_hash,
            },
        }

    def _not_applicable(self) -> Dict[str, Any]:
        return {
            "applicable_count": 0,
            "completed_count": 0,
            "failed_count": 0,
            "schema_version": SCHEMA_VERSION,
            "config_hash": self._config_hash(),
            "data": {},
        }