"""앞 화 명부를 추출에 실어 **같은 것을 같은 신원으로** 잇는다.

## 왜

번호를 프로젝트 장부에서 발급하면(`app.core.entity_identity`) 2화가 1화를
**덮는 일**은 없어진다. 그러나 같은 인물이 화마다 **새 번호**를 받는다 —
실측 `da049582` 에서 같은 사람이 `C02 기사 최씨`(3화)와 `C06 최씨`(2화)로
두 행이 됐다. 이어 붙이려면 추출하는 모델이 앞 화를 **봐야** 한다.

## 계약 (Codex 2026-09-04 합의)

1. 모델은 **명부에서 고르거나** 「새것」이라고만 답한다. 임의의 `short_id` 를
   쓰지 못한다 — 허용 목록이 **그 호출에 실린 명부**의 runtime enum 이다.
2. 서버가 대조한다. 명부 밖 값은 인정하지 않고 **새것**으로 돌린다.
3. 두 행이 같은 기존 ID 를 주장하면 **둘 다 새것**이다(fail-closed).
   ★잘못 쪼개는 편이 잘못 합치는 것보다 안전하다 — 합치면 앞 화의 그림과
   설명이 남의 것이 되는데, 쪼개면 행이 하나 늘 뿐이다.
4. 명부에는 이름만이 아니라 **별명**(`EntityAlias`)과 **구조 앵커**를 싣는다.
   ★지금 앵커가 있는 갈래는 **아웃룩(입는 인물)뿐**이다. `location_part` 는
   이 판의 이관 대상이 아니라 앵커를 안 만든다 — 대상으로 넓힐 때 같이 만든다.
5. 지시문은 **팩**에서 온다. 소스에 박으면 버전도 hash 도 audit 도 없다.

★명부가 비면(첫 화) 이 모듈은 **한 바이트도 안 바꾼다** — 스키마도, 프롬프트도.
"""
from __future__ import annotations

import copy
import hashlib
import logging
from typing import Any, Dict, List, Optional, Sequence, Tuple

logger = logging.getLogger(__name__)

#: 체크포인트에 장부가 앉는 칸. **한 곳에서만** 적는다.
LEDGER_KEY = "episode_carry_ledger"

#: 계약이 바뀌면 올린다 — 지문에 접혀 resume 이 옛 체크포인트를 안 건너뛴다.
CARRY_CONTRACT_VERSION = 1

#: 엔티티 행에 실리는 칸. **한 곳에서만** 적는다.
FIELD = "prior_short_id"

#: 「앞 화에 없다 = 새것」을 뜻하는 값. ★빈 문자열을 쓰면 「모델이 안 채웠다」와
#:  구별이 안 된다. 이 값을 enum 에 **넣어** 모델이 명시적으로 고르게 한다.
NEW_SENTINEL = "NEW"

#: 이관 장부의 행선지.
CARRY_REUSED = "reused"        # 명부의 것을 물려받았다
CARRY_NEW = "new"              # 새것 — 번호는 발급기가 준다
CARRY_REJECTED = "rejected"    # 명부 밖 값을 냈다 → 새것으로 돌림
CARRY_CONTESTED = "contested"  # 둘 이상이 같은 ID 를 주장 → 둘 다 새것

_MODULE = "episode_carry"
PROMPT_PACK_VERSION = "1.202609040000"
STEM_HEAD = "prior_roster_head"
STEM_RULE = "prior_roster_rule"
STEMS = (STEM_HEAD, STEM_RULE)


def load_pack(*, db=None, version: Optional[str] = None) -> Dict[str, Any]:
    """명부 지문 팩. `raw_content_hash` 를 소비 지문에 접어야 한다."""
    from app.modules.prompt_loader import resolve_effective

    ver = version or PROMPT_PACK_VERSION
    resolved = {st: resolve_effective(_MODULE, st, kind="prompt",
                                      version=ver, db=db) for st in STEMS}
    manifest = "|".join(
        f"{st}:{r['source']}:{r['version']}:{r['raw_content_hash']}"
        for st, r in sorted(resolved.items()))
    return {"module": _MODULE, "version": ver, "stems": resolved,
            "pack_manifest_hash": hashlib.sha256(
                manifest.encode("utf-8")).hexdigest()[:16]}


