from fastapi import FastAPI, Request
from fastapi.responses import HTMLResponse
from fastapi.middleware.cors import CORSMiddleware
from apscheduler.schedulers.background import BackgroundScheduler
from datetime import datetime
import httpx
import os
import threading
app = FastAPI(title="HF Spaces Auto-Ping API")
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_methods=["*"],
allow_headers=["*"],
)
# ── Shared state ────────────────────────────────────────────────────────────
state = {
"total_requests": 0,
"auto_pings": 0,
"manual_requests": 0,
"last_ping": None,
"ping_log": [], # last 20 ping results
"lock": threading.Lock(),
}
SELF_URL = os.getenv("SPACE_HOST", "http://localhost:7860")
# ── Helpers ─────────────────────────────────────────────────────────────────
def increment(kind: str):
with state["lock"]:
state["total_requests"] += 1
state[kind] += 1
def auto_ping():
"""Called by the scheduler every 60 seconds."""
try:
resp = httpx.get(f"{SELF_URL}/ping", timeout=10)
status = resp.status_code
except Exception as e:
status = f"ERROR: {e}"
ts = datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S UTC")
entry = {"time": ts, "status": status}
with state["lock"]:
state["ping_log"].insert(0, entry)
state["ping_log"] = state["ping_log"][:20]
state["last_ping"] = ts
# ── Scheduler ────────────────────────────────────────────────────────────────
scheduler = BackgroundScheduler()
scheduler.add_job(auto_ping, "interval", minutes=1, id="auto_ping")
scheduler.start()
# ── Routes ───────────────────────────────────────────────────────────────────
@app.middleware("http")
async def count_requests(request: Request, call_next):
# Don't double-count the /ping endpoint hit by the scheduler
# (it increments itself); count everything else here.
if request.url.path not in ("/ping",):
increment("manual_requests")
response = await call_next(request)
return response
@app.get("/ping")
def ping():
"""Internal health-check endpoint hit by the scheduler."""
increment("auto_pings")
return {"status": "ok", "timestamp": datetime.utcnow().isoformat()}
@app.get("/stats")
def stats():
"""Return live counters as JSON."""
with state["lock"]:
return {
"total_requests": state["total_requests"],
"auto_pings": state["auto_pings"],
"manual_requests": state["manual_requests"],
"last_auto_ping": state["last_ping"],
"ping_log": state["ping_log"],
}
@app.get("/", response_class=HTMLResponse)
def dashboard():
"""Live dashboard – auto-refreshes every 10 s."""
html = """
HF Spaces · Request Monitor
🚀 HF Spaces · Request Monitor
Auto-pings every 60 s | Dashboard refreshes every 10 s
📋 Last 20 Auto-Ping Results
| # | Timestamp (UTC) | Status |
|---|
| Loading… |
Next refresh in 10s
"""
return HTMLResponse(content=html)