""" HuggingFace Spaces Keep-Alive Service ====================================== A minimal service that periodically pings Zelin bot Spaces to prevent them from sleeping due to inactivity. Spaces monitored: - TomatitoToho/agent-nav-3d (Zelin Minecraft bot) - TomatitoToho/zelin-v6x (Zelin Discord bot) This Space also exposes /health so the Zelin spaces can ping it back, creating a mutually-assured-wakefulness triangle. """ import os import time import threading import logging from datetime import datetime, timezone import requests from flask import Flask, jsonify # --------------------------------------------------------------------------- # Configuration # --------------------------------------------------------------------------- SPACES = { "agent-nav-3d": { "url": "https://tomatitotoho-agent-nav-3d.hf.space/health", "label": "Zelin Minecraft Bot", "emoji": "⛏️", "hf_url": "https://huggingface.co/spaces/TomatitoToho/agent-nav-3d", }, "zelin-v6x": { "url": "https://tomatitotoho-zelin-v6x.hf.space/health", "label": "Zelin Discord Bot", "emoji": "🤖", "hf_url": "https://huggingface.co/spaces/TomatitoToho/zelin-v6x", }, "keepalive": { "url": "https://tomatitotoho-keepalive.hf.space/health", "label": "Keep-Alive Service (self)", "emoji": "💓", "hf_url": "https://huggingface.co/spaces/TomatitoToho/keepalive", }, } PING_INTERVAL_SECONDS = int(os.environ.get("PING_INTERVAL", 240)) # 4 min default PING_TIMEOUT_SECONDS = int(os.environ.get("PING_TIMEOUT", 15)) PORT = int(os.environ.get("PORT", 7860)) # --------------------------------------------------------------------------- # Logging # --------------------------------------------------------------------------- logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s", datefmt="%Y-%m-%d %H:%M:%S", ) logger = logging.getLogger("keepalive") # --------------------------------------------------------------------------- # Flask app # --------------------------------------------------------------------------- app = Flask(__name__) # --------------------------------------------------------------------------- # In-memory status store (thread-safe via lock) # --------------------------------------------------------------------------- status_lock = threading.Lock() space_status: dict[str, dict] = {} ping_history: list[dict] = [] # last 50 pings total_pings = 0 failed_pings = 0 service_start_time = datetime.now(timezone.utc) def _init_status(): for key, meta in SPACES.items(): space_status[key] = { "key": key, "label": meta["label"], "emoji": meta["emoji"], "hf_url": meta["hf_url"], "last_ping_time": None, "last_status": "never_pinged", "last_status_code": None, "last_latency_ms": None, "consecutive_failures": 0, "total_successes": 0, "total_failures": 0, } _init_status() # --------------------------------------------------------------------------- # Ping logic # --------------------------------------------------------------------------- def ping_space(key: str) -> dict: """Ping a single space and return the result dict.""" meta = SPACES[key] url = meta["url"] result = { "key": key, "timestamp": datetime.now(timezone.utc).isoformat(), "status": "unknown", "status_code": None, "latency_ms": None, "error": None, } try: start = time.monotonic() resp = requests.get(url, timeout=PING_TIMEOUT_SECONDS) latency = (time.monotonic() - start) * 1000 result["status_code"] = resp.status_code result["latency_ms"] = round(latency, 1) if 200 <= resp.status_code < 400: result["status"] = "alive" else: result["status"] = "unhealthy" result["error"] = f"HTTP {resp.status_code}" except requests.exceptions.Timeout: result["status"] = "timeout" result["error"] = f"Timed out after {PING_TIMEOUT_SECONDS}s" except requests.exceptions.ConnectionError as exc: result["status"] = "connection_error" result["error"] = str(exc)[:120] except Exception as exc: # noqa: BLE001 result["status"] = "error" result["error"] = str(exc)[:120] return result def ping_all_spaces(): """Ping every configured space and update in-memory status.""" global total_pings, failed_pings # noqa: PLW0603 logger.info("Pinging all spaces ...") for key in SPACES: result = ping_space(key) with status_lock: total_pings += 1 st = space_status[key] st["last_ping_time"] = result["timestamp"] st["last_status"] = result["status"] st["last_status_code"] = result["status_code"] st["last_latency_ms"] = result["latency_ms"] if result["status"] == "alive": st["consecutive_failures"] = 0 st["total_successes"] += 1 else: st["consecutive_failures"] += 1 st["total_failures"] += 1 failed_pings += 1 # Log with appropriate severity if st["consecutive_failures"] >= 5: logger.warning( "⚠️ %s has been unreachable for %d consecutive pings!", key, st["consecutive_failures"], ) else: logger.info( "❌ %s: %s (%s)", key, result["status"], result.get("error", ""), ) # Append to history (keep last 50) ping_history.append(result) if len(ping_history) > 50: ping_history.pop(0) logger.info("Ping round complete.") # --------------------------------------------------------------------------- # Background scheduler (simple threaded timer — no extra deps needed) # --------------------------------------------------------------------------- def _scheduler_loop(): """Daemon thread that pings all spaces on a fixed interval.""" # First ping immediately ping_all_spaces() while True: time.sleep(PING_INTERVAL_SECONDS) try: ping_all_spaces() except Exception: # noqa: BLE001 logger.exception("Unexpected error in scheduler loop") scheduler_thread = threading.Thread(target=_scheduler_loop, daemon=True) scheduler_thread.start() logger.info( "Scheduler started — pinging every %d seconds", PING_INTERVAL_SECONDS, ) # --------------------------------------------------------------------------- # Routes # --------------------------------------------------------------------------- @app.route("/") def dashboard(): """HTML status dashboard.""" with status_lock: rows = "" for key, st in space_status.items(): if st["last_status"] == "alive": badge_color = "#22c55e" badge_text = "ALIVE" elif st["last_status"] == "never_pinged": badge_color = "#6b7280" badge_text = "PENDING" elif st["consecutive_failures"] >= 5: badge_color = "#ef4444" badge_text = "DOWN" else: badge_color = "#f59e0b" badge_text = "ISSUES" latency_str = ( f'{st["last_latency_ms"]:.0f}ms' if st["last_latency_ms"] is not None else "—" ) last_time = st["last_ping_time"] or "—" if last_time != "—": # Trim to just the time portion for readability try: last_time = last_time.split("T")[1][:8] + " UTC" except (IndexError, ValueError): pass rows += f""" {st["emoji"]} {st["label"]}
{key} {badge_text} {latency_str} {last_time} {st["total_successes"]}/{st["total_successes"]+st["total_failures"]} {st["consecutive_failures"]} """ uptime_seconds = int( (datetime.now(timezone.utc) - service_start_time).total_seconds() ) uptime_hours = uptime_seconds // 3600 uptime_mins = (uptime_seconds % 3600) // 60 history_rows = "" for entry in reversed(ping_history[-15:]): ts = entry["timestamp"].split("T")[1][:8] if "T" in entry["timestamp"] else entry["timestamp"] icon = "✅" if entry["status"] == "alive" else "❌" latency = f'{entry["latency_ms"]:.0f}ms' if entry["latency_ms"] else "—" err = f' ({entry["error"][:60]})' if entry.get("error") else "" history_rows += f"{ts}{icon} {entry['key']}{latency}{entry['status']}{err}" html = f""" Zelin Keep-Alive Monitor