def pack_fingerprint(*, db=None, version: Optional[str] = None) -> Dict[str, Any]:
    """소비 지문에 접을 좌표. ★상수만 올리고 bytes 를 안 접으면 안 움직인다."""
    pack = load_pack(db=db, version=version)
    return {"carry_contract": CARRY_CONTRACT_VERSION,
            "carry_pack": pack["version"],
            "carry_pack_hash": pack["pack_manifest_hash"]}


def _text(stem: str, *, db=None, version: Optional[str] = None) -> str:
    return load_pack(db=db, version=version)["stems"][stem]["content"]


def _carry_anchor_rows(db, project_id: str, episode_id: str, *,
                       include_self: bool = False) -> List[Any]:
    """명부 앵커가 볼 **배정 범위** — 앞선 화수 + 이 화, **화별로** 판정.

    ★★★소비자용 `episode_outlook_rows`(정확히 이 화)와 **다른 물음**이다
     (Codex BLOCK 2026-09-04 3차).

     명부에는 앞 화 canon 이 실린다. 그런데 앵커를 「이 화의 배정」으로만
     보면, **2화를 처음 추출할 때는 2화 아웃룩 배정이 아직 없어서** 1화의
     `O01` 이 명부에 나오면서도 착용자가 비어 버린다 — 같은 이름의 옷을
     가르는 핵심 정보가 사라진다. 반대로 프로젝트 전체를 보면 **미래 화의
     착용자**가 섞인다. 그래서 범위는 명부와 **똑같이** 「앞선 화수 + 이 화」다.

    ★★★그리고 exact-first 는 **화마다** 따진다 (Codex BLOCK 4차).

     범위 전체로 한 번에 접으면 — 014 뒤 2화만 sync 돼 exact 가 생기고 1화는
     아직 NULL 인 **흔한 중간 상태**에서, 3화 명부가 **1화 앵커를 통째로
     잃는다.**

     legacy 판정의 「양 끝이 걸렸나」도 **그 화의** active 집합으로 본다.
     화들을 합쳐 보면 `C01` 은 1화에만, `O02` 는 2화에만 active 인데도
     「범위 안에 둘 다 있다」로 통과한다 — **같은 화에서 함께 active 였다는
     근거가 없다.**
    """
    from sqlalchemy import text as sql_text

    from app.core.entity_identity import active_episode_canon_ids
    from app.models.project import CharacterOutlook

    # ★★범위는 **명부와 같아야 한다** (Codex BLOCK 2026-09-04). 명부에서
    #  이 화를 뺐는데 앵커만 이 화를 보면, 늦게 얼린 스냅샷의 앵커 줄이
    #  달라져서 「얼리는 시점과 무관」이 아웃룩에서 다시 깨진다.
    eligible = [r[0] for r in db.execute(sql_text(
        "SELECT e.id FROM episode e WHERE e.project_id = :pid AND ("
        + ("  e.id = :eid OR " if include_self else "") +
        "  e.episode_number < ("
        "    SELECT episode_number FROM episode WHERE id = :eid)) "
        "ORDER BY e.episode_number"
    ), {"pid": project_id, "eid": episode_id}).fetchall()]
    if not eligible:
        return []

    rows = db.query(CharacterOutlook).filter(
        CharacterOutlook.project_id == project_id).all()
    by_ep: Dict[str, List[Any]] = {}
    legacy: List[Any] = []
    for r in rows:
        if r.episode_id is None:
            legacy.append(r)
        else:
            by_ep.setdefault(r.episode_id, []).append(r)

    out: Dict[str, Any] = {}
    for eid in eligible:
        exact = by_ep.get(eid) or []
        if exact:
            for r in exact:
                out[r.id] = r
            continue
        if not legacy:
            continue
        # ★**그 화의** active 집합. 화들을 합치지 않는다.
        active = set(active_episode_canon_ids(db, project_id, eid))
        for r in legacy:
            if r.character_id in active and r.outlook_id in active:
                out[r.id] = r
    return list(out.values())


