Update app.py

app.py

@@ -1,230 +1,230 @@
-import json
-import time
-import requests
-import schedule
-import gradio as gr
-import pandas as pd
-from tqdm import tqdm
-from bs4 import BeautifulSoup
-from urllib.parse import urlparse
-
-TIMEOUT = 15
-DELAY = 1
-
-
-def get_studios(username: str):
-    # 请求负载
-    payload = {
-        "PageNumber": 1,
-        "PageSize": 1000,
-        "Name": "",
-        "SortBy": "gmt_modified",
-        "Order": "desc",
-    }
-    try:
-        # 发送PUT请求
-        response = requests.put(
-            f"https://www.modelscope.cn/api/v1/studios/{username}/list",
-            data=json.dumps(payload),
-            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"
-            },
-            timeout=TIMEOUT,
-        )
-        # 检查请求是否成功
-        response.raise_for_status()
-        # 解析JSON响应
-        spaces: list = response.json()["Data"]["Studios"]
-        if spaces:
-            studios = []
-            for space in spaces:
-                studios.append(
-                    f"https://www.modelscope.cn/studios/{username}/{space['Name']}"
-                )
-
-            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:
-        print(f"请求发生错误: {err}")
-
-    return []
-
-
-def get_spaces(username: str):
-    try:
-        # 发送GET请求
-        response = requests.get(
-            "https://huggingface.co/spaces-json",
-            params={"sort": "trending", "search": username},
-            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"
-            },
-            timeout=TIMEOUT,
-        )
-        # 检查请求是否成功
-        response.raise_for_status()
-        # 解析JSON响应
-        spaces: list = response.json()["spaces"]
-        studios = []
-        for space in spaces:
-            if space["author"] == username:
-                studios.append(f"https://huggingface.co/spaces/{space['id']}")
-
-        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:
-        print(f"请求发生错误: {err}")
-
-    return []
-
-
-def activate_space(url: str):
-    success = "success"
-    try:
-        # 发送GET请求获取页面内容
-        response = requests.get(url, timeout=TIMEOUT)
-        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)
-
-    except requests.exceptions.Timeout as errt:
-        print(f"请求超时: {errt}, retrying...")
-        time.sleep(DELAY)
-        return activate_space(url)
-
-    except requests.RequestException as e:
-        success = f"{e}"
-
-    except Exception as e:
-        success = f"{e}"
-
-    return success
-
-
-def activate(hf_users: str, ms_users: str):
-    hf_usernames = hf_users.split(";")
-    ms_usernames = ms_users.split(";")
-    spaces = []
-    for user in tqdm(hf_usernames, desc="Collecting spaces..."):
-        username = user.strip()
-        if username:
-            spaces += get_spaces(username)
-            time.sleep(DELAY)
-
-    for user in tqdm(ms_usernames, desc="Collecting studios..."):
-        username = user.strip()
-        if username:
-            spaces += get_studios(username)
-            time.sleep(DELAY)
-
-    output = []
-    for space in tqdm(spaces, desc="Activating spaces..."):
-        output.append({"space": space, "status": activate_space(space)})
-        time.sleep(DELAY)
-
-    print("Activation complete!")
-    return pd.DataFrame(output)
-
-
-def monitor(hf_users: str, ms_users: str, trigger_time="12:30"):
-    if schedule.get_jobs():
-        return
-
-    print(f"监控开启中...每日触发时间：{trigger_time}")
-    schedule.every().day.at(trigger_time).do(lambda: activate(hf_users, ms_users))
-    while True:
-        schedule.run_pending()
-        time.sleep(DELAY)
-
-
-def list_tasks():
-    jobs = schedule.get_jobs()
-    if jobs:
-        return f"{jobs}".replace("[", "").replace("]", "")
-
-    return "None"
-
-
-with gr.Blocks() as iface:
-    gr.Interface(
-        title="Start keeping all spaces active daily",
-        fn=monitor,
-        inputs=[
-            gr.Textbox(label="HuggingFace", placeholder="Usernames joint by ;"),
-            gr.Textbox(label="ModelScope", placeholder="Usernames joint by ;"),
-        ],
-        outputs=None,
-        allow_flagging=False,
-    )
-    gr.Interface(
-        title="See current task status",
-        fn=list_tasks,
-        inputs=None,
-        outputs=gr.Textbox(label="Current task details"),
-        allow_flagging=False,
-    )
-    gr.Interface(
-        title="Test activation for all spaces once",
-        fn=activate,
-        inputs=[
-            gr.Textbox(label="HuggingFace", placeholder="Usernames joint by ;"),
-            gr.Textbox(label="ModelScope", placeholder="Usernames joint by ;"),
-        ],
-        outputs=gr.Dataframe(label="Activated spaces"),
-        allow_flagging=False,
-    )
-
-iface.launch()
+import json
+import time
+import requests
+import schedule
+import gradio as gr
+import pandas as pd
+from tqdm import tqdm
+from bs4 import BeautifulSoup
+from urllib.parse import urlparse
+
+TIMEOUT = 15
+DELAY = 1
+
+
+def get_studios(username: str):
+    # 请求负载
+    payload = {
+        "PageNumber": 1,
+        "PageSize": 1000,
+        "Name": "",
+        "SortBy": "gmt_modified",
+        "Order": "desc",
+    }
+    try:
+        # 发送PUT请求
+        response = requests.put(
+            f"https://www.modelscope.cn/api/v1/studios/{username}/list",
+            data=json.dumps(payload),
+            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"
+            },
+            timeout=TIMEOUT,
+        )
+        # 检查请求是否成功
+        response.raise_for_status()
+        # 解析JSON响应
+        spaces: list = response.json()["Data"]["Studios"]
+        if spaces:
+            studios = []
+            for space in spaces:
+                studios.append(
+                    f"https://www.modelscope.cn/studios/{username}/{space['Name']}"
+                )
+
+            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:
+        print(f"请求发生错误: {err}")
+
+    return []
+
+
+def get_spaces(username: str):
+    try:
+        # 发送GET请求
+        response = requests.get(
+            "https://huggingface.co/spaces-json",
+            params={"sort": "trending", "search": username},
+            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"
+            },
+            timeout=TIMEOUT,
+        )
+        # 检查请求是否成功
+        response.raise_for_status()
+        # 解析JSON响应
+        spaces: list = response.json()["spaces"]
+        studios = []
+        for space in spaces:
+            if space["author"] == username:
+                studios.append(f"https://huggingface.co/spaces/{space['id']}")
+
+        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:
+        print(f"请求发生错误: {err}")
+
+    return []
+
+
+def activate_space(url: str):
+    success = "success"
+    try:
+        # 发送GET请求获取页面内容
+        response = requests.get(url, timeout=TIMEOUT)
+        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)
+
+    except requests.exceptions.Timeout as errt:
+        print(f"请求超时: {errt}, retrying...")
+        time.sleep(DELAY)
+        return activate_space(url)
+
+    except requests.RequestException as e:
+        success = f"{e}"
+
+    except Exception as e:
+        success = f"{e}"
+
+    return success
+
+
+def activate(hf_users: str, ms_users: str):
+    hf_usernames = hf_users.split(";")
+    ms_usernames = ms_users.split(";")
+    spaces = []
+    for user in tqdm(hf_usernames, desc="Collecting spaces..."):
+        username = user.strip()
+        if username:
+            spaces += get_spaces(username)
+            time.sleep(DELAY)
+
+    for user in tqdm(ms_usernames, desc="Collecting studios..."):
+        username = user.strip()
+        if username:
+            spaces += get_studios(username)
+            time.sleep(DELAY)
+
+    output = []
+    for space in tqdm(spaces, desc="Activating spaces..."):
+        output.append({"space": space, "status": activate_space(space)})
+        time.sleep(DELAY)
+
+    print("Activation complete!")
+    return pd.DataFrame(output)
+
+
+def monitor(hf_users: str, ms_users: str, period=1):
+    if schedule.get_jobs():
+        return
+
+    print(f"监控开启中...每日触发")
+    schedule.every(period).days.do(lambda: activate(hf_users, ms_users))
+    while True:
+        schedule.run_pending()
+        time.sleep(DELAY)
+
+
+def list_tasks():
+    jobs = schedule.get_jobs()
+    if jobs:
+        return f"{jobs}".replace("[", "").replace("]", "")
+
+    return "None"
+
+
+with gr.Blocks() as iface:
+    gr.Interface(
+        title="Start keeping all spaces active daily",
+        fn=monitor,
+        inputs=[
+            gr.Textbox(label="HuggingFace", placeholder="Usernames joint by ;"),
+            gr.Textbox(label="ModelScope", placeholder="Usernames joint by ;"),
+        ],
+        outputs=None,
+        allow_flagging=False,
+    )
+    gr.Interface(
+        title="See current task status",
+        fn=list_tasks,
+        inputs=None,
+        outputs=gr.Textbox(label="Current task details"),
+        allow_flagging=False,
+    )
+    gr.Interface(
+        title="Test activation for all spaces once",
+        fn=activate,
+        inputs=[
+            gr.Textbox(label="HuggingFace", placeholder="Usernames joint by ;"),
+            gr.Textbox(label="ModelScope", placeholder="Usernames joint by ;"),
+        ],
+        outputs=gr.Dataframe(label="Activated spaces"),
+        allow_flagging=False,
+    )
+
+iface.launch()