#!/usr/bin/env python3
"""Muse Spark 1.3 Contributor 를 ②선정 판정에 세워 본다 (2026-09-03).

`build_judge_ab_gallery.py` 가 보여 주는 그 판(에피소드의 cross-model
선정 판정)에 심판 한 자리를 더한다. **기존 두 심판은 다시 부르지 않는다**
— 기록에 있는 판정을 그대로 읽어 비교 표에만 담고, 새로 사는 것은
`meta/muse-spark-1.3-contributor` 하나뿐이다. (`judge3_flash_pilot.py`
가 08-13 파일럿에 flash 를 더한 방식과 같다.)

## 계약 — 원본과 같아야 비교가 성립한다

- 지시문 = **프로덕션 함수가 지금 내주는 그것**. 손으로 옮겨 적지 않고
  `resolve_judge_texts(2, "judge_still", STILL_JUDGE_PACK_VERSION)` 를
  부른다. 팩 `15.202608272340` 은 08-27 이후 커밋이 없어 그 주행이 보낸
  bytes 와 같다(확인함).
- 스키마 = `build_judge_schema(roll_labels(2), with_physics=True)`.
- 머리말 = **샷마다 갈린다.** bgfirst 2택1 샷은 중립 머리말
  (`load_bgfirst_judge_header`), 표준 멀티롤 샷은
  `make_gemini_judge_fn` 의 기본값. 상수를 여기 다시 적지 않고 함수
  기본값에서 읽는다 — 저쪽이 바뀌면 여기도 따라 바뀌어야 한다.
- 참조 = 기록의 `refs`(판정용 공유 참조). 인라인으로 나가 경로가
  `<bytes:N>` 로만 남은 것은 **asset_id → DB → 파일**로 되찾고, 파일
  길이가 기록의 N 과 다르면 그 샷을 버린다. 참조를 비운 채 판정하면
  원 판정과 다른 계약이다.
- 순서 = 정순·역순 **둘 다**. 프로덕션은 모델과 순서를 묶어 한 모델당
  한 순서만 보지만(그 대가는 `_judge_cross_model_order` 가 적어 뒀다),
  새 심판은 양쪽을 다 보게 해 「이 모델이 자리에 흔들리나」를 잰다.
  역순 응답은 프로덕션과 같은 함수로 canonical 라벨에 되돌린다.

## 돈

논리 호출 = 샷수 × 순서수. 클라이언트는 스키마 실패에 한 번 되묻고
(`max_retry=1`), `response_format` 400 이면 빼고 한 번 더 — 그래서
물리 왕복은 논리 하나당 **최대 4회**다. 아래 `--max-calls` 는 논리
호출 상한이고, 실제로 나간 것은 `llm_call_log` 에서 센다(클라이언트가
왕복마다 남긴다).

## 프로덕션 무접촉

records.json·체크포인트·DB 를 쓰지 않는다. 읽기만 하고 산출은
`artifact/` 아래 새 디렉토리에만 만든다.

사용:
  muse_judge_pilot.py <project_id> <episode_id>            # 판정(재개 가능)
  muse_judge_pilot.py <pid> <eid> --limit 1                # 한 샷만 먼저
  muse_judge_pilot.py <pid> <eid> --orders forward         # 정순만
"""
from __future__ import annotations

import argparse
import inspect
import json
import pathlib
import sys
from typing import Any, Dict, List, Tuple

sys.path.insert(0, str(pathlib.Path(__file__).resolve().parents[2]))
sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent))
import _db  # noqa: E402

ROOT = pathlib.Path(__file__).resolve().parents[3]

#: 이번에 재는 심판. OpenRouter 카탈로그 확인(2026-09-03):
#: 입력 text+image+…, 출력 text, ctx 1,048,576, reasoning **mandatory**.
#:
#: ★contributor 등급은 **계정이 유료 모델 학습을 허용해야** 열린다
#:  (404 `paid-model-training-violation-by-account`). 허용하지 않은
#:  계정에서는 일반판 `meta/muse-spark-1.3` 을 쓴다 — 같은 모델·같은
#:  판정이고 값만 다르다. 어느 쪽으로 샀는지는 결과에 적힌다.
MUSE_MODEL = "meta/muse-spark-1.3-contributor"

