[deploy] app.py

app.py

@@ -0,0 +1,153 @@
+"""
+📡 Space Pinger
+================
+Pings all AIencoder HuggingFace Spaces every 4 minutes to keep them alive.
+Auto-discovers spaces via HF API so it stays up to date.
+"""
+
+import gradio as gr
+import requests, threading, time, os
+from datetime import datetime
+from huggingface_hub import HfApi
+
+HF_TOKEN = os.environ.get("HF_TOKEN", "")
+OWNER    = "AIencoder"
+INTERVAL = 240  # 4 minutes (HF sleeps after 5 min of inactivity)
+
+state = {
+    "running": False,
+    "log": "Ready. Press Start to begin pinging.\n",
+    "spaces": [],
+    "ping_count": 0,
+    "last_round": "Never",
+}
+lock = threading.Lock()
+
+
+def log(msg):
+    ts = datetime.now().strftime("%H:%M:%S")
+    with lock:
+        state["log"] += f"[{ts}] {msg}\n"
+        lines = state["log"].split("\n")
+        if len(lines) > 300:
+            state["log"] = "\n".join(lines[-300:])
+
+
+def get_spaces():
+    """Fetch all spaces for the owner."""
+    try:
+        api = HfApi(token=HF_TOKEN)
+        spaces = list(api.list_spaces(author=OWNER, token=HF_TOKEN))
+        urls = []
+        for s in spaces:
+            space_id = s.id
+            # Skip self to avoid circular pinging
+            if space_id == f"{OWNER}/space-pinger":
+                continue
+            # Convert to URL: owner/space-name -> https://owner-space-name.hf.space
+            slug = space_id.replace("/", "-").replace("_", "-").lower()
+            url = f"https://{slug}.hf.space"
+            urls.append((space_id, url))
+        return urls
+    except Exception as e:
+        log(f"⚠️ Could not fetch spaces: {e}")
+        return state["spaces"]
+
+
+def ping_all():
+    """Ping every space once."""
+    spaces = get_spaces()
+    state["spaces"] = spaces
+    log(f"\n🔄 Pinging {len(spaces)} spaces...")
+    
+    ok, fail = 0, 0
+    for space_id, url in spaces:
+        try:
+            r = requests.get(url, timeout=15, allow_redirects=True)
+            status = r.status_code
+            if status < 400:
+                log(f"  ✅ {space_id} ({status})")
+                ok += 1
+            else:
+                log(f"  ⚠️ {space_id} ({status})")
+                fail += 1
+        except Exception as e:
+            log(f"  ❌ {space_id} — {type(e).__name__}")
+            fail += 1
+
+    state["ping_count"] += 1
+    state["last_round"] = datetime.now().strftime("%H:%M:%S")
+    log(f"Round {state['ping_count']} done — ✅ {ok} up, ❌ {fail} down. Next in {INTERVAL//60} min.")
+
+
+def ping_loop():
+    """Background loop."""
+    while state["running"]:
+        ping_all()
+        # Sleep in small chunks so stop signal is responsive
+        for _ in range(INTERVAL):
+            if not state["running"]:
+                break
+            time.sleep(1)
+    log("⛔ Pinger stopped.")
+
+
+def start():
+    if state["running"]:
+        return "⚠️ Already running!"
+    state["running"] = True
+    state["log"] = "🚀 Pinger starting...\n"
+    threading.Thread(target=ping_loop, daemon=True).start()
+    return "✅ Pinger started! Pinging every 4 minutes."
+
+
+def stop():
+    state["running"] = False
+    return "⛔ Stop signal sent."
+
+
+def get_log():
+    with lock:
+        return state["log"]
+
+
+def get_status():
+    spaces = state["spaces"]
+    status = "🟢 Running" if state["running"] else "⚫ Idle"
+    space_list = "\n".join(f"- {sid}" for sid, _ in spaces) if spaces else "_Not fetched yet_"
+    return (
+        f"**Status:** {status}  \n"
+        f"**Ping rounds completed:** {state['ping_count']}  \n"
+        f"**Last round:** {state['last_round']}  \n"
+        f"**Interval:** {INTERVAL//60} minutes  \n\n"
+        f"**Spaces being pinged ({len(spaces)}):**\n{space_list}"
+    )
+
+
+with gr.Blocks(title="📡 Space Pinger", theme=gr.themes.Soft()) as app:
+    gr.Markdown("# 📡 Space Pinger\nKeeps all AIencoder Spaces alive by pinging every 4 minutes.")
+
+    with gr.Row():
+        start_btn = gr.Button("🚀 Start Pinger", variant="primary", scale=2)
+        stop_btn  = gr.Button("⛔ Stop", variant="stop", scale=1)
+
+    status_out = gr.Markdown("*Press Start.*")
+
+    with gr.Row():
+        with gr.Column(scale=1):
+            gr.Markdown("### 📊 Status")
+            status_box = gr.Markdown("*No data yet.*")
+            gr.Button("🔄 Refresh").click(get_status, outputs=status_box)
+        with gr.Column(scale=2):
+            gr.Markdown("### 📋 Log")
+            log_box = gr.Textbox(lines=20, max_lines=20, interactive=False)
+            gr.Button("🔄 Refresh Log").click(get_log, outputs=log_box)
+
+    start_btn.click(start, outputs=status_out)
+    stop_btn.click(stop, outputs=status_out)
+
+    timer = gr.Timer(value=10)
+    timer.tick(get_log, outputs=log_box)
+    timer.tick(get_status, outputs=status_box)
+
+app.launch(server_name="0.0.0.0", server_port=7860)