support self activate

app.py

@@ -2,29 +2,48 @@ import os
 import re
 import json
 import time
+import random
+import string
 import requests
 import schedule
 import gradio as gr
 import pandas as pd
 from tqdm import tqdm
 from functools import partial
-from bs4 import BeautifulSoup
 from datetime import datetime, timedelta
-from urllib.parse import urlparse
 
 TIMEOUT = 15
 DELAY = 1
 
 
-def fix_datetime(text: str):
-    datetime_pattern = r"\b\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}\b"
+def start_monitor(url: str):
+    payload = {
+        "data": ["", ""],
+        "event_data": None,  # 使用None来表示null
+        "fn_index": 0,
+        "trigger_id": 11,
+        "session_hash": "".join(
+            random.choice(string.ascii_lowercase) for _ in range(11)
+        ),
+    }
+    response = requests.post(f"{url}/queue/join?", json=payload)
+    # 检查请求是否成功
+    if response.status_code == 200:
+        print("请求成功")
+        return response.text
+
+    return None
+
 
-    def add_six_hours(match):
-        datetime_str = match.group(0)
-        dt = datetime.strptime(datetime_str, "%Y-%m-%d %H:%M:%S")
-        dt_plus_six = dt + timedelta(hours=6)
-        return dt_plus_six.strftime("%Y-%m-%d %H:%M:%S")
+def add_six_hours(match):
+    datetime_str = match.group(0)
+    dt = datetime.strptime(datetime_str, "%Y-%m-%d %H:%M:%S")
+    dt_plus_six = dt + timedelta(hours=6)
+    return dt_plus_six.strftime("%Y-%m-%d %H:%M:%S")
 
+
+def fix_datetime(text: str):
+    datetime_pattern = r"\b\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}\b"
     return re.sub(datetime_pattern, add_six_hours, text)
 
 
@@ -60,16 +79,12 @@ def get_studios(username: str):
 
             return studios
 
-    except requests.exceptions.HTTPError as errh:
-        print(f"HTTP错误发生: {errh}")
-    except requests.exceptions.ConnectionError as errc:
-        print(f"连接错误发生: {errc}")
     except requests.exceptions.Timeout as errt:
         print(f"请求超时: {errt}, retrying...")
         time.sleep(DELAY)
         return get_studios(username)
 
-    except requests.exceptions.RequestException as err:
+    except Exception as err:
         print(f"请求发生错误: {err}")
 
     return []
@@ -93,19 +108,16 @@ def get_spaces(username: str):
         studios = []
         for space in spaces:
             if space["author"] == username:
-                studios.append(f"https://huggingface.co/spaces/{space['id']}")
+                studios.append(f"https://{username}-{space['id']}.hf.space")
 
         return studios
 
-    except requests.exceptions.HTTPError as errh:
-        print(f"HTTP错误发生: {errh}")
-
     except requests.exceptions.Timeout as errt:
         print(f"请求超时: {errt}, retrying...")
         time.sleep(DELAY)
         return get_spaces(username)
 
-    except requests.exceptions.RequestException as err:
+    except Exception as err:
         print(f"请求发生错误: {err}")
 
     return []
@@ -117,48 +129,13 @@ def activate_space(url: str):
         # 发送GET请求获取页面内容
         response = (
             requests.get(url, timeout=TIMEOUT)
-            if "https://huggingface.co/spaces" in url
+            if ".hf.space" in url
             else requests.put(url)
         )
-        response.raise_for_status()  # 确保请求成功
-        web_page_content = response.text
-
-        # 使用BeautifulSoup解析页面
-        soup = BeautifulSoup(web_page_content, "html.parser")
-
-        # 寻找class为btn-lg的按钮
-        button = soup.find("button", class_="btn-lg")
-        if not button:
-            # print("未找到class为'btn-lg'的按钮。")
-            return success
-
-        # 获取表单的action属性和method属性
-        form = button.find_parent("form")
-        if not form:
-            return "按钮未在表单内找到。"
-
-        form_action = form.get("action")
-        form_method = form.get("method", "GET").upper()  # 默认为GET
-        host = "https://" + urlparse(url).hostname
-        if not host in form_action:
-            form_action = host + form_action
-
-        # 提取表单数据
-        form_data = {
-            input_tag.get("name"): input_tag.get("value")
-            for input_tag in form.find_all("input")
-            if input_tag.has_attr("name")
-        }
-
-        # 发送请求提交表单
-        if form_method == "POST":
-            response = requests.post(form_action, data=form_data)
-        else:
-            response = requests.get(form_action, params=form_data)
-
-        # 打印响应内容
-        print("提交表单后的页面内容:")
-        print(response.text)
+        response.raise_for_status()
+
+        if "-keep-spaces-active.hf.space" in url:
+            start_monitor(url)
 
     except requests.exceptions.Timeout as errt:
         print(f"请求超时: {errt}, retrying...")
@@ -179,7 +156,7 @@ def activate(hf_users: str, ms_users: str):
         hf_users = os.getenv("hf_users")
 
     if not ms_users:
-        ms_users = os.getenv("ms_users")
+        ms_users = hf_users
 
     hf_usernames = hf_users.split(";")
     ms_usernames = ms_users.split(";")
@@ -213,7 +190,7 @@ def monitor(hf_users: str, ms_users: str, period=3):
         hf_users = os.getenv("hf_users")
 
     if not ms_users:
-        ms_users = os.getenv("ms_users")
+        ms_users = hf_users
 
     print(f"监控开启中...每 {period} 小时触发")
     fixed_activate = partial(activate, hf_users=hf_users, ms_users=ms_users)
@@ -234,7 +211,7 @@ def list_tasks():
     return "None"
 
 
-with gr.Blocks() as iface:
+with gr.Blocks() as demo:
     gr.Interface(
         title="Start keeping all spaces active periodically",
         fn=monitor,
@@ -275,4 +252,4 @@ with gr.Blocks() as iface:
         allow_flagging="never",
     )
 
-iface.launch()
+demo.launch()