Spaces:
Running
Running
File size: 3,005 Bytes
5470817 c78b448 5470817 b4d058e 5470817 466b6f5 c78b448 5470817 c78b448 5470817 |
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 |
from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from fastapi.staticfiles import StaticFiles
from apscheduler.schedulers.background import BackgroundScheduler
from datetime import datetime
import os
from dotenv import load_dotenv
from huggingface_hub import HfApi
import json
# Load environment variables
load_dotenv()
# API configuration
API_HOST = os.getenv("API_HOST", "0.0.0.0")
API_PORT = int(os.getenv("API_PORT", "3002"))
app = FastAPI()
# Add CORS middleware
app.add_middleware(
CORSMiddleware,
allow_origins=[
"http://localhost:5173", # Vite dev server
f"http://localhost:{API_PORT}", # API port
"https://huggingface.co", # HF main domain
"https://*.hf.space", # HF Spaces domains
],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# Cache storage
cache = {
"data": None,
"last_updated": None
}
# HF API configuration
HF_TOKEN = os.getenv("HUGGING_FACE_HUB_TOKEN")
REPO_ID = os.getenv("HUGGING_FACE_STORAGE_REPO")
FILE_PATH = os.getenv("HUGGING_FACE_STORAGE_FILE_PATH")
CACHE_DURATION_MINUTES = int(os.getenv("UPDATE_INTERVAL_MINUTES", "15"))
# Initialize HF API client
hf_api = HfApi(token=HF_TOKEN)
def fetch_leaderboards():
"""Fetch leaderboards data from Hugging Face"""
try:
# Download the JSON file directly
json_path = hf_api.hf_hub_download(
repo_id=REPO_ID,
filename=FILE_PATH,
repo_type="dataset"
)
with open(json_path, 'r') as f:
cache["data"] = json.load(f)
cache["last_updated"] = datetime.now()
print(f"Cache updated at {cache['last_updated']}")
except Exception as e:
print(f"Error fetching data: {str(e)}")
if not cache["data"]: # Only raise if we don't have any cached data
raise HTTPException(status_code=500, detail="Failed to fetch leaderboards data")
# Initialize scheduler
scheduler = BackgroundScheduler()
scheduler.add_job(fetch_leaderboards, 'interval', minutes=CACHE_DURATION_MINUTES)
scheduler.start()
# Initial fetch
fetch_leaderboards()
@app.get("/api/leaderboards")
async def get_leaderboards():
"""Get leaderboards data from cache"""
if not cache["data"]:
fetch_leaderboards()
return {
"data": cache["data"],
"last_updated": cache["last_updated"].isoformat() if cache["last_updated"] else None
}
@app.get("/api/health")
async def health_check():
"""Health check endpoint"""
return {
"status": "healthy",
"cache_status": "initialized" if cache["data"] else "empty",
"last_updated": cache["last_updated"].isoformat() if cache["last_updated"] else None
}
# Mount static files
app.mount("/", StaticFiles(directory="static", html=True), name="static")
if __name__ == "__main__":
import uvicorn
uvicorn.run("server:app", host=API_HOST, port=API_PORT, reload=True) |