app.py

# -*- coding: utf-8 -*-
"""Open Materials Challenge — 접수와 순위표.

**이 서비스가 하지 않는 일**: 채점. 스페이스에는 Materials Project 자료도 GPU 도 없다.
제출을 받아 원장(비공개 데이터셋)에 적고, 워커가 채점해 넣은 결과를 보여줄 뿐이다.
여기서 물성을 판정하는 척하면 참가자는 근거 없는 반려를 받게 된다.

순위표는 TTL 당 한 번만 만들어 미리 압축해 두고, 캐시는 단일 갱신자로 돌린다.
첫 페이지는 async 로 서빙하고, 표는 500 행씩 나눠 그린다.
"""
import base64
import gzip
import hashlib
import hmac
import json
import os
import re
import secrets
import time
import urllib.parse
import urllib.request
import uuid

from fastapi import FastAPI, HTTPException, Request
from fastapi.middleware.gzip import GZipMiddleware
from fastapi.responses import (FileResponse, JSONResponse, RedirectResponse,
                               Response)
from pydantic import BaseModel

import gates
import seasons
import store

HERE = os.path.dirname(os.path.abspath(__file__))
DATA = os.path.join(HERE, "data")
# 페이지 폴링 주기(20초)보다 크게 잡는다.
CACHE = store.Cached(ttl=int(os.environ.get("OMC_CACHE_TTL", "60")))
SALT = os.environ.get("OMC_SALT", "omc-season1")

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 = "omc_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("OMC_SESSION_KEY") or secrets.token_hex(16)

DAILY_CAP = int(os.environ.get("OMC_DAILY_CAP", "30"))
CAP_TZ_OFFSET = 9 * 3600          # 참가자가 가정할 하루 경계 (KST)

app = FastAPI(title="Open Materials Challenge")
app.add_middleware(GZipMiddleware, minimum_size=1024)


class Submission(BaseModel):
    formula: str                  # 조성. 예: Li3YCl6, LiZr2(PO4)3
    display_name: str             # 순위표에 표시될 이름
    model_name: str = ""          # 어떤 모델이 제안했는지. 자유 기재
    rationale: str = ""           # 설계 근거 (선택)
    structure: str = ""           # CIF (선택) - 내면 채점이 빠르고 정확해진다
    season: int = 0
    # 조성을 공개할지는 제출자가 정한다. 되돌릴 수 없는 쪽이 기본값이어야 한다 -
    # 한 번 공개된 조성은 되감을 수 없고, 공개는 특허성을 깎는다.
    visibility: str = "private"


# ---------------------------------------------------------------- 원장 접근
def _submissions(s):
    return CACHE.get("ids:%d" % s["number"],
                     lambda: store.listdir(seasons.path(s, "submissions")))


def _board(s):
    return CACHE.get("board:%d" % s["number"],
                     lambda: store.read(seasons.path(s, "leaderboard.json"),
                                        default={}) or {})


def public_id(key):
    return "OMC-" + hashlib.sha256((SALT + key).encode()).hexdigest()[:10].upper()


# 공개 표에 나가도 되는 것만 적는다.
# **차단목록이 아니라 허용목록이다.** 기록에 새 필드가 늘어도 여기 적지 않는 한
# 밖으로 나가지 않는다. 반대로 하면 필드가 늘 때마다 누군가 지우는 것을 기억해야 하고,
# 한 번 잊으면 그대로 새어 나간다.
_SAFE_KEYS = (
    "candidate_id", "display_name", "hf_user", "model_name", "visibility",
    "submitted_at", "season", "scored_at", "status", "gate", "hold_kind",
    "axes", "detail", "notes", "note", "total", "total_max", "held_axes",
    "known_in_mp", "is_anchor", "tier", "rank", "div_rank", "division",
    "track", "track_rank",
)

# 공개를 고른 제출에만 함께 내보낸다. 비공개면 이 목록은 통째로 빠진다.
_PUBLIC_ONLY = ("composition", "measured", "n_atoms", "structure", "rationale")

# 두 번째 그물. 허용목록이 이미 막고 있지만, 나중에 누가 다른 경로로 항목을 만들어도
# 응답 직전에 한 번 더 걷어낸다. 한 겹이 뚫려도 다른 한 겹이 남는다.
_MASK_DROP = _PUBLIC_ONLY