def _anchor_for(db, project_id: str, owner: str,
                canon_ids: Sequence[str],
                episode_id: Optional[str] = None, *,
                include_self: bool = False) -> Dict[str, str]:
    """구조 앵커 — 그 행이 **무엇에 매달려 있는가**.

    ★같은 이름이 여러 개일 때 이것 하나로 갈린다.
    ★지금 구현된 갈래는 **아웃룩뿐**(입는 인물). 다른 갈래는 빈 값이고,
     그것이 계약이다 — 있는 척하면 다음 사람이 앵커를 믿고 합친다.
    ★범위는 `_carry_anchor_rows` 가 정한다 — 명부와 **같은 범위**여야
     앞 화 착용자는 남고 미래 화 착용자는 안 섞인다.
    """
    if not canon_ids or owner != "outlook" or not episode_id:
        return {}
    want = set(canon_ids)
    rows = [r for r in _carry_anchor_rows(db, project_id, episode_id,
                                          include_self=include_self)
            if r.outlook_id in want]
    if not rows:
        return {}
    from sqlalchemy import text as sql_text

    short_by_id = {
        r[0]: r[1] for r in db.execute(sql_text(
            "SELECT id, short_id FROM entity_canon "
            "WHERE id = ANY(:ids) AND short_id IS NOT NULL"
        ), {"ids": sorted({r.character_id for r in rows})}).fetchall()
    }
    out: Dict[str, List[str]] = {}
    for r in rows:
        sid = short_by_id.get(r.character_id)
        if sid:
            out.setdefault(r.outlook_id, []).append(sid)
    return {k: "·".join(sorted(set(v))) for k, v in out.items()}