#: 판정류 긴 구조화 출력의 미달-잘림 방지 예산. 프로덕션 OpenRouter
#: 심판이 쓰는 값과 같다(`multiroll_gemini.py` :1650 부근 8000).
#: ★이 모델은 reasoning 이 강제라 이 예산 안에서 추론까지 나온다 —
#: 첫 샷 결과의 usage 를 보고 모자라면 여기서 올린다(올리면 기록한다).
MAX_TOKENS = 8000

DEFAULT_OUT = ROOT / "artifact" / "20260903_muse_judge_pilot"


# ── 계약 조립 — 전부 프로덕션 함수에서 온다 ──────────────────────────

def _judge_contract() -> Tuple[str, Dict[str, Any], List[str]]:
    from app.modules.pipeline.multiroll_gemini import (
        STILL_JUDGE_PACK_VERSION, judge_pack_content_hash,
        resolve_judge_pack_version, resolve_judge_texts,
    )
    from app.modules.pipeline.multiroll_select import (
        build_judge_schema, roll_labels,
    )

    labels = roll_labels(2)
    texts = resolve_judge_texts(2, judge_name="judge_still",
                                pack_version=STILL_JUDGE_PACK_VERSION)
    schema = build_judge_schema(labels, with_physics=True)
    print(f"판정 팩: {resolve_judge_pack_version(STILL_JUDGE_PACK_VERSION)}"
          f" · 내용 지문 {judge_pack_content_hash(STILL_JUDGE_PACK_VERSION)}"
          f" · judge_sys {len(texts['judge_sys'])}자")
    return texts["judge_sys"], schema, list(labels)


def _headers() -> Tuple[str, str]:
    """(기본 머리말, bgfirst 머리말) — 둘 다 **저쪽에서** 읽는다."""
    from app.modules.pipeline.multiroll_gemini import make_gemini_judge_fn
    from app.modules.pipeline.still_recipe import load_bgfirst_judge_header

    default = inspect.signature(
        make_gemini_judge_fn).parameters["prompt_header"].default
    if not isinstance(default, str) or not default.strip():
        raise RuntimeError(
            "make_gemini_judge_fn.prompt_header 기본값을 못 읽었다 — "
            "머리말을 지어내지 않는다")
    return default, load_bgfirst_judge_header()


def _resolve_ref_path(ref: Dict[str, Any]) -> pathlib.Path:
    """참조 한 장의 실제 파일.

    경로가 그대로면 그것을 쓰고, 인라인으로 나가 `<bytes:N>` 만 남았으면
    asset_id 로 DB 에서 찾는다. 찾은 파일의 **길이가 N 과 같은지**까지
    본다 — 서로 다른 두 사실이 맞아떨어져야 「그 파일이 맞다」이다.
    """
    raw = str(ref.get("path") or "")
    if not raw.startswith("<bytes:"):
        p = pathlib.Path(raw)
        if not p.is_file():
            raise FileNotFoundError(f"참조 파일이 없다: {raw}")
        return p
    want = int(raw[len("<bytes:"):-1])
    asset_id = ref.get("asset_id")
    if not asset_id:
        raise ValueError(f"인라인 참조인데 asset_id 가 없다: {raw}")
    got = _db.rows("SELECT file_path FROM image_asset WHERE id = :i",
                   {"i": asset_id})
    if not got:
        raise LookupError(f"image_asset 에 {asset_id} 가 없다")
    p = pathlib.Path(str(got[0][0]))
    if not p.is_absolute():
        p = ROOT / p
    if not p.is_file():
        raise FileNotFoundError(f"asset {asset_id} 의 파일이 없다: {p}")
    size = p.stat().st_size
    if size != want:
        raise ValueError(
            f"asset {asset_id} 길이 불일치 — 기록 {want} ≠ 파일 {size}")
    return p


def _parts_for(rec: Dict[str, Any], rdir: pathlib.Path, tag: str,
               labels: List[str], order: List[str],
               head_default: str, head_bgfirst: str) -> List[Dict[str, Any]]:
    """프로덕션이 보낸 것과 같은 모양의 parts — 제시 순서만 인자로 받는다."""
    from app.modules.pipeline.multiroll_gemini import png_part, ref_parts

    head = head_bgfirst if rec.get("bgfirst") else head_default
    prompt = rec.get("prompt")
    if not isinstance(prompt, str) or not prompt.strip():
        raise ValueError(f"{tag}: 기록에 prompt 가 없다")
    labeled = [(str(r.get("label") or ""), _resolve_ref_path(r))
               for r in (rec.get("refs") or [])]

    parts: List[Dict[str, Any]] = [
        {"type": "text", "text": head + "\n" + prompt}]
    parts += ref_parts(labeled)
    by_label = {lab: rdir / f"{tag}_{lab.lower()}.png" for lab in labels}
    for disp, src in zip(labels, order):
        f = by_label[src]
        if not f.is_file():
            raise FileNotFoundError(f"후보 이미지가 없다: {f}")
        parts.append({"type": "text", "text": f"Candidate {disp}:"})
        parts.append(png_part(f))
    return parts


