Spaces:
Running
Running
File size: 3,755 Bytes
4f95823 | 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 | """
keepalive.py — server-side "Keep Awake" for B24 Browser.
SCOPE (read this before wiring it up):
This pings a URL periodically from the backend, so a tracked tab stays
"active" even while the user's phone is asleep or the app is backgrounded
or fully closed — as long as this backend itself keeps running (note: a
free HF Space can also idle-sleep, so this isn't infinite either).
What this does NOT do: it does not simulate mouse movement, keystrokes,
or run the page's real JavaScript. It's a plain periodic HTTP request.
That's enough for many simple session-timeout mechanisms (dashboards,
admin panels, some web tools) but will NOT fool sites that check real
interaction via WebSocket heartbeats or JS-side activity checks — those
need a full headless browser per session, which isn't safe to run at
scale on limited RAM, and many such platforms (e.g. hosted notebook /
compute services) explicitly prohibit keep-alive workarounds in their
ToS. Keep this scoped to "don't lose my session on a page I'm using
while I multitask", not "trick any site indefinitely" — and maybe warn
users in the UI if a URL looks like a known compute/notebook host.
"""
import time
import threading
import requests
MAX_SESSIONS_PER_USER = 3
MAX_DURATION_SECONDS = 6 * 60 * 60 # 6h safety cap, then auto-stop
MIN_INTERVAL_SECONDS = 30
_sessions = {} # session_id -> dict(user_id, url, interval, started_at, stop_event, thread)
_lock = threading.Lock()
def _run(session_id, url, interval, headers, stop_event, started_at):
while not stop_event.is_set():
if time.time() - started_at > MAX_DURATION_SECONDS:
break
try:
requests.get(url, headers=headers or {}, timeout=10)
except requests.RequestException:
pass # transient failure — just try again next tick
stop_event.wait(interval)
with _lock:
_sessions.pop(session_id, None)
def start(user_id: str, url: str, interval: int = 60, headers: dict = None):
"""Begin pinging `url` every `interval` seconds on behalf of user_id."""
interval = max(interval, MIN_INTERVAL_SECONDS)
with _lock:
active_for_user = [s for s in _sessions.values() if s["user_id"] == user_id]
if len(active_for_user) >= MAX_SESSIONS_PER_USER:
return {"error": f"max {MAX_SESSIONS_PER_USER} keep-alive sessions per user"}
session_id = f"{user_id}:{int(time.time() * 1000)}"
stop_event = threading.Event()
thread = threading.Thread(
target=_run,
args=(session_id, url, interval, headers, stop_event, time.time()),
daemon=True,
)
_sessions[session_id] = {
"user_id": user_id,
"url": url,
"interval": interval,
"started_at": time.time(),
"stop_event": stop_event,
"thread": thread,
}
thread.start()
return {"session_id": session_id, "interval": interval, "max_duration": MAX_DURATION_SECONDS}
def stop(user_id: str, session_id: str):
with _lock:
session = _sessions.get(session_id)
if not session or session["user_id"] != user_id:
return {"error": "session not found"}
session["stop_event"].set()
return {"stopped": True}
def status(user_id: str):
with _lock:
return [
{
"session_id": sid,
"url": s["url"],
"interval": s["interval"],
"elapsed_seconds": int(time.time() - s["started_at"]),
"remaining_seconds": max(0, MAX_DURATION_SECONDS - int(time.time() - s["started_at"])),
}
for sid, s in _sessions.items()
if s["user_id"] == user_id
]
|