def build_prior_roster(
    db, project_id: str, owner: str, episode_id: Optional[str] = None, *,
    include_self: bool = False,
) -> Tuple[List[str], List[str]]:
    """**앞 화들**(그리고 force 일 때만 이 화 자신)의 신원 목록.

    ★★★``episode_id`` 를 주면 **앞선 화수**의 것만 싣는다 (Codex BLOCK
     2026-09-04). 프로젝트 전체를 실으면 1화를 다시 분석할 때 2·3화 신원이
     **과거로 새어** 들어가고, 어느 화에도 안 붙은 고아까지 후보가 된다.
    ★``episode_id`` 를 안 주면 예전처럼 프로젝트 전체다 — 도구·조회용이다.

    ★★★``include_self`` — **이 화 자신**을 넣을지 (2026-09-04 실측 수정).

     넣는 까닭: 같은 화를 `force` 로 다시 돌릴 때 모델이 제 앞 판 산출을 못
     보면 전부 「새것」이 되어 **재실행마다 번호가 늘어나고**, 뒤 화가 이미
     가리키던 옛 번호와 **같은 사람이 둘로 갈린다**.

     그런데 늘 넣으면 명부 내용이 **언제 처음 얼리느냐**에 따라 달라진다.
     이 화가 아직 안 돌았으면 비어 있고, 돌고 난 뒤 처음 물으면 제 canon 이
     들어찬다. 그러면 이 PR 이전에 **이미 분석된 화**는 재개할 때마다
     지문이 어긋나 `entity_all_*` 과 `outlook_phase1` 을 **다시 산다**
     (실측: 1화 재개가 10초 만에 `config_hash mismatch` 로 섰다).

     그래서 **force 로 스냅샷을 다시 뜰 때만** 넣는다. 그 자리에서는 이 화가
     이미 돈 것이 확실하고, 어차피 다시 사는 판이라 지문이 움직여도 손해가
     없다. 평상시 경로는 앞 화만 보므로 **얼리는 시점과 무관**하다.

    Returns:
        (프롬프트 줄들, 허용 `short_id` 목록). 없으면 **둘 다 비었다** —
        그러면 첫 화는 한 바이트도 안 달라진다.
    """
    from sqlalchemy import text as sql_text

    from app.core.entity_identity import NULL_OUTLOOK_SHORT_ID

    if episode_id:
        rows = db.execute(sql_text(
            "SELECT DISTINCT c.id, c.short_id, c.name, "
            "       COALESCE(c.description, '') "
            "FROM entity_canon c "
            "JOIN entity_episode_link l ON l.canon_id = c.id "
            "JOIN episode e ON e.id = l.episode_id "
            "WHERE c.project_id = :pid AND c.entity_type = :et "
            "  AND c.short_id IS NOT NULL "
            "  AND (" + ("e.id = :eid OR " if include_self else "") +
            "      e.episode_number < ("
            "        SELECT episode_number FROM episode WHERE id = :eid)) "
            "ORDER BY c.short_id"
        ), {"pid": project_id, "et": owner, "eid": episode_id}).fetchall()
    else:
        rows = db.execute(sql_text(
            "SELECT id, short_id, name, COALESCE(description, '') "
            "FROM entity_canon "
            "WHERE project_id = :pid AND entity_type = :et "
            "  AND short_id IS NOT NULL "
            "ORDER BY short_id"
        ), {"pid": project_id, "et": owner}).fetchall()
    # ★예약값은 명부에서 뺀다 — 실체가 아니라 「배정 없음」 표식이다.
    rows = [r for r in rows if r[1] != NULL_OUTLOOK_SHORT_ID]
    if not rows:
        return [], []

    ids = [r[0] for r in rows]
    alias_rows = db.execute(sql_text(
        "SELECT canon_id, alias FROM entity_alias WHERE canon_id = ANY(:ids)"
    ), {"ids": ids}).fetchall()
    aliases: Dict[str, List[str]] = {}
    for cid, alias in alias_rows:
        aliases.setdefault(cid, []).append(alias)

    anchors = _anchor_for(db, project_id, owner, ids, episode_id,
                          include_self=include_self)

    lines: List[str] = []
    allowed: List[str] = []
    for cid, sid, name, desc in rows:
        allowed.append(sid)
        parts = [f"- {sid} | {name}"]
        alt = aliases.get(cid)
        if alt:
            parts.append(f"다른 이름: {', '.join(sorted(set(alt)))}")
        anc = anchors.get(cid)
        if anc:
            parts.append(f"딸린 곳: {anc}")
        # ★설명은 한 줄만 — 명부가 길어지면 본문을 밀어낸다.
        one = " ".join(str(desc).split())
        if one:
            parts.append(one[:120])
        lines.append(" | ".join(parts))
    return lines, allowed


#: 이 화가 **본 명부**를 얼려 두는 자리. ★체크포인트와 나란히 둔다.
SNAPSHOT_NAME = "_carry_snapshot.json"


def _snapshot_path(project_id: str, episode_id: str):
    import pathlib as _pl

    from app.core.config import settings

    return (_pl.Path(settings.projects_dir) / project_id / "checkpoints"
            / "episodes" / episode_id / SNAPSHOT_NAME)


