""" Module de serveur API pour le leaderboard parser. Ce module contient la configuration FastAPI et les endpoints pour le mode serveur. """ import datetime import threading import logging import os from fastapi import FastAPI, HTTPException from fastapi.responses import JSONResponse, HTMLResponse from src.file_utils import format_datetime # Initialiser le logger logger = logging.getLogger("leaderboard-parser") # Variables globales pour suivre l'état du serveur processing_status = "idle" processing_error = None last_run_time = None # Initialiser l'application FastAPI app = FastAPI(title="Leaderboard Parser API") # Cette fonction sera importée depuis main.py process_leaderboards = None def initialize_server(process_function): """ Initialise le serveur avec la fonction de traitement des leaderboards. Cette fonction doit être appelée avant de démarrer le serveur. Args: process_function: Fonction qui traite les leaderboards """ global process_leaderboards, last_run_time process_leaderboards = process_function # Initialiser last_run_time à la date actuelle last_run_time = datetime.datetime.now() logger.info(f"Serveur initialisé avec la fonction de traitement. last_run_time initialisé à {last_run_time.isoformat()}") # Endpoints API @app.get("/", response_class=HTMLResponse) async def root(): """Root endpoint serving a simple HTML page with status info""" global processing_status, last_run_time, processing_error next_run = format_datetime(last_run_time + datetime.timedelta(hours=int(os.environ.get("LEADERBOARD_REPROCESS_INTERVAL_HOURS", 24)))) if last_run_time else "Not yet scheduled" last_run = format_datetime(last_run_time) if last_run_time else "Never" # Calculate timestamp for next run to use in JavaScript next_run_timestamp = "" if last_run_time: next_run_time = last_run_time + datetime.timedelta(hours=int(os.environ.get("LEADERBOARD_REPROCESS_INTERVAL_HOURS", 24))) next_run_timestamp = next_run_time.timestamp() * 1000 # Convert to milliseconds for JavaScript html_content = f""" Leaderboard Parser Status

Leaderboard Parser

Status {processing_status}

Last run {last_run}

{f'

Error {processing_error}

' if processing_error else ''}

Calculating time remaining...

""" return HTMLResponse(content=html_content) @app.get("/status") async def get_status(): """Get the current status of the parser""" global processing_status, last_run_time, processing_error return { "status": processing_status, "last_run": format_datetime(last_run_time) if last_run_time else None, "next_run": format_datetime(last_run_time + datetime.timedelta(hours=int(os.environ.get("LEADERBOARD_REPROCESS_INTERVAL_HOURS", 24)))) if last_run_time else None, "error": processing_error } @app.post("/run") async def trigger_run(): """Manually trigger a leaderboard processing run""" global processing_status, process_leaderboards if not process_leaderboards: raise HTTPException(status_code=500, detail="Server not properly initialized") if processing_status == "running": raise HTTPException(status_code=409, detail="Processing is already running") # Start processing in a separate thread threading.Thread(target=lambda: process_leaderboards()).start() return {"status": "started", "message": "Processing started"} @app.get("/health") async def health_check(): """Health check endpoint for HF Spaces to verify the app is running""" return {"status": "ok"}