"""프롬프트 로더 — DB 우선, 파일 fallback.

1. DB에서 module+name+is_active=true 조회
2. 없으면 파일에서 최신 버전 로드
3. 새 프롬프트 저장 시 DB + 파일 동시 저장

## Version pack semantics (problems.md #6)

Module 디렉토리는 여러 version 디렉토리를 가질 수 있고 (예: ``shot_extract/
9.x``, ``shot_extract/10.x``, ``shot_extract/11.x``), 각 version 안에 stem
파일이 들어 있다. 이 로더는 stem 별로 ``_version_sort_key`` 기준 가장 최신
version 디렉토리를 독립적으로 탐색하므로, **같은 module 안에서 stem 이 서로
다른 version 에서 로드될 수 있다** (e.g. ``system.md`` 는 v11, ``schema.json``
은 v10 — 각자가 자기 stem 을 가진 가장 최신 version 우선).

이 동작은 backward-compat 으로 유지하되 다음 두 단계 가시화/강제 layer 를
제공한다.

1. **observability** (default): 같은 module 안에서 stem 들의 effective version
   이 일치하지 않으면 ``logger.warning`` (process 당 1회) + ``get_effective
   _source`` response 의 ``module_pack.drift`` 에 stem→version 매핑 노출.
2. **opt-in strict** (``PROMPT_VERSION_PACK_STRICT=true``): module 의 가장 최신
   version 디렉토리 안에 stem 이 없으면 ``RuntimeError`` 발생. 운영자가 prompt
   정리 후 default 로 전환하기 위한 toggle.

   strict 모드 적용 boundary (review I3/I4):
     - **module 디렉토리 자체가 없으면** strict 가 raise 하지 않고 기존
       ``FileNotFoundError`` 가 caller 에서 발생한다 (typo 등은 strict 와 별개
       경로로 진단). 같은 module 의 stem-version 정렬에만 적용.
     - **db= 인자로 DB row 가 hit 되면** ``_load_from_file`` 자체를 호출하지
       않으므로 strict 검증이 우회된다. 거의 모든 caller 가 file-only (db 미전달)
       이지만 ``location_floor_plan_step`` 등 db-aware caller 는 stale DB row
       가 stem 을 살리는 시나리오에서는 strict toggle 영향을 받지 않는다.
"""

import json
import logging
import os
from pathlib import Path
from threading import Lock
from typing import Any, Dict, List, Literal, Optional, Tuple

logger = logging.getLogger(__name__)

# Process 당 한 번만 logger.warning emit 하기 위한 module/stem-version 캐시.
_DRIFT_WARNED_LOCK = Lock()
_DRIFT_WARNED: set[tuple[str, str, str]] = set()  # (module, stem, version)


def _version_sort_key(name: str):
    """버전 디렉토리 정렬 키 — '12.202603271830' → (12, '202603271830').

    Primary segment 는 numeric, secondary 는 lexical 비교. 본 프로젝트 컨벤션
    (`버전.YYYYMMDDHHmm`, 12-digit timestamp) 에서는 secondary 길이가 고정이라
    lexical 정렬이 시간 순서와 일치한다.

    **주의 — 비표준 secondary 형식 시 lexical drift 재발**:
      - ``9.99`` vs ``9.100`` 형식은 secondary lexical 비교에서 ``"99" > "100"``
        (잘못된 정렬). 본 프로젝트는 12-digit timestamp 만 사용하므로 안전.
      - 다른 secondary 형식 (예: dotted semver `9.1.2`) 도입 시 추가 numeric
        cast 필요.
    """
    parts = name.split(".", 1)
    try:
        return (int(parts[0]), parts[1] if len(parts) > 1 else "")
    except (ValueError, IndexError):
        return (0, name)


_AllowedWhereExtra = Literal["", "AND schema_json IS NOT NULL"]