def carry_snapshot(db, project_id: str, episode_id: str, owner: str, *,
                   refresh: bool = False) -> Dict[str, Any]:
    """이 화가 쓰는 명부 — **한 번 얼리면 안 바뀐다**.

    ★★★왜 얼리나 (Codex BLOCK 2026-09-04 재지적).

    명부를 매번 현재 DB 에서 읽으면, **이 화가 제 canon 을 만든 뒤** 명부가
    달라진다. 그러면 같은 체크포인트의 지문이 스스로 어긋나서 **재개가 유료
    상류를 다시 산다.** 무료 스텝의 지문 어긋남이 유료 하류를 다시 사는
    부류다.

    그래서 이 화가 **처음 물을 때** 얼려 파일로 남기고, 그 뒤로는 그것만 본다.
    `refresh=True`(force 재실행)일 때만 다시 뜬다.

    Returns: ``{"lines": [...], "allowed": [...], "digest": "..."}``
    """
    import json as _json

    path = _snapshot_path(project_id, episode_id)
    data: Dict[str, Any] = {}
    if path.is_file():
        try:
            data = _json.loads(path.read_text(encoding="utf-8"))
        except Exception as exc:  # noqa: BLE001
            # ★못 읽으면 **선다** — 조용히 다시 뜨면 지문이 흔들린다.
            raise RuntimeError(
                f"이 화의 명부 스냅샷을 못 읽는다: {path} ({exc}). 고치거나 "
                f"지운 뒤 force 로 다시 돌려라") from exc
    if owner in data and not refresh:
        got = data[owner]
        # ★읽을 때 모양을 본다 — 프롬프트 bytes 와 config_hash 가 갈리면
        #  「같은 것을 본다」는 계약이 조용히 깨진다.
        if (not isinstance(got, dict)
                or not isinstance(got.get("lines"), list)
                or not isinstance(got.get("allowed"), list)):
            raise RuntimeError(
                f"명부 스냅샷의 모양이 깨졌다: {path} (owner={owner}). "
                f"지운 뒤 force 로 다시 돌려라")
        want = (hashlib.sha256("\n".join(got["lines"]).encode("utf-8"))
                .hexdigest()[:16] if got["allowed"] else "")
        if got.get("digest") != want:
            raise RuntimeError(
                f"명부 스냅샷의 지문이 내용과 안 맞는다: {path} "
                f"(owner={owner}, 적힌 것={got.get('digest')!r}, 계산={want!r})")
        return got

    # ★다시 뜨는 판(force)에서만 이 화 자신을 넣는다 — 위 docstring 참고.
    lines, allowed = build_prior_roster(db, project_id, owner, episode_id,
                                        include_self=refresh)
    entry = {"lines": lines, "allowed": allowed,
             "digest": (hashlib.sha256("\n".join(lines).encode("utf-8"))
                        .hexdigest()[:16] if allowed else "")}
    data[owner] = entry
    # ★원자적으로 쓴다 — 도중에 죽으면 잘린 파일이 남고, 그 뒤 재개가
    #  「못 읽는다」로 영영 막힌다 (Codex 2026-09-04).
    from app.core.checkpoint_io import atomic_write_json

    path.parent.mkdir(parents=True, exist_ok=True)
    atomic_write_json(path, data)
    return entry


def roster_digest(db, project_id: str, owner: str,
                  episode_id: Optional[str] = None) -> str:
    """이 화가 **실제로 보는 명부**의 지문. 없으면 빈 문자열.

    ★★팩 버전만 접으면 명부 **내용**(이름·별명·앵커)이 바뀌어도 지문이 안
     움직여 옛 체크포인트가 그대로 재사용된다 (Codex BLOCK 2026-09-04).
    ★반대로 프로젝트 전체를 접으면 뒤 화가 생길 때마다 앞 화가 제 것이 아닌
     변화로 계속 stale 된다 — 그래서 **앞 화 + 이 화** 범위와 같아야 한다.
    """
    if not episode_id:
        lines, allowed = build_prior_roster(db, project_id, owner)
        return (hashlib.sha256("\n".join(lines).encode("utf-8")).hexdigest()[:16]
                if allowed else "")
    # ★얼린 것을 본다 — 안 그러면 이 화가 canon 을 만든 뒤 제 지문이 바뀐다.
    return carry_snapshot(db, project_id, episode_id, owner)["digest"]


