Spaces:
Paused
Paused
| # -*- coding: utf-8 -*- | |
| """νλ κ΄λ¬Έ μμΌμ΄λ Space β μ μ νμ΄μ§ + λμ μ μ (νλΌμ΄λΉ λ°μ΄ν°μ ) | |
| π΄ μ μ μ Space λμ€ν¬κ° μλλΌ **λ°μ΄ν°μ repo** μ μ΄λ€. Space λ μ¬μμνλ©΄ λμ€ν¬κ° λ μκ°λ€. | |
| """ | |
| import json, os, threading | |
| from fastapi import FastAPI, Request | |
| from fastapi.responses import HTMLResponse, JSONResponse | |
| from huggingface_hub import HfApi, hf_hub_download | |
| DS = os.environ.get("RECORD_DATASET", "FINAL-Bench/gate-tetris-record") | |
| TOK = os.environ.get("HF_TOKEN") | |
| api = HfApi(token=TOK) | |
| LOCK = threading.Lock() | |
| BLANK = {"matches": 0, "wins": {"plain": 0, "jev": 0, "ztc": 0}, | |
| "blocked": 0, "recent": []} | |
| app = FastAPI() | |
| HTML = open(os.path.join(os.path.dirname(__file__), "index.html"), encoding="utf-8").read() | |
| def load(): | |
| try: | |
| p = hf_hub_download(DS, "tally.json", repo_type="dataset", token=TOK, | |
| force_download=True) | |
| with open(p, encoding="utf-8") as f: | |
| t = json.load(f) | |
| except Exception: | |
| t = {} | |
| out = json.loads(json.dumps(BLANK)) | |
| out["matches"] = int(t.get("matches", 0) or 0) | |
| for k in ("plain", "jev", "ztc"): | |
| out["wins"][k] = int((t.get("wins") or {}).get(k, 0) or 0) | |
| out["blocked"] = int(t.get("blocked", 0) or 0) | |
| out["recent"] = list(t.get("recent") or []) | |
| return out | |
| def save(t): | |
| api.upload_file( | |
| path_or_fileobj=json.dumps(t, ensure_ascii=False, indent=1).encode("utf-8"), | |
| path_in_repo="tally.json", repo_id=DS, repo_type="dataset", | |
| commit_message="record one match") | |
| def index(): | |
| return HTML | |
| def get_record(): | |
| return JSONResponse(load()) | |
| async def post_record(req: Request): | |
| body = await req.json() | |
| win = body.get("winner") | |
| if win not in ("plain", "jev", "ztc"): | |
| return JSONResponse({"error": "bad winner"}, status_code=400) | |
| sc = body.get("score") or {} | |
| ac = body.get("acted") or {} | |
| mode = "human" if body.get("mode") == "human" else "auto" | |
| blocked = int(body.get("blocked") or 0) | |
| with LOCK: | |
| t = load() | |
| t["matches"] += 1 | |
| t["wins"][win] += 1 | |
| t["blocked"] += max(0, blocked) | |
| row = {"mode": mode, "winner": win, | |
| "score": {k: int(sc.get(k, 0) or 0) for k in ("plain", "jev", "ztc")}, | |
| "acted": {k: int(ac.get(k, 0) or 0) for k in ("plain", "jev", "ztc")}, | |
| "blocked": max(0, blocked)} | |
| t["recent"] = ([row] + t["recent"])[:50] | |
| try: | |
| save(t) | |
| except Exception as e: | |
| out = dict(t); out["error"] = "save failed: %s" % e | |
| return JSONResponse(out, status_code=200) | |
| return JSONResponse(t) | |
| def health(): | |
| return {"ok": True, "dataset": DS, "token": bool(TOK)} | |