major: init

.gitattributes

@@ -0,0 +1,35 @@
+*.7z filter=lfs diff=lfs merge=lfs -text
+*.arrow filter=lfs diff=lfs merge=lfs -text
+*.bin filter=lfs diff=lfs merge=lfs -text
+*.bz2 filter=lfs diff=lfs merge=lfs -text
+*.ckpt filter=lfs diff=lfs merge=lfs -text
+*.ftz filter=lfs diff=lfs merge=lfs -text
+*.gz filter=lfs diff=lfs merge=lfs -text
+*.h5 filter=lfs diff=lfs merge=lfs -text
+*.joblib filter=lfs diff=lfs merge=lfs -text
+*.lfs.* filter=lfs diff=lfs merge=lfs -text
+*.mlmodel filter=lfs diff=lfs merge=lfs -text
+*.model filter=lfs diff=lfs merge=lfs -text
+*.msgpack filter=lfs diff=lfs merge=lfs -text
+*.npy filter=lfs diff=lfs merge=lfs -text
+*.npz filter=lfs diff=lfs merge=lfs -text
+*.onnx filter=lfs diff=lfs merge=lfs -text
+*.ot filter=lfs diff=lfs merge=lfs -text
+*.parquet filter=lfs diff=lfs merge=lfs -text
+*.pb filter=lfs diff=lfs merge=lfs -text
+*.pickle filter=lfs diff=lfs merge=lfs -text
+*.pkl filter=lfs diff=lfs merge=lfs -text
+*.pt filter=lfs diff=lfs merge=lfs -text
+*.pth filter=lfs diff=lfs merge=lfs -text
+*.rar filter=lfs diff=lfs merge=lfs -text
+*.safetensors filter=lfs diff=lfs merge=lfs -text
+saved_model/**/* filter=lfs diff=lfs merge=lfs -text
+*.tar.* filter=lfs diff=lfs merge=lfs -text
+*.tar filter=lfs diff=lfs merge=lfs -text
+*.tflite filter=lfs diff=lfs merge=lfs -text
+*.tgz filter=lfs diff=lfs merge=lfs -text
+*.wasm filter=lfs diff=lfs merge=lfs -text
+*.xz filter=lfs diff=lfs merge=lfs -text
+*.zip filter=lfs diff=lfs merge=lfs -text
+*.zst filter=lfs diff=lfs merge=lfs -text
+*tfevents* filter=lfs diff=lfs merge=lfs -text

README.md

@@ -0,0 +1,13 @@
+---
+title: Civitai Asset Migration and Archiving Tool
+emoji: 🤗
+colorFrom: yellow
+colorTo: blue
+sdk: gradio
+sdk_version: 5.12.0
+app_file: app.py
+pinned: false
+license: mit
+---
+
+Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference

app.py

