replace

README.md

@@ -1,6 +1,6 @@
 ---
-title: Keep Demos Alive
-emoji: 🕓
+title: Online Tools II
+emoji: 🛠️
 colorFrom: blue
 colorTo: yellow
 sdk: gradio
@@ -8,5 +8,5 @@ sdk_version: 5.22.0
 app_file: app.py
 pinned: false
 license: apache-2.0
-short_description: Keep all your spaces/studios alive
+short_description: Online tool collection 2
 ---

app.py

@@ -1,101 +1,36 @@
-import time
-import schedule
-import threading
 import gradio as gr
-from datetime import datetime
-from modules.smtp import send_email
-from modules.activate import trigger
-from modules.times import fix_datetime, time_diff
-from modules.restart import self_restart, test_hf_restart, test_ms_restart
-from utils import START_TIME, PERIOD, DELAY
+from modules.qr import qrcode
+from modules.smtp import smtp_tester
+from modules.trans import translator
+from modules.url import url_shortner
+from utils import EN_US
 
+ZH2EN = {
+    "# 在线工具合集2": "# Online Tools Collection II",
+    "二维码生成": "QR Code",
+    "SMTP 测试": "SMTP Test",
+    "翻译器": "Translator",
+    "短链接生成": "URL Shortner",
+}
 
-def run_schedule():
-    global START_TIME
-    START_TIME = datetime.now()
-    trigger()
-    while True:
-        schedule.run_pending()
-        time.sleep(DELAY)
 
-
-def monitor(period=PERIOD):
-    print(f"Monitor is on and triggered every {period}h...")
-    schedule.every(47).hours.do(self_restart)
-    schedule.every(int(period)).hours.do(trigger)
-    threading.Thread(target=run_schedule, daemon=True).start()
-
-
-# UI func
-def tasklst():
-    status = "Success"
-    logs = None
-    try:
-        logs = f"\nHas been running for {time_diff(START_TIME, datetime.now())}\n"
-        jobs = schedule.get_jobs()
-        for job in jobs:
-            prev_run = ""
-            last_run = fix_datetime(job.last_run)
-            if last_run:
-                prev_run = f"last run: {last_run}, "
-
-            next_run = fix_datetime(job.next_run)
-            logs += f"\nEvery {job.interval}h do {job.job_func.__name__} ({prev_run}next run: {next_run})\n"
-
-    except Exception as e:
-        status = f"{e}"
-
-    return status, logs
+def _L(zh_txt: str):
+    return ZH2EN[zh_txt] if EN_US else zh_txt
 
 
 if __name__ == "__main__":
-    monitor()
     with gr.Blocks() as demo:
-        gr.Markdown("# Keep Spaces/Studios Alive")
-        with gr.Row():
-            with gr.Column():
-                task_btn = gr.Button("See current task status")
-                once_btn = gr.Button("Trigger once manually")
-                smtp_btn = gr.Button("SMTP test")
-                with gr.Tab("HF token test"):
-                    hf_txt = gr.Textbox(
-                        label="Restart a space with permissions",
-                        placeholder="username/repo",
-                        value="Genius-Society/online_tools",
-                    )
-                    with gr.Row():
-                        clhf_btn = gr.Button("Clear")
-                        hf_btn = gr.Button("Restart to test token validity")
+        gr.Markdown(_L("# 在线工具合集2"))
+        with gr.Tab(_L("二维码生成")):
+            qrcode()
 
-                with gr.Tab("MS cookie test"):
-                    ms_txt = gr.Textbox(
-                        label="Restart a studio with permissions",
-                        placeholder="username/repo",
-                        value="Genius-Society/online_tools",
-                    )
-                    with gr.Row():
-                        clms_btn = gr.Button("Clear")
-                        ms_btn = gr.Button("Restart to test cookie validity")
+        with gr.Tab(_L("SMTP 测试")):
+            smtp_tester()
 
-            with gr.Column():
-                status_bar = gr.Textbox(label="Status", show_copy_button=True)
-                gr.Markdown("Logs")
-                logs_bar = gr.Markdown(container=True, show_copy_button=True)
+        with gr.Tab(_L("翻译器")):
+            translator()
 
