"""era_research — 시대·브랜드 인지 대상 사전 이미지 조사 (2026-08-14 사용자 확정).

"그냥 이미지 생성으로는 그 시대나 장면을 제대로 묘사하기 어려운 경우
서울역과 같이 사전 이미지 조사가 필요해, 무조건" — 지폐·브랜드 가전·
신발·패션·차량·1980년대 열차 실내 같은 **시청자 인지 대상**은 생성 전에
실물 사진을 조사해 참조로 동봉한다. 서울역 재생성(regen_period_bg)에서
통한 구조(검색→후보→VLM 선택→참조)의 범용 정식 배선이다.

층위 셋(전부 `era_research_enabled` 기본 OFF=byte-identical):
  B1 엔티티 참조 생성(소품·장소 — 인물 제외) / B2 groupbg 플레이트 /
  B3 샷 레벨(엔티티로 커버되지 않는 시대 인지 대상).

계약 요점:
- 판별+원어 질의 저작 = **1콜**(대상 텍스트가 짧고 이미 원어라 seed 의
  장문 언어-격리 브리프 2콜 기계가 불요). 판별 기준은 팩 assess_sys 가
  소유 — 코드에는 대상 어휘가 없다(통칭 하드코딩 금지). "당연한 것은
  조사하지 않는다·대부분 빈 목록"이 계약 본문에 명문.
- 질의는 무조건 원어 + language_lock(원어 잠금문을 지시문 맨 앞에) —
  `search_grounded_ref.search_reference_images` 재사용(나간 질의 기록
  포함). 다운로드·안전 정책·VLM 선택도 같은 모듈 부품 재사용.
- 선택 심판 = GPT LVM 단독(모델 분업표 "이미지 비교 선택=GPT"). seed 의
  이중 평균은 후보가 구조물일 때의 계약 — 여기서는 단일 대상 사용성
  판정이라 1심으로 줄인다.
- 산출 meta(질의·나간 질의·선택 URL·sha256)는 호출측이 record/CP 에
  영속한다 — 기록 없는 유료 조사는 없다.
"""
from __future__ import annotations

import hashlib
import logging
from pathlib import Path
from typing import Any, Dict, List, Optional

from app.modules.prompt_loader import load_prompt, pack_dir_content_hash

logger = logging.getLogger(__name__)

_MODULE = "era_research"

PACK_VERSION_MAP = {
    "1": "1.202608141330",
}
ERA_RESEARCH_PACK_VERSION = "1"
# 정책 문자열 — ON 시 지문 기여. 판별·선택 계약이 바뀌면 버전을 올린다.
# v2 (2026-08-14 #119①): 적용 범위가 계약에 들어간다 — 샷별 bgfirst 판·
# confined 합성까지 조사가 붙는 커버리지 확장 + records 사이드카 캐시.
ERA_RESEARCH_POLICY_VERSION = "era_research_v2_plate_confined"
# 한 대상 텍스트에서 조사할 대상 상한 — "보통 하나"(2026-08-03 관례).
# 판별이 여럿을 물어도 프레임을 지배하는 것부터 이 수만 조사한다.
MAX_SUBJECTS_PER_CALL = 2
# 후보 다운로드 상한 — search_grounded_ref MAX_PICK_CANDIDATES 와 별도로
# 조사당 비용을 묶는다(후보가 많아도 선택 품질이 늘지 않는 실측 관례).
MAX_CANDIDATES = 4

ASSESS_MODEL = "gemini-flash"   # 판별·질의 저작 — 저비용 보조 분업
PICK_MODEL = "gpt"              # 후보 선택 — 이미지 비교=GPT LVM 분업


def resolve_era_pack(selector: str = ERA_RESEARCH_PACK_VERSION) -> str:
    try:
        return PACK_VERSION_MAP[selector]
    except KeyError:
        raise ValueError(
            f"unknown era_research pack selector {selector!r} "
            f"(known: {sorted(PACK_VERSION_MAP)})")


def era_pack_content_hash(selector: str = ERA_RESEARCH_PACK_VERSION) -> str:
    return pack_dir_content_hash(_MODULE, resolve_era_pack(selector))