@@ -0,0 +1,275 @@
+import gradio as gr
+from gradio_modal import Modal
+from civitai_api import (search_civitai, download_civitai, select_civitai_item, add_civitai_item, get_civitai_tag, select_civitai_all_item,
+                           update_civitai_selection, update_civitai_checkbox, from_civitai_checkbox)
+from civitai_constants import (TYPE, BASEMODEL, SORT, PERIOD, FILETYPE)
+import time
+
+css = """
+.title { font-size: 3em; align-items: center; text-align: center; }
+.info { align-items: center; text-align: center; }
+.desc [src$='#float'] { float: right; margin: 20px; }
+#modal-window{ position: absolute; left: 50%; top: 50%; transform: translate(-50%, -50%); max-height: inherit; }
+.modal-container{ max-width:60vw; height:100vh; }
+.prose.desc{ padding: 0px 20px; }
+/* Video hover styles */
+.media-container {
+    position: relative;
+    width: 100%;
+    height: 100%;
+}
+.result{ padding: 16px; }
+
+.media-container img,
+.media-container video {
+    width: 100%;
+    height: 100%;
+    object-fit: contain;
+}
+
+.media-container video {
+    display: none;
+    position: absolute;
+    top: 0;
+    left: 0;
+    z-index: 2;
+}
+
+.gallery-item:hover .media-container video {
+    display: block;
+}
+
+.gallery-item:hover .media-container img {
+    opacity: 0.3;
+}
+
+.gallery-item {
+    position: relative;
+    cursor: pointer;
+    overflow: hidden;
+}
+"""
+
+# js = """
+# (() => {
+#     function setupGalleryHover() {
+#         const gallery = document.getElementById('gallery');
+#         if (!gallery) return;
+
+#         function handleMediaContainer(container) {
+#             const video = container.querySelector('video');
+#             if (!video) return;
+            
+#             video.muted = true;
+#             video.preload = "metadata";
+            
+#             container.addEventListener('mouseenter', () => {
+#                 if (video.paused) {
+#                     video.play().catch(() => {});
+#                 }
+#             });
+            
+#             container.addEventListener('mouseleave', () => {
+#                 if (!video.paused) {
+#                     video.pause();
+#                     video.currentTime = 0;
+#                 }
+#             });
+#         }
+
+#         function setupContainers() {
+#             const containers = gallery.querySelectorAll('.media-container');
+#             containers.forEach(handleMediaContainer);
+#         }
+
+#         setupContainers();
+        
+#         const observer = new MutationObserver(setupContainers);
+#         observer.observe(gallery, { childList: true, subtree: true });
+#     }
+
+#     if (document.readyState === 'loading') {
+#         document.addEventListener('DOMContentLoaded', setupGalleryHover);
+#     } else {
+#         setupGalleryHover();
+#     }
+# })();
+# """
+
+common_sense = """
+## The Common Sense Agreement
+
+This tool was created in response to recent censorship and model deactivations on Civitai. This Space should be treated as a tool for artists and creators to easily backup and archive models -- this is NOT a tool to blindly scrape and download every asset your heart desires. For those that wish to do so, may do so on their own local machines. Users of the tool are responsible for their own actions - you have been warned.
+
+By continuing on to the tool, you are assumed to have **acknowledged, understood, and are in agreeance** with the following common sense keypoints:
+
+1. **Storage is not free.** You are responsible for understanding your storage limits and the costs that will be incurred if you exceed your tier's free allocation.
+2. **Bandwidth is not free.** Misuse of the tool will likely be frowned upon by both Huggingface as well as Civitai, potentially leading to negative consequences for all. Don't be that guy. There are better ways if you must do so. 
+3. **Not all models are created equally.** Respect the licensing agreements and the wishes of creators, both big and small. 
+4. **Respect the FOSS community**, give back where you can, and be aware of your surroundings. Protect this industry as best you can, otherwise prepare for a wicked spanking when AGI is here.
+
+### For those with common sense, please continue.
+"""
+
+
+with gr.Blocks(fill_width=True, css=css, delete_cache=(3600, 3600)) as demo:
+    gr.Markdown("# 🚨 CivitAI Models & LoRA Conservation Project 📦", elem_classes="title")
+    
+    state = gr.State(value={})
+    
+    # Create browser state for API keys with error handling
+    api_keys_state = gr.BrowserState(["", ""], storage_key="civitai2hf_api_keys")
+    
+    with gr.Row():
+        show_btn = gr.Button("Show Modal")
+    with gr.Tabs() as tabs:
+        with gr.TabItem("Search & Download"):
+            with gr.Row():
+                # Left column for filters (1/3 width)
+                with gr.Column(scale=1):
+                    with gr.Group():
+                        gr.Markdown("## 2️⃣ Search Filters", container=True)
+                        with gr.Accordion("Type & Model", open=True):
+                            search_civitai_type = gr.CheckboxGroup(label="Type", choices=TYPE, value=["Checkpoint", "LORA"])
+                            search_civitai_basemodel = gr.CheckboxGroup(label="Base Model", choices=BASEMODEL, value=[])
+                            search_civitai_filetype = gr.CheckboxGroup(label="File type", visible=False,  choices=FILETYPE, value=["Model"])
+                        with gr.Accordion("Search Options", open=True):
+                            search_civitai_sort = gr.Radio(label="Sort", choices=SORT, value=SORT[0])
+                            search_civitai_period = gr.Radio(label="Period", choices=PERIOD, value="Month")
+                            search_civitai_limit =  gr.Slider(0, 100, value=50, step=10, label="Limit", info="Maximum items to query per page via API.")
+                            search_civitai_page = gr.Slider(0, 10, 1, step=1, label="Num Pages", info="If 0, retrieve all pages")
+
+                # Right column for results and download (2/3 width)
+                with gr.Column(scale=2):
+                    with gr.Group():
+                        gr.Markdown("## 3️⃣ Search Query", container=True)
+                        with gr.Row():
+                            search_civitai_query = gr.Textbox(label="Query", placeholder="flux", lines=1)
+                            search_civitai_tag = gr.Dropdown(label="Tag", choices=get_civitai_tag(), value=get_civitai_tag()[0], allow_custom_value=True)
+                            search_civitai_user = gr.Textbox(label="Username", lines=1)
+                        search_civitai_submit = gr.Button("Search on Civitai", variant="primary")
+                    
+                    with gr.Group():
+                        gr.Markdown("## 4️⃣ Select Models to Backup from Search Results", container=True)
+                        with gr.Row():
+                            search_civitai_desc = gr.Markdown(value="", visible=False, elem_classes="desc")
+                            search_civitai_json = gr.JSON(value={}, visible=False)
+                        with gr.Row(equal_height=True):
+                            with gr.Column(scale=9):
+                                with gr.Accordion("Gallery View ⚠️ Surprise NSFW + RAM Warning ⚠️", open=False):
+                                    search_civitai_gallery = gr.Gallery(
+                                        [], 
+                                        label="Select from Results", 
+                                        allow_preview=False, 
+                                        columns=5, 
+                                        elem_id="gallery", 
+                                        object_fit="contain", 
+                                        show_share_button=False, 
+                                        interactive=True,
+                                        preview=False,
+                                        height="auto"
+                                    )
+                                with gr.Accordion("List View", open=False):
+                                    search_civitai_result_checkbox = gr.CheckboxGroup(label="", choices=[], value=[])
+                                search_civitai_result = gr.Dropdown(label="Selected Models from Search", choices=[("", "")], value=[],
+                                                                    allow_custom_value=True, visible=True, multiselect=True)
+                                search_civitai_result_info = gr.Markdown("Standing by.", elem_classes="info")
+                        with gr.Row():
+                            search_civitai_add = gr.Button("5️⃣ Add Selected Models to DL List", variant="primary")
+                            search_civitai_select_all = gr.Button("Select All", variant="secondary")
+                    
+                    with gr.Group():
+                        gr.Markdown("## 6️⃣🚨 Ensure you have set your correct API key settings before moving on.", container=True)
+                        dl_url = gr.Textbox(label="Download URL(s)", placeholder="https://civitai.com/api/download/models/28907\n...", value="", lines=3, max_lines=255, interactive=True)
+                        with gr.Row():
+                            newrepo_id = gr.Textbox(label="Your Repo ID", placeholder="yourid/yourrepo", value="", max_lines=1)
+                            newrepo_type = gr.Radio(label="Repo Type", choices=["model", "dataset"], value="model")
+                            with gr.Group():
+                                is_private = gr.Checkbox(label="Private Repo", value=False)
+                                is_rename = gr.Checkbox(label="Auto rename", value=True)
+                        with gr.Row():
+                            is_info = gr.Checkbox(label="Upload Civitai Metadata Files to HF", value=True)
+                        
+                        run_button = gr.Button(value="7️⃣ Create Repo & Backup Models", variant="primary")
+                        uploaded_urls = gr.CheckboxGroup(visible=False, choices=[], value=[]) # hidden
+                        urls_md = gr.Markdown("<br><br>", elem_classes="result")
+                        urls_remain = gr.Textbox("Remaining URLs", value="", show_copy_button=True, visible=False)
+
+        with gr.TabItem("Settings 1️⃣"):
+            with gr.Column():
+                with gr.Group():
+                    with gr.Row():
+                        with gr.Column():
+                            civitai_key = gr.Textbox(label="Your Civitai Key", value="", max_lines=1)
+                            gr.Markdown("Your Civitai API key is available at [https://civitai.com/user/account](https://civitai.com/user/account).", elem_classes="info")
+                        with gr.Column():
+                            hf_token = gr.Textbox(label="Your HF write token", placeholder="hf_...", value="", max_lines=1)
+                            gr.Markdown("Your token is available at [hf.co/settings/tokens](https://huggingface.co/settings/tokens).", elem_classes="info")
+                    
+                    # Add save status message
+                    saved_message = gr.Markdown("✅ API keys saved to browser storage", visible=False)
+                    
+                    # Load saved keys when the app starts with error handling
+                    @demo.load(inputs=[api_keys_state], outputs=[civitai_key, hf_token])
+                    def load_api_keys(saved_keys):
+                        try:
+                            if not saved_keys or not isinstance(saved_keys, list) or len(saved_keys) != 2:
+                                return "", ""
+                            # Ensure we're not trying to use None values
+                            civitai = saved_keys[0] or ""
+                            hf = saved_keys[1] or ""
+                            return civitai, hf
+                        except Exception as e:
+                            print(f"Error loading API keys: {e}")
+                            return "", ""
+                    
+                    # Save keys when they change with improved error handling
+                    @gr.on([civitai_key.change, hf_token.change], inputs=[civitai_key, hf_token], outputs=[api_keys_state, saved_message])
+                    def save_api_keys(civitai_key, hf_token):
+                        try:
+                            # Ensure we're storing strings, not None
+                            civitai = str(civitai_key or "")
+                            hf = str(hf_token or "")
+                            timestamp = time.strftime("%I:%M:%S %p")
+                            
+                            if civitai or hf:
+                                return [civitai, hf], gr.Markdown(
+                                    f"✅ API keys saved to browser storage at {timestamp}",
+                                    visible=True
+                                )
+                            return ["", ""], gr.Markdown("", visible=False)
+                        except Exception as e:
+                            print(f"Error saving API keys: {e}")
+                            return ["", ""], gr.Markdown("Error saving API keys", visible=True)
+
+    with Modal(visible=True, elem_id="modal-window") as modal:
+        gr.Markdown(common_sense)
+    gr.on(
+        triggers=[run_button.click],
+        fn=download_civitai,
+        inputs=[dl_url, civitai_key, hf_token, uploaded_urls, newrepo_id, newrepo_type, is_private, is_info, is_rename],
+        outputs=[uploaded_urls, urls_md, urls_remain],
+        queue=True,
+    )
+    gr.on(
+        triggers=[search_civitai_submit.click, search_civitai_query.submit, search_civitai_user.submit],
+        fn=search_civitai,
+        inputs=[search_civitai_query, search_civitai_type, search_civitai_basemodel, search_civitai_sort,
+                search_civitai_period, search_civitai_tag, search_civitai_user, search_civitai_limit,
+                search_civitai_page, search_civitai_filetype, civitai_key, search_civitai_gallery, state],
+        outputs=[search_civitai_result, search_civitai_desc, search_civitai_submit, search_civitai_query, search_civitai_gallery,
+                 search_civitai_result_checkbox, search_civitai_result_info, state],
+        queue=False,
+        show_api=False,
+    )
+    search_civitai_result.change(select_civitai_item, [search_civitai_result, state], [search_civitai_desc, search_civitai_json, state], queue=False, show_api=False)\
+    .success(update_civitai_checkbox, [search_civitai_result], [search_civitai_result_checkbox], queue=True, show_api=False)
+    search_civitai_result_checkbox.select(from_civitai_checkbox, [search_civitai_result_checkbox], [search_civitai_result], queue=False, show_api=False)
+    search_civitai_add.click(add_civitai_item, [search_civitai_result, dl_url], [dl_url], queue=False, show_api=False)
+    search_civitai_select_all.click(select_civitai_all_item, [search_civitai_select_all, state], [search_civitai_select_all, search_civitai_result], queue=False, show_api=False)
+    search_civitai_gallery.select(update_civitai_selection, [search_civitai_result, state], [search_civitai_result], queue=False, show_api=False)
+    
+    show_btn.click(lambda: Modal(visible=True), None, modal)
+
+demo.queue()
+demo.launch(ssr_mode=False)

