| 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_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 |
| try: |
| ensure_dirs() |
| shutil.copy(SEED_LEADERBOARD, LEADERBOARD_FILE) |
| print(f"[SEED] Copied leaderboard data from {SEED_LEADERBOARD} to {LEADERBOARD_FILE}") |
| except OSError as e: |
| print(f"[WARN] Cannot seed leaderboard at {LEADERBOARD_FILE}: {e}. Using bundled seed read-only.") |
|
|
|
|
| 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() |
| source = LEADERBOARD_FILE if LEADERBOARD_FILE.exists() else SEED_LEADERBOARD |
| if not source.exists(): |
| return [] |
| entries = [] |
| with open(source, "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") |
|
|
|
|
| |
|
|
| 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) |
|
|
