Hugging Face's logo

Spaces:
AIencoder
/
space-pinger
Sleeping

App Files Community
space-pinger
File size: 4,634 Bytes
4df2067
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
"""
📡 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)