civitai_api.py

@@ -0,0 +1,446 @@
+import gradio as gr
+from huggingface_hub import HfApi, hf_hub_url
+import os
+from pathlib import Path
+import gc
+import requests
+from requests.adapters import HTTPAdapter
+from urllib3.util import Retry
+from civitai_constants import PERIOD, SORT
+from utils import (get_token, set_token, is_repo_exists, get_user_agent, get_download_file,
+                   list_uniq, list_sub, duplicate_hf_repo, HF_SUBFOLDER_NAME, get_state, set_state)
+import re
+from PIL import Image
+import json
+import pandas as pd
+import tempfile
+import hashlib
+
+# Huge shoutout to @John6666, saved me many hours.
+
+TEMP_DIR = tempfile.mkdtemp()
+
+
+def parse_urls(s):
+    url_pattern = "https?://[\\w/:%#\\$&\\?\\(\\)~\\.=\\+\\-]+"
+    try:
+        urls = re.findall(url_pattern, s)
+        return list(urls)
+    except Exception:
+        return []
+
+
+def parse_repos(s):
+    repo_pattern = r'[^\w_\-\.]?([\w_\-\.]+/[\w_\-\.]+)[^\w_\-\.]?'
+    try:
+        s = re.sub("https?://[\\w/:%#\\$&\\?\\(\\)~\\.=\\+\\-]+", "", s)
+        repos = re.findall(repo_pattern, s)
+        return list(repos)
+    except Exception:
+        return []
+
+
+def to_urls(l: list[str]):
+    return "\n".join(l)
+
+
+def uniq_urls(s):
+    return to_urls(list_uniq(parse_urls(s) + parse_repos(s)))
+
+
+def upload_safetensors_to_repo(filename, repo_id, repo_type, is_private, progress=gr.Progress(track_tqdm=True)):
+    output_filename = Path(filename).name
+    hf_token = get_token()
+    api = HfApi(token=hf_token)
+    try:
+        if not is_repo_exists(repo_id, repo_type): api.create_repo(repo_id=repo_id, repo_type=repo_type, token=hf_token, private=is_private)
+        progress(0, desc=f"Start uploading... {filename} to {repo_id}")
+        api.upload_file(path_or_fileobj=filename, path_in_repo=output_filename, repo_type=repo_type, revision="main", token=hf_token, repo_id=repo_id)
+        progress(1, desc="Uploaded.")
+        url = hf_hub_url(repo_id=repo_id, repo_type=repo_type, filename=output_filename)
+    except Exception as e:
+        print(f"Error: Failed to upload to {repo_id}. {e}")
+        gr.Warning(f"Error: Failed to upload to {repo_id}. {e}")
+        return None
+    finally:
+        if Path(filename).exists(): Path(filename).unlink()
+    return url
+
+
+def is_same_file(filename: str, cmp_sha256: str, cmp_size: int):
+    if cmp_sha256:
+        sha256_hash = hashlib.sha256()
+        with open(filename, "rb") as f:
+            for byte_block in iter(lambda: f.read(4096), b""):
+                sha256_hash.update(byte_block)
+        sha256 = sha256_hash.hexdigest()
+    else: sha256 = ""
+    size = os.path.getsize(filename)
+    if size == cmp_size and sha256 == cmp_sha256: return True
+    else: return False
+
+
+def get_safe_filename(filename, repo_id, repo_type):
+    hf_token = get_token()
+    api = HfApi(token=hf_token)
+    new_filename = filename
+    try:
+        i = 1
+        while api.file_exists(repo_id=repo_id, filename=Path(new_filename).name, repo_type=repo_type, token=hf_token):
+            infos = api.get_paths_info(repo_id=repo_id, paths=[Path(new_filename).name], repo_type=repo_type, token=hf_token)
+            if infos and len(infos) == 1:
+                repo_fs = infos[0].size
+                repo_sha256 = infos[0].lfs.sha256 if infos[0].lfs is not None else ""
+                if is_same_file(filename, repo_sha256, repo_fs): break
+            new_filename = str(Path(Path(filename).parent, f"{Path(filename).stem}_{i}{Path(filename).suffix}"))
+            i += 1
+        if filename != new_filename:
+            print(f"{Path(filename).name} is already exists but file content is different. renaming to {Path(new_filename).name}.")
+            Path(filename).rename(new_filename)
+    except Exception as e:
+        print(f"Error occured when renaming {filename}. {e}")
+    finally:
+        return new_filename
+
+
+def download_file(dl_url, civitai_key, progress=gr.Progress(track_tqdm=True)):
+    download_dir = TEMP_DIR
+    progress(0, desc=f"Start downloading... {dl_url}")
+    output_filename = get_download_file(download_dir, dl_url, civitai_key)
+    return output_filename
+
+
+def save_civitai_info(dl_url, filename, civitai_key="", progress=gr.Progress(track_tqdm=True)):
+    json_str, html_str, image_path = get_civitai_json(dl_url, True, filename, civitai_key)
+    if not json_str: return  "", "", ""
+    json_path = str(Path(TEMP_DIR, Path(filename).stem + ".json"))
+    html_path = str(Path(TEMP_DIR, Path(filename).stem + ".html"))
+    try:
+        with open(json_path, 'w') as f:
+            json.dump(json_str, f, indent=2)
+        with open(html_path, mode='w', encoding="utf-8") as f:
+            f.write(html_str)
+        return json_path, html_path, image_path
+    except Exception as e:
+        print(f"Error: Failed to save info file {json_path}, {html_path} {e}")
+        return "", "", ""
+
+
+def upload_info_to_repo(dl_url, filename, repo_id, repo_type, is_private, civitai_key="", progress=gr.Progress(track_tqdm=True)):
+    def upload_file(api, filename, repo_id, repo_type, hf_token):
+        if not Path(filename).exists(): return
+        api.upload_file(path_or_fileobj=filename, path_in_repo=Path(filename).name, repo_type=repo_type, revision="main", token=hf_token, repo_id=repo_id)
+        Path(filename).unlink()
+
+    hf_token = get_token()
+    api = HfApi(token=hf_token)
+    try:
+        if not is_repo_exists(repo_id, repo_type): api.create_repo(repo_id=repo_id, repo_type=repo_type, token=hf_token, private=is_private)
+        progress(0, desc=f"Downloading info... {filename}")
+        json_path, html_path, image_path = save_civitai_info(dl_url, filename, civitai_key)
+        progress(0, desc=f"Start uploading info... {filename} to {repo_id}")
+        if not json_path: return
+        else: upload_file(api, json_path, repo_id, repo_type, hf_token)
+        if html_path: upload_file(api, html_path, repo_id, repo_type, hf_token)
+        if image_path: upload_file(api, image_path, repo_id, repo_type, hf_token)
+        progress(1, desc="Info uploaded.")
+        return
+    except Exception as e:
+        print(f"Error: Failed to upload info to {repo_id}. {e}")
+        gr.Warning(f"Error: Failed to upload info to {repo_id}. {e}")
+        return
+
+
+def download_civitai(dl_url, civitai_key, hf_token, urls, 
+                     newrepo_id, repo_type="model", is_private=True, is_info=False, is_rename=True, progress=gr.Progress(track_tqdm=True)):
+    if hf_token: set_token(hf_token)
+    else: set_token(os.environ.get("HF_TOKEN")) # default huggingface write token
+    if not civitai_key: civitai_key = os.environ.get("CIVITAI_API_KEY") # default Civitai API key
+    if not newrepo_id: newrepo_id = os.environ.get("HF_REPO") # default repo to upload
+    if not get_token() or not civitai_key: raise gr.Error("HF write token and Civitai API key is required.")
+    if not urls: urls = []
+    dl_urls = parse_urls(dl_url)
+    remain_urls = dl_urls.copy()
+    try:
+        md = f'### Your repo: [{newrepo_id}]({"https://huggingface.co/datasets/" if repo_type == "dataset" else "https://huggingface.co/"}{newrepo_id})\n'
+        for u in dl_urls:
+            file = download_file(u, civitai_key)
+            if not Path(file).exists() or not Path(file).is_file(): continue
+            if is_rename: file = get_safe_filename(file, newrepo_id, repo_type)
+            url = upload_safetensors_to_repo(file, newrepo_id, repo_type, is_private)
+            if url:
+                if is_info: upload_info_to_repo(u, file, newrepo_id, repo_type, is_private, civitai_key)
+                urls.append(url)
+                remain_urls.remove(u)
+                md += f"- Uploaded [{str(u)}]({str(u)})\n"
+        dp_repos = parse_repos(dl_url)
+        for r in dp_repos:
+            url = duplicate_hf_repo(r, newrepo_id, "model", repo_type, is_private, HF_SUBFOLDER_NAME[1])
+            if url: urls.append(url)
+        return gr.update(value=urls, choices=urls), gr.update(value=md), gr.update(value="\n".join(remain_urls), visible=False)
+    except Exception as e:
+        gr.Info(f"Error occured: {e}")
+        return gr.update(value=urls, choices=urls), gr.update(value=md), gr.update(value="\n".join(remain_urls), visible=True)
+    finally:
+        gc.collect()
+
+
+def search_on_civitai(query: str, types: list[str], allow_model: list[str] = [], limit: int = 100,
+                      sort: str = "Highest Rated", period: str = "AllTime", tag: str = "", user: str = "", page: int = 1,
+                      filetype: list[str] = [], api_key: str = "", progress=gr.Progress(track_tqdm=True)):
+    user_agent = get_user_agent()
+    headers = {'User-Agent': user_agent, 'content-type': 'application/json'}
+    if api_key: headers['Authorization'] = f'Bearer {{{api_key}}}'
+    base_url = 'https://civitai.com/api/v1/models'
+    params = {'sort': sort, 'period': period, 'limit': int(limit), 'nsfw': 'true'}
+    if len(types) != 0: params["types"] = types
+    if query: params["query"] = query
+    if tag: params["tag"] = tag
+    if user: params["username"] = user
+    if page != 0: params["page"] = int(page)
+    session = requests.Session()
+    retries = Retry(total=5, backoff_factor=1, status_forcelist=[500, 502, 503, 504])
+    session.mount("https://", HTTPAdapter(max_retries=retries))
+    rs = []
+    try:
+        if page == 0:
+            progress(0, desc="Searching page 1...")
+            print("Searching page 1...")
+            r = session.get(base_url, params=params | {'page': 1}, headers=headers, stream=True, timeout=(7.0, 30))
+            rs.append(r)
+            if r.ok:
+                json = r.json()
+                next_url = json['metadata']['nextPage'] if 'metadata' in json and 'nextPage' in json['metadata'] else None
+                i = 2
+                while(next_url is not None):
+                    progress(0, desc=f"Searching page {i}...")
+                    print(f"Searching page {i}...")
+                    r = session.get(next_url, headers=headers, stream=True, timeout=(7.0, 30))
+                    rs.append(r)
+                    if r.ok:
+                        json = r.json()
+                        next_url = json['metadata']['nextPage'] if 'metadata' in json and 'nextPage' in json['metadata'] else None
+                    else: next_url = None
+                    i += 1
+        else:
+            progress(0, desc="Searching page 1...")
+            print("Searching page 1...")
+            r = session.get(base_url, params=params, headers=headers, stream=True, timeout=(7.0, 30))
+            rs.append(r)
+    except requests.exceptions.ConnectTimeout:
+        print("Request timed out.")
+    except Exception as e:
+        print(e)
+    items = []
+    for r in rs:
+        if not r.ok: continue
+        json = r.json()
+        if 'items' not in json: continue
+        for j in json['items']:
+            for model in j['modelVersions']:
+                item = {}
+                if len(allow_model) != 0 and model['baseModel'] not in set(allow_model): continue
+                item['name'] = j['name']
+                item['creator'] = j['creator']['username'] if 'creator' in j.keys() and 'username' in j['creator'].keys() else ""
+                item['tags'] = j['tags'] if 'tags' in j.keys() else []
+                item['model_name'] = model['name'] if 'name' in model.keys() else ""
+                item['base_model'] = model['baseModel'] if 'baseModel' in model.keys() else ""
+                item['description'] = model['description'] if 'description' in model.keys() else ""
+                item['md'] = ""
+                
+                # Handle both images and videos
+                if 'images' in model.keys() and len(model["images"]) != 0:
+                    first_media = model["images"][0]
+                    item['img_url'] = first_media["url"]
+                    item['is_video'] = first_media.get("type", "image") == "video"
+                    item['video_url'] = first_media.get("meta", {}).get("video", "") if item['is_video'] else ""
+                    
+                    if item['is_video']:
+                        item['md'] += f'<video src="{item["img_url"]}" poster="{item["img_url"]}" muted loop autoplay width="300" height="480" style="float:right;padding:16px;"></video><br>'
+                    else:
+                        item['md'] += f'<img src="{item["img_url"]}#float" alt="thumbnail" width="150" height="240"><br>'
+                else:
+                    item['img_url'] = "/home/user/app/null.png"
+                    item['is_video'] = False
+                    item['video_url'] = ""
+                
+                item['md'] += f'''Model URL: [https://civitai.com/models/{j["id"]}](https://civitai.com/models/{j["id"]})<br>Model Name: {item["name"]}<br>
+                    Creator: {item["creator"]}<br>Tags: {", ".join(item["tags"])}<br>Base Model: {item["base_model"]}<br>Description: {item["description"]}'''
+                if 'files' in model.keys():
+                    for f in model['files']:
+                        i = item.copy()
+                        i['dl_url'] = f['downloadUrl']
+                        if len(filetype) != 0 and f['type'] not in set(filetype): continue
+                        items.append(i)
+                else:
+                    item['dl_url'] = model['downloadUrl']
+                    items.append(item)
+    return items if len(items) > 0 else None
+
+
+def search_civitai(query, types, base_model=[], sort=SORT[0], period=PERIOD[0], tag="", user="", limit=100, page=1,
+                   filetype=[], api_key="", gallery=[], state={}, progress=gr.Progress(track_tqdm=True)):
+    civitai_last_results = {}
+    set_state(state, "civitai_last_choices", [("", "")])
+    set_state(state, "civitai_last_gallery", [])
+    set_state(state, "civitai_last_results", civitai_last_results)
+    results_info = "No item found."
+    items = search_on_civitai(query, types, base_model, int(limit), sort, period, tag, user, int(page), filetype, api_key)
+    if not items: return gr.update(choices=[("", "")], value=[], visible=True),\
+          gr.update(value="", visible=False), gr.update(), gr.update(), gr.update(), gr.update(), results_info, state
+    choices = []
+    gallery = []
+    for item in items:
+        base_model_name = "Pony🐴" if item['base_model'] == "Pony" else item['base_model']
+        name = f"{item['name']} (for {base_model_name} / By: {item['creator']})"
+        value = item['dl_url']
+        choices.append((name, value))
+        
+        # For gallery, use tuples with HTML that includes both image and video
+        if item.get('is_video') and item.get('video_url'):
+            # Create an HTML element that contains both image and video
+            media_html = f"""
+            <div class="media-container">
+                <img src="{item['img_url']}" alt="{name}">
+                <video src="{item['video_url']}" muted loop poster="{item['img_url']}"></video>
+            </div>
+            """
+            gallery.append((item['img_url'], name))  # Keep using image as thumbnail
+        else:
+            gallery.append((item['img_url'], name))
+            
+        civitai_last_results[value] = item
+    if len(choices) >= 1: 
+        results_info = f"{int(len(choices))} items found."
+    else: 
+        choices = [("", "")]
+        
+    md = ""
+    set_state(state, "civitai_last_choices", choices)
+    set_state(state, "civitai_last_gallery", gallery)
+    set_state(state, "civitai_last_results", civitai_last_results)
+    
+    return gr.update(choices=choices, value=[], visible=True),\
+           gr.update(value=md, visible=True),\
+           gr.update(),\
+           gr.update(),\
+           gr.update(value=gallery),\
+           gr.update(choices=choices, value=[]),\
+           results_info,\
+           state
+
+
+def get_civitai_json(dl_url: str, is_html: bool=False, image_baseurl: str="", api_key=""):
+    if not image_baseurl: image_baseurl = dl_url
+    default = ("", "", "") if is_html else ""
+    if "https://civitai.com/api/download/models/" not in dl_url: return default
+    user_agent = get_user_agent()
+    headers = {'User-Agent': user_agent, 'content-type': 'application/json'}
+    if api_key: headers['Authorization'] = f'Bearer {{{api_key}}}'
+    base_url = 'https://civitai.com/api/v1/model-versions/'
+    params = {}
+    session = requests.Session()
+    retries = Retry(total=5, backoff_factor=1, status_forcelist=[500, 502, 503, 504])
+    session.mount("https://", HTTPAdapter(max_retries=retries))
+    model_id = re.sub('https://civitai.com/api/download/models/(\\d+)(?:.+)?', '\\1', dl_url)
+    url = base_url + model_id
+    #url = base_url + str(dl_url.split("/")[-1])
+    try:
+        r = session.get(url, params=params, headers=headers, stream=True, timeout=(5.0, 15))
+        if not r.ok: return default
+        json = dict(r.json()).copy()
+        html = ""
+        image = ""
+        if "modelId" in json.keys():
+            url = f"https://civitai.com/models/{json['modelId']}"
+            r = session.get(url, params=params, headers=headers, stream=True, timeout=(5.0, 15))
+            if not r.ok: return json, html, image
+            html = r.text
+        if 'images' in json.keys() and len(json["images"]) != 0:
+            url = json["images"][0]["url"]
+            r = session.get(url, params=params, headers=headers, stream=True, timeout=(5.0, 15))
+            if not r.ok: return json, html, image
+            image_temp = str(Path(TEMP_DIR, "image" + Path(url.split("/")[-1]).suffix))
+            image = str(Path(TEMP_DIR, Path(image_baseurl.split("/")[-1]).stem + ".png"))
+            with open(image_temp, 'wb') as f:
+                f.write(r.content)
+            Image.open(image_temp).convert('RGBA').save(image)
+        return json, html, image
+    except Exception as e:
+        print(e)
+        return default
+
+
+def get_civitai_tag():
+    default = [""]
+    user_agent = get_user_agent()
+    headers = {'User-Agent': user_agent, 'content-type': 'application/json'}
+    base_url = 'https://civitai.com/api/v1/tags'
+    params = {'limit': 200}
+    session = requests.Session()
+    retries = Retry(total=5, backoff_factor=1, status_forcelist=[500, 502, 503, 504])
+    session.mount("https://", HTTPAdapter(max_retries=retries))
+    url = base_url
+    try:
+        r = session.get(url, params=params, headers=headers, stream=True, timeout=(7.0, 15))
+        if not r.ok: return default
+        j = dict(r.json()).copy()
+        if "items" not in j.keys(): return default
+        items = []
+        for item in j["items"]:
+            items.append([str(item.get("name", "")), int(item.get("modelCount", 0))])
+        df = pd.DataFrame(items)
+        df.sort_values(1, ascending=False)
+        tags = df.values.tolist()
+        tags = [""] + [l[0] for l in tags]
+        return tags
+    except Exception as e:
+        print(e)
+        return default
+
+
+def select_civitai_item(results: list[str], state: dict):
+    json = {}
+    if "http" not in "".join(results) or len(results) == 0: return gr.update(value="", visible=True), gr.update(value=json, visible=False), state
+    result = get_state(state, "civitai_last_results")
+    last_selects = get_state(state, "civitai_last_selects")
+    selects = list_sub(results, last_selects if last_selects else [])
+    md = result.get(selects[-1]).get('md', "") if result and isinstance(result, dict) and len(selects) > 0 else ""
+    set_state(state, "civitai_last_selects", results)
+    return gr.update(value=md, visible=True), gr.update(value=json, visible=False), state
+
+
+def add_civitai_item(results: list[str], dl_url: str):
+    if "http" not in "".join(results): return gr.update(value=dl_url)
+    new_url = dl_url if dl_url else ""
+    for result in results:
+        if "http" not in result: continue
+        new_url += f"\n{result}" if new_url else f"{result}"
+    new_url = uniq_urls(new_url)
+    return gr.update(value=new_url)
+
+
+def select_civitai_all_item(button_name: str, state: dict):
+    if button_name not in ["Select All", "Deselect All"]: return gr.update(value=button_name), gr.Update(visible=True)
+    civitai_last_choices = get_state(state, "civitai_last_choices")
+    selected = [t[1] for t in civitai_last_choices if t[1] != ""] if button_name == "Select All" else []
+    new_button_name = "Select All" if button_name == "Deselect All" else "Deselect All"
+    return gr.update(value=new_button_name), gr.update(value=selected, choices=civitai_last_choices)
+
+
+def update_civitai_selection(evt: gr.SelectData, value: list[str], state: dict):
+    try:
+        civitai_last_choices = get_state(state, "civitai_last_choices")
+        selected_index = evt.index
+        selected = list_uniq([v for v in value if v != ""] + [civitai_last_choices[selected_index][1]])
+        return gr.update(value=selected)
+    except Exception:
+        return gr.update()
+
+
+def update_civitai_checkbox(selected: list[str]):
+    return gr.update(value=selected)
+
+
+def from_civitai_checkbox(selected: list[str]):
+    return gr.update(value=selected)

