"""대상 **텍스트 조사** — 사진을 찾기 전에 「그게 무엇인가」와 「이미지에서 무엇을 확인할까」를 만든다.

★사용자 지정 순서 (2026-09-03):
    ① 텍스트(웹) 검색으로 두 가지를 파악한다 — 그 대상이 **무엇인지**(「됫박」처럼 쓰는 사람도
       모를 수 있다)와, 그것을 토대로 **무엇을 검색하면 될지** · **이미지에서 무엇을 확인할지**
       (VLM 에게 어찌 생겼는지 판단시킬 문구)
    ② 처음엔 좁고 조밀하게 이미지 검색           → `narrow_queries`
    ③ VLM 이 ①의 문구로 판단                      → `appearance_criteria` (심판 CRITERIA)
    ④ 없으면 뼈대만 남겨 넓게 재검색               → `rough_queries`
    ⑤ 마지막엔 반드시 한 장                        → `coarse_type_pick` 의 closest 선택

이 모듈은 `acquire_one(write_brief=…)` 에 들어가는 **저작기**를 낸다 — 옛 `grounding_search_brief`
저작기와 같은 자리, 같은 반환 모양(`search_directive_native` · `search_terms_native` ·
`language_lock_native`)에 조사 결과(`research`)를 얹는다. 대상당 조사는 **한 번**이고 2라운드는
그 결과의 rough 질의를 쓴다(재개 때는 앞 라운드에 적힌 조사를 되쓴다 — 다시 사지 않는다).

★작품 명사·나라/장소 목록·regex 없음. 팩은 일반 지시문이고 값은 전부 런타임(뼈대·좌표·근거)에서 온다.
★HITL 0 — 여기 결과는 사람 없이 그대로 다음 단계로 간다.
"""
from __future__ import annotations

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

logger = logging.getLogger(__name__)

PROMPT_MODULE = "grounding_target_research"
RESEARCH_PACK_VERSION = "1.202609030300"
SYSTEM_STEM = "system"
SCHEMA_STEM = "schema"
RESEARCH_CONTRACT_VERSION = "1.202609030300"
#: 웹 검색이 되는 텍스트 모델 — `search_grounded_ref` 의 검색 orchestrator 와 같은 물리 모델
RESEARCH_MODEL_ALIAS = "gpt"
PER_CALL_DEADLINE_SECONDS = 300.0
STEP_TAG = "grounding_target_research"


class ResearchInputsMissing(ValueError):
    """조사할 뼈대가 없다 — 부류 이름이 비었다. ★빈 질의로 조사하면 아무거나 온다."""


class ResearchFailed(RuntimeError):
    """조사 호출이 답을 못 냈다(전송·파싱). ★조용히 빈 결과로 넘어가지 않는다."""


def load_pack(*, db=None, version: Optional[str] = None) -> Dict[str, Any]:
    from app.modules.prompt_loader import resolve_effective
    ver = version or RESEARCH_PACK_VERSION
    stems = {
        SYSTEM_STEM: resolve_effective(PROMPT_MODULE, SYSTEM_STEM, kind="prompt", version=ver),
        SCHEMA_STEM: resolve_effective(PROMPT_MODULE, SCHEMA_STEM, kind="schema", version=ver),
    }
    missing = [k for k, v in stems.items() if not v or not v.get("content")]
    if missing:
        raise ValueError(f"grounding_target_research 팩 {ver} 에 stem 이 없다: {missing}")
    return {"version": ver, "stems": stems}


def pack_content_sha(pack: Optional[Dict[str, Any]] = None) -> str:
    pack = pack or load_pack()
    h = hashlib.sha256()
    for stem in (SYSTEM_STEM, SCHEMA_STEM):
        h.update(str(pack["stems"][stem]["content"]).encode("utf-8"))
        h.update(b"\x00")
    return h.hexdigest()[:16]