def _select_latest_active_row(
    db,
    module: str,
    name: str,
    where_extra: _AllowedWhereExtra = "",
    version: Optional[str] = None,
):
    """prompt_template 중 (module, name, is_active=true) + extra where 매칭 row
    들을 모두 가져와 ``_version_sort_key`` 기준 최댓값 row 반환. 없으면 None.

    SQL ``ORDER BY version DESC`` 는 lexical 이라 ``9.x`` > ``10.x`` 로 잘못된
    정렬이 가능 (problems.md #7). Python in-memory 정렬로 numeric-aware.

    where_extra: ``_AllowedWhereExtra`` 닫힌 enum 으로 좁힘 — 신규 절을 추가
    하려면 enum 에 명시 필요 (정적 보호). 직접 concat 방식이라 user-supplied
    값을 받지 못하게 type-level 차단.

    W19A versioned-load (review §3.6): ``version`` 이 주어지면 그 version 과
    정확히 일치하는 row 만 후보로 두고 sort. 없으면 None (latest 로 silent
    fallback 하지 않음 — caller 의 명시적 pin 을 깨지 않기 위해).
    """
    from sqlalchemy import text as sql_text
    sql = (
        "SELECT * FROM prompt_template "
        "WHERE module = :m AND name = :n AND is_active = true "
        f"{where_extra}"
    )
    rows = db.execute(sql_text(sql), {"m": module, "n": name}).fetchall()
    if not rows:
        return None
    if version is not None:
        rows = [r for r in rows if r.version == version]
        if not rows:
            return None
    rows.sort(key=lambda r: _version_sort_key(r.version), reverse=True)
    return rows[0]


PROMPTS_BASE = (
    Path(__file__).resolve().parent.parent.parent.parent
    / "prompts" / "_base"
)


from functools import lru_cache  # noqa: E402 — 아래 해시 전용


@lru_cache(maxsize=64)
def pack_dir_content_hash(module: str, resolved: str) -> str:
    """해석된 프롬프트 팩 디렉토리의 파일 bytes 해시 16 hex (#77 공용).

    버전 '문자열' 스탬프는 셀렉터 상수·해석 코드·서버 메모리 중 어느 층이
    낡아도 안 움직인다 — 이 해시는 **이 프로세스가 실제 로드하는 디렉토리의
    bytes** 를 지문에 접는다(2026-08-08 최종 스틸 253/311 재사용 사고의
    수정을 multiroll_judge 전용에서 모든 팩 공용으로 승격).

    팩 디렉토리는 불변 관례(덮어쓰기 금지·새 버전 디렉토리)라 프로세스
    수명 동안 캐시한다. 부재·빈 디렉토리는 fail-closed — 없는 팩을 빈
    해시로 접으면 "팩이 사라져도 지문이 안 움직이는" 같은 부류의 구멍이
    다시 생긴다.
    """
    import hashlib

    root = PROMPTS_BASE / module / resolved
    files = (sorted(p for p in root.rglob("*") if p.is_file())
             if root.is_dir() else [])
    if not files:
        raise ValueError(
            f"{module} 팩 디렉토리가 없거나 비어 있다: {root}")
    h = hashlib.sha256()
    for p in files:
        h.update(str(p.relative_to(root)).encode("utf-8"))
        h.update(b"\x00")
        h.update(hashlib.sha256(p.read_bytes()).digest())
    return h.hexdigest()[:16]


def pack_stem_content_hash(module: str, resolved: str, stem: str) -> str:
    """해석된 팩 디렉토리 안 **한 스템 파일**의 bytes 해시 16 hex.

    (2026-08-13 Codex R2): 소비 범위가 스템 하나인 절을 디렉토리 전체
    해시로 지문에 접으면, 무관한 동거 스템 개정·selector 이동까지 전량
    유료 재생성을 일으킨다 — 그 스템 bytes 만 접는다. 부재=fail-closed
    (pack_dir_content_hash 와 같은 이유).
    """
    import hashlib

    p = PROMPTS_BASE / module / resolved / f"{stem}.md"
    if not p.is_file():
        raise ValueError(f"{module} 팩 스템이 없다: {p}")
    return hashlib.sha256(p.read_bytes()).hexdigest()[:16]