def build_assess_schema() -> Dict[str, Any]:
    return {
        "type": "object",
        "properties": {
            "subjects": {
                "type": "array", "maxItems": 4,
                "items": {
                    "type": "object",
                    "properties": {
                        "subject_native": {"type": "string", "minLength": 2},
                        "search_terms_native": {
                            "type": "array", "minItems": 2, "maxItems": 4,
                            "items": {"type": "string", "minLength": 2},
                        },
                        "language_lock_native": {
                            "type": "string", "minLength": 10},
                        "reason_ko": {"type": "string", "minLength": 2},
                    },
                    "required": ["subject_native", "search_terms_native",
                                 "language_lock_native", "reason_ko"],
                    "additionalProperties": False,
                },
            },
        },
        "required": ["subjects"],
        "additionalProperties": False,
    }


def assess_subjects(
    *,
    step_tag: str,
    subject_text: str,
    world_facts_block: str,
    project_config: Optional[Dict[str, Any]] = None,
    opik_metadata: Optional[Dict[str, Any]] = None,
) -> List[Dict[str, Any]]:
    """조사 필요 대상 판별 + 원어 질의 저작 — 1콜.

    빈 목록 = 조사 불요(대부분의 입력). 상한은 호출측이 아니라 여기서
    자른다(MAX_SUBJECTS_PER_CALL) — 판별이 길게 물어도 지출이 안 는다.
    """
    from app.modules.llm.llm_client import call_structured

    assess_sys = load_prompt(
        _MODULE, "assess_sys", version=resolve_era_pack()).strip()
    parts = [{
        "type": "text",
        "text": ("SUBJECT TEXT:\n" + subject_text.strip()
                 + "\n\nWORLD FACTS (region and era — authoritative):\n"
                 + (world_facts_block or "").strip()),
    }]
    pc = {**(project_config or {}), step_tag: {"model": ASSESS_MODEL}}
    data = call_structured(
        step_tag, assess_sys, parts, build_assess_schema(),
        project_config=pc, schema_name=step_tag,
        opik_metadata=opik_metadata,
    )
    subjects = [s for s in (data.get("subjects") or [])
                if isinstance(s, dict)]
    return subjects[:MAX_SUBJECTS_PER_CALL]


def build_ref_role(subject_native: str) -> str:
    """참조 역할문 — 형태 권위만 갖고 구도·인물 권위는 없다(뼈대/표면 관례)."""
    return load_prompt(
        _MODULE, "ref_role", version=resolve_era_pack(),
        subject=subject_native).strip()


