Create app.py

app.py

@@ -0,0 +1,277 @@
+"""
+Space Keeper - Keeps HuggingFace Spaces alive during hackathon evaluation
+Pings all Spaces in an organization on a schedule to prevent them from sleeping.
+"""
+
+import os
+import json
+import threading
+from datetime import datetime, timezone
+from pathlib import Path
+
+import gradio as gr
+import requests
+from huggingface_hub import HfApi
+from apscheduler.schedulers.background import BackgroundScheduler
+from apscheduler.triggers.interval import IntervalTrigger
+
+# Configuration
+ORG_NAME = os.environ.get("ORG_NAME", "mcp-hackathon")
+PING_INTERVAL_HOURS = int(os.environ.get("PING_INTERVAL_HOURS", "6"))
+REQUEST_TIMEOUT = int(os.environ.get("REQUEST_TIMEOUT", "30"))
+LOG_FILE = Path("run_logs.json")
+MAX_LOG_ENTRIES = 100  # Keep last N runs
+
+# Global state
+scheduler = BackgroundScheduler()
+api = HfApi()
+
+
+def load_logs() -> list:
+    """Load logs from file."""
+    if LOG_FILE.exists():
+        try:
+            with open(LOG_FILE, "r") as f:
+                return json.load(f)
+        except Exception:
+            return []
+    return []
+
+
+def save_logs(logs: list):
+    """Save logs to file, keeping only the most recent entries."""
+    logs = logs[-MAX_LOG_ENTRIES:]
+    with open(LOG_FILE, "w") as f:
+        json.dump(logs, f, indent=2)
+
+
+def ping_space(space_id: str) -> dict:
+    """Ping a single Space and return the result."""
+    url = f"https://huggingface.co/spaces/{space_id}"
+    try:
+        response = requests.get(url, timeout=REQUEST_TIMEOUT)
+        return {
+            "space_id": space_id,
+            "status": "success",
+            "status_code": response.status_code,
+            "error": None
+        }
+    except requests.Timeout:
+        return {
+            "space_id": space_id,
+            "status": "timeout",
+            "status_code": None,
+            "error": f"Request timed out after {REQUEST_TIMEOUT}s"
+        }
+    except Exception as e:
+        return {
+            "space_id": space_id,
+            "status": "error",
+            "status_code": None,
+            "error": str(e)
+        }
+
+
+def run_ping_job(triggered_by: str = "scheduler") -> dict:
+    """Run the ping job for all Spaces in the org."""
+    start_time = datetime.now(timezone.utc)
+    
+    # Get all Spaces in the org
+    try:
+        spaces = list(api.list_spaces(author=ORG_NAME))
+    except Exception as e:
+        run_result = {
+            "timestamp": start_time.isoformat(),
+            "triggered_by": triggered_by,
+            "status": "error",
+            "error": f"Failed to list Spaces: {str(e)}",
+            "total_spaces": 0,
+            "successful": 0,
+            "failed": 0,
+            "duration_seconds": 0,
+            "results": []
+        }
+        logs = load_logs()
+        logs.append(run_result)
+        save_logs(logs)
+        return run_result
+    
+    # Ping each Space
+    results = []
+    for space in spaces:
+        result = ping_space(space.id)
+        results.append(result)
+    
+    end_time = datetime.now(timezone.utc)
+    duration = (end_time - start_time).total_seconds()
+    
+    successful = sum(1 for r in results if r["status"] == "success")
+    failed = len(results) - successful
+    
+    run_result = {
+        "timestamp": start_time.isoformat(),
+        "triggered_by": triggered_by,
+        "status": "completed",
+        "error": None,
+        "total_spaces": len(results),
+        "successful": successful,
+        "failed": failed,
+        "duration_seconds": round(duration, 2),
+        "results": results
+    }
+    
+    # Save to logs
+    logs = load_logs()
+    logs.append(run_result)
+    save_logs(logs)
+    
+    return run_result
+
+
+def scheduled_job():
+    """Wrapper for the scheduled job."""
+    print(f"[{datetime.now(timezone.utc).isoformat()}] Running scheduled ping job...")
+    run_ping_job(triggered_by="scheduler")
+
+
+def manual_trigger():
+    """Manually trigger a ping run."""
+    result = run_ping_job(triggered_by="manual")
+    return format_run_result(result), get_logs_display()
+
+
+def format_run_result(result: dict) -> str:
+    """Format a single run result for display."""
+    if result["status"] == "error":
+        return f"""## ❌ Run Failed
+
+**Time:** {result['timestamp']}
+**Triggered by:** {result['triggered_by']}
+**Error:** {result['error']}
+"""
+    
+    status_emoji = "✅" if result["failed"] == 0 else "⚠️"
+    
+    output = f"""## {status_emoji} Run Completed
+
+**Time:** {result['timestamp']}
+**Triggered by:** {result['triggered_by']}
+**Duration:** {result['duration_seconds']}s
+**Spaces pinged:** {result['total_spaces']} ({result['successful']} successful, {result['failed']} failed)
+"""
+    
+    if result["failed"] > 0:
+        output += "\n### Failed Spaces:\n"
+        for r in result["results"]:
+            if r["status"] != "success":
+                output += f"- `{r['space_id']}`: {r['error']}\n"
+    
+    return output
+
+
+def get_logs_display() -> str:
+    """Get formatted logs for display."""
+    logs = load_logs()
+    
+    if not logs:
+        return "No runs recorded yet. Click 'Run Now' to trigger a manual run."
+    
+    # Reverse to show most recent first
+    logs = list(reversed(logs))
+    
+    output = "# Run History\n\n"
+    
+    for i, run in enumerate(logs[:20]):  # Show last 20 runs
+        timestamp = run["timestamp"]
+        triggered_by = run["triggered_by"]
+        
+        if run["status"] == "error":
+            output += f"### ❌ {timestamp}\n"
+            output += f"Triggered by: {triggered_by} | Error: {run['error']}\n\n"
+        else:
+            status_emoji = "✅" if run["failed"] == 0 else "⚠️"
+            output += f"### {status_emoji} {timestamp}\n"
+            output += f"Triggered by: {triggered_by} | "
+            output += f"Spaces: {run['total_spaces']} | "
+            output += f"Success: {run['successful']} | "
+            output += f"Failed: {run['failed']} | "
+            output += f"Duration: {run['duration_seconds']}s\n"
+            
+            if run["failed"] > 0:
+                failed_spaces = [r["space_id"] for r in run["results"] if r["status"] != "success"]
+                output += f"Failed: {', '.join(failed_spaces)}\n"
+            output += "\n"
+    
+    if len(logs) > 20:
+        output += f"\n*...and {len(logs) - 20} more runs (showing last 20)*\n"
+    
+    return output
+
+
+def get_status() -> str:
+    """Get current scheduler status."""
+    next_run = scheduler.get_jobs()[0].next_run_time if scheduler.get_jobs() else None
+    
+    return f"""## Space Keeper Status
+
+**Organization:** `{ORG_NAME}`
+**Ping Interval:** Every {PING_INTERVAL_HOURS} hours
+**Request Timeout:** {REQUEST_TIMEOUT} seconds
+**Next Scheduled Run:** {next_run.strftime('%Y-%m-%d %H:%M:%S UTC') if next_run else 'Not scheduled'}
+
+---
+
+This Space automatically pings all Spaces in the `{ORG_NAME}` organization to keep them from sleeping during the evaluation period.
+"""
+
+
+def refresh_logs():
+    """Refresh the logs display."""
+    return get_logs_display()
+
+
+# Build the Gradio interface
+with gr.Blocks(title="Space Keeper", theme=gr.themes.Soft()) as demo:
+    gr.Markdown("# 🔄 Space Keeper")
+    gr.Markdown("Keeps HuggingFace Spaces alive by pinging them on a schedule.")
+    
+    with gr.Row():
+        with gr.Column(scale=1):
+            status_display = gr.Markdown(get_status())
+            
+            with gr.Row():
+                run_btn = gr.Button("🚀 Run Now", variant="primary", size="lg")
+                refresh_btn = gr.Button("🔄 Refresh Logs", size="lg")
+            
+            last_run_display = gr.Markdown("Click 'Run Now' to trigger a manual ping run.")
+        
+        with gr.Column(scale=2):
+            logs_display = gr.Markdown(get_logs_display())
+    
+    # Event handlers
+    run_btn.click(
+        fn=manual_trigger,
+        outputs=[last_run_display, logs_display]
+    )
+    
+    refresh_btn.click(
+        fn=refresh_logs,
+        outputs=[logs_display]
+    )
+
+
+# Start the scheduler
+scheduler.add_job(
+    scheduled_job,
+    trigger=IntervalTrigger(hours=PING_INTERVAL_HOURS),
+    id="ping_job",
+    name="Ping all Spaces",
+    replace_existing=True
+)
+scheduler.start()
+
+print(f"Space Keeper started for org: {ORG_NAME}")
+print(f"Ping interval: {PING_INTERVAL_HOURS} hours")
+
+if __name__ == "__main__":
+    demo.launch()