# -*- coding: utf-8 -*-
"""LEADBOARD — 신약 예측 도구를 분야별로 같은 잣대에 세우는 시험대.
**이 서비스가 하는 일**
· 카테고리·부문 카드를 서빙한다
· 테스트셋(구조만)을 내려준다
· 제출을 받아 비공개 원장에 적는다
· 워커가 굴려 놓은 순위표를 보여준다
**하지 않는 일: 채점.**
정답은 이 컨테이너에 없다. 컨테이너 이미지는 누구나 받을 수 있으므로, 여기에 라벨을 두면
그 순간 1조가 무너진다. 채점은 원장을 폴링하는 별도 워커가 로컬 정답 파일로 한다.
부문 카드에는 **채점을 검증하는 데 필요한 모든 공개 정보**가 들어 있다 —
분할 등급 · 정답 등급 · 잡음 바닥 · 기준선 성적 · 데이터 지문.
우리 점수를 믿어달라고 하지 않기 위해서다.
"""
import base64
import glob
import hashlib
import hmac
import io
import json
import os
import re
import secrets
import time
import urllib.error
import urllib.parse
import urllib.request
from fastapi import FastAPI, HTTPException, Request
from fastapi.middleware.gzip import GZipMiddleware
from fastapi.responses import (FileResponse, JSONResponse, RedirectResponse,
Response)
from pydantic import BaseModel
HERE = os.path.dirname(os.path.abspath(__file__))
DATA = os.path.join(HERE, "data")
SPEC = "v1.1"
LEDGER = os.environ.get("LB_LEDGER_REPO", "FINAL-Bench/leadboard-submissions")
LEDGER_API = "https://huggingface.co/api/datasets/%s" % LEDGER
LEDGER_RAW = "https://huggingface.co/datasets/%s/resolve/main" % LEDGER
HF_TOKEN = os.environ.get("HF_TOKEN", "")
OAUTH_ID = os.environ.get("OAUTH_CLIENT_ID", "")
OAUTH_SECRET = os.environ.get("OAUTH_CLIENT_SECRET", "")
OAUTH_ISS = os.environ.get("OPENID_PROVIDER_URL", "https://huggingface.co")
SPACE_HOST = os.environ.get("SPACE_HOST", "")
COOKIE = "lb_session"
IN_FRAME = bool(SPACE_HOST)
COOKIE_KW = ({"samesite": "none", "secure": True} if IN_FRAME
else {"samesite": "lax", "secure": False})
SESSION_KEY = os.environ.get("LB_SESSION_KEY") or secrets.token_hex(16)
DAILY_CAP = int(os.environ.get("LB_DAILY_CAP", "5"))
app = FastAPI(title="LEADBOARD")
app.add_middleware(GZipMiddleware, minimum_size=1024)
_C = {}
# 하루 한도 카운터. 컨테이너 재기동이면 비는데, 그래도 폭주는 막는다.
# 정확한 회계가 필요해지면 원장 쪽으로 옮긴다.
_CAP = {}
def cached(key, ttl, produce):
hit = _C.get(key)
if hit and time.time() - hit[0] < ttl:
return hit[1]
try:
v = produce()
except Exception:
if hit:
return hit[1]
raise
_C[key] = (time.time(), v)
return v
# ------------------------------------------------------------------ 부문 카드
def _sha(path):
h = hashlib.sha256()
with open(path, "rb") as f:
for b in iter(lambda: f.read(65536), b""):
h.update(b)
return h.hexdigest()[:16]
def _board_meta():
"""부문 한 줄 설명. 이름과 표적 기호만으로는 무엇을 재는지 알 수 없다."""
p = os.path.join(HERE, "board_meta.json")
try:
return json.load(io.open(p, encoding="utf-8")).get("boards", {})
except Exception:
return {}
def _load_boards():
meta = _board_meta()
out = {}
for p in sorted(glob.glob(os.path.join(DATA, "*_card.json"))):
try:
d = json.load(io.open(p, encoding="utf-8"))
except Exception:
continue
nf = d.get("noise_floor") or {}
base = d.get("baselines") or {}
# 분류 전용 부문에는 MAE 기준선이 없다. 그럴 때는 AUROC 가 가장 높은 것을 최선으로 본다.
# 없는 값을 0 으로 치면 그 부문이 "가장 정확한 곳"으로 표에 오른다.
with_mae = {k: v for k, v in base.items() if v.get("mae") is not None}
if with_mae:
best = min(with_mae, key=lambda k: with_mae[k]["mae"])
d["metric"] = "mae"
elif base:
best = max(base, key=lambda k: base[k].get("auroc") or 0)
d["metric"] = "auroc"
else:
best = None
d["metric"] = None
d["dataset_sha"] = _sha(p)
d["spec"] = SPEC
d["best_baseline"] = best
# 모델 오차가 실험 오차의 몇 배인가. 1 에 붙을수록 측정 한계다.
if best and nf.get("sd_single") and d["metric"] == "mae":
d["error_over_noise"] = round(base[best]["mae"] / nf["sd_single"], 2)
if ("상수 예측" in base and "Morgan+LightGBM" in base
and base["Morgan+LightGBM"].get("mae") is not None
and base["상수 예측"].get("mae") is not None):
d["beats_constant"] = base["Morgan+LightGBM"]["mae"] < base["상수 예측"]["mae"]
if d.get("n_test"):
d["near_pct"] = round(100.0 * d.get("near_threshold", 0) / d["n_test"], 1)
m = meta.get(d["board"]) or {}
d["blurb"] = m.get("ko", "")
d["blurb_en"] = m.get("en", "")
d["open"] = True
out[d["board"]] = d
return out
def boards():
return cached("boards", 300, _load_boards)
def _load_references():
"""현업 도구를 이 부문 훈련 자료로 재학습해 얻은 성적.
다른 자료로 학습한 모델을 그대로 옮겨 재면, 훈련 자료의 차이가 방법의 차이로 읽힌다.
그래서 **방법만 가져오고 데이터는 이 부문 것을 쓴다.** 그래야 표에 오른 숫자가
"이 방법이 이 문제에서 어디까지 가는가"를 뜻한다.
"""
# 참조 방법은 여럿일 수 있다. 파일 하나에 방법 하나를 담고, 여기서 다 모은다.
out = {}
for p in sorted(glob.glob(os.path.join(DATA, "reference_*.json"))):
try:
d = json.load(io.open(p, encoding="utf-8"))
except Exception:
continue
for e in d.get("entries", []):
out.setdefault(e["board"], []).append(
dict(e, method=d.get("method", "reference"),
method_en=d.get("method_en") or d.get("method", "reference"),
ref_note=d.get("note"), ref_note_en=d.get("note_en")))
return out
def references():
return cached("refs", 300, _load_references)
def categories():
def build():
cats = json.load(io.open(os.path.join(HERE, "categories.json"), encoding="utf-8"))
bd = boards()
for c in cats["categories"]:
c["open"] = len([b for b in c["boards"] if b in bd])
c["board_cards"] = [bd[b] for b in c["boards"] if b in bd]
return cats
return cached("cats", 300, build)
# ------------------------------------------------------------------ 원장
def _hdr():
return {"Authorization": "Bearer " + HF_TOKEN, "User-Agent": "LEADBOARD/1.0"}
def ledger_read(path, default=None):
try:
with urllib.request.urlopen(urllib.request.Request(
"%s/%s" % (LEDGER_RAW, path), headers=_hdr()), timeout=60) as r:
return json.loads(r.read())
except Exception:
return default
def ledger_write(path, obj, summary):
blob = base64.b64encode(json.dumps(obj, ensure_ascii=False).encode()).decode()
lines = [json.dumps({"key": "header", "value": {"summary": summary}}),
json.dumps({"key": "file", "value": {"path": path, "content": blob,
"encoding": "base64"}})]
req = urllib.request.Request(LEDGER_API + "/commit/main",
data=("\n".join(lines) + "\n").encode(),
headers=dict(_hdr(), **{"Content-Type": "application/x-ndjson"}))
with urllib.request.urlopen(req, timeout=180) as r:
return json.loads(r.read())
# ------------------------------------------------------------------ 세션
def sign(v):
return hmac.new(SESSION_KEY.encode(), v.encode(), hashlib.sha256).hexdigest()[:32]
def set_session(resp, user):
raw = json.dumps(user, ensure_ascii=False)
b = base64.urlsafe_b64encode(raw.encode()).decode()
resp.set_cookie(COOKIE, "%s.%s" % (b, sign(b)), max_age=86400 * 7,
httponly=True, **COOKIE_KW)
def who(req: Request):
c = req.cookies.get(COOKIE) or ""
if "." not in c:
return None
b, sg = c.rsplit(".", 1)
if not hmac.compare_digest(sg, sign(b)):
return None
try:
return json.loads(base64.urlsafe_b64decode(b.encode()).decode())
except Exception:
return None
# ------------------------------------------------------------------ 라우트
@app.get("/")
def index():
return FileResponse(os.path.join(HERE, "index.html"))
@app.get("/i18n.js")
def i18n():
"""문자열 사전. 화면 코드와 분리해 두어야 영어판이 조용히 뒤처지지 않는다."""
return FileResponse(os.path.join(HERE, "i18n.js"),
media_type="application/javascript")
@app.get("/api/categories")
def api_categories():
c = categories()
bd = boards()
return {"spec": SPEC, "categories": c["categories"],
"totals": {"planned": sum(x["planned"] for x in c["categories"]),
"open": len(bd), "ledger": bool(HF_TOKEN)}}
@app.get("/api/leaders")
def api_leaders():
"""부문마다 현재 1위 한 줄. 첫 화면에서 전체를 한눈에 보기 위한 것이다.
부문별로 순위표를 따로 부르면 왕복이 부문 수만큼 늘어난다. 여기서 한 번에 모은다.
**참가 제출이 없으면 비워서 보낸다** - 기준선을 1위 자리에 앉히지 않는다.
기준선은 넘어야 할 선이지 우승자가 아니다.
"""
def build():
out = []
for name, b in boards().items():
rolled = ledger_read("leaderboard/%s.json" % name.lower(), {}) or {}
ent = (rolled.get("entries") or [])
top = ent[0] if ent else None
nf = (b.get("noise_floor") or {}).get("sd_single")
base = b.get("baselines") or {}
row = {"board": name, "n_test": b.get("n_test"),
"blurb": b.get("blurb"), "blurb_en": b.get("blurb_en"),
"split_grade": b.get("split_grade"), "answer_grade": b.get("answer_grade"),
"noise_floor": nf, "entries": len(ent),
"baseline_best": (min((v["mae"] for v in base.values() if v.get("mae") is not None),
default=None)),
"baseline_best_auroc": (max((v["auroc"] for v in base.values()
if v.get("auroc") is not None), default=None)),
"leader": None}
if top:
row["leader"] = {"method": top.get("method"), "user": top.get("user"),
"mae": top.get("mae"), "auroc": top.get("auroc"),
"verified": bool(top.get("verified")), "leak": top.get("leak")}
out.append(row)
out.sort(key=lambda r: (r["leader"] is None, -(r["n_test"] or 0)))
return out
rows = cached("leaders", 60, build)
return {"spec": SPEC, "n": len(rows), "rows": rows,
"held": sum(1 for r in rows if r["leader"])}
@app.get("/api/board/{name}")
def api_board(name: str):
b = boards().get(name)
if not b:
raise HTTPException(404, "그런 부문이 없다")
return b
@app.get("/api/board/{name}/testset")
def api_testset(name: str):
"""테스트셋. 구조만 나간다 - 라벨은 이 컨테이너에 존재하지 않는다."""
p = os.path.join(DATA, "%s_test.json" % name.lower())
if not os.path.exists(p):
raise HTTPException(404, "테스트셋이 아직 없다")
return FileResponse(p, media_type="application/json",
filename="%s_test.json" % name.lower())
@app.get("/api/board/{name}/leaderboard")
def api_leaderboard(name: str):
"""순위표. 기준선은 항상 포함된다 (3조).
참가자 항목은 워커가 굴려 놓은 것을 그대로 보여준다. 여기서 계산하지 않는다 -
계산하려면 정답이 있어야 하고, 정답은 여기 없다.
"""
b = boards().get(name)
if not b:
raise HTTPException(404, "그런 부문이 없다")
rolled = cached("lb:" + name, 60,
lambda: ledger_read("leaderboard/%s.json" % name.lower(),
{"entries": [], "updated": None})) or {}
rows = []
for k, v in (b.get("baselines") or {}).items():
rows.append({"method": k, "user": "—", "kind": "baseline",
"mae": v["mae"], "auroc": v["auroc"], "prauc": v["prauc"]})
# 현업 도구 참조 항목. 기준선과 참가 제출 사이에 놓는다 - 학습하지 않은 선도 아니고
# 이번 회차의 참가자도 아니다. 참가자가 자기 위치를 가늠할 세 번째 좌표다.
for r in references().get(name, []):
rows.append({"method": r["method"], "method_en": r.get("method_en"),
"user": "—", "kind": "reference",
"mae": r.get("mae"), "auroc": r.get("auroc"),
"prauc": r.get("prauc"), "leak": r.get("leak"),
"note": r.get("ref_note"), "note_en": r.get("ref_note_en")})
for e in rolled.get("entries", []):
rows.append(dict(e, kind=e.get("kind", "entry")))
# 부문 주지표로 세운다. 분류 전용 부문에서 mae 로 세우면 전부 동률이 된다.
if b.get("metric") == "auroc":
rows.sort(key=lambda r: -(r.get("auroc") or 0))
else:
rows.sort(key=lambda r: (r.get("mae") is None, r.get("mae") or 9e9))
nf = (b.get("noise_floor") or {}).get("sd_single")
best = rows[0].get("mae") if rows else None
# 4조: 최고점에서 잡음 바닥 안에 든 항목은 같은 계단으로 묶는다.
for r in rows:
r["within_noise"] = bool(nf and best is not None and r.get("mae") is not None
and r["mae"] - best < nf)
rank = 0
for r in rows:
# 순위는 참가 제출에만 매긴다. 기준선과 참조 도구는 표에 서되 등수를 갖지 않는다 -
# 우리가 올린 것이 1위 자리를 차지하면 참가자에게 겨룰 자리가 없다.
if r["kind"] in ("baseline", "reference"):
r["rank"] = None
else:
rank += 1
r["rank"] = rank
return {"board": name, "noise_floor": nf, "rows": rows,
"updated": rolled.get("updated"),
"note": "잡음 바닥 안에 든 항목은 순위 차이로 주장하지 않는다 (4조)"}
# ------------------------------------------------------------------ 로그인
@app.get("/login")
def login(request: Request):
if not (OAUTH_ID and OAUTH_SECRET):
return _err_page("이 시험대에 로그인이 아직 구성되지 않았습니다.")
nxt = request.query_params.get("next", "/")
st = base64.urlsafe_b64encode(json.dumps({"n": nxt, "r": secrets.token_hex(8)}).encode()).decode()
q = urllib.parse.urlencode({
"client_id": OAUTH_ID, "redirect_uri": _redirect(request),
"response_type": "code", "scope": "openid profile", "state": st})
return RedirectResponse("%s/oauth/authorize?%s" % (OAUTH_ISS, q))
def _redirect(request: Request):
"""플랫폼이 등록해 주는 콜백 주소는 **/auth/callback** 이다.
여기를 /auth 로 두면 토큰 교환에서 redirect_uri 불일치로 거부되고, 그 예외가
그대로 500 이 되어 화면 전체가 죽는다. 로그인 한 번 눌렀다가 사이트가 사라진다.
"""
if SPACE_HOST:
return "https://%s/auth/callback" % SPACE_HOST
return str(request.base_url).rstrip("/") + "/auth/callback"
def _err_page(msg, detail=""):
"""로그인이 실패해도 화면은 살아 있어야 한다. 흰 배경에 Internal Server Error 만
남으면 이용자는 사이트가 죽은 줄 안다."""
return Response(
""
"로그인 실패"
"로그인을 마치지 못했습니다
%s
%s"
"← 시험대로 돌아가기
"
% (msg, ("%s
" % detail[:300]) if detail else ""),
media_type="text/html; charset=utf-8", status_code=200)
@app.get("/auth/callback")
def auth(request: Request):
code = request.query_params.get("code")
st = request.query_params.get("state") or ""
nxt = "/"
try:
nxt = json.loads(base64.urlsafe_b64decode(st.encode()).decode()).get("n", "/")
except Exception:
pass
if not code:
return RedirectResponse(nxt)
try:
body = urllib.parse.urlencode({
"client_id": OAUTH_ID, "client_secret": OAUTH_SECRET,
"grant_type": "authorization_code", "code": code,
"redirect_uri": _redirect(request)}).encode()
with urllib.request.urlopen(urllib.request.Request(
OAUTH_ISS + "/oauth/token", data=body,
headers={"Content-Type": "application/x-www-form-urlencoded"}),
timeout=60) as r:
tok = json.loads(r.read())
with urllib.request.urlopen(urllib.request.Request(
OAUTH_ISS + "/oauth/userinfo",
headers={"Authorization": "Bearer " + tok["access_token"]}),
timeout=60) as r:
ui = json.loads(r.read())
except urllib.error.HTTPError as e:
return _err_page("인증 제공자가 요청을 거부했습니다. 다시 시도해 주십시오.",
"%d %s" % (e.code, e.read().decode("utf-8", "replace")))
except Exception as e:
return _err_page("인증 중 통신에 실패했습니다. 잠시 후 다시 시도해 주십시오.",
"%s: %s" % (type(e).__name__, e))
resp = RedirectResponse(nxt)
set_session(resp, {"user": ui.get("preferred_username") or ui.get("sub"),
"name": ui.get("name", ""), "pic": ui.get("picture", "")})
return resp
@app.get("/auth")
def auth_legacy(request: Request):
"""예전 주소로 들어온 콜백도 받아 준다. 링크가 어딘가 남아 있을 수 있다."""
return auth(request)
@app.get("/logout")
def logout():
r = RedirectResponse("/")
r.delete_cookie(COOKIE, **COOKIE_KW)
return r
@app.get("/api/me")
def api_me(request: Request):
u = who(request)
return {"user": u, "login_enabled": bool(OAUTH_ID), "daily_cap": DAILY_CAP}
# ------------------------------------------------------------------ 제출
class Submission(BaseModel):
board: str
method: str # 방법 이름. 순위표에 이렇게 표시된다
predictions: dict # {compound_id: 예측값}
training_data: str = "" # 5조 누출검사에 쓴다
pretrained_on: str = ""
code_url: str = ""
container: str = "" # 있으면 [검증됨] 심사 대상 (6조)
@app.post("/api/submit")
def submit(s: Submission, request: Request):
u = who(request)
if not u:
raise HTTPException(401, "제출하려면 로그인해야 한다")
if not HF_TOKEN:
raise HTTPException(503, "원장이 구성되지 않았다")
b = boards().get(s.board)
if not b:
raise HTTPException(404, "그런 부문이 없다")
if not re.fullmatch(r"[\w .\-+/()]{2,60}", s.method or ""):
raise HTTPException(400, "방법 이름은 2~60자여야 한다")
# 하루 한도. 8조가 점수 공개를 막아도 제출 자체는 비용이 들고, 무제한이면
# 원장이 잠긴다. 경계는 참가자가 가정할 하루(KST)로 잡는다.
day = int((time.time() + 9 * 3600) // 86400)
key = "cap:%s:%d" % (u["user"], day)
used = _CAP.get(key, 0)
if used >= DAILY_CAP:
raise HTTPException(429, "오늘 제출 한도 %d회를 다 썼다. 한국시간 자정에 초기화된다"
% DAILY_CAP)
n = b.get("n_test", 0)
got = len(s.predictions or {})
if got < n:
raise HTTPException(400, "예측이 %d개 필요한데 %d개다. 빈 항목은 채워 보내라 "
"- 임의로 메우면 그 값이 점수에 들어간다" % (n, got))
# **모양도 본다.** 축이 여럿인 부문에 숫자 하나를 보내면 채점기는 전부 missing 으로
# 처리하고 점수가 비어서 돌아온다 - 참가자는 무엇이 틀렸는지 알 길이 없다.
# 개수만 세고 통과시키면 그 침묵이 우리 몫이 된다.
ax = b.get("axes")
if ax:
bad = [k for k, v in (s.predictions or {}).items()
if not (isinstance(v, (list, tuple)) and len(v) == ax)]
if bad:
raise HTTPException(400, "이 부문은 화합물마다 숫자 %d개짜리 배열이 필요하다. "
"%d건이 그 모양이 아니다 (예: %s). 형식은 부문 안내를 보라"
% (ax, len(bad), bad[0]))
else:
bad = [k for k, v in (s.predictions or {}).items()
if isinstance(v, (list, tuple))]
if bad:
raise HTTPException(400, "이 부문은 화합물마다 숫자 하나가 필요한데 배열이 왔다 "
"(%d건, 예: %s)" % (len(bad), bad[0]))
sid = hashlib.sha256(("%s|%s|%s|%f" % (s.board, u["user"], s.method, time.time()))
.encode()).hexdigest()[:16].upper()
rec = {"submission_id": sid, "board": s.board, "hf_user": u["user"],
"method": s.method, "predictions": s.predictions,
"training_data": s.training_data, "pretrained_on": s.pretrained_on,
"code_url": s.code_url, "container": s.container,
"submitted_at": int(time.time()), "spec": SPEC,
"dataset_sha": b.get("dataset_sha")}
ledger_write("submissions/%s/%s.json" % (s.board.lower(), sid), rec,
"submit %s %s" % (s.board, sid))
_CAP[key] = used + 1
return {"ok": True, "submission_id": sid,
"remaining_today": DAILY_CAP - _CAP[key],
"note": "채점은 워커가 한다. 8조에 따라 직전 최고점을 잡음 바닥 이상으로 "
"넘었을 때만 새 점수가 공개된다."}
@app.get("/api/health")
def health():
return {"ok": True, "spec": SPEC, "boards": len(boards()),
"ledger": bool(HF_TOKEN), "login": bool(OAUTH_ID),
"scoring_here": False}