def _ask(step_tag: str, judge_sys: str, parts: List[Dict[str, Any]],
         schema: Dict[str, Any], meta: Dict[str, Any],
         model: str) -> Dict[str, Any]:
    from app.modules.llm.openrouter_vlm_client import (
        ask_openrouter_structured,
    )

    return ask_openrouter_structured(
        step_tag, judge_sys, parts, schema,
        model=model, max_tokens=MAX_TOKENS, opik_metadata=meta)


# ── 장부 — 실제로 나간 왕복은 클라이언트가 남긴 것에서 센다 ─────────

def _spend_rows(model: str) -> List[Tuple[Any, ...]]:
    """★칼럼 이름은 실제 표에서 가져온다 — `model_name`·`input_tokens`·
    `output_tokens` 다(`model`·`usage` 가 아니다). 없는 칸을 물으면
    조회가 죽고, 죽은 조회를 「0건」으로 읽으면 「안 샀다」가 된다."""
    return _db.rows(
        "SELECT status, count(*), "
        "  coalesce(sum(input_tokens), 0), coalesce(sum(output_tokens), 0) "
        "FROM llm_call_log WHERE model_name = :m GROUP BY status",
        {"m": model})


def _spend_total(model: str) -> int:
    return sum(int(r[1]) for r in _spend_rows(model))



def _load_results(res_path: pathlib.Path) -> Dict[str, Any]:
    """결과 적재 — 키는 **모델 → 샷 → 순서**.

    2026-09-03 이전 판은 `샷 → 순서` 로 평평했다. 그 모양이면 각 칸이
    적어 둔 `model` 을 보고 모델별로 옮겨 담는다 — 버리지 않는다.
    """
    if not res_path.is_file():
        return {}
    data = json.loads(res_path.read_text("utf-8"))
    if not data:
        return {}
    # 새 모양이면 최상위 키가 모델 id 이고 그 아래가 샷이다.
    probe = next(iter(data.values()))
    if isinstance(probe, dict) and probe and all(
            isinstance(v, dict) and ("model" in v or "ok" in v or "error" in v)
            for v in probe.values()):
        # 최상위가 샷, 그 아래가 순서 슬롯 — 옛 평평한 판이다.
        out: Dict[str, Any] = {}
        for tag, orders in data.items():
            for order_name, slot in orders.items():
                mdl = str(slot.get("model") or "unknown")
                out.setdefault(mdl, {}).setdefault(tag, {})[order_name] = slot
        print(f"옛 평평한 결과를 모델별로 옮겨 담았다: "
              f"{ {m: len(v) for m, v in out.items()} }")
        return out
    return data


# ── 본체 ────────────────────────────────────────────────────────────

