app.py

"""
📡 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
            # 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)