civitai_constants.py

@@ -0,0 +1,26 @@
+# CivitAI API Constants
+
+TYPE = [
+    "Checkpoint", "TextualInversion", "Hypernetwork", "AestheticGradient", 
+    "LORA", "LoCon", "DoRA", "Controlnet", "Upscaler", "Motion", "VAE", 
+    "Poses", "Wildcards", "Workflows", "Other"
+]
+
+FILETYPE = [
+    "Model", "VAE", "Config", "Training Data"
+]
+
+BASEMODEL = [
+    "Pony", "Illustrious", "SDXL 1.0", "SD 1.5", "Flux.1 D", "Flux.1 S", 
+    "SD 3.5", "CogVideoX", "SVD", "SVD XT", "Wan Video", "Mochi", "LTXV", 
+    "Hunyuan Video", "HiDream", "Other"
+]
+
+SORT = [
+    "Highest Rated", "Most Downloaded", "Most Liked", "Most Discussed", 
+    "Most Collected", "Most Buzz", "Newest"
+]
+
+PERIOD = [
+    "AllTime", "Year", "Month", "Week", "Day"
+] 

packages.txt

@@ -0,0 +1 @@
+git-lfs aria2

requirements.txt

@@ -0,0 +1,3 @@
+huggingface-hub
+gdown
+gradio_modal