-            task_btn.click(fn=tasklst, outputs=[status_bar, logs_bar])
-            once_btn.click(fn=trigger, outputs=[status_bar, logs_bar])
-            smtp_btn.click(fn=send_email, outputs=[status_bar, logs_bar])
-            hf_btn.click(
-                fn=test_hf_restart,
-                inputs=[hf_txt],
-                outputs=[status_bar, logs_bar],
-            )
-            clhf_btn.click(fn=lambda: None, outputs=hf_txt)
-            ms_btn.click(
-                fn=test_ms_restart,
-                inputs=[ms_txt],
-                outputs=[status_bar, logs_bar],
-            )
-            clms_btn.click(fn=lambda: None, outputs=ms_txt)
+        with gr.Tab(_L("短链接生成")):
+            url_shortner()
 
     demo.launch()

modules/activate.py

@@ -1,193 +0,0 @@
-import json
-import time
-import requests
-import threading
-from tqdm import tqdm
-from datetime import datetime
-from huggingface_hub import HfApi
-from modules.smtp import send_email
-from modules.times import fix_datetime
-from modules.restart import restart_repos
-from utils import HF_DOMAIN, HEADER, MS_HEADER, TIMEOUT, MS_DOMAIN, DELAY, USERS
-
-
-def get_space_status(repo_id: str):
-    response: list = requests.get(
-        url=f"{HF_DOMAIN}/api/spaces/semantic-search",
-        params={"q": repo_id, "includeNonRunning": "true"},
-        headers=HEADER,
-        timeout=TIMEOUT,
-    ).json()
-
-    if response:
-        for repo in response:
-            if repo["id"] == repo_id:
-                return repo["runtime"]["stage"]
-
-    return "ERROR"
-
-
-def get_spaces(username: str):
-    sleepings, errors = [], []
-    try:
-        spaces = HfApi().list_spaces(author=username)
-        for space in spaces:
-            status = get_space_status(space.id)
-            if status == "SLEEPING":
-                space_id = space.id.replace("/", "-").replace("_", "-").lower()
-                if space.sdk == "gradio":
-                    sleepings.append(f"https://{space_id}.hf.space")
-                else:
-                    sleepings.append(f"https://{space_id}.static.hf.space")
-
-            elif "ERROR" in status:
-                errors.append(f"{HF_DOMAIN}/spaces/{space.id}")
-
-        return sleepings, errors
-
-    except Exception as e:
-        print(f"An error occurred in the request: {e}")
-
-    return [], []
-
-
-def activate_space(url: str):
-    try:
-        response = requests.get(url, headers=HEADER, timeout=TIMEOUT)
-        response.raise_for_status()
-
-    except Exception as e:
-        print(e)
-
-
-def check_ms_login():
-    try:
-        response = requests.get(
-            "https://www.modelscope.cn/api/v1/users/login/info",
-            headers=MS_HEADER,
-            timeout=TIMEOUT,
-        )
-        response.raise_for_status()
-
-    except Exception as e:
-        send_email(f"ModelScope cookie 失效: {e}")
-
-
-def get_studios(username: str):
-    try:
-        response = requests.put(
-            f"{MS_DOMAIN}/api/v1/studios/{username}/list",
-            data=json.dumps(
-                {
-                    "PageNumber": 1,
-                    "PageSize": 1000,
-                    "Name": "",
-                    "SortBy": "gmt_modified",
-                    "Order": "desc",
-                }
-            ),
-            headers=MS_HEADER,
-            timeout=TIMEOUT,
-        )
-        response.raise_for_status()
-        spaces: list = response.json()["Data"]["Studios"]
-        if spaces:
-            studios, errors = [], []
-            for space in spaces:
-                status = space["Status"]
-                if status == "Expired":
-                    studios.append(f"{username}/{space['Name']}")
-                elif status == "Failed":
-                    errors.append(f"{MS_DOMAIN}/studios/{username}/{space['Name']}")
-
-            return studios, errors
-
-    except requests.exceptions.Timeout as e:
-        print(f"Timeout: {e}, retrying...")
-        time.sleep(DELAY)
-        return get_studios(username)
-
-    except Exception as e:
-        print(f"Requesting error: {e}")
-
-    return [], []
-
-
-def activate_studio(repo: str, holding_delay=5):
-    repo_page = f"{MS_DOMAIN}/studios/{repo}"
-    status_api = f"{MS_DOMAIN}/api/v1/studio/{repo}/status"
-    start_expired_api = f"{MS_DOMAIN}/api/v1/studio/{repo}/start_expired"
-    try:
-        response = requests.put(start_expired_api, headers=MS_HEADER, timeout=TIMEOUT)
-        response.raise_for_status()
-        while (
-            requests.get(status_api, headers=MS_HEADER, timeout=TIMEOUT).json()["Data"][
-                "Status"
-            ]
-            != "Running"
-        ):
-            requests.get(repo_page, headers=MS_HEADER, timeout=TIMEOUT)
-            time.sleep(holding_delay)
-
-    except requests.exceptions.Timeout as e:
-        print(f"Failed to activate {repo}: {e}, retrying...")
-        activate_studio(repo)
-
-    except Exception as e:
-        print(e)
-
-
-# UI func
-def trigger(users=USERS):
-    status = "Success"
-    logs = None
-    try:
-        spaces, studios, failures = [], [], []
-        usernames = users.split(";")
-        for user in tqdm(usernames, desc="Collecting spaces"):
-            username = user.strip()
-            if username:
-                sleeps, errors = get_spaces(username)
-                spaces += sleeps
-                failures += errors
-                time.sleep(DELAY)
-
-        for space in tqdm(spaces, desc="Activating spaces"):
-            activate_space(space)
-            time.sleep(DELAY)
-
-        check_ms_login()
-        for user in tqdm(usernames, desc="Collecting studios"):
-            username = user.strip()
-            if username:
-                sleeps, errors = get_studios(username)
-                studios += sleeps
-                failures += errors
-                time.sleep(DELAY)
-
-        for studio in tqdm(studios, desc="Activating studios"):
-            threading.Thread(
-                target=activate_studio, args=(studio,), daemon=True
-            ).start()
-            time.sleep(DELAY)
-
-        logs = (
-            "\n".join(spaces + studios)
-            + f"\n[{fix_datetime(datetime.now())}] Activation complete!\n"
-        )
-        print(logs)
-        content = ""
-        for failure in failures:
-            errepo: str = failure
-            errepo = errepo.replace(HF_DOMAIN, "").replace(MS_DOMAIN, "")
-            content += f"<br><a href='{failure}'>{errepo[1:]}</a><br>"
-
-        if content:
-            send_email(content)
-
-        restart_repos()
-
-    except Exception as e:
-        status = f"{e}"
-
-    return status, logs

