add "wake up" capability

app.py

@@ -7,7 +7,7 @@ import threading
 import gradio as gr
 from fastapi.responses import JSONResponse, RedirectResponse
 from fastapi import FastAPI, Request, Response, status
-from utils import manage_urls
+from utils import manage_urls, wake_up_space, get_wake_up_action_name
 
 # Charger les variables d'environnement
 load_dotenv()
@@ -64,8 +64,12 @@ def ping_loop():
     while True:
         for url in url_list:
             try:
-                requests.get(url, timeout=5)
-                print(f"[ping] ✅ {url}")
+                if get_wake_up_action_name(url):
+                    print(f"[wake-up] 🚀 {url}")
+                    wake_up_space(url)
+                else:
+                    requests.get(url, timeout=5)
+                    print(f"[ping] ✅ {url}")
             except Exception as e:
                 print(f"[ping] ❌ {url} | {e}")
         time.sleep(3600)  # 1h

requirements.txt

@@ -3,3 +3,4 @@ fastapi==0.115.12
 uvicorn==0.34.2
 requests==2.32.3
 python-dotenv==1.1.0
+beautifulsoup4==4.13.4

utils.py

@@ -1,3 +1,9 @@
+import requests
+from typing import Optional
+from bs4 import BeautifulSoup
+import time
+import re
+
 # Fonction Gradio pour ajouter / retirer des URLs
 def manage_urls(action, url):
     global url_list
@@ -18,3 +24,60 @@ def manage_urls(action, url):
 
     # Toujours retourner la liste actuelle formatée
     return message, "\n".join(url_list)
+
+def wake_up_space(space_url: str):
+    action_name = get_wake_up_action_name(space_url)
+
+    if not action_name:
+        print(f"[wake-up] ❌ Invalid URL: {space_url}")
+        return
+    
+    try:
+        session = requests.Session()
+        response = session.get(space_url, timeout=10)
+        response.raise_for_status()
+        
+        # Vérifier si le formulaire de réveil est présent
+        soup = BeautifulSoup(response.text, "html.parser")
+        form = soup.find("form", action=action_name)
+
+        if form:
+            print("[wake-up] 💤 Space is sleeping, trying to restart it...")
+            csrf_token_input = form.find("input", {"name": "csrf"})
+            csrf_token = csrf_token_input["value"] if csrf_token_input else ""
+
+            post_response = session.post(
+                f"https://huggingface.co{action_name}",
+                headers={
+                    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/58.0.3029.110 Safari/537.3"
+                },
+                cookies=session.cookies,
+                data={"csrf": csrf_token},
+                timeout=10
+            )
+
+            if post_response.status_code == 200:
+                print("[wake-up] 🚀 Restart triggered successfully. Waiting for the Space to be ready...")
+                time.sleep(60)  # attendre que le Space redémarre
+            else:
+                print(f"[wake-up] ❌ Failed to restart Space: {post_response.status_code}")
+        else:
+            print("[wake-up] ✅ Space is already awake.")
+
+    except Exception as e:
+        if "403 Forbidden" in str(e):
+            # This is ok, the space is already awake
+            print("[wake-up] ✅ Space is already awake.")
+        else:
+            print(f"[wake-up] ❌ Error waking up space: {e}")
+
+def get_wake_up_action_name(space_url: str) -> Optional[str]:
+    space_pattern = 'https://([\\w]+)-([\\w-]+).hf.space'
+    pattern = re.compile(space_pattern, flags=re.IGNORECASE)
+
+    if match := pattern.match(space_url):
+        # Extract the owner and the space name
+        owner = match.group(1)
+        space_name = match.group(2)
+
+        return f'/spaces/{owner}/{space_name}/start'