def build_roster_block(
    db, project_id: str, owner: str, episode_id: Optional[str] = None, *,
    refresh: bool = False, pack_db=None, version: Optional[str] = None,
) -> Tuple[str, List[str]]:
    """프롬프트에 붙일 **한 덩어리**와 허용 ID.

    ★목록과 문안을 **함께** 낸다. 따로 두면 한쪽만 붙는 판이 생기고, 그러면
    모델이 채울 수 없는 칸을 required 로 요구하게 된다.
    """
    if episode_id:
        snap = carry_snapshot(db, project_id, episode_id, owner, refresh=refresh)
        lines, allowed = snap["lines"], snap["allowed"]
    else:
        lines, allowed = build_prior_roster(db, project_id, owner)
    if not allowed:
        return "", []
    head = _text(STEM_HEAD, db=pack_db, version=version).strip()
    rule = _text(STEM_RULE, db=pack_db, version=version).strip()
    body = "\n".join([head, "", *lines, "", rule])
    return "\n\n" + body, allowed


def patch_schema_with_prior_ids(schema: Dict[str, Any],
                                allowed: Sequence[str]) -> Dict[str, Any]:
    """산출에 **`prior_short_id` runtime enum** 을 더한다.

    ★허용 목록은 **그 호출에 실린 명부** + `NEW` 뿐이다. 팩에 안 박는다.
    ★`allowed` 가 비면 스키마를 안 건드린다 — 첫 화 불변.
    """
    if not allowed:
        return schema
    out = copy.deepcopy(schema)
    enum = [*allowed, NEW_SENTINEL]
    for key in list(out.get("properties", {}).keys()):
        arr = out["properties"][key]
        if arr.get("type") != "array" or "items" not in arr:
            continue
        props = arr["items"].setdefault("properties", {})
        req = arr["items"].setdefault("required", [])
        props[FIELD] = {"type": "string", "enum": enum}
        if FIELD not in req:
            req.append(FIELD)
    return out


def apply_prior_ids(
    entities: Sequence[Dict[str, Any]], allowed: Sequence[str],
) -> Dict[str, Any]:
    """모델이 고른 앞 화 신원을 **대조해서** 행에 옮긴다.

    - 명부 안의 값 → 그 행의 `short_id` 가 된다(물려받음).
    - 명부 밖 · `NEW` · 빈 값 → 그대로 두어 발급기가 새 번호를 준다.
    - 같은 ID 를 **둘 이상**이 주장 → **모두** 새것 (fail-closed).

    Returns:
        장부 — `{short_id or index: 행선지}` 와 갈래별 개수. ★계산해 놓고
        안 남기면 「다퉜다」와 「원래 새것」이 구별이 안 된다.
    """
    allow = set(allowed)
    ledger: Dict[str, str] = {}
    counts = {CARRY_REUSED: 0, CARRY_NEW: 0,
              CARRY_REJECTED: 0, CARRY_CONTESTED: 0}
    if not allow:
        for e in entities:
            e.pop(FIELD, None)
        counts[CARRY_NEW] = len(entities)
        return {"ledger": ledger, "counts": counts}

    # 1차 — 누가 무엇을 주장했나
    claims: Dict[str, List[int]] = {}
    for i, e in enumerate(entities):
        raw = str(e.get(FIELD) or "").strip()
        if not raw or raw == NEW_SENTINEL:
            continue
        if raw not in allow:
            ledger[f"#{i}:{raw}"] = CARRY_REJECTED
            counts[CARRY_REJECTED] += 1
            logger.warning(
                "episode_carry: 명부에 없는 앞 화 ID %r — 새것으로 돌린다", raw)
            continue
        claims.setdefault(raw, []).append(i)

    # 2차 — 다툰 것은 아무도 못 가진다
    for sid, idxs in claims.items():
        if len(idxs) > 1:
            ledger[sid] = CARRY_CONTESTED
            counts[CARRY_CONTESTED] += 1
            logger.warning(
                "episode_carry: %s 를 %d 행이 주장 — 합치지 않고 모두 새것으로 "
                "둔다(fail-closed)", sid, len(idxs))
            continue
        entities[idxs[0]]["short_id"] = sid
        ledger[sid] = CARRY_REUSED
        counts[CARRY_REUSED] += 1

    for e in entities:
        e.pop(FIELD, None)
        if not str(e.get("short_id") or "").strip():
            counts[CARRY_NEW] += 1
    return {"ledger": ledger, "counts": counts}