def main() -> int:
    ap = argparse.ArgumentParser(add_help=True)
    ap.add_argument("project_id")
    ap.add_argument("episode_id")
    ap.add_argument("--out", default=str(DEFAULT_OUT))
    ap.add_argument("--limit", type=int, default=0,
                    help="앞에서 N 샷만 (0=전부)")
    ap.add_argument("--orders", default="forward,reverse")
    ap.add_argument("--model", default=MUSE_MODEL,
                    help="OpenRouter 모델 id (기본=contributor 등급)")
    ap.add_argument("--max-calls", type=int, default=12,
                    help="논리 호출 상한 — 넘으면 선다")
    args = ap.parse_args()

    from app.core.config import settings
    from app.modules.llm.openrouter_vlm_client import openrouter_configured
    from app.modules.pipeline.multiroll_select import (
        flip_display_to_canonical, normalize_flip_verdict,
    )

    if not openrouter_configured():
        print("★OPENROUTER_API_KEY 가 없다 — backend/.env 를 봐라")
        return 2

    rdir = (pathlib.Path(settings.projects_dir) / args.project_id
            / "images" / args.episode_id / "scene" / "recipe")
    recs = json.loads((rdir / "records.json").read_text("utf-8"))
    shots = sorted(k for k, v in recs.items()
                   if isinstance(v, dict) and v.get("cross_model_order"))
    if not shots:
        print("★내 조회로는 cross_model_order 를 가진 샷을 못 찾았다 — "
              "「없다」로 읽지 마라")
        return 1
    if args.limit:
        shots = shots[:args.limit]
    orders = [o.strip() for o in args.orders.split(",") if o.strip()]
    for o in orders:
        if o not in ("forward", "reverse"):
            raise ValueError(f"모르는 순서: {o}")

    judge_sys, schema, labels = _judge_contract()
    head_default, head_bgfirst = _headers()
    outdir = pathlib.Path(args.out)
    outdir.mkdir(parents=True, exist_ok=True)
    res_path = outdir / "results.json"
    results: Dict[str, Any] = _load_results(res_path)

    # 계약을 산출과 함께 남긴다 — 다음 사람이 「무엇을 보냈나」를 다시
    # 조립하지 않아도 되게. 지시문은 파일로, 스키마도 파일로.
    (outdir / "judge_sys.txt").write_text(judge_sys, encoding="utf-8")
    (outdir / "judge_schema.json").write_text(
        json.dumps(schema, ensure_ascii=False, indent=1), encoding="utf-8")
    (outdir / "headers.json").write_text(json.dumps(
        {"default": head_default, "bgfirst": head_bgfirst},
        ensure_ascii=False, indent=1), encoding="utf-8")

    # ★신원은 (모델, 샷, 순서) 셋이다. 앞의 두 개로만 키를 잡으면 다른
    #  모델로 같은 판을 다시 살 때 **앞 모델 것을 덮어쓴다**.
    mine = results.setdefault(args.model, {})

    def _done(t: str, o: str) -> bool:
        return bool((mine.get(t) or {}).get(o, {}).get("ok"))

    todo = [(t, o) for t in shots for o in orders if not _done(t, o)]
    print(f"대상 {len(shots)}샷 × 순서 {len(orders)} = 논리 {len(todo)}회"
          f" (이미 있는 것 건너뜀) · 모델 {args.model}")
    if len(todo) > args.max_calls:
        print(f"★상한 초과 — 논리 {len(todo)} > --max-calls {args.max_calls}."
              f" 상한을 올리든지 --limit 으로 줄여라. 아무것도 안 샀다.")
        return 3

    before = _spend_total(args.model)
    rev_map = flip_display_to_canonical(labels)
    fwd_map = {lab: lab for lab in labels}

    for tag, order_name in todo:
        rec = recs[tag]
        order = list(labels) if order_name == "forward" else list(
            reversed(labels))
        slot: Dict[str, Any] = {"model": args.model, "order": order_name,
                                "display_to_canonical":
                                    fwd_map if order_name == "forward"
                                    else rev_map}
        try:
            parts = _parts_for(rec, rdir, tag, labels, order,
                               head_default, head_bgfirst)
            raw = _ask("muse_judge_pilot", judge_sys, parts, schema,
                       {"project_id": args.project_id,
                        "episode_id": args.episode_id,
                        "shot": tag, "order": order_name,
                        "pilot": "muse_judge_pilot"},
                       args.model)
            slot["raw"] = raw
            slot["normalized"] = normalize_flip_verdict(
                raw, slot["display_to_canonical"], labels)
            slot["ok"] = True
            print(f"  {tag} {order_name}: winner "
                  f"{slot['normalized']['winner']}")
        except Exception as exc:  # noqa: BLE001
            slot["ok"] = False
            slot["error"] = repr(exc)[:600]
            print(f"  {tag} {order_name}: 실패 — {exc!r}"[:220])
        mine.setdefault(tag, {})[order_name] = slot
        res_path.write_text(
            json.dumps(results, ensure_ascii=False, indent=1),
            encoding="utf-8")

    after = _spend_total(args.model)
    print(f"\n장부(llm_call_log, model={args.model}): 이번에 늘어난 왕복 "
          f"{after - before}건 (논리 {len(todo)}회)")
    for st, n, tin, tout in _spend_rows(args.model):
        print(f"  {st}: {n}건 · 입력 {tin} · 출력 {tout} 토큰")
    print(f"산출: {res_path}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())