def mask(rec):
    """공개 표가 볼 수 있는 것만 남긴다.

    비공개 제출에서는 **원소 종류까지만** 보여준다. 어떤 화학계가 상위인지는 대회의
    공개 지식이어야 하지만, 조성비는 제출자의 것이다.

    조성을 그대로 남기면 안 되는 것은 물론이고, 계산값(생성에너지 등)도 남기면 안 된다 -
    원소 집합과 함께 두면 후보 조성을 훑어 맞출 수 있어 사실상 조성을 알려주는 것과 같다.
    """
    pub = rec.get("visibility") == "public"
    comp = rec.get("composition") or {}
    out = {k: rec.get(k) for k in _SAFE_KEYS if k in rec}
    # 구역은 조성에서 뽑아 두고, 조성 자체는 내보내지 않는다.
    out["division"] = division_of(comp)
    if pub:
        out["formula"] = rec.get("formula")
        for k in _PUBLIC_ONLY:
            if k in rec:
                out[k] = rec[k]
        out["masked"] = False
    else:
        out["formula"] = "·".join(sorted(comp)) if comp else "비공개"
        out["masked"] = True
    return out


def strip_masked(entries):
    """마스킹된 항목에서 조성을 되돌릴 수 있는 필드를 전부 없앤다."""
    for e in entries:
        if e.get("masked"):
            for k in _MASK_DROP:
                e.pop(k, None)
    return entries


# ---------------------------------------------------------------- 세션
def _sign(payload):
    raw = base64.urlsafe_b64encode(json.dumps(payload).encode()).decode().rstrip("=")
    sig = hmac.new(SESSION_KEY.encode(), raw.encode(), hashlib.sha256).hexdigest()[:32]
    return raw + "." + sig


def _unsign(token):
    try:
        raw, sig = (token or "").rsplit(".", 1)
    except ValueError:
        return None
    good = hmac.new(SESSION_KEY.encode(), raw.encode(), hashlib.sha256).hexdigest()[:32]
    if not hmac.compare_digest(sig, good):
        return None
    pad = "=" * (-len(raw) % 4)
    try:
        return json.loads(base64.urlsafe_b64decode(raw + pad))
    except Exception:
        return None


def current_user(request: Request):
    return _unsign(request.cookies.get(COOKIE))


def _redirect_uri(request: Request):
    if SPACE_HOST:
        return "https://%s/auth/callback" % SPACE_HOST
    return str(request.base_url).rstrip("/") + "/auth/callback"


# ---------------------------------------------------------------- 화면
@app.get("/")
async def index():
    # 정적 페이지는 요청 스레드풀을 타지 않게 async 로 둔다.
    return FileResponse(os.path.join(HERE, "index.html"))


@app.get("/api/me")
def me(request: Request):
    u = current_user(request)
    return {"signed_in": bool(u), "user": u,
            "oauth_configured": bool(OAUTH_ID and OAUTH_SECRET)}


@app.get("/login")
def login(request: Request):
    if not (OAUTH_ID and OAUTH_SECRET):
        raise HTTPException(503, "OAuth 가 설정되지 않았습니다.")
    state = secrets.token_urlsafe(16)
    q = urllib.parse.urlencode({
        "client_id": OAUTH_ID, "redirect_uri": _redirect_uri(request),
        "response_type": "code", "scope": "openid profile", "state": state})
    r = RedirectResponse("%s/oauth/authorize?%s" % (OAUTH_ISS, q))
    r.set_cookie("omc_state", state, max_age=600, httponly=True, **COOKIE_KW)
    return r


@app.get("/auth/callback")
def callback(request: Request, code: str = "", state: str = ""):
    if not code or state != request.cookies.get("omc_state"):
        return RedirectResponse("/?auth=failed")
    body = urllib.parse.urlencode({
        "client_id": OAUTH_ID, "client_secret": OAUTH_SECRET,
        "grant_type": "authorization_code", "code": code,
        "redirect_uri": _redirect_uri(request)}).encode()
    try:
        req = urllib.request.Request("%s/oauth/token" % OAUTH_ISS, data=body)
        with urllib.request.urlopen(req, timeout=30) as x:
            tok = json.loads(x.read())
        req = urllib.request.Request(
            "%s/oauth/userinfo" % OAUTH_ISS,
            headers={"Authorization": "Bearer " + tok["access_token"]})
        with urllib.request.urlopen(req, timeout=30) as x:
            info = json.loads(x.read())
    except Exception:
        return RedirectResponse("/?auth=failed")
    u = {"sub": info.get("sub"), "name": info.get("preferred_username") or info.get("name"),
         "picture": info.get("picture")}
    r = RedirectResponse("/?auth=ok")
    r.set_cookie(COOKIE, _sign(u), max_age=30 * 86400, httponly=True, **COOKIE_KW)
    r.delete_cookie("omc_state")
    return r