modules/qr.py

@@ -0,0 +1,47 @@
+import gradio as gr
+from utils import download_file, EN_US, API_QR, TMP_DIR
+
+ZH2EN = {
+    "二维码输出尺寸": "Image size",
+    "输入文本": "Input text",
+    "输出二维码": "QR code",
+    "输入文字在线生成二维码": "Enter text to generate a QR code.",
+    "状态栏": "Status",
+}
+
+
+def _L(zh_txt: str):
+    return ZH2EN[zh_txt] if EN_US else zh_txt
+
+
+def infer(img_size: int, input_txt: str):
+    status = "Success"
+    img = None
+    try:
+        if (not input_txt) or input_txt == "0":
+            raise ValueError("Please input valid text!")
+
+        img = download_file(
+            f"{API_QR}/?size={img_size}x{img_size}&data={input_txt}",
+            f"{TMP_DIR}/qrcode.jpg",
+        )
+
+    except Exception as e:
+        status = f"{e}"
+
+    return status, img
+
+
+def qrcode():
+    return gr.Interface(
+        fn=infer,
+        inputs=[
+            gr.Slider(35, 1000, 217, label=_L("二维码输出尺寸")),
+            gr.Textbox(label=_L("输入文本"), placeholder=_L("输入文字在线生成二维码")),
+        ],
+        outputs=[
+            gr.Textbox(label=_L("状态栏"), show_copy_button=True),
+            gr.Image(label=_L("输出二维码"), show_share_button=False),
+        ],
+        flagging_mode="never",
+    )

modules/restart.py

