Spaces:
Running
Running
| 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) | |