"""GROUNDING-V2 — Sol 분류기. **검색 호출이 없다** (계획 §2-2).

설계: 계약 §2·§3, 계획 §6·§8.

이 모듈은 **순수 함수 + LLM 호출 하나**만 담는다. DB 쓰기·스텝 오케스트레이션 없음
(``search_grounded_ref.py`` 와 같은 태도).

★프롬프트는 **v2 전용 module namespace** 를 **명시 version 으로 pin** 해서 읽는다.
``PROMPT_VERSION_PACK_STRICT`` 는 process-global 이라 namespace 별로 못 켠다
(``prompt_loader._is_version_pack_strict`` 는 인자가 없다) — 그래서 strict 대신
``resolve_effective`` 로 **실효 source 와 raw hash 를 직접 확인**한다.

★hash 는 둘이다. 섞으면 안 된다:
  ``raw_content_hash``  로드한 원본 source bytes
  ``payload_hash``      **실제로 provider 에 나간** system/user/schema payload
format 인자가 바뀌어도 지문이 안 움직이는 구멍을 막는다.
"""
from __future__ import annotations

import hashlib
import json
import logging
from typing import Any, Dict, List, Optional, Sequence

from app.modules.prompt_loader import resolve_effective

logger = logging.getLogger(__name__)

CLASSIFIER_CONTRACT_VERSION = 4

_MODULE = "grounding_classify"
#: ★명시 pin. latest 자동 선택에 기대지 않는다 — 기존 module 에 버전을 더하면
#: legacy 도 새 prompt 를 읽는 문제(계획 §4-6)와 같은 부류를 v2 에서 원천 차단한다.
# ★14 = 폐기 축 제거. `discriminability`(A)와 `difficulty`(세로축)를 schema·
#  지시문에서 **뺐다** — 검색을 안 하는 단계가 세상 사실을 답하던 자리이고,
#  그 답이 조사 여부를 가로막았다(계약 §2, 2026-08-30).
PROMPT_PACK_VERSION = "15.202608302200"
#: 이 팩의 stem 은 **둘뿐이고 역할이 다르다.** 목록이 아니라 이름 둘로 둔다 —
#: 목록을 만들면 「무엇이든 더 넣을 수 있는 칸」이 되고, 그 순간 닫힌 세계가 아니게 된다.
SYSTEM_STEM = "system"
SCHEMA_STEM = "classify_schema"

#: 초기 판정자 = Sol (사용자 지시). step 이름으로 모델이 정해진다.
STEP_NAME = "grounding_classify"


#: ★판정을 **한 모델에 안 맡긴다.** held-out 실측에서 한 모델 3표본이 6축 중
#: 3축을 틀렸고, 틀린 셋 중 둘이 `difficulty` 를 낮게 본 것이었다 — 그 축은
#: 「**다른 모델**이 만들 수 있나」를 묻는데 근거 없이 짐작으로 답한다.
#: 같은 모델을 여러 번 물으면 **같은 짐작이 반복될 뿐**이다(Codex).
#:
#: ★**이 단계는 VLM 이 아니다.** 그림을 한 장도 안 보내고 원문·엔티티 설명만
#:  읽는다. 그래서 관찰 자리의 「VLM 은 gemini + grok 만」 규칙과 무관하고,
#:  여기 판정자는 **텍스트 판정자**다 — 사용자 지정 `gpt` + `grok`.
#:
#: ★2026-08-30 저녁: `generation_difficulty` 를 **되살렸다**(팩 15). 폐기한
#:  `difficulty` 와 다르다 — 5단계 ordinal 이 아니라 `hard/not_hard/uncertain`
#:  **3값**이고, **route 를 안 가른다**. 여는 것은 둘뿐이다:
#:  ①저빈도 요소 필터에서 살린다 ②참고 사진을 찾아 온다.
#:  ★`discriminability`(대다수가 알아보는가)는 **계속 폐기**다 — 사용자 규칙에
#:   없는 제3의 문이다.
#:
#: ★2026-08-30: 이 단계는 이제 **이미지 모델 좌표를 아예 안 받는다.**
#:  `target_image_*` 는 폐기한 `difficulty`(「이 이미지 모델이 근거 없이 맞힐
#:  수 있나」)의 전제였다. 축이 없어졌는데 좌표만 남기면 ①user prompt 가
#:  스키마에 없는 칸을 답하라고 시키고 ②이미지 모델을 바꾼 것만으로 같은
#:  분류를 다시 사게 된다.
DEFAULT_JUDGES = ("gpt", "grok")