def build_user(skeleton: Dict[str, Any], evidence: Sequence[str]) -> str:
    """조사 입력. ★값은 전부 런타임 — 뼈대(owner/부류/검색어/언어 잠금/시대/지역)와 근거 문장."""
    kind = str(skeleton.get("coarse_type_label") or "").strip()
    if not kind:
        raise ResearchInputsMissing("조사할 부류 이름이 비었다")
    lock = str(skeleton.get("language_lock_native") or "").strip()
    lines = [
        f"KIND (registered as): {skeleton.get('owner_type') or ''}",
        f"COARSE LABEL: {kind}",
        "SEARCH TERMS (proposed by the manuscript reader): "
        + (", ".join(str(t) for t in (skeleton.get("terms_native") or ()) if str(t).strip()) or "(none)"),
        f"LANGUAGE LOCK: {lock or '(not declared — use the language of the evidence)'}",
        f"ERA: {skeleton.get('era') or '(not declared)'}",
        f"REGION: {skeleton.get('region') or '(not declared)'}",
        "SOURCE EVIDENCE (manuscript sentences, verbatim):",
    ]
    ev = [str(e) for e in evidence if str(e).strip()]
    lines += [f"- {e}" for e in ev] or ["- (none)"]
    return "\n".join(lines)


def _extract(text: str) -> Optional[Dict[str, Any]]:
    from app.modules.pipeline.grounding_claims_search import _extract_json_object
    return _extract_json_object(text)


def _sources_of(resp: Any) -> List[str]:
    from app.modules.pipeline.grounding_claims_search import _url_citations
    try:
        return [str(u) for u in (_url_citations(resp) or ())]
    except Exception:                                   # noqa: BLE001
        return []


