Spaces:
Running
Running
File size: 3,615 Bytes
01f4cb5 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 | import json
import os
from datetime import datetime, timedelta
from pathlib import Path
from typing import Any
STORAGE_DIR = Path("/data")
SUBMISSIONS_DIR = STORAGE_DIR / "submissions"
LEADERBOARD_FILE = STORAGE_DIR / "leaderboard.jsonl"
RATE_LIMIT_FILE = STORAGE_DIR / "rate_limits.json"
# Seed data bundled with the app (used on first boot)
SEED_LEADERBOARD = Path(__file__).parent.parent / "assets" / "leaderboard.jsonl"
def ensure_dirs():
STORAGE_DIR.mkdir(parents=True, exist_ok=True)
SUBMISSIONS_DIR.mkdir(parents=True, exist_ok=True)
def _seed_leaderboard():
"""Copy bundled leaderboard data to /data on first boot."""
if LEADERBOARD_FILE.exists():
return
if SEED_LEADERBOARD.exists():
import shutil
shutil.copy(SEED_LEADERBOARD, LEADERBOARD_FILE)
print(f"[SEED] Copied leaderboard data from {SEED_LEADERBOARD} to {LEADERBOARD_FILE}")
def save_submission(submission_id: str, payload: dict) -> str:
"""Save raw submission JSON to local storage."""
ensure_dirs()
file_path = SUBMISSIONS_DIR / f"{submission_id}.json"
with open(file_path, "w", encoding="utf-8") as f:
json.dump(payload, f, ensure_ascii=False, indent=2)
return str(file_path)
def list_submissions() -> list[dict]:
"""List all submission metadata."""
ensure_dirs()
results = []
for f in sorted(SUBMISSIONS_DIR.glob("*.json")):
try:
with open(f, "r", encoding="utf-8") as fp:
data = json.load(fp)
meta = data.get("meta", {})
meta["file"] = str(f.name)
results.append(meta)
except Exception:
continue
return results
def load_leaderboard() -> list[dict]:
"""Load current leaderboard data from local storage."""
_seed_leaderboard()
if not LEADERBOARD_FILE.exists():
return []
entries = []
with open(LEADERBOARD_FILE, "r", encoding="utf-8") as f:
for line in f:
line = line.strip()
if line:
entries.append(json.loads(line))
return entries
def save_leaderboard(entries: list[dict]) -> None:
"""Overwrite leaderboard file with the current entries."""
ensure_dirs()
with open(LEADERBOARD_FILE, "w", encoding="utf-8") as f:
for entry in entries:
f.write(json.dumps(entry, ensure_ascii=False) + "\n")
# ---- Rate limiting ----
def check_rate_limit(email: str, cooldown_minutes: int = 60) -> tuple[bool, str]:
"""Check if email is allowed to submit. Returns (allowed, message)."""
ensure_dirs()
limits = {}
if RATE_LIMIT_FILE.exists():
with open(RATE_LIMIT_FILE, "r", encoding="utf-8") as f:
limits = json.load(f)
last_str = limits.get(email)
if last_str:
last_time = datetime.fromisoformat(last_str)
next_allowed = last_time + timedelta(minutes=cooldown_minutes)
if datetime.utcnow() < next_allowed:
remaining = int((next_allowed - datetime.utcnow()).total_seconds() / 60)
return False, f"This email has already submitted within the last hour. Please wait {remaining} minutes."
return True, ""
def record_submission_time(email: str) -> None:
"""Record the current submission time for an email."""
ensure_dirs()
limits = {}
if RATE_LIMIT_FILE.exists():
with open(RATE_LIMIT_FILE, "r", encoding="utf-8") as f:
limits = json.load(f)
limits[email] = datetime.utcnow().isoformat()
with open(RATE_LIMIT_FILE, "w", encoding="utf-8") as f:
json.dump(limits, f, ensure_ascii=False, indent=2)
|