app.py

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 datetime import datetime, timedelta

TIMEOUT = 15
DELAY = 1


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("请求失败")


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)


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/api/v1/studio/{username}/{space['Name']}/start_expired"
                )

            return studios

    except requests.exceptions.Timeout as errt:
        print(f"请求超时: {errt}, retrying...")
        time.sleep(DELAY)
        return get_studios(username)

    except Exception 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://{space['id'].replace('/', '-')}.hf.space")

        return studios

    except requests.exceptions.Timeout as errt:
        print(f"请求超时: {errt}, retrying...")
        time.sleep(DELAY)
        return get_spaces(username)

    except Exception as err:
        print(f"请求发生错误: {err}")

    return []


def activate_space(url: str):
    success = "success"
    try:
        # 发送GET请求获取页面内容
        response = (
            requests.get(url, timeout=TIMEOUT)
            if ".hf.space" in url
            else requests.put(url)
        )
        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...")
        time.sleep(DELAY)
        return activate_space(url)

    except requests.RequestException as e:
        success = "success" if "500 Server Error:" in f"{e}" else f"{e}"

    except Exception as e:
        success = f"{e}"

    return success


def activate(hf_users: str, ms_users: str):
    if not hf_users:
        hf_users = os.getenv("hf_users")

    if not ms_users:
        ms_users = hf_users

    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=3):
    if schedule.get_jobs():
        return

    if not hf_users:
        hf_users = os.getenv("hf_users")

    if not ms_users:
        ms_users = hf_users

    print(f"监控开启中...每 {period} 小时触发")
    fixed_activate = partial(activate, hf_users=hf_users, ms_users=ms_users)
    schedule.every(period).hours.do(fixed_activate)
    while True:
        schedule.run_pending()
        time.sleep(DELAY)


def list_tasks():
    jobs = schedule.get_jobs()
    if jobs:
        details = f"{jobs}".replace("[", "").replace("]", "")
        return fix_datetime(
            details.split("functools.")[0] + "(" + details.split(") (")[-1]
        )

    return "None"


with gr.Blocks() as demo:
    gr.Interface(
        title="Start keeping all spaces active periodically",
        fn=monitor,
        inputs=[
            gr.Textbox(
                label="HuggingFace",
                placeholder="Usernames joint by ;",
            ),
            gr.Textbox(
                label="ModelScope",
                placeholder="Usernames joint by ;",
            ),
        ],
        outputs=None,
        allow_flagging="never",
    )
    gr.Interface(
        title="See current task status",
        fn=list_tasks,
        inputs=None,
        outputs=gr.Textbox(label="Current task details"),
        allow_flagging="never",
    )
    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="never",
    )

demo.launch()