@@ -1,80 +0,0 @@
-import requests
-from huggingface_hub import HfApi
-from modules.smtp import send_email
-from utils import HF_DOMAIN, MS_DOMAIN, HF_TK, MS_HEADER, TIMEOUT, REPOS
-
-HF_API = HfApi(token=HF_TK)
-
-
-def restart_private_studio(repo: str):
-    response = requests.put(
-        f"{MS_DOMAIN}/api/v1/studio/{repo}/reset_restart",
-        headers=MS_HEADER,
-        timeout=TIMEOUT,
-    )
-    response.raise_for_status()
-    return f"\n{response.text}\n"
-
-
-def restart_repos():
-    if REPOS:
-        repos = REPOS.split(";")
-        for repo in repos:
-            private_repo = repo.strip()
-            print(f"Restarting {private_repo} space & studio...")
-            restart_private_studio(private_repo)
-            HF_API.restart_space(private_repo)
-
-
-def self_restart(me="kakamond/keeps_alive"):
-    try:
-        spaces = HF_API.list_spaces(author=me.split("/")[0])
-        for space in spaces:
-            if space.id != me and not space.disabled:
-                HF_API.restart_space(space.id)
-
-        HF_API.restart_space(me)
-
-    except Exception as e:
-        send_email(f"Failed to self-restart: {e}")
-
-
-# UI func
-def test_hf_restart(repo: str):
-    status = "Success"
-    logs = None
-    try:
-        if not repo:
-            raise ValueError("Empty repo!")
-
-        if f"{HF_DOMAIN}/spaces/" in repo:
-            repo = repo.split("/spaces/")[1].split("/")[:2]
-            repo = "/".join(repo)
-
-        logs = f"\n{HF_API.restart_space(repo)}\n"
-        logs += f"\nSuccessfully restart {HF_DOMAIN}/spaces/{repo}\n"
-
-    except Exception as e:
-        status = f"{e}"
-
-    return status, logs
-
-
-def test_ms_restart(repo: str):
-    status = "Success"
-    logs = None
-    try:
-        if not repo:
-            raise ValueError("Empty repo!")
-
-        if f"{MS_DOMAIN}/studios/" in repo:
-            repo = repo.split("/studios/")[1].split("/")[:2]
-            repo = "/".join(repo)
-
-        logs = restart_private_studio(repo)
-        logs += f"\nSuccessfully restart {MS_DOMAIN}/studios/{repo}\n"
-
-    except Exception as e:
-        status = f"{e}"
-
-    return status, logs

modules/smtp.py

@@ -1,61 +1,66 @@
 import requests
-from utils import EMAIL, SMTP, TAG, HOST, PORT
-
-
-def email_api(
-    content: str,
-    title="Runtime error detected",
-    header="Please fix following repo(s):",
-    email=EMAIL,
-    password=SMTP,
-    target=TAG,
-    host=HOST,
-    port=PORT,
-):
-    body = f"""
-    <html>
-        <body>
-            <h2>{header}</h2>
-            {content}
-        </body>
-    </html>
-    """
-    response = requests.get(
-        "http://api.mmp.cc/api/mail",
-        params={
-            "host": host,
-            "Port": port,
-            "key": password,  # apikey
-            "email": email,  # from
-            "mail": target,  # to
-            "title": title,  # subject
-            "name": "Keep Spaces Alive",  # nickname
-            "text": body,  # content
-        },
-    )
-    if response.status_code == 200:
-        result: dict = response.json()
-        if result.get("status") == "success":
-            return "Email sent successfully!"
-        else:
-            raise ConnectionError(f"Failed to send email: {result.get('message')}")
+import gradio as gr
+from utils import API_SMTP, EN_US
 
-    else:
-        raise ConnectionError(
-            f"Request failed with status code: {response.status_code}"
-        )
+ZH2EN = {
+    "收信人邮箱": "To email",
+    "标题": "Title",
+    "测试标题": "Test title",
+    "正文": "Content",
+    "SMTP 在线测试工具": "SMTP online tester",
+    "发信人昵称": "Sender name",
+    "测试昵称": "Test nickname",
+    "发信人邮箱": "From email",
+    "应用密钥": "API key",
+    "SMTP 服务器": "SMTP host",
+    "端口": "Port",
+    "发送状态": "Status",
+}
+
+
+def _L(zh_txt: str):
+    return ZH2EN[zh_txt] if EN_US else zh_txt
 
 
-# UI func
-def send_email(content="Test SMTP"):
-    status = "Success"
-    logs = None
+def infer(target, title, content, name, email, password, host, port):
     try:
-        status = email_api(content)
-        logs = f"\nEmail content: {content}\n"
+        response = requests.get(
+            API_SMTP,
+            params={
+                "host": host,
+                "Port": port,
+                "key": password,  # apikey
+                "email": email,  # from
+                "mail": target,  # to
+                "title": title,  # subject
+                "name": name,  # nickname
+                "text": content,  # content
+            },
+        )
+        if response.status_code == 200:
+            result: dict = response.json()
+            return result.get("status")
+
+        else:
+            raise ConnectionError(f"{response.status_code}")
 
     except Exception as e:
-        status = f"{e}"
-        print(status)
+        return f"{e}"
 
-    return status, logs
+
+def smtp_tester():
+    return gr.Interface(
+        fn=infer,
+        inputs=[
+            gr.Textbox(label=_L("收信人邮箱"), placeholder="Recipient"),
+            gr.Textbox(label=_L("标题"), value=_L("测试标题")),
+            gr.TextArea(label=_L("正文"), value=_L("SMTP 在线测试工具")),
+            gr.Textbox(label=_L("发信人昵称"), value=_L("测试昵称")),
+            gr.Textbox(label=_L("发信人邮箱"), placeholder="Sender"),
+            gr.Textbox(label=_L("应用密钥"), placeholder="SMTP password"),
+            gr.Textbox(label=_L("SMTP 服务器"), value="smtp.163.com"),
+            gr.Slider(label=_L("端口"), minimum=0, maximum=65535, step=1, value=25),
+        ],
+        outputs=gr.TextArea(label=_L("发送状态"), show_copy_button=True),
+        flagging_mode="never",
+    )

modules/times.py

@@ -1,21 +0,0 @@
-from datetime import datetime
-from zoneinfo import ZoneInfo
-from tzlocal import get_localzone
-
-
-def fix_datetime(naive_time: datetime, target_tz=ZoneInfo("Asia/Shanghai")):
-    if not naive_time:
-        return None
-
-    local_tz = get_localzone()
-    aware_local = naive_time.replace(tzinfo=local_tz)
-    return aware_local.astimezone(target_tz).strftime("%Y-%m-%d %H:%M:%S")
-
-
-def time_diff(time1: datetime, time2: datetime):
-    t_diff = time2 - time1
-    days = t_diff.days
-    hours = t_diff.seconds // 3600
-    minutes = (t_diff.seconds % 3600) // 60
-    seconds = t_diff.seconds % 60
-    return f"{days} d {hours} h {minutes} m {seconds} s"

modules/trans.py

@@ -0,0 +1,69 @@
+import json
+import requests
+import gradio as gr
+from utils import API_TRANS, KEY_TRANS, EN_US
+
+ZH2EN = {
+    "输入文本区域": "Input text area",
+    "在这里输入文本...": "Type the text here...",
+    "模式": "Mode",
+    "翻译结果": "Translation results",
+    "状态栏": "Status",
+}
+
+
+def _L(zh_txt: str):
+    return ZH2EN[zh_txt] if EN_US else zh_txt
+
+
+def infer(source, direction):
+    status = "Success"
+    result = None
+    try:
+        if not source or not direction:
+            raise ValueError("请输入有效文本!")
+
+        response = requests.request(
+            "POST",
+            API_TRANS,
+            data=json.dumps(
+                {
+                    "source": source,
+                    "trans_type": direction,
+                    "request_id": "demo",
+                    "detect": True,
+                }
+            ),
+            headers={
+                "content-type": "application/json",
+                "x-authorization": f"token {KEY_TRANS}",
+            },
+        )
+
+        result = json.loads(response.text)["target"]
+
+    except Exception as e:
+        status = f"{e}"
+
+    return status, result
+
+
+def translator():
+    return gr.Interface(
+        fn=infer,
+        inputs=[
+            gr.TextArea(label=_L("输入文本区域"), placeholder=_L("在这里输入文本...")),
+            gr.Textbox(label=_L("模式"), value="auto2en"),
+        ],
+        outputs=[
+            gr.Textbox(label=_L("状态栏"), show_copy_button=True),
+            gr.TextArea(label=_L("翻译结果"), show_copy_button=True),
+        ],
+        flagging_mode="never",
+        examples=[
+            ["这是最好的翻译服务。", "auto2ja"],
+            ["これは最高の翻訳サービスです。", "auto2en"],
+            ["This is the best translation service.", "auto2zh"],
+        ],
+        cache_examples=False,
+    )

modules/url.py