def research_target(client: Any, *, skeleton: Dict[str, Any], evidence: Sequence[str],
                    model: Optional[str] = None, opik_metadata: Optional[Dict[str, Any]] = None,
                    pack: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
    """한 대상을 웹으로 조사한다. ★한 호출 · 구조화 답 · 출처. 실패는 `ResearchFailed`.

    반환: `what_it_is` · `appearance_criteria` · `narrow_queries` · `rough_queries` ·
    `search_directive_native` · `language_lock_native` · `sources` · `provenance`.
    """
    from app.core.research_call_budget import bind_current_research_budget
    from app.modules.llm.image_tracer import record_provider_call
    from app.modules.pipeline.grounding_claims_search import _armed_create
    from app.modules.pipeline.llm_deadline import call_with_deadline
    from app.modules.pipeline.search_grounded_ref import (SEARCH_ORCHESTRATOR,
                                                           build_web_search_tool)

    pack = pack or load_pack()
    system = str(pack["stems"][SYSTEM_STEM]["content"]).strip()
    schema = pack["stems"][SCHEMA_STEM]["content"]
    schema_body = json.loads(schema) if isinstance(schema, str) else schema
    user = build_user(skeleton, evidence)
    model = model or SEARCH_ORCHESTRATOR
    meta = {**(opik_metadata or {}), "pack": pack["version"], "contract": RESEARCH_CONTRACT_VERSION}
    t0 = time.monotonic()
    try:
        resp = call_with_deadline(
            bind_current_research_budget(_armed_create), client, dict(
                model=model, instructions=system,
                input=[{"role": "user", "content": [{"type": "input_text", "text": user}]}],
                tools=[build_web_search_tool(want_images=False)],
                include=["web_search_call.action.sources"], store=False,
                text={"format": {"type": "json_schema", "name": "grounding_target_research",
                                 "strict": True, "schema": schema_body}}),
            deadline_seconds=PER_CALL_DEADLINE_SECONDS)
    except Exception as exc:                            # noqa: BLE001
        ms = int((time.monotonic() - t0) * 1000)
        record_provider_call(step=STEP_TAG, model=model, prompt=f"{system}\n---\n{user}",
                             status="error", duration_ms=ms, meta=meta,
                             operation="web_search", provider="openai", error=str(exc)[:500])
        raise ResearchFailed(f"조사 호출 실패: {exc}") from exc
    ms = int((time.monotonic() - t0) * 1000)
    raw = str(getattr(resp, "output_text", "") or "")
    parsed = _extract(raw)
    record_provider_call(step=STEP_TAG, model=model, prompt=f"{system}\n---\n{user}",
                         status="ok" if parsed else "error", duration_ms=ms, meta=meta,
                         operation="web_search", provider="openai",
                         error="" if parsed else "JSON 이 아니다")
    if not parsed:
        raise ResearchFailed("조사 답이 JSON 객체가 아니다")
    got = validate_research(parsed)          # ★빈 핵심 칸은 실패다 — 조용히 넘기지 않는다 (Codex BLOCK 2)
    return {
        **got,
        "language_lock_native": str(parsed.get("language_lock_native") or "").strip()
        or str(skeleton.get("language_lock_native") or ""),
        "sources": [str(s) for s in (parsed.get("sources") or ()) if str(s).strip()] or _sources_of(resp),
        "provenance": {"provider": "openai", "model": model, "pack": pack["version"],
                       "contract": RESEARCH_CONTRACT_VERSION, "duration_ms": ms},
    }


REQUIRED_TEXT = ("what_it_is", "appearance_criteria", "search_directive_native")
REQUIRED_LISTS = ("narrow_queries", "rough_queries")


def validate_research(parsed: Dict[str, Any]) -> Dict[str, Any]:
    """조사 답의 **구조** 검증. ★Codex BLOCK 2 (2026-09-03): 핵심 칸이 빈 문자열이어도 지나가면 VLM 은
    CRITERIA 없이 돌고 2라운드는 넓은 검색이 아니라 1라운드 반복이 된다. 빈 것은 실패(재시도 가능)다.
    rough 가 비었다고 narrow 로 **대체하지 않는다**. 뜻은 안 본다 — 비었는가만."""
    out: Dict[str, Any] = {}
    missing: List[str] = []
    for k in REQUIRED_TEXT:
        v = str((parsed or {}).get(k) or "").strip()
        if not v:
            missing.append(k)
        out[k] = v
    for k in REQUIRED_LISTS:
        vs = [str(q).strip() for q in ((parsed or {}).get(k) or ()) if str(q).strip()]
        if not vs:
            missing.append(k)
        out[k] = vs
    if missing:
        raise ResearchFailed(f"조사 답에 빈 칸이 있다 — {missing}. 다시 조사해야 한다(재시도 가능)")
    return out


def evidence_fingerprint(evidence: Sequence[str]) -> str:
    """검증된 원고 근거의 **안정된 지문** — 정렬한 근거 문장의 해시. ★Codex BLOCK 1 (2026-09-03):
    조사가 원고 근거를 읽는 이상 근거는 실질 입력이다. 뼈대가 같고 근거의 뜻이 달라졌는데 옛 조사를
    되쓰면 안 된다. 묘사 산문(visual_brief)이 아니라 **원문 인용**만 접으므로 merge 가 문장을 조금
    다르게 합쳐도 신원은 안 흔들린다."""
    h = hashlib.sha256()
    for q in sorted({str(e).strip() for e in (evidence or ()) if str(e).strip()}):
        h.update(q.encode("utf-8"))
        h.update(b"\x00")
    return h.hexdigest()[:16]


def dominant_script(text: str) -> str:
    """글의 **지배 문자 체계** — 낱글자의 Unicode 이름 첫 낱말(HANGUL · LATIN · CJK · HIRAGANA …)을 센다.
    ★언어·나라 목록이 아니다 — 문자 속성 조회다. 검색 연산자(`site:` …)·URL·숫자·기호는 세지 않는다."""
    import re
    import unicodedata
    t = re.sub(r"\b[a-z]+:\S+", " ", str(text or ""))
    t = re.sub(r"https?://\S+", " ", t)
    counts: Dict[str, int] = {}
    for ch in t:
        if not ch.isalpha():
            continue
        try:
            name = unicodedata.name(ch)
        except ValueError:
            continue
        head = name.split(" ")[0]
        counts[head] = counts.get(head, 0) + 1
    if not counts:
        return ""
    return max(sorted(counts), key=lambda k: counts[k])


def queries_in_lock(lock: str, queries: Sequence[str], *,
                    native_terms: Sequence[str] = ()) -> Dict[str, List[str]]:
    """★provider 앞 문 (Codex 2026-09-03 ④): 우리가 만든 질의의 지배 문자 체계가 **대상의 구조화 잠금**
    (`language_lock_native` + `terms_native`)의 지배 체계와 같아야 한다 — 언어 이름→문자 표를 두지 않는다.
    기대 체계를 못 읽으면(비었거나 기호·짧은 부호뿐) 문을 안 건다. 반환: {"kept": [...], "dropped": [...]}."""
    basis = " ".join([str(lock or ""), *(str(t) for t in (native_terms or ()) if str(t).strip())])
    want = dominant_script(basis)
    # ★잠금이 「ko」같은 짧은 부호이고 native term 도 없으면 문자 체계를 읽을 수 없다 — 문을 안 건다.
    letters = sum(1 for ch in basis if ch.isalpha())
    if not want or letters < 4:
        return {"kept": list(queries), "dropped": []}
    kept, dropped = [], []
    for q in queries:
        got = dominant_script(q)
        (kept if (not got or got == want) else dropped).append(q)
    return {"kept": kept, "dropped": dropped}


def _with_coordinates(directive: str, *, era: str, region: str) -> str:
    """지시문에 선언 좌표가 빠졌으면 **기계적으로** 앞에 붙인다 — 뜻을 판단하지 않는다.
    (검색은 좌표 없는 질의로 옛것·옆 나라 것을 가져온다 — 실측 2026-09-02.)"""
    head = " ".join(x for x in (str(region or "").strip(), str(era or "").strip()) if x)
    body = str(directive or "").strip()
    if not head:
        return body
    missing = [x for x in (region, era) if x and str(x).strip() and str(x).strip() not in body]
    return f"{head} — {body}" if missing else body


def brief_from_research(research: Dict[str, Any], *, narrow: bool, era: str, region: str,
                        lock: str, native_terms: Sequence[str] = ()) -> Dict[str, Any]:
    """조사 결과 → `acquire_one` 이 받는 저작 모양. ★1라운드=좁은 질의 · 2라운드(narrow)=뼈대 질의."""
    # ★질의마다 선언 좌표를 **글자 그대로** 싣는다 — 빠졌으면 기계적으로 앞에 붙인다(뜻 판단 없음).
    #  좌표 없는 질의는 앞선 시기·옆 나라 것을 가져온다(실측 2026-09-02) · provider 앞 문이 이것을 센다.
    raw_terms = list(research["rough_queries"] if narrow else research["narrow_queries"])
    coordinated = [_with_coordinates(q, era=era, region=region) for q in raw_terms]
    # ★언어 잠금 문 — **좌표를 지난 최종 outbound 질의**에 건다(Codex 2026-09-03). 기대 문자 체계는 대상의
    #  구조화 잠금(`language_lock_native` · `terms_native`)에서 파생 — 선언값이 부호("ko")여도 native term 이 정한다.
    #  어긴 질의는 빼고, 남는 것이 없으면 실패(재시도 가능). 실측 2026-09-03: P06 r1 에 영어 낱말이 섞였다.
    gate = queries_in_lock(str(lock or research.get("language_lock_native") or ""), coordinated,
                           native_terms=native_terms)
    if gate["dropped"] and not gate["kept"]:
        raise ResearchFailed(f"조사가 낸 질의가 전부 언어 잠금({lock!r})을 어겼다 — {gate['dropped'][:3]}")
    terms = list(gate["kept"])
    directive = _with_coordinates(research.get("search_directive_native") or " ".join(terms[:1]),
                                  era=era, region=region)
    return {
        "search_directive_native": directive,
        "search_terms_native": terms,
        "language_lock_native": str(research.get("language_lock_native") or lock or ""),
        "vlm_criteria": str(research.get("appearance_criteria") or ""),
        "language_lock_dropped": list(gate["dropped"]),
        "research": {k: research.get(k) for k in ("what_it_is", "appearance_criteria", "narrow_queries",
                                                   "rough_queries", "sources", "provenance",
                                                   "evidence_fingerprint")},
    }


def identity_inputs(*, era: str, region: str) -> Dict[str, Any]:
    """저작 신원에 접는 값 — 조사 팩·심판 팩·좌표. ★필드 목록은 여기 한 곳."""
    from app.modules.pipeline import coarse_type_pick as ctp
    return {"research_pack": RESEARCH_PACK_VERSION, "research_contract": RESEARCH_CONTRACT_VERSION,
            "research_pack_sha": pack_content_sha(), "pick_pack": ctp.PROMPT_PACK_VERSION,
            "era": str(era or ""), "region": str(region or "")}


def target_identity(target: Dict[str, Any], *, era: str, region: str,
                    evidence: Optional[Sequence[str]] = None) -> str:
    """대상 신원 = 뼈대 + 조사/심판 팩 + **원고 근거 지문**.
    ★묘사 산문(visual_brief)은 신원이 아니다 — 뼈대만 (Codex 2026-09-02). ★그러나 조사가 읽는
    원고 인용(source_quotes)은 실질 입력이라 **안정된 지문**으로 접는다 (Codex BLOCK 1 2026-09-03):
    인용이 바뀌면 신원이 바뀌고 웹 조사가 다시 돈다."""
    from app.modules.pipeline import grounding_search_brief as gsb
    sk = gsb.skeleton_search_input(target, era=era, region=region)
    ev = list(evidence) if evidence is not None else [
        str(x) for x in (target.get("source_quotes") or ()) if str(x).strip()]
    raw = json.dumps({"skeleton": {k: sk[k] for k in ("owner_type", "coarse_type_label", "terms_native",
                                                       "language_lock_native", "era", "region")},
                      "evidence_fingerprint": evidence_fingerprint(ev),
                      **identity_inputs(era=era, region=region)},
                     sort_keys=True, ensure_ascii=False)
    return hashlib.sha256(raw.encode("utf-8")).hexdigest()[:24]


def make_writer(*, world_facts: str, source_text: str, era: str = "", region: str = "",
                evidence_of: Optional[Callable[[Dict[str, Any]], Sequence[str]]] = None,
                client: Any = None,
                research_call: Optional[Callable[..., Dict[str, Any]]] = None,
                opik_metadata: Optional[Dict[str, Any]] = None
                ) -> Callable[..., Dict[str, Any]]:
    """`acquire_one(write_brief=…)` 에 넣을 **조사 저작기**. ★옛 `grounding_search_brief.make_writer`
    와 같은 자리 · 같은 속성 계약(`identity_inputs`·`target_identity`·`coordinates`·`era_tokens`·legacy).

    - 대상당 조사 **한 번**(같은 프로세스에서는 메모 · 재개 때는 `target["_prior_research"]` 를 되쓴다)
    - `narrow=False` → 좁은 질의 · `narrow=True` → 뼈대(rough) 질의. 둘 다 같은 CRITERIA 를 얹는다.
    Args:
        evidence_of: 대상 → 원고 근거 문장들(전문). 없으면 `target["source_quotes"]` 를 본다.
        research_call: 시험이 갈아 끼우는 자리 — `research_target` 와 같은 서명.
    """
    from app.modules.pipeline import grounding_search_brief as gsb
    from app.modules.pipeline.reference_acquisition_rounds import era_tokens_of

    memo: Dict[str, Dict[str, Any]] = {}
    fn = research_call or research_target

    def _client():
        nonlocal client
        if client is None:
            from app.core.openai_keys import openai_client as _mk
            client = _mk()
        return client

    def _evidence(target: Dict[str, Any]) -> List[str]:
        if evidence_of is not None:
            return [str(x) for x in (evidence_of(target) or ())]
        return [str(x) for x in (target.get("source_quotes") or ()) if str(x).strip()]

    def _write(target: Dict[str, Any], *, narrow: bool = False) -> Dict[str, Any]:
        sk = gsb.skeleton_search_input(target, era=era, region=region)
        sid = str(target.get("subject_id") or "")
        ev = _evidence(target)
        fp = evidence_fingerprint(ev)
        got = memo.get((sid, fp))                    # ★열쇠는 (대상, 근거 지문) — 같은 sid 라도 근거가 바뀌면 다시
        prior = target.get("_prior_research") if isinstance(target.get("_prior_research"), dict) else None
        if got is None and prior and prior.get("narrow_queries") \
                and str(prior.get("evidence_fingerprint") or "") == fp:
            got = dict(prior)                                 # ★재개 — 같은 근거의 조사만 되쓴다
        if got is None:
            got = dict(fn(_client(), skeleton=sk, evidence=ev,
                          opik_metadata={**(opik_metadata or {}), "subject_id": sid}))
            got["evidence_fingerprint"] = fp
        memo[(sid, fp)] = got
        return brief_from_research(got, narrow=narrow, era=era, region=region,
                                   lock=sk.get("language_lock_native") or "",
                                   native_terms=list(sk.get("terms_native") or ()))

    _write.identity_inputs = identity_inputs(era=era, region=region)
    _write.era_tokens = era_tokens_of(era)
    _write.coordinates = gsb.coordinates_of({"era": era, "region": region})
    _write.target_identity = lambda t: target_identity(t, era=era, region=region,
                                                      evidence=_evidence(t))
    # ★★legacy alias 를 **내지 않는다** (Codex 긴급 BLOCK 2026-09-03 02:20): 옛 뼈대 신원을 legacy 로
    #  내면 중앙 조립이 옛 `selected` 구매를 새 신원에 alias 해서 **새 웹 조사가 0회**가 된다 — 실측
    #  (attempt d38e6456ca2b · alias 24). 옛 줄에는 조사 팩·근거 지문이 없으므로 이번 cutover 는
    #  전부 다시 조사해야 한다. (옛 브리프 저작기의 alias 는 그 저작기에만 남는다.)
    _write.legacy_identity_inputs = None
    _write.legacy_target_identity = None
    _write.research_memo = memo
    return _write