app/ledger.py

# -*- coding: utf-8 -*-
"""원장 — 참가자·포지션·점수를 담는다.



파일 기반이다. HF Space 재시작 시 사라지므로 운영에서는 외부 저장소로 옮긴다.

지금은 구조를 확정하는 게 목적이고, 저장 위치만 바꾸면 되도록 인터페이스를 좁게 둔다.



🔴 포지션은 절대 덮어쓰지 않는다. 매 제출을 append 한다.

   덮어쓰면 "언제 무엇을 냈는가"가 사라지고, 그게 사라지면 채점을 재현할 수 없다.

"""
import os, json, time, hashlib, threading
from datetime import datetime, timezone

def _root():
    """원장 위치. 영속 스토리지가 있으면 거기, 없으면 컨테이너 안.



    🔴 HF Space에 영속 스토리지를 안 붙이면 /data 가 없다. 그때 죽어 버리면

       화면 자체가 안 뜬다. 떠 있되 '재시작하면 사라진다'는 사실이 보이게 한다.

       원장이 휘발되는 것과 서비스가 안 뜨는 것은 둘 다 나쁘지만, 후자는

       무엇이 문제인지조차 알 수 없다.

    """
    want = os.environ.get("FINCHAL_DATA")
    if want:
        try:
            os.makedirs(want, exist_ok=True)
            t = os.path.join(want, ".w")
            open(t, "w").close(); os.remove(t)
            return want
        except Exception:
            pass
    return os.path.join(os.path.dirname(os.path.abspath(__file__)), "_data")


ROOT = _root()
LOCK = threading.Lock()


# ─────────────────────── 원장 미러 ───────────────────────
# 🔴 HF Space의 /data 는 영속 스토리지를 붙이지 않으면 재시작에 사라진다.
#    실측했다 — 재시작 후 부팅 표식이 없어졌다(boots 1 -> 1).
#    유료 스토리지 대신 비공개 데이터셋 저장소에 원장을 미러링한다.
#    시즌 4개월치 기록이 컨테이너 재시작 한 번에 날아가면 대회가 끝난다.
MIRROR = os.environ.get("FINCHAL_MIRROR")        # 예: FINAL-Bench/finchal-ledger
_MIR = {"ok": False, "dirty": False, "at": 0.0, "n": 0, "err": None}


def _api():
    from huggingface_hub import HfApi
    tok = os.environ.get("HF_TOKEN") or os.environ.get("FINCHAL_HF_TOKEN")
    return HfApi(token=tok) if tok else None


def restore():
    """부팅 시 미러에서 원장을 되살린다. 이게 실패하면 미러에 쓰지 않는다.



    🔑 복원 실패 후 flush 하면 빈 원장이 미러를 덮어 시즌이 통째로 사라진다.

       한 번의 잘못된 덮어쓰기가 4개월을 지운다.

    """
    if not MIRROR:
        return "미러 없음"
    try:
        from huggingface_hub import snapshot_download
        a = _api()
        if a is None:
            _MIR["err"] = "HF_TOKEN 없음"
            return _MIR["err"]
        try:
            a.repo_info(MIRROR, repo_type="dataset")
        except Exception:
            a.create_repo(MIRROR, repo_type="dataset", private=True, exist_ok=True)
        # 🔴 기본 캐시로 받으면 경로에 .cache 가 들어간다. 그걸 걸러내는
        #    필터를 두면 **받은 파일이 전부 걸러진다** — 실제로 그렇게 되어
        #    "복원 0개"가 나왔다. 목적지를 우리가 정해 그 함정을 없앤다.
        tmp = os.path.join(ROOT, ".restore")
        d = snapshot_download(MIRROR, repo_type="dataset", token=a.token,
                              local_dir=tmp)
        import shutil, glob
        os.makedirs(os.path.join(ROOT, "positions"), exist_ok=True)
        n = 0
        for src in glob.glob(os.path.join(d, "**", "*"), recursive=True):
            if not os.path.isfile(src):
                continue
            rel = os.path.relpath(src, d)
            if rel.startswith(".git") or rel.startswith(".huggingface"):
                continue
            dst = os.path.join(ROOT, rel)
            os.makedirs(os.path.dirname(dst), exist_ok=True)
            shutil.copyfile(src, dst)
            n += 1
        shutil.rmtree(tmp, ignore_errors=True)
        _MIR["ok"] = True
        _MIR["n"] = n
        return "복원 %d개" % n
    except Exception as e:
        _MIR["err"] = "%s: %s" % (type(e).__name__, e)
        return _MIR["err"]


def flush(force=False):
    """원장을 미러에 올린다. 제출마다 올리면 커밋이 폭주하므로 모아서 보낸다."""
    if not (MIRROR and _MIR["ok"]) or (not _MIR["dirty"] and not force):
        return False
    try:
        a = _api()
        a.upload_folder(repo_id=MIRROR, repo_type="dataset", folder_path=ROOT,
                        commit_message="ledger sync",
                        ignore_patterns=["*.tmp", ".cache/*", ".restore/*",
                                         ".huggingface/*"])
        _MIR["dirty"] = False
        _MIR["at"] = time.time()
        return True
    except Exception as e:
        _MIR["err"] = "%s: %s" % (type(e).__name__, e)
        return False


def mirror_status():
    return {"repo": MIRROR, "restored": _MIR["ok"], "files": _MIR["n"],
            "pending": _MIR["dirty"], "error": _MIR["err"]}