def classify_samples(
    subjects: Sequence[Dict[str, Any]],
    *,
    samples: int = 3,
    judges: Optional[Sequence[str]] = None,
    on_sample: Optional[Any] = None,
    reuse: Optional[Any] = None,
    **kwargs: Any,
) -> Dict[str, Any]:
    """같은 입력을 ``samples`` 회 판정해 **subject 별 표본 목록**을 돌려준다.

    ★한 표본으로 route 를 정할 수 없다 — 판정기를 결정적으로 만들 수 없다
    (``_NO_TEMPERATURE_ALIASES``). 접는 것은 planner 의
    ``decide_route_from_samples`` 가 한다.

    ★production 호출부는 ``GroundingPlanStep`` 이다(§2-3.5 에서 배선했다).
    ``payload_hash``·``prompt_version``·judge alias·physical model 이 비었거나
    표본마다 다르면 그 스텝의 ``check_provenance`` 가 **fail-closed** 한다.
    ★``v2`` 에서만 불린다 — `legacy`·`shadow_plan` 은 스텝이 안 돈다.

    Returns:
        {"by_subject": {subject_id: [record, ...]}, "runs": [fingerprint, ...],
         "logical_calls": int, "search_calls": 0}
    """
    if samples < 1:
        raise ValueError(f"samples 는 1 이상이어야 한다 (받은 값 {samples})")
    panel = list(judges) if judges is not None else list(DEFAULT_JUDGES)
    if not panel:
        raise ValueError("판정자가 없다 — 한 모델이라도 있어야 한다")
    by_subject: Dict[str, List[Dict[str, Any]]] = {
        s["research_subject_id"]: [] for s in subjects}
    runs: List[Any] = []
    reused = 0
    base_cfg = dict(kwargs.pop("project_config", None) or {})
    for alias in panel:
        # ★판정자를 **명시로** 정한다. step 기본값에 기대면 어느 모델이 답했는지
        #  기록만 보고는 못 되짚는다.
        cfg = {**base_cfg, STEP_NAME: {**(base_cfg.get(STEP_NAME) or {}),
                                       "model": alias}}
        for n in range(samples):
            # ★★**이미 산 것은 다시 안 산다.** 재개를 호출부가 따로 구현하면
            #  판정자·표본 순서를 두 곳에 적게 되고, 한쪽만 고쳐진다.
            #  `search_claims` 가 batch 를 재사용하는 것과 같은 자리다.
            got = reuse(alias, n) if reuse else None
            out = got if got is not None else classify(
                subjects, project_config=cfg, **kwargs)
            if got is not None:
                reused += 1
            runs.append(out.get("fingerprint"))
            for rec in out.get("records") or []:
                sid = rec.get("research_subject_id")
                if sid in by_subject:
                    by_subject[sid].append({**rec, "judge_alias": alias})
            # ★★**산 것을 판정보다 먼저 넘긴다** — `search_claims(on_batch=)` 와
            #  같은 자리다. 한 판(판정자 × 표본)이 다 끝난 뒤에만 넘기면,
            #  후반이 끊길 때 **앞서 산 호출까지 통째로 다시 산다**.
            # ★재사용한 것은 **안 넘긴다** — 이미 파일에 있다. 넘기면 재개
            #  때마다 같은 것을 다시 쓰고, 「몇 개 샀나」가 거짓이 된다.
            if on_sample and got is None:
                on_sample({"judge_alias": alias, "sample_index": n, **out})
    return {
        "contract_version": CLASSIFIER_CONTRACT_VERSION,
        "by_subject": by_subject,
        "runs": runs,
        "judges": panel,
        # ★**산 것**과 **돈 것**을 따로 센다. 하나로 뭉치면 재개한 판이
        #  「N회 샀다」로 보고된다.
        "reused_calls": reused,
        "bought_calls": (samples * len(panel) - reused) if subjects else 0,
        "logical_calls": (samples * len(panel)) if subjects else 0,
        "search_calls": 0,
    }