def load_prompt(
    module: str,
    name: str,
    db=None,
    version: Optional[str] = None,
    **kwargs,
) -> str:
    """프롬프트 텍스트 로드 (DB 우선, 파일 fallback).

    Args:
        module: 모듈명 (scene_extractor_v2, outlook_extractor 등)
        name: 파일명 stem (system, turn_scene_detail 등)
        db: SQLAlchemy session (없으면 파일만 사용)
        version: pack 디렉토리 / DB row version 정확히 일치하는 source 만
            허용. None 이면 기존 latest-pack 동작 (numeric desc) 유지.
            W19A 가 floor_plan_prompt 의 v5↔v6 selector 를 안전하게 분기하기
            위해 도입 (review §3.6). 다른 module 은 caller 가 version 을
            전달하지 않는 한 동작 변경 없음.
        **kwargs: 프롬프트 텍스트의 format 변수

    Returns: 포맷된 프롬프트 텍스트
    """
    content = None

    # 1. DB에서 로드 (numeric-aware version 정렬 — lexical "9.x" > "10.x" 버그 회피)
    if db:
        try:
            row = _select_latest_active_row(db, module, name, version=version)
            if row:
                content = row.content
        except Exception as exc:
            logger.warning("DB prompt load failed (%s/%s): %s", module, name, exc)

    # 2. 파일 fallback
    if content is None:
        content = _load_from_file(module, name, version=version)

    if content is None:
        raise FileNotFoundError(
            f"Prompt not found: {module}/{name}"
            + (f" (version={version})" if version is not None else "")
        )

    return content.format(**kwargs) if kwargs else content


def load_schema(
    module: str,
    name: str,
    db=None,
    version: Optional[str] = None,
) -> Dict[str, Any]:
    """JSON 스키마 로드 (DB 우선, 파일 fallback).

    Args:
        module: 모듈명
        name: 스키마 파일명 stem (extract_schema 등, .json 확장자 제외)
        version: pack 디렉토리 / DB row version 정확히 일치하는 source 만
            허용. None 이면 기존 latest-pack 동작 유지. ``load_prompt`` 와
            동일한 의미.
    """
    # 1. DB에서 로드 (numeric-aware version 정렬)
    if db:
        try:
            row = _select_latest_active_row(
                db, module, name,
                where_extra="AND schema_json IS NOT NULL",
                version=version,
            )
            if row and row.schema_json:
                return json.loads(row.schema_json)
        except Exception as exc:
            logger.warning("DB schema load failed (%s/%s): %s", module, name, exc)

    # 2. 파일 fallback
    content = _load_from_file(module, name, ext=".json", version=version)
    if content:
        return json.loads(content)

    raise FileNotFoundError(
        f"Schema not found: {module}/{name}"
        + (f" (version={version})" if version is not None else "")
    )


def _list_module_versions(module: str) -> List[str]:
    """module 디렉토리의 version 디렉토리 이름들 (numeric desc 정렬). 없으면 [].

    같은 정렬을 ``_load_from_file`` / ``get_effective_source`` /
    ``_resolve_stem_version_in_pack`` 가 공유한다.
    """
    module_dir = PROMPTS_BASE / module
    if not module_dir.exists():
        return []
    # ``.`` 로 시작하는 디렉토리는 판이 아니다 — 도구가 남긴 것이다
    # (실측: ``multiroll_judge/.pytest_cache`` · ``shot_continuity/
    # .pytest_cache``). ``_version_sort_key`` 가 이런 이름을 ``(0, name)``
    # 으로 떨어뜨려 최신 선택은 무사했지만, ``get_effective_source`` 의
    # ``module_pack.versions`` 에는 판인 척 실려 admin UI 에 노출됐다.
    return sorted(
        [d.name for d in module_dir.iterdir()
         if d.is_dir() and not d.name.startswith(".")],
        key=_version_sort_key,
        reverse=True,
    )


def _is_version_pack_strict() -> bool:
    """version pack strict 모드 평가 (review I1).

    우선순위: ``PROMPT_VERSION_PACK_STRICT`` ENV (있으면 그대로 사용) →
    ``settings.prompt_version_pack_strict`` (Pydantic, .env 자동 로드).
    ENV 직접 평가 분기는 import-cycle/legacy 호환을 위해 보존한다 — 운영자
    instant-toggle (export 후 reload 없이) 도 가능.
    """
    raw = os.environ.get("PROMPT_VERSION_PACK_STRICT")
    if raw is not None:
        return raw.strip().lower() in ("1", "true", "yes", "on")
    try:
        from app.core.config import settings
        return bool(getattr(settings, "prompt_version_pack_strict", False))
    except Exception:
        return False


def _emit_pack_drift_warning(
    module: str, stem: str, version: str, latest_version: str,
) -> None:
    """동일 (module, stem, version) 조합에 대해 process 당 1회만 warning 발행."""
    key = (module, stem, version)
    with _DRIFT_WARNED_LOCK:
        if key in _DRIFT_WARNED:
            return
        _DRIFT_WARNED.add(key)
    logger.warning(
        "Prompt version pack drift: module=%s stem=%s loaded_from=%s "
        "(latest_in_module=%s). To enforce single-pack semantics, set "
        "PROMPT_VERSION_PACK_STRICT=true and align stem files within the latest "
        "version directory.",
        module, stem, version, latest_version,
    )


def _resolve_stem_in_pack(
    module: str, name: str, ext: str = ".md",
) -> Tuple[Optional[Path], Optional[str], List[str]]:
    """stem 파일을 numeric-desc version 순회로 찾고 (path, found_version, all_versions) 반환.

    - found_version 이 ``all_versions[0]`` (latest) 와 다르면 drift 후보.
    - strict 모드면 latest 에 stem 없을 때 caller 가 RuntimeError 발생시킨다.
    """
    module_dir = PROMPTS_BASE / module
    if not module_dir.exists():
        return None, None, []
    versions = _list_module_versions(module)
    for ver in versions:
        fpath = module_dir / ver / f"{name}{ext}"
        if fpath.exists():
            return fpath, ver, versions
    return None, None, versions


def _load_from_file(
    module: str,
    name: str,
    ext: str = ".md",
    version: Optional[str] = None,
) -> Optional[str]:
    """파일에서 stem 의 최신 매칭 version 을 읽어 반환.

    같은 module 안에서 stem 이 서로 다른 version 에서 로드될 수 있다
    (backward-compat). drift 시 logger.warning + ``module_pack.drift`` 가시화.
    ``PROMPT_VERSION_PACK_STRICT`` 가 truthy 면 latest pack 에 stem 이 없을 때
    RuntimeError 를 raise 하여 운영자가 prompt 정리 후 single-pack 으로 전환할
    수 있게 한다.

    W19A versioned-load (review §3.6): ``version`` 이 주어지면 그 pack
    디렉토리 안에서만 stem 을 찾고, 없으면 None 반환. latest 로 silent
    fallback 하지 않는다 (caller 의 명시적 pin 을 깨지 않음). 이 모드는
    strict drift 검사를 적용하지 않는다 — 명시적 version 선택은 운영자가
    이미 pack 을 알고 있다는 신호.
    """
    if version is not None:
        fpath = PROMPTS_BASE / module / version / f"{name}{ext}"
        if not fpath.exists():
            return None
        return fpath.read_text(encoding="utf-8").strip()

    fpath, found_ver, versions = _resolve_stem_in_pack(module, name, ext=ext)
    if fpath is None:
        if _is_version_pack_strict() and versions:
            raise RuntimeError(
                f"Prompt version pack strict: stem '{name}{ext}' not found in any "
                f"version of module '{module}'. Latest pack='{versions[0]}'. "
                "Add the stem to the latest version directory or unset "
                "PROMPT_VERSION_PACK_STRICT to use lenient mode."
            )
        return None

    latest_ver = versions[0] if versions else None
    if latest_ver and found_ver != latest_ver:
        if _is_version_pack_strict():
            raise RuntimeError(
                f"Prompt version pack strict: stem '{name}{ext}' missing in latest "
                f"pack '{latest_ver}' of module '{module}' (resolved from "
                f"'{found_ver}' under lenient mode). Add the stem to the latest "
                "version directory or unset PROMPT_VERSION_PACK_STRICT."
            )
        _emit_pack_drift_warning(module, f"{name}{ext}", found_ver, latest_ver)

    return fpath.read_text(encoding="utf-8").strip()


def get_effective_source(module: str, name: str, db=None) -> Dict[str, Any]:
    """``load_prompt`` 가 실제로 사용할 source 의 provenance 표시 (problems.md #8).

    admin UI 가 prompt_template DB row 만 보고 active prompt 판단하던 문제를
    해소하기 위한 effective view. **caller 별 db 전달 여부에 따라 winner 가
    다르므로 두 mode 를 동시에 표시** — db-aware caller (db 전달, 예: location
    _floor_plan_step) 와 file-only caller (db 미 전달, 예: scene_extractor_v2,
    background_master_plan 등 대부분의 pipeline module).

    Returns:
        {
          "module": str, "name": str,
          "candidates": {
            "db":   {"id", "version", "is_active", "has_schema"} | None,
            "file": {"version", "path"} | None,
          },
          "effective": {
            "with_db":    {"winner": "db"|"file"|None, "fallback_reason": ...},
            "without_db": {"winner": "file"|None, "fallback_reason": ...},
          },
          "module_pack": {
            "versions": [str],          # numeric desc 정렬, 모든 version 디렉토리
            "latest": str | None,
            "strict_enabled": bool,     # PROMPT_VERSION_PACK_STRICT 평가 결과
            "stem_drift": bool,         # file_candidate.version != latest 면 True
          },
        }

    winner 결정:
      - with_db: DB 후보 존재 → "db", 없으면 file → "file", 둘 다 없으면 None
      - without_db: DB 무시, file 만 — file → "file", 없으면 None

    ``module_pack`` (problems.md #6) 는 admin UI 가 stem-version drift 와 strict
    toggle 상태를 한눈에 볼 수 있게 한다. 이 view 는 ``.md`` 파일 후보만
    고려하므로, schema (``.json``) drift 는 별도 ``get_effective_schema_source``
    로 확인한다.
    """
    db_candidate: Optional[Dict[str, Any]] = None
    file_candidate: Optional[Dict[str, Any]] = None

    # DB 후보 (db 전달 시만)
    if db is not None:
        try:
            row = _select_latest_active_row(db, module, name)
            if row:
                db_candidate = {
                    "id": getattr(row, "id", None),
                    "version": row.version,
                    "is_active": True,
                    "has_schema": bool(getattr(row, "schema_json", None)),
                }
        except Exception as exc:
            logger.warning(
                "DB prompt effective lookup failed (%s/%s): %s", module, name, exc,
            )

    # File 후보 (.md)
    pack_versions: List[str] = []
    module_dir = PROMPTS_BASE / module
    if module_dir.exists():
        try:
            pack_versions = _list_module_versions(module)
            for ver in pack_versions:
                fpath = module_dir / ver / f"{name}.md"
                if fpath.exists():
                    file_candidate = {
                        "version": ver,
                        "path": str(fpath),
                    }
                    break
        except Exception as exc:
            logger.warning(
                "File prompt effective lookup failed (%s/%s): %s", module, name, exc,
            )

    # with_db: DB 우선, fallback file
    if db_candidate is not None:
        with_db = {"winner": "db", "fallback_reason": None}
    elif file_candidate is not None:
        with_db = {
            "winner": "file",
            "fallback_reason": "no_db_session" if db is None else "no_active_db_row",
        }
    else:
        with_db = {"winner": None, "fallback_reason": "no_file"}

    # without_db: file 만
    if file_candidate is not None:
        without_db = {"winner": "file", "fallback_reason": None}
    else:
        without_db = {"winner": None, "fallback_reason": "no_file"}

    pack_latest = pack_versions[0] if pack_versions else None
    stem_drift = bool(
        file_candidate is not None
        and pack_latest is not None
        and file_candidate.get("version") != pack_latest
    )

    return {
        "module": module,
        "name": name,
        "candidates": {
            "db": db_candidate,
            "file": file_candidate,
        },
        "effective": {
            "with_db": with_db,
            "without_db": without_db,
        },
        "module_pack": {
            "versions": pack_versions,
            "latest": pack_latest,
            "strict_enabled": _is_version_pack_strict(),
            "stem_drift": stem_drift,
        },
    }


def get_active_version(module: str, name: str, db=None) -> Optional[str]:
    """현재 활성 버전 반환."""
    if db:
        try:
            row = _select_latest_active_row(db, module, name)
            if row:
                return row.version
        except Exception as exc:
            logger.warning("prompt_template DB lookup failed for %s/%s: %s — 파일 fallback", module, name, exc)

    # 파일 fallback
    versions = _list_module_versions(module)
    return versions[0] if versions else None