utils.py

@@ -0,0 +1,299 @@
+import gradio as gr
+from huggingface_hub import HfApi, HfFolder, hf_hub_download, snapshot_download
+import os
+from pathlib import Path
+import shutil
+import gc
+import re
+import urllib.parse
+import subprocess
+import time
+from typing import Any
+
+
+def get_token():
+    try:
+        token = HfFolder.get_token()
+    except Exception:
+        token = ""
+    return token
+
+
+def set_token(token):
+    try:
+        HfFolder.save_token(token)
+    except Exception:
+        print(f"Error: Failed to save token.")
+
+
+def get_state(state: dict, key: str):
+    if key in state.keys(): return state[key]
+    else:
+        print(f"State '{key}' not found.")
+        return None
+
+
+def set_state(state: dict, key: str, value: Any):
+    state[key] = value
+
+
+def get_user_agent():
+    return 'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:127.0) Gecko/20100101 Firefox/127.0'
+
+
+def is_repo_exists(repo_id: str, repo_type: str="model"):
+    hf_token = get_token()
+    api = HfApi(token=hf_token)
+    try:
+        if api.repo_exists(repo_id=repo_id, repo_type=repo_type, token=hf_token): return True
+        else: return False
+    except Exception as e:
+        print(f"Error: Failed to connect {repo_id} ({repo_type}). {e}")
+        return True # for safe
+
+
+MODEL_TYPE_CLASS = {
+    "diffusers:StableDiffusionPipeline": "SD 1.5",
+    "diffusers:StableDiffusionXLPipeline": "SDXL",
+    "diffusers:FluxPipeline": "FLUX",
+}
+
+
+def get_model_type(repo_id: str):
+    hf_token = get_token()
+    api = HfApi(token=hf_token)
+    lora_filename = "pytorch_lora_weights.safetensors"
+    diffusers_filename = "model_index.json"
+    default = "SDXL"
+    try:
+        if api.file_exists(repo_id=repo_id, filename=lora_filename, token=hf_token): return "LoRA"
+        if not api.file_exists(repo_id=repo_id, filename=diffusers_filename, token=hf_token): return "None"
+        model = api.model_info(repo_id=repo_id, token=hf_token)
+        tags = model.tags
+        for tag in tags:
+            if tag in MODEL_TYPE_CLASS.keys(): return MODEL_TYPE_CLASS.get(tag, default)
+    except Exception:
+        return default
+    return default
+
+
+def list_uniq(l):
+    return sorted(set(l), key=l.index)
+
+
+def list_sub(a, b):
+    return [e for e in a if e not in b]
+
+
+def is_repo_name(s):
+    return re.fullmatch(r'^[\w_\-\.]+/[\w_\-\.]+$', s)
+
+
+def get_hf_url(repo_id: str, repo_type: str="model"):
+    if repo_type == "dataset": url = f"https://huggingface.co/datasets/{repo_id}"
+    elif repo_type == "space": url = f"https://huggingface.co/spaces/{repo_id}"
+    else: url = f"https://huggingface.co/{repo_id}"
+    return url
+
+
+def split_hf_url(url: str):
+    try:
+        s = list(re.findall(r'^(?:https?://huggingface.co/)(?:(datasets|spaces)/)?(.+?/.+?)/\w+?/.+?/(?:(.+)/)?(.+?.\w+)(?:\?download=true)?$', url)[0])
+        if len(s) < 4: return "", "", "", ""
+        repo_id = s[1]
+        if s[0] == "datasets": repo_type = "dataset"
+        elif s[0] == "spaces": repo_type = "space"
+        else: repo_type = "model"
+        subfolder = urllib.parse.unquote(s[2]) if s[2] else None
+        filename = urllib.parse.unquote(s[3])
+        return repo_id, filename, subfolder, repo_type
+    except Exception as e:
+        print(e)
+
+
+def download_hf_file(directory, url, progress=gr.Progress(track_tqdm=True)):
+    hf_token = get_token()
+    repo_id, filename, subfolder, repo_type = split_hf_url(url)
+    try:
+        print(f"Downloading {url} to {directory}")
+        if subfolder is not None: path = hf_hub_download(repo_id=repo_id, filename=filename, subfolder=subfolder, repo_type=repo_type, local_dir=directory, token=hf_token)
+        else: path = hf_hub_download(repo_id=repo_id, filename=filename, repo_type=repo_type, local_dir=directory, token=hf_token)
+        return path
+    except Exception as e:
+        print(f"Failed to download: {e}")
+        return None
+
+
+def download_thing(directory, url, civitai_api_key="", progress=gr.Progress(track_tqdm=True)): # requires aria2, gdown
+    try:
+        url = url.strip()
+        if "drive.google.com" in url:
+            original_dir = os.getcwd()
+            os.chdir(directory)
+            subprocess.run(f"gdown --fuzzy {url}", shell=True)
+            os.chdir(original_dir)
+        elif "huggingface.co" in url:
+            url = url.replace("?download=true", "")
+            if "/blob/" in url: url = url.replace("/blob/", "/resolve/")
+            download_hf_file(directory, url)
+        elif "civitai.com" in url:
+            if civitai_api_key:
+                url = f"'{url}&token={civitai_api_key}'" if "?" in url else f"{url}?token={civitai_api_key}"
+                print(f"Downloading {url}")
+                subprocess.run(f"aria2c --console-log-level=error --summary-interval=10 -c -x 16 -k 1M -s 16 -d {directory} {url}", shell=True)
+            else:
+                print("You need an API key to download Civitai models.")
+        else:
+            os.system(f"aria2c --console-log-level=error --summary-interval=10 -c -x 16 -k 1M -s 16 -d {directory} {url}")
+    except Exception as e:
+        print(f"Failed to download: {e}")
+
+
+def get_local_file_list(dir_path, recursive=False):
+    file_list = []
+    pattern = "**/*.*" if recursive else "*/*.*"
+    for file in Path(dir_path).glob(pattern):
+        if file.is_file():
+            file_path = str(file)
+            file_list.append(file_path)
+    return file_list
+
+
+def get_download_file(temp_dir, url, civitai_key, progress=gr.Progress(track_tqdm=True)):
+    try:
+        if not "http" in url and is_repo_name(url) and not Path(url).exists():
+            print(f"Use HF Repo: {url}")
+            new_file = url
+        elif not "http" in url and Path(url).exists():
+            print(f"Use local file: {url}")
+            new_file = url
+        elif Path(f"{temp_dir}/{url.split('/')[-1]}").exists():
+            print(f"File to download alreday exists: {url}")
+            new_file = f"{temp_dir}/{url.split('/')[-1]}"
+        else:
+            print(f"Start downloading: {url}")
+            recursive = False if "huggingface.co" in url else True
+            before = get_local_file_list(temp_dir, recursive)
+            download_thing(temp_dir, url.strip(), civitai_key)
+            after = get_local_file_list(temp_dir, recursive)
+            new_file = list_sub(after, before)[0] if list_sub(after, before) else ""
+        if not new_file:
+            print(f"Download failed: {url}")
+            return ""
+        print(f"Download completed: {url}")
+        return new_file
+    except Exception as e:
+        print(f"Download failed: {url} {e}")
+        return ""
+
+
+def download_repo(repo_id: str, dir_path: str, progress=gr.Progress(track_tqdm=True)): # for diffusers repo
+    hf_token = get_token()
+    try:
+        snapshot_download(repo_id=repo_id, local_dir=dir_path, token=hf_token, allow_patterns=["*.safetensors", "*.bin"],
+                          ignore_patterns=["*.fp16.*", "/*.safetensors", "/*.bin"], force_download=True)
+        return True
+    except Exception as e:
+        print(f"Error: Failed to download {repo_id}. {e}")
+        gr.Warning(f"Error: Failed to download {repo_id}. {e}")
+        return False
+
+
+def upload_repo(repo_id: str, dir_path: str, is_private: bool, is_pr: bool=False, progress=gr.Progress(track_tqdm=True)): # for diffusers repo
+    hf_token = get_token()
+    api = HfApi(token=hf_token)
+    try:
+        progress(0, desc="Start uploading...")
+        api.create_repo(repo_id=repo_id, token=hf_token, private=is_private, exist_ok=True)
+        api.upload_folder(repo_id=repo_id, folder_path=dir_path, path_in_repo="", create_pr=is_pr, token=hf_token)
+        progress(1, desc="Uploaded.")
+        return get_hf_url(repo_id, "model")
+    except Exception as e:
+        print(f"Error: Failed to upload to {repo_id}. {e}")
+        return ""
+
+
+def gate_repo(repo_id: str, gated_str: str, repo_type: str="model"):
+    hf_token = get_token()
+    api = HfApi(token=hf_token)
+    try:
+        if gated_str == "auto": gated = "auto"
+        elif gated_str == "manual": gated = "manual"
+        else: gated = False
+        api.update_repo_settings(repo_id=repo_id, gated=gated, repo_type=repo_type, token=hf_token)
+    except Exception as e:
+        print(f"Error: Failed to update settings {repo_id}. {e}")
+
+
+HF_SUBFOLDER_NAME = ["None", "user_repo"]
+
+
+def duplicate_hf_repo(src_repo: str, dst_repo: str, src_repo_type: str, dst_repo_type: str,
+                      is_private: bool, subfolder_type: str=HF_SUBFOLDER_NAME[1], progress=gr.Progress(track_tqdm=True)):
+    hf_token = get_token()
+    api = HfApi(token=hf_token)
+    try:
+        if subfolder_type == "user_repo": subfolder = src_repo.replace("/", "_")
+        else: subfolder = ""
+        progress(0, desc="Start duplicating...")
+        api.create_repo(repo_id=dst_repo, repo_type=dst_repo_type, private=is_private, exist_ok=True, token=hf_token)
+        for path in api.list_repo_files(repo_id=src_repo, repo_type=src_repo_type, token=hf_token):
+            file = hf_hub_download(repo_id=src_repo, filename=path, repo_type=src_repo_type, token=hf_token)
+            if not Path(file).exists(): continue
+            if Path(file).is_dir(): # unused for now
+                api.upload_folder(repo_id=dst_repo, folder_path=file, path_in_repo=f"{subfolder}/{path}" if subfolder else path,
+                                  repo_type=dst_repo_type, token=hf_token)
+            elif Path(file).is_file():
+                api.upload_file(repo_id=dst_repo, path_or_fileobj=file, path_in_repo=f"{subfolder}/{path}" if subfolder else path,
+                                repo_type=dst_repo_type, token=hf_token)
+            if Path(file).exists(): Path(file).unlink()
+        progress(1, desc="Duplicated.")
+        return f"{get_hf_url(dst_repo, dst_repo_type)}/tree/main/{subfolder}" if subfolder else get_hf_url(dst_repo, dst_repo_type)
+    except Exception as e:
+        print(f"Error: Failed to duplicate repo {src_repo} to {dst_repo}. {e}")
+        return ""
+
+
+BASE_DIR = str(Path(__file__).resolve().parent.resolve())
+CIVITAI_API_KEY = os.environ.get("CIVITAI_API_KEY")
+
+
+def get_file(url: str, path: str): # requires aria2, gdown
+    print(f"Downloading {url} to {path}...")
+    get_download_file(path, url, CIVITAI_API_KEY)
+
+
+def git_clone(url: str, path: str, pip: bool=False, addcmd: str=""): # requires git
+    os.makedirs(str(Path(BASE_DIR, path)), exist_ok=True)
+    os.chdir(Path(BASE_DIR, path))
+    print(f"Cloning {url} to {path}...")
+    cmd = f'git clone {url}'
+    print(f'Running {cmd} at {Path.cwd()}')
+    i = subprocess.run(cmd, shell=True).returncode
+    if i != 0: print(f'Error occured at running {cmd}')
+    p = url.split("/")[-1]
+    if not Path(p).exists: return
+    if pip:
+        os.chdir(Path(BASE_DIR, path, p))
+        cmd = f'pip install -r requirements.txt'
+        print(f'Running {cmd} at {Path.cwd()}')
+        i = subprocess.run(cmd, shell=True).returncode
+        if i != 0: print(f'Error occured at running {cmd}')
+    if addcmd:
+        os.chdir(Path(BASE_DIR, path, p))
+        cmd = addcmd
+        print(f'Running {cmd} at {Path.cwd()}')
+        i = subprocess.run(cmd, shell=True).returncode
+        if i != 0: print(f'Error occured at running {cmd}')
+
+
+def run(cmd: str, timeout: float=0):
+    print(f'Running {cmd} at {Path.cwd()}')
+    if timeout == 0:
+        i = subprocess.run(cmd, shell=True).returncode
+        if i != 0: print(f'Error occured at running {cmd}')
+    else:
+        p = subprocess.Popen(cmd, shell=True)
+        time.sleep(timeout)
+        p.terminate()
+        print(f'Terminated in {timeout} seconds')