def research_reference(
    *,
    subject: Dict[str, Any],
    world_facts_block: str,
    out_path: Path,
    step_tag: str,
    project_config: Optional[Dict[str, Any]] = None,
    opik_metadata: Optional[Dict[str, Any]] = None,
    openai_client: Any = None,
) -> Optional[Dict[str, Any]]:
    """한 대상을 조사해 참조 1장을 확보한다 — 실패는 None(생성은 계속).

    검색→후보 다운로드(안전 정책)→GPT 선택→out_path 저장. meta 반환:
    {subject, terms, queries, picked_index, picked_url, sha256, file}.
    조사 실패가 생성을 막지 않는 것이 계약이다(참조 없이 그리던 기존
    동작으로 그대로 진행) — 다만 결과 meta 가 없으면 기록도 없으므로
    호출측은 None 도 감사 필드(researched=false)로 남겨야 한다.
    """
    from app.modules.llm.llm_client import call_structured
    from app.modules.pipeline.multiroll_gemini import png_part
    from app.modules.pipeline.search_grounded_ref import (
        build_pick_schema,
        build_pick_user_head,
        download_candidate,
        resolve_ref_pack_version,
        search_reference_images,
    )

    if openai_client is None:
        from app.core.openai_keys import openai_client as _mk

        openai_client = _mk()

    terms = [str(t) for t in (subject.get("search_terms_native") or [])]
    lock = str(subject.get("language_lock_native") or "")
    name = str(subject.get("subject_native") or "").strip()
    if not (name and terms):
        return None
    try:
        res = search_reference_images(
            openai_client,
            directive_native=name,
            terms_native=terms,
            language_lock_native=lock,
        )
    except Exception as exc:  # noqa: BLE001 — 조사 실패는 생성 비차단
        logger.warning("era_research[%s]: 검색 실패 — %r", step_tag, exc)
        return None
    images = list(res.get("images") or [])[:MAX_CANDIDATES]
    if not images:
        logger.info("era_research[%s]: 후보 0 — %s", step_tag, name)
        return None

    cand_dir = out_path.parent / f".{out_path.stem}_cand"
    cand_paths: List[Path] = []
    cand_urls: List[str] = []
    for i, im in enumerate(images, 1):
        dest = cand_dir / f"cand{i}.png"
        if download_candidate(
                str(im.get("image_url") or ""), dest,
                fallback_url=str(im.get("thumbnail_url") or "")):
            cand_paths.append(dest)
            cand_urls.append(str(im.get("image_url") or ""))
    if not cand_paths:
        logger.info("era_research[%s]: 다운로드 전멸 — %s", step_tag, name)
        return None

    pick_sys = load_prompt(
        "search_grounded_ref", "pick_system",
        version=resolve_ref_pack_version()).strip()
    parts: List[Dict[str, Any]] = [{
        "type": "text",
        "text": build_pick_user_head(
            structure_desc=name, world_facts_block=world_facts_block),
    }]
    for i, p in enumerate(cand_paths, 1):
        parts.append({"type": "text", "text": f"PHOTOGRAPH {i}:"})
        parts.append(png_part(p))
    pick_tag = f"{step_tag}_pick"
    pc = {**(project_config or {}), pick_tag: {"model": PICK_MODEL}}
    try:
        verdict = call_structured(
            pick_tag, pick_sys, parts,
            build_pick_schema(max_items=len(cand_paths)),
            project_config=pc, schema_name=pick_tag,
            opik_metadata=opik_metadata,
        )
    except Exception as exc:  # noqa: BLE001 — 선택 실패도 비차단
        logger.warning("era_research[%s]: 선택 실패 — %r", step_tag, exc)
        return None
    chosen = int(verdict.get("chosen_index") or 0)
    if not (1 <= chosen <= len(cand_paths)):
        logger.info("era_research[%s]: 사용 가능 후보 없음 — %s",
                    step_tag, name)
        return None
    data = cand_paths[chosen - 1].read_bytes()
    out_path.parent.mkdir(parents=True, exist_ok=True)
    out_path.write_bytes(data)
    return {
        "subject": name,
        "terms": terms,
        "queries": list(res.get("queries") or []),
        "candidates": len(cand_paths),
        "picked_index": chosen,
        "picked_url": cand_urls[chosen - 1],
        "picked_reason_ko": str(verdict.get("chosen_reason_ko") or ""),
        "sha256": hashlib.sha256(data).hexdigest(),
        "file": out_path.name,
    }


def _cache_sha(*parts: str) -> str:
    basis = "\n".join(p.strip() for p in parts)
    return hashlib.sha256(basis.encode("utf-8")).hexdigest()[:16]