def _call_structured(**kwargs) -> Dict[str, Any]:
    """``llm_client`` 로 가는 **유일한 문**.

    import 를 여기까지 미루는 이유 — litellm 은 **import 시점에** 가격표를 바깥에서
    받아온다. 검색도 모델 호출도 아니지만, 이 단계의 통과 조건이 「바깥 호출 0」이라
    실제로 호출하지 않는 경로에서는 그것조차 안 나가야 한다. 문이 하나뿐이라
    시험도 이 함수만 갈아 끼우면 되고 **`llm_client` 를 import 할 일이 없다**.
    """
    from app.modules.llm.llm_client import call_structured

    return call_structured(**kwargs)


def _sha16(text: str) -> str:
    return hashlib.sha256(text.encode("utf-8")).hexdigest()[:16]


def load_pack(*, db=None, version: Optional[str] = None) -> Dict[str, Any]:
    """팩을 통째로 읽고 **completeness 를 확인**한다.

    stem 하나라도 빠지면 ``FileNotFoundError`` — 반쪽 팩으로 조용히 돌지 않는다.
    """
    ver = version or PROMPT_PACK_VERSION
    # 둘 다 없으면 resolve_effective 가 FileNotFoundError 로 선다 — 반쪽 팩으로
    # 조용히 돌지 않는다는 것이 여기서 말하는 completeness 다.
    resolved: Dict[str, Dict[str, Any]] = {
        SYSTEM_STEM: resolve_effective(
            _MODULE, SYSTEM_STEM, kind="prompt", version=ver, db=db),
        SCHEMA_STEM: resolve_effective(
            _MODULE, SCHEMA_STEM, kind="schema", version=ver, db=db),
    }
    return {
        "module": _MODULE,
        "version": ver,
        "stems": resolved,
        # 팩 지문 — stem 별 (source, version, raw hash) 를 순서 고정해 접는다.
        "pack_manifest_hash": _sha16("|".join(
            f"{stem}:{r['source']}:{r['version']}:{r['raw_content_hash']}"
            for stem, r in sorted(resolved.items())
        )),
    }


def build_user_prompt(
    subjects: Sequence[Dict[str, Any]],
    *,
    era: str = "",
    region: str = "",
) -> str:
    """후보 목록 → user prompt. ★원문 근거를 자르지 않는다.

    ★2026-08-30: **이미지 모델 좌표를 안 싣는다.** 그 블록과 「`difficulty` 는
    이 모델이 맞힐 수 있는지로 답하세요」는 폐기한 축의 지시문이었다. 스키마에
    없는 칸을 답하라고 시키면 지시문이 다른 칸의 답까지 흔든다.
    """
    lines: List[str] = [
        "## 목표 맥락",
        f"- 시대: {era or '★알 수 없음'}",
        f"- 지역: {region or '★알 수 없음'}",
        "",
        "★판정은 **이 시대·지역의 실물**을 기준으로 합니다. 후보 설명은 아직 조사 전에",
        "쓰인 것이라 시대가 안 적혀 있을 수 있습니다 — **설명에 시대가 없다고 해서**",
        "`generic` 으로 답하지 마세요. 위 맥락을 적용해 판정하세요.",
        "",
        "아래 후보들을 판정하세요. `research_subject_id` 를 그대로 되돌려 주세요.",
        "",
    ]
    for s in subjects:
        lines.append(f"- research_subject_id: {s['research_subject_id']}")
        lines.append(f"  대상: {s.get('surface_form', '')}")
        lines.append(f"  종류: {s.get('owner_type', '')}")
        if s.get("source_anchor"):
            lines.append(f"  원문 위치: {s['source_anchor']}")
        if s.get("source_quote"):
            lines.append(f"  원문: {s['source_quote']}")
        if s.get("world_context"):
            lines.append(f"  작품 세계 정보: {s['world_context']}")
        lines.append("")
    return "\n".join(lines).rstrip()