Zelin Keep-Alive Monitor

Pinging every {PING_INTERVAL_SECONDS}s · Uptime: {uptime_hours}h {uptime_mins}m · Total pings: {total_pings} · Failed: {failed_pings}

{len(SPACES)}
Spaces Monitored
{PING_INTERVAL_SECONDS}s
Ping Interval
{total_pings}
Total Pings
{failed_pings}
Failed Pings

Space Status

{rows}
SpaceStatusLatency Last PingSuccess RateFail Streak

Recent Ping History (last 15)

{history_rows}
TimeSpaceLatencyResult

Keep-Alive Service · /health · /ping · /api/status

""" return html @app.route("/health") def health(): """Simple health check endpoint.""" return jsonify({"status": "ok", "service": "keepalive", "timestamp": datetime.now(timezone.utc).isoformat()}), 200 @app.route("/ping") def ping(): """Trigger an on-demand ping of all spaces and return results.""" results = {} for key in SPACES: result = ping_space(key) # Update status store with status_lock: st = space_status[key] st["last_ping_time"] = result["timestamp"] st["last_status"] = result["status"] st["last_status_code"] = result["status_code"] st["last_latency_ms"] = result["latency_ms"] if result["status"] == "alive": st["consecutive_failures"] = 0 st["total_successes"] += 1 else: st["consecutive_failures"] += 1 st["total_failures"] += 1 ping_history.append(result) if len(ping_history) > 50: ping_history.pop(0) results[key] = result return jsonify({"triggered_by": "manual", "spaces": results}), 200 @app.route("/api/status") def api_status(): """Full JSON status dump.""" with status_lock: return jsonify({ "service": "keepalive", "uptime_seconds": int( (datetime.now(timezone.utc) - service_start_time).total_seconds() ), "ping_interval_seconds": PING_INTERVAL_SECONDS, "total_pings": total_pings, "failed_pings": failed_pings, "spaces": space_status, "recent_history": ping_history[-15:], }), 200 # --------------------------------------------------------------------------- # Entry point # --------------------------------------------------------------------------- if __name__ == "__main__": logger.info("Starting Keep-Alive service on port %d", PORT) app.run(host="0.0.0.0", port=PORT)