def boot_check():
    """원장이 재시작을 견디는지 **실측**한다.



    🔴 '쓰기가 되는가'로 판정하면 안 된다. 영속 스토리지가 없어도 /data 는

       만들어지고 써진다 — 재시작할 때 사라질 뿐이다. 그래서 쓰기 검사는

       항상 통과하는 가짜 신호다. 실제로 확인하려면 이전 부팅의 흔적이

       남아 있는지를 봐야 한다.



    반환: (판정, 부팅횟수)

      "확인됨"   이전 부팅 기록이 살아남았다 = 영속

      "미확인"   첫 부팅이라 아직 알 수 없다

      "휘발됨"   기록을 남겼는데 사라졌다 = 재시작마다 원장이 날아간다

    """
    f = os.path.join(ROOT, "_boot.json")
    prev = _read(f, None)
    n = (prev or {}).get("n", 0) + 1
    if prev is None:
        # 컨테이너는 시작됐는데 흔적이 없다. 첫 부팅인지 날아간 건지는
        # 이 시점에 구분되지 않는다. 판정은 다음 부팅으로 미룬다.
        verdict = "미확인"
    else:
        verdict = "확인됨"
    try:
        _write(f, {"n": n, "at": now().isoformat(), "root": ROOT})
    except Exception:
        pass
    return verdict, n


def _p(*a):
    p = os.path.join(ROOT, *a)
    os.makedirs(os.path.dirname(p), exist_ok=True)
    return p


def now():
    return datetime.now(timezone.utc).replace(microsecond=0)


def _read(path, default):
    try:
        with open(path, encoding="utf-8") as f:
            return json.load(f)
    except Exception:
        return default


def _write(path, obj):
    tmp = path + ".tmp"
    with open(tmp, "w", encoding="utf-8") as f:
        json.dump(obj, f, ensure_ascii=False)
    os.replace(tmp, path)          # 원자적 교체 — 중간에 죽어도 반쪽 파일이 안 남는다


# ─────────────────────── 참가자 ───────────────────────
def make_key(user_id):
    raw = "%s|%s|finchal" % (user_id, os.environ.get("FINCHAL_SALT", "s1"))
    return "fc_" + hashlib.sha256(raw.encode()).hexdigest()[:24]


def register(user_id, strategy, ptype, assets):
    """참가 등록. 이미 있으면 전략명·유형만 갱신하고 키는 유지한다."""
    with LOCK:
        u = _read(_p("users.json"), {})
        key = make_key(user_id)
        prev = u.get(user_id, {})
        u[user_id] = {
            "user": user_id, "key": key, "strategy": strategy, "type": ptype,
            "assets": sorted(set(assets)),
            "joined": prev.get("joined", now().isoformat()),
            "visits": prev.get("visits", []),
        }
        _write(_p("users.json"), u)
    _MIR["dirty"] = True
    return u[user_id]


def user_by_key(key):
    for v in _read(_p("users.json"), {}).values():
        if v.get("key") == key:
            return v
    return None


def touch(user_id):
    """접속 기록 — 사람 부문의 '주 1회 접속' 요건 판정에 쓴다."""
    with LOCK:
        u = _read(_p("users.json"), {})
        if user_id in u:
            v = u[user_id].setdefault("visits", [])
            d = now().date().isoformat()
            if not v or v[-1] != d:
                v.append(d)
                v[:] = v[-400:]
                _write(_p("users.json"), u)


def all_users():
    return list(_read(_p("users.json"), {}).values())


# ─────────────────────── 포지션 ───────────────────────
def submit(user_id, asset, position, source="api"):
    """포지션 제출. 덮어쓰지 않고 줄 단위로 쌓는다."""
    pos = max(-1.0, min(1.0, float(position)))
    rec = {"t": now().isoformat(), "u": user_id, "a": asset,
           "p": pos, "src": source}
    with LOCK:
        with open(_p("positions", "%s.jsonl" % asset), "a", encoding="utf-8") as f:
            f.write(json.dumps(rec, ensure_ascii=False) + "\n")
    _MIR["dirty"] = True
    touch(user_id)
    return rec


def seed(user_id, asset, position, at, source="seed"):
    """지정한 시각으로 포지션을 기록한다. 기준선을 시즌 개막 시점에 심을 때만 쓴다.



    🔴 참가자에게는 절대 열어 주지 않는다. 과거 시각으로 포지션을 넣을 수 있으면

       결과를 보고 나서 유리한 시점에 진입한 것으로 꾸밀 수 있다.

    """
    pos = max(-1.0, min(1.0, float(position)))
    rec = {"t": str(at), "u": user_id, "a": asset, "p": pos, "src": source}
    with LOCK:
        with open(_p("positions", "%s.jsonl" % asset), "a", encoding="utf-8") as f:
            f.write(json.dumps(rec, ensure_ascii=False) + "\n")
    _MIR["dirty"] = True
    return rec


def positions(asset, user_id=None):
    out = []
    p = _p("positions", "%s.jsonl" % asset)
    if not os.path.exists(p):
        return out
    with open(p, encoding="utf-8") as f:
        for line in f:
            try:
                r = json.loads(line)
            except Exception:
                continue
            if user_id is None or r.get("u") == user_id:
                out.append(r)
    return out


def latest(asset, user_id):
    ps = positions(asset, user_id)
    return ps[-1] if ps else None


# ─────────────────────── 점수 스냅샷 ───────────────────────
def save_scores(asset, rows):
    _write(_p("scores", "%s.json" % asset),
           {"at": now().isoformat(), "rows": rows})


def load_scores(asset):
    return _read(_p("scores", "%s.json" % asset), {"at": None, "rows": []})