@@ -0,0 +1,58 @@
+import json
+import requests
+import gradio as gr
+from utils import is_valid_url, HEADER, EN_US, API_URL
+
+ZH2EN = {
+    "输入长链接": "Input a long URL",
+    "输出短链接": "Output short URL",
+    "预览短链接": "Preview short URL",
+    "将长链接转换为短的、易于共享的链接": "Convert long urls into short, easy-to-share links",
+    "状态栏": "Status",
+}
+
+
+def _L(zh_txt: str):
+    return ZH2EN[zh_txt] if EN_US else zh_txt
+
+
+# outer func
+def infer(longUrl: str):
+    status = "Success"
+    shortUrl = preview = None
+    try:
+        response = requests.post(API_URL, json={"url": longUrl}, headers=HEADER)
+        if response.status_code == 200:
+            shortUrl = json.loads(response.text)["shortUrl"]
+        else:
+            raise ConnectionError(response.text)
+
+        if is_valid_url(shortUrl):
+            preview = f'<a href="{shortUrl}" target="_blank">{shortUrl}</a>'
+
+    except Exception as e:
+        status = f"{e}"
+
+    return status, shortUrl, preview
+
+
+def url_shortner():
+    return gr.Interface(
+        fn=infer,
+        inputs=gr.Textbox(
+            label=_L("输入长链接"),
+            placeholder=_L("将长链接转换为短的、易于共享的链接"),
+        ),
+        outputs=[
+            gr.Textbox(label=_L("状态栏"), show_copy_button=True),
+            gr.Textbox(label=_L("输出短链接"), show_copy_button=True),
+            gr.HTML(
+                container=True,
+                show_label=True,
+                label=_L("预览短链接"),
+            ),
+        ],
+        flagging_mode="never",
+        examples=["https://www.bing.com", "https://www.baidu.com"],
+        cache_examples=False,
+    )

utils.py

@@ -1,33 +1,50 @@
 import os
+import re
+import shutil
+import requests
 
-PERIOD = os.getenv("period")
-USERS = os.getenv("users")
-REPOS = os.getenv("repos")
-EMAIL = os.getenv("email")
-SMTP = os.getenv("smtp")
-HF_TK = os.getenv("hftk")
-MS_CK = os.getenv("msck")
-TAG = os.getenv("target")
-if not TAG:
-    TAG = EMAIL
-
-HOST = os.getenv("host")
-PORT = os.getenv("port")
-if not (HOST and PORT):
-    HOST = "smtp.qq.com"
-    PORT = 587
-else:
-    PORT = int(PORT)
-
-DELAY = 1
-TIMEOUT = None
-START_TIME = None
-HF_DOMAIN = "https://huggingface.co"
-MS_DOMAIN = "https://www.modelscope.cn"
+EN_US = os.getenv("LANG") != "zh_CN.UTF-8"
+API_QR = os.getenv("api_qr")
+API_URL = os.getenv("api_url")
+API_SMTP = os.getenv("api_smtp")
+API_TRANS = os.getenv("api_caiyun")
+KEY_TRANS = os.getenv("apikey_caiyun")
+if not (API_SMTP and API_TRANS and KEY_TRANS):
+    print("请检查环境变量")
+    exit()
+
+
+TMP_DIR = "./__pycache__"
 HEADER = {
-    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/58.0.3029.110 Safari/537"
-}
-MS_HEADER = {
-    "User-Agent": HEADER["User-Agent"],
-    "Cookie": MS_CK,
+    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/132.0.0.0 Safari/537.36 Edg/132.0.0.0",
 }
+
+
+def clean_dir(dir_path: str):
+    if os.path.exists(dir_path):
+        shutil.rmtree(dir_path)
+
+    os.makedirs(dir_path)
+
+
+def is_valid_url(url):
+    # 定义 URL 的正则表达式
+    pattern = re.compile(
+        r"^(https?://)?"  # 协议（http 或 https，可选）
+        r"([a-zA-Z0-9-]+\.)+[a-zA-Z]{2,}"  # 域名
+        r"(:\d+)?"  # 端口号（可选）
+        r"(/[^ ]*)?$"  # 路径（可选）
+    )
+    # 使用正则表达式匹配 URL
+    return bool(pattern.match(url))
+
+
+def download_file(url, local_filename):
+    clean_dir(os.path.dirname(local_filename))
+    response = requests.get(url, stream=True)
+    response.raise_for_status()
+    with open(local_filename, "wb") as f:
+        for chunk in response.iter_content(chunk_size=8192):
+            f.write(chunk)
+
+    return local_filename