@app.get("/logout")
def logout():
    r = RedirectResponse("/")
    r.delete_cookie(COOKIE)
    return r


@app.get("/api/seasons")
def season_list():
    return {"seasons": seasons.all_public()}


@app.get("/api/season")
def season(request: Request):
    return seasons.public(seasons.get(request.query_params.get("season")))


# ---------------------------------------------------------------- 제출
@app.post("/api/submit")
def submit(s: Submission, request: Request):
    season = seasons.get(s.season or None)
    if not season.get("open"):
        raise HTTPException(400, "이 시즌은 제출을 받지 않습니다.")
    user = current_user(request)
    if (OAUTH_ID and OAUTH_SECRET) and not user:
        raise HTTPException(401, "제출하려면 Hugging Face 로그인이 필요합니다.")

    g = seasons.gate(season)
    ok, why, info = gates.check(s.formula, cif=s.structure or None,
                                require_elements=g.get("require_elements", ("Li",)),
                                max_atoms=g.get("max_atoms", 60))
    if not ok:
        raise HTTPException(400, why)

    comp = info["composition"]
    # 열쇠는 **기약 조성**으로 만든다. Li3YCl6 와 Li9Y3Cl18 은 같은 물질이다.
    # 원본 조성으로 만들면 배수만 바꿔 같은 물질을 몇 번이고 올릴 수 있다.
    red = info.get("reduced") or comp
    key = "-".join("%s%g" % (k, red[k]) for k in sorted(red))
    cid = public_id(key)

    ids = _submissions(season)
    if cid in ids:
        raise HTTPException(409, "이미 제출된 조성입니다 (%s)." % cid)

    if user:
        today = int((time.time() + CAP_TZ_OFFSET) // 86400)
        mine = CACHE.get("cap:%s:%d:%d" % (user.get("sub"), season["number"], today),
                         lambda: 0)
        if mine >= DAILY_CAP:
            raise HTTPException(429, "하루 제출 상한(%d건)에 도달했습니다." % DAILY_CAP)

    rec = {"candidate_id": cid, "formula": s.formula.strip(),
           "composition": comp, "n_atoms": info["n_atoms"],
           "display_name": (s.display_name or "").strip()[:60],
           "model_name": (s.model_name or "").strip()[:80],
           "rationale": (s.rationale or "").strip()[:1000],
           "visibility": "public" if s.visibility == "public" else "private",
           "has_structure": bool(s.structure),
           "hf_user": (user or {}).get("name", ""),
           "hf_sub": (user or {}).get("sub", ""),
           "season": season["number"],
           "submitted_at": int(time.time()),
           "uuid": uuid.uuid4().hex}
    store.write(seasons.path(season, "submissions/%s.json" % cid), rec,
                summary="submit %s" % cid)
    if s.structure:
        # 구조는 따로 둔다. 목록을 훑을 때 같이 끌려오면 순위표가 무거워진다.
        store.write(seasons.path(season, "structures/%s.json" % cid),
                    {"candidate_id": cid, "cif": s.structure[:400_000]},
                    summary="structure %s" % cid)
    CACHE.drop("ids:%d" % season["number"])

    scored = set((_board(season).get("entries") or {}).keys())
    ahead = len([i for i in _submissions(season) if i not in scored])
    return {"accepted": True, "candidate_id": cid, "queue_position": ahead,
            "note": "채점은 별도 계산 작업으로 처리되며 완료까지 시간이 걸립니다."}


# ---------------------------------------------------------------- 순위표
def _anchors(season):
    p = os.path.join(DATA, season.get("anchors") or "")
    if not os.path.exists(p):
        return []
    out = []
    for a in json.load(open(p, encoding="utf-8")):
        if not a.get("admitted"):
            continue
        out.append({"candidate_id": a["label"], "is_anchor": True,
                    "display_name": "기준물질", "model_name": "",
                    "formula": a.get("formula", ""),
                    "total": a.get("total"), "total_max": a.get("total_max"),
                    "tier": a.get("tier", 1),
                    "axes": {k: v["points"] for k, v in (a.get("axes") or {}).items()},
                    "detail": {k: v["detail"] for k, v in (a.get("axes") or {}).items()},
                    "division": _division_from_formula(a.get("formula", "")),
                    "note": a.get("note", "")})
    return out


def _division_from_formula(f):
    """기준물질에는 조성 사전이 없고 화학식 문자열만 있다.

    직접 파싱하지 않는다 - Li10Ge(PS6)2 같은 괄호 표기에서 순진한 정규식은 P1S6 을
    내놓는다. 그러면 이 물질이 황화물이 아닌 것으로 분류된다.
    """
    if not f:
        return "other"
    try:
        from pymatgen.core import Composition
        return division_of({str(el): n for el, n
                            in Composition(f).get_el_amt_dict().items()})
    except Exception:
        return "other"


# 음이온 골격이 물질의 성질을 크게 가른다. 같은 계열끼리 견주어야 후보의 위치가 읽힌다.
# 부문을 나누되 **점수는 그대로 둔다** - 어떤 제출도 무효가 되지 않는다.
_ANION_ORDER = ["F", "Cl", "Br", "I", "S", "Se", "O", "N", "P", "H"]
DIVISIONS = {"F": "fluoride", "Cl": "halide", "Br": "halide", "I": "halide",
             "S": "sulfide", "Se": "sulfide", "O": "oxide", "N": "nitride",
             "P": "oxide", "H": "hydride"}


def division_of(comp):
    """골격 음이온으로 부문을 정한다.

    섞여 있으면 **가장 많은 것**을 고른다. Li6PS5Cl 은 S 가 5, Cl 이 1 이므로 황화물이고,
    실제로 아지로다이트는 황화물로 분류된다. 개수가 같으면 위 순서에서 앞선 것을 쓴다.
    """
    if not comp:
        return "other"
    best, bestn = None, -1
    for el in _ANION_ORDER:
        n = comp.get(el, 0)
        if n > bestn:
            best, bestn = el, n
    if bestn <= 0:
        return "other"
    return DIVISIONS.get(best, "other")


def _leaderboard_body(season):
    rows = [mask(r) for r in (_board(season).get("entries") or {}).values()]
    anchors = _anchors(season)
    merged = rows + anchors
    # 점수가 없는 것은 맨 뒤로. 그 안에서는 낸 순서를 지킨다.
    # 전 축이 채워진 것부터, 그 안에서 총점 순. 축이 빈 것은 뒤로 - 총점이 있어도
    # 만점이 달라 같은 줄에 세울 수 없다.
    merged.sort(key=lambda e: (e.get("tier", 1),
                               1 if (e.get("total") is None or e.get("held_axes")) else 0,
                               -(e.get("total") if e.get("total") is not None else 0),
                               e.get("submitted_at") or 0))
    n = 0
    div_n = {}
    trk_n = {}
    for e in merged:
        # 참가 제출은 mask() 가 조성에서 구역을 뽑아 넣어 두었다. 기준물질은 mask 를
        # 거치지 않으므로 여기서 채운다 - 기본값으로 덮으면 전부 other 가 된다.
        if not e.get("division"):
            e["division"] = division_of(e.get("composition") or {})
        # 등수는 **전 축이 채워진 것**에만 준다. 축이 하나라도 비면 만점이 달라 총점을
        # 나란히 세울 수 없다 - 점수는 남기되 순위에는 넣지 않는다.
        unscored = e.get("total") is None or bool(e.get("held_axes"))
        if e.get("is_anchor") or e.get("tier", 1) != 1 or unscored:
            e["rank"] = None          # 기준물질은 척도를 표시할 뿐 등수를 갖지 않는다
            e["div_rank"] = None
            e["track"] = None if e.get("is_anchor") else ("mp" if e.get("known_in_mp") else "novel")
            e["track_rank"] = None
        else:
            n += 1
            e["rank"] = n
            # 부문 순위는 전체 순위와 따로 매긴다. 점수는 손대지 않는다.
            d = e["division"]
            div_n[d] = div_n.get(d, 0) + 1
            e["div_rank"] = div_n[d]
            # **트랙 순위.** 같은 조성을 두 경로로 채점해 재보니 신규 조성 쪽이
            # 계통적으로 손해였다(32건 중 28건, 평균 1~2점, 최대 3.5점). MP 에 값이
            # 있으면 그 값을 쓰고 없으면 우리가 계산하는데, 두 자의 눈금이 다르다.
            # 눈금이 다른 것을 한 줄로 세우면 "새로운 조성일수록 불리한" 순위표가 된다 -
            # 신규성에 배점을 걸어 놓고 그러면 그 배점이 거짓말이 된다.
            tr = "mp" if e.get("known_in_mp") else "novel"
            e["track"] = tr
            trk_n[tr] = trk_n.get(tr, 0) + 1
            e["track_rank"] = trk_n[tr]
    tally = {}
    for e in merged:
        if e.get("is_anchor"):
            continue
        tally[e["division"]] = tally.get(e["division"], 0) + 1
    n_held = sum(1 for e in merged
                 if not e.get("is_anchor")
                 and (e.get("total") is None or e.get("held_axes")))
    strip_masked(merged)          # 구역까지 매긴 뒤, 응답으로 나가기 직전에 걷어낸다
    return {"season": seasons.public(season), "entries": merged,
            "divisions": tally,
            "counts": {"scored": len(rows) - n_held, "held": n_held,
                       "anchors": len(anchors)}}


def _leaderboard_bytes(season):
    raw = json.dumps(_leaderboard_body(season), ensure_ascii=False).encode("utf-8")
    return raw, gzip.compress(raw, 6)


@app.get("/api/leaderboard")
def leaderboard(request: Request):
    """응답은 시즌에만 의존하므로 TTL 당 한 번만 만들고 한 번만 압축한다."""
    season = seasons.get(request.query_params.get("season"))
    raw, gz = CACHE.get("lbresp:%d" % season["number"],
                        lambda: _leaderboard_bytes(season))
    if "gzip" in (request.headers.get("accept-encoding") or ""):
        return Response(content=gz, media_type="application/json",
                        headers={"Content-Encoding": "gzip", "Vary": "Accept-Encoding"})
    return Response(content=raw, media_type="application/json",
                    headers={"Vary": "Accept-Encoding"})


_FILLER = {"ai", "model", "llm", "the", "v", "ver", "version", "latest", "preview",
           "chat", "instruct", "it", "api", "official", "new"}


def canon_model(name):
    """표기가 갈린 같은 모델을 하나로 묶는다.

    'Anthropic Claude Opus 5' 와 'claude-opus-5' 는 같은 것이고, 따로 세면 둘 다
    실제보다 적게 쓰인 것처럼 보인다. 다만 변종을 구분하는 낱말(Pro/Flash 등)은
    지우지 않는다 - 그것까지 뭉치면 다른 모델이 한 칸에 들어간다.
    """
    s = re.sub(r"[^a-z0-9]+", " ", (name or "").lower()).strip()
    toks = [t for t in s.split() if t and t not in _FILLER]
    return " ".join(toks) or (name or "").strip().lower()


@app.get("/api/models")
def models(request: Request):
    """모델별 성적: 어느 모델이 가장 좋은 후보를 냈는가, 그리고 얼마나 쓰이는가.

    두 수를 **따로** 낸다. 많이 쓰인다고 잘하는 것이 아니다 - 다들 집는 모델은 점수와
    무관하게 제출 수가 쌓이고, 세 번 쓰인 모델이 1위 후보를 갖고 있을 수 있다.
    평균만 내면 둘 다 가려진다. 그래서 최고·평균·건수를 나란히 둔다.
    기준물질은 제외한다 - 참가자의 모델이 만든 것이 아니다.
    """
    season = seasons.get(request.query_params.get("season"))
    agg = {}
    for r in (_board(season).get("entries") or {}).values():
        if r.get("total") is None:
            continue
        name = (r.get("model_name") or "").strip() or "미기재 / unspecified"
        key = canon_model(name)
        a = agg.setdefault(key, {"n": 0, "best": None, "sum": 0.0,
                                 "top_candidate": None, "spellings": {}})
        a["spellings"][name] = a["spellings"].get(name, 0) + 1
        a["n"] += 1
        a["sum"] += r["total"]
        if a["best"] is None or r["total"] > a["best"]:
            a["best"] = r["total"]
            a["top_candidate"] = r.get("candidate_id")
    out = []
    for k, a in agg.items():
        # 참가자들이 실제로 가장 많이 쓴 표기를 이름으로 삼는다
        label = max(a["spellings"].items(), key=lambda kv: kv[1])[0]
        out.append({"model": label, "n": a["n"], "best": round(a["best"], 2),
                    "mean": round(a["sum"] / a["n"], 2),
                    "top_candidate": a["top_candidate"]})
    out.sort(key=lambda x: (-x["best"], -x["n"]))
    return {"models": out, "counts": {"models": len(out),
                                      "entries": sum(m["n"] for m in out)}}


@app.get("/api/queue")
def queue(request: Request):
    season = seasons.get(request.query_params.get("season"))
    ids = _submissions(season)
    scored = set((_board(season).get("entries") or {}).keys())
    return {"queued": len([i for i in ids if i not in scored]),
            "scored": len(scored), "total": len(ids)}


@app.get("/api/health")
def health():
    return {"ok": True, "seasons": [s["number"] for s in seasons.SEASONS]}