File size: 3,858 Bytes
dd21b8c
 
 
 
 
a4ed0ef
 
dd21b8c
a4ed0ef
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
c6c3b30
a10794c
 
a4ed0ef
 
a10794c
 
a4ed0ef
a10794c
 
 
 
 
a4ed0ef
a10794c
 
 
 
 
 
 
a4ed0ef
dd21b8c
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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
import requests
from typing import Optional
from bs4 import BeautifulSoup
import time
import re
import os
from dotenv import load_dotenv

load_dotenv()

def get_url_list():
    # Liste d'URLs à pinger
    default_urls = os.getenv("DEFAULT_URLS", "")
    url_list = [url.strip() for url in default_urls.split(",") if url.strip()]

    return url_list

def add_url(url: str) -> None:
    """
    Add a new URL to the list of URLs.
    """
    if url not in get_url_list():
        os.environ["DEFAULT_URLS"] = ",".join([os.environ["DEFAULT_URLS"], url])
        print(f"Added URL: {url}")
    else:
        print(f"URL already exists: {url}")

def remove_url(url: str) -> None:
    """
    Remove a URL from the list of URLs.
    """
    url_list = get_url_list()
    if url in url_list:
        url_list.remove(url)
        os.environ["DEFAULT_URLS"] = ",".join(url_list)
        print(f"Removed URL: {url}")
    else:
        print(f"URL not found: {url}")

# Fonction Gradio pour ajouter / retirer des URLs
def manage_urls(action, url):
    url_list = get_url_list()

    if action == "Ajouter":
        if url and url not in url_list:
            add_url(url)
            message = f"✅ Ajouté: {url}"
        else:
            message = "⚠️ URL déjà présente ou vide"
    elif action == "Supprimer":
        if url in url_list:
            remove_url(url)
            message = f"❌ Supprimé: {url}"
        else:
            message = "⚠️ URL non trouvée"
    else:
        message = "❓ Action invalide"

    # Toujours retourner la liste actuelle formatée
    return message, "\n".join(get_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'