def assess_and_research_cached(
    *,
    step_tag: str,
    subject_text: str,
    world_facts_block: str,
    out_dir: Path,
    cache_get: Any,
    cache_put: Any,
    project_config: Optional[Dict[str, Any]] = None,
    openai_client: Any = None,
    failed_memo: Optional[set] = None,
    outcome: Optional[Dict[str, Any]] = None,
) -> Optional[Dict[str, Any]]:
    """판별+조사를 records 사이드카에 캐시 — 같은 장소의 샷들이 1회만
    지출한다 (#119①: 샷별 판 98개가 장소별로 겹치는 실측 대응).

    failed_memo (Codex BLOCK-1): 호출측이 걷기(1바퀴) 단위로 소유하는
    실패 sentinel — 같은 걷기에서 실패한 키는 재시도하지 않는다(같은
    장소 다음 샷마다 유료 사슬 재구매 차단). 메모리 전용이라 다음 resume
    걷기는 자연 재시도. 실패 감사는 era_fail:: record 로 영속(게이트로
    읽지 않는다).

    cache_get(key)->dict|None / cache_put(key, dict) 는 호출측(records)
    소유 — 이 모듈은 저장소를 모른다. 캐시 키 = 대상 텍스트·세계관·정책·
    팩의 sha16. **판별 결과는 빈 목록(비대상)도 캐시**해 재판별 지출을
    막고, **조사 실패는 캐시하지 않는다**(비차단 계약 그대로 — 다음
    방문이 재시도한다). 참조 파일은 out_dir/eraref_<sha16>.png 로 내용
    주소화되어 샷 간 공유된다. 반환 meta 에는 "path"(절대 경로)가 붙는다.
    """
    pack = resolve_era_pack()

    def _mark_failed() -> None:
        # (era R1 BLOCK-1) 호출측이 "조사 미성립(실패)"과 "비대상(정당한
        # era 없음)"을 가르는 신호 — 실패면 기존 성공 산출을 강등·재생성
        # 하지 않고 보존할 수 있게 한다.
        if outcome is not None:
            outcome["failed"] = True

    def _fail_put(sha: str, payload: Dict[str, Any]) -> None:
        # (era R1 BLOCK-3) 실패 감사는 시도마다 **단조 변화**해야 한다 —
        # 같은 bytes 로 덮어쓰면 resume 걷기의 반복 유료 실패가 records
        # 변화 0 으로 보여 #77 지출 감지(_jit_tag_snapshot)에 안 잡히고
        # 무상한 재검색이 열린다. attempts 증가가 bytes 를 움직인다.
        key = f"era_fail::{sha}"
        prev = cache_get(key)
        attempts = (prev.get("attempts", 0) + 1
                    if isinstance(prev, dict) else 1)
        cache_put(key, {**payload, "attempts": attempts})

    if outcome is not None:
        outcome.setdefault("failed", False)
    # Codex BLOCK-3: 팩 **내용** 해시까지 캐시 신원 — selector 미변경
    # 내용 drift 가 옛 결과를 재사용하지 않게.
    a_sha = _cache_sha(subject_text, world_facts_block,
                       ERA_RESEARCH_POLICY_VERSION, pack,
                       era_pack_content_hash())
    a_key = f"era_assess::{a_sha}"
    if failed_memo is not None and a_key in failed_memo:
        _mark_failed()
        return None
    cached = cache_get(a_key)
    if isinstance(cached, dict) and "subjects" in cached:
        subjects = [s for s in (cached.get("subjects") or [])
                    if isinstance(s, dict)]
    else:
        try:
            subjects = assess_subjects(
                step_tag=step_tag, subject_text=subject_text,
                world_facts_block=world_facts_block,
                project_config=project_config,
            )
        except Exception as exc:  # noqa: BLE001 — 판별 실패 비차단
            logger.warning(
                "era_research[%s]: 판별 실패 — %r", step_tag, exc)
            if failed_memo is not None:
                failed_memo.add(a_key)
            _fail_put(a_sha, {"stage": "assess", "error": repr(exc)})
            _mark_failed()
            return None
        cache_put(a_key, {"subjects": subjects})
    if not subjects:
        return None  # 비대상 — 실패 아님(정당한 era 없음, 캐시됨)
    subject = subjects[0]
    name = str(subject.get("subject_native") or "").strip()
    if not name:
        _mark_failed()  # 판별 산출 결손 — 비대상이 아니라 미성립
        return None
    # BLOCK-3: 같은 이름이라도 저작 질의가 다르면 다른 조사 — 판별 신원
    # (a_sha)과 질의 전체를 참조 신원에 접는다.
    # (era R1 BLOCK-2) 검색·선택 산출을 지배하는 팩 계약(search_contract
    # _sha — pick_system 등)도 참조 신원에 접는다: 이것 없이는 picker 팩을
    # 바꿔도 era_ref 캐시 적중이 조사·선택을 영구 우회해 #77 계약이 깨진다.
    from app.modules.pipeline.search_grounded_ref import search_contract_sha

    r_sha = _cache_sha(
        a_sha, name,
        "\n".join(str(t) for t in
                  (subject.get("search_terms_native") or [])),
        str(subject.get("language_lock_native") or ""),
        search_contract_sha())
    r_key = f"era_ref::{r_sha}"
    out_path = out_dir / f"eraref_{r_sha}.png"
    if failed_memo is not None and r_key in failed_memo:
        _mark_failed()
        return None
    rc = cache_get(r_key)
    if (isinstance(rc, dict) and rc.get("sha256")
            and out_path.is_file() and out_path.stat().st_size > 0
            and hashlib.sha256(out_path.read_bytes()).hexdigest()
            == rc.get("sha256")):
        return {**rc, "path": str(out_path)}
    meta = research_reference(
        subject=subject, world_facts_block=world_facts_block,
        out_path=out_path, step_tag=step_tag,
        project_config=project_config, openai_client=openai_client,
    )
    if not meta:
        # BLOCK-1: 같은 걷기에서는 이 대상 재시도 금지 + 실패 감사 영속
        # (다음 resume 걷기가 재시도 — era_fail:: 은 게이트가 아니다).
        if failed_memo is not None:
            failed_memo.add(r_key)
        _fail_put(r_sha, {"stage": "research", "subject": name})
        _mark_failed()
        return None
    cache_put(r_key, meta)
    return {**meta, "path": str(out_path)}