def payload_identity(system: str, user: str, schema: Any) -> str:
    """★이 호출의 신원. **한 곳에서만** 만든다 — 두 곳에 적으면 한쪽만 고쳐진다.

    `grounding_claims_search.payload_identity` 와 같은 자리다.
    """
    return _sha16("\x1e".join((
        system, user, json.dumps(schema, sort_keys=True, ensure_ascii=False))))


def planned_payload(subjects: Sequence[Dict[str, Any]], *, era: str = "",
                    region: str = "", db=None,
                    pack_version: Optional[str] = None) -> str:
    """**보내기 전에** 이 호출의 신원을 낸다. ★무료.

    ★`classify` 와 **같은 함수들**로 만든다 — `load_pack` · `build_user_prompt` ·
    `payload_identity`. 손으로 다시 만들면 두 벌이 되고, 그러면 재생 gate 가
    잠그는 것과 실제로 나가는 것이 갈린다 (Codex).
    """
    pack = load_pack(db=db, version=pack_version)
    return payload_identity(
        pack["stems"][SYSTEM_STEM]["content"],
        build_user_prompt(subjects, era=era, region=region),
        pack["stems"][SCHEMA_STEM]["content"])


def classify(
    subjects: Sequence[Dict[str, Any]],
    *,
    era: str = "",
    region: str = "",
    project_config: Optional[Dict] = None,
    opik_metadata: Optional[Dict] = None,
    db=None,
    pack_version: Optional[str] = None,
    strict_single_attempt: bool = False,
) -> Dict[str, Any]:
    """후보들을 분류한다. ★**검색을 하지 않는다.**

    ★``strict_single_attempt`` 는 **측정용**이다. fallback 과 Router 재시도를 봉인해
    **물리 시도 1회**만 허용한다 — 안 봉인하면 「논리 호출 N회」로 보고한 것이
    실제로는 그보다 많은 전송이고, 반복 측정의 각 판이 서로 다른 모델·다른 시도
    수를 탄 것이 된다. 실패는 예외로 올려 **미확정**으로 만든다.

    ★``era``/``region`` 은 **반드시 넘긴다.** 안 넘기면 분류기가 *조사 전에 상상으로
    쓰인 묘사*만 보고 판정한다 — 실측에서 같은 실물이 「시대가 적힌 에피소드」와
    「안 적힌 에피소드」에서 research/skip 으로 갈렸다.

    ★``target_image_*`` 는 **더 이상 받지 않는다** (2026-08-30). 그 좌표를
    쓰던 축(`difficulty`)이 없어졌다.
    """
    if not subjects:
        return {
            "contract_version": CLASSIFIER_CONTRACT_VERSION,
            "records": [], "fingerprint": None, "search_calls": 0,
        }

    # ★입력 id 중복을 먼저 막는다 — 뒤에서 bijection 을 요구하려면 입력이 먼저 유일해야 한다.
    ids = [s["research_subject_id"] for s in subjects]
    dup_in = {i for i in ids if ids.count(i) > 1}
    if dup_in:
        raise ValueError(
            f"classify: 입력에 같은 research_subject_id 가 여러 번 있다: {sorted(dup_in)}")

    pack = load_pack(db=db, version=pack_version)
    system_prompt = pack["stems"][SYSTEM_STEM]["content"]
    schema = pack["stems"][SCHEMA_STEM]["content"]
    user_prompt = build_user_prompt(
        subjects,
        era=era, region=region,
    )

    # ★실제로 나간 payload 의 지문 — raw source hash 와 별개로 남긴다.
    #  ★신원은 **한 함수**가 만든다(`payload_identity`). 여기서 따로 조립하면
    #   재생 gate 가 잠그는 것과 실제로 나가는 것이 갈린다.
    payload_hash = payload_identity(system_prompt, user_prompt, schema)

    usage: Dict[str, Any] = {}
    result = _call_structured(
        step=STEP_NAME,
        system_prompt=system_prompt,
        user_prompt=user_prompt,
        response_schema=schema,
        project_config=project_config,
        schema_name="grounding_classifications",
        opik_metadata=opik_metadata,
        usage_sink=usage,
        **({"enable_fallback": False, "num_retries": 0}
           if strict_single_attempt else {}),
    )

    # ★실제로 판정한 모델을 기록한다. call_structured 는 Tier 3 에서 다른 provider 로
    #  넘어갈 수 있으므로 **설정이 아니라 응답이 말하는 것**을 남긴다.
    judge = {
        "judge_step": STEP_NAME,
        # ★**부탁한** 판정자를 따로 남긴다. 아래 `judge_model_alias` 는 **응답이
        #  말한 것**이라, tier fallback 으로 gpt 가 gemini 로 넘어가면 둘 다
        #  gemini 로 기록되고 「패널이 한 모델로 무너진 것」이 안 보인다.
        "requested_judge_alias": str(
            ((project_config or {}).get(STEP_NAME) or {}).get("model") or ""),
        "judge_model_alias": usage.get("alias"),
        "judge_physical_model": usage.get("physical_model"),
    }

    by_id = {s["research_subject_id"]: s for s in subjects}
    # ★id 별 정확히 한 건만 받는다. 두 건이 살아남으면 다음 유료 검색 단계가
    #  같은 subject 를 중복 구매한다.
    returned: Dict[str, List[Dict[str, Any]]] = {}
    unknown_ids: List[Any] = []
    for item in (result or {}).get("classifications", []) or []:
        sid = item.get("research_subject_id")
        if sid not in by_id:
            unknown_ids.append(sid)
            continue
        returned.setdefault(sid, []).append(item)
    if unknown_ids:
        logger.warning(
            "grounding_classify: 모르는 subject id %r 가 돌아왔다", unknown_ids)

    records: List[Dict[str, Any]] = []
    for sid in by_id:
        got = returned.get(sid, [])
        if len(got) == 1:
            records.append({**got[0], **judge})
            continue
        # 0건(미제출) 이든 2건 이상(중복) 이든 **subject 당 한 줄의 미확정**으로 닫는다.
        records.append({
            "research_subject_id": sid,
            "classifier_missing": len(got) == 0,
            "classifier_duplicate_count": len(got) if len(got) > 1 else None,
            **judge,
        })

    return {
        "contract_version": CLASSIFIER_CONTRACT_VERSION,
        "records": records,
        "fingerprint": {
            "prompt_module": _MODULE,
            "prompt_version": pack["version"],
            "pack_manifest_hash": pack["pack_manifest_hash"],
            "stem_sources": {
                stem: {
                    "source": r["source"], "locator": r["locator"],
                    "version": r["version"], "raw_content_hash": r["raw_content_hash"],
                } for stem, r in pack["stems"].items()
            },
            "payload_hash": payload_hash,
            "judge_step": judge["judge_step"],
            "requested_judge_alias": judge["requested_judge_alias"],
            "judge_model_alias": judge["judge_model_alias"],
            "judge_physical_model": judge["judge_physical_model"],
            "era": era,
            "region": region,
        },
        "search_calls": 0,  # ★계약이다. 이 단계는 무료다.
    }