Upload 6 files

README.md

@@ -1,12 +1,13 @@
 ---
-title: Civitai To Hf
-emoji: 📈
-colorFrom: indigo
-colorTo: indigo
+title: CivitAI to HF Downloader Alpha
+emoji: 🤗
+colorFrom: yellow
+colorTo: blue
 sdk: gradio
-sdk_version: 4.44.1
+sdk_version: 4.44.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,57 @@
+import gradio as gr
+from civitai_to_hf import search_civitai, download_civitai, select_civitai_item, CIVITAI_TYPE, CIVITAI_BASEMODEL, CIVITAI_SORT, CIVITAI_PERIOD
+
+css = """
+.title { text-align: center; !important; }
+"""
+
+with gr.Blocks(theme="NoCrypt/miku@>=1.2.2", fill_width=True, css=css, delete_cache=(60, 3600)) as demo:
+    with gr.Column():
+        gr.Markdown("# Civitai to HF Downloader Alpha", elem_classes="title")
+        with gr.Accordion("Search Civitai", open=False):
+            with gr.Row():
+                search_civitai_type = gr.CheckboxGroup(label="Type", choices=CIVITAI_TYPE, value=[])
+                search_civitai_basemodel = gr.CheckboxGroup(label="Base model", choices=CIVITAI_BASEMODEL, value=[])
+            with gr.Row():
+                search_civitai_sort = gr.Radio(label="Sort", choices=CIVITAI_SORT, value=CIVITAI_SORT[0])
+                search_civitai_period = gr.Radio(label="Period", choices=CIVITAI_PERIOD, value=CIVITAI_PERIOD[0])
+            with gr.Row():
+                search_civitai_query = gr.Textbox(label="Query", placeholder="oomuro sakurako...", lines=1)
+                search_civitai_tag = gr.Textbox(label="Tag", lines=1)
+                search_civitai_submit = gr.Button("Search on Civitai")
+            with gr.Row():
+                search_civitai_result = gr.Dropdown(label="Search Results", choices=[("", "")], value="", allow_custom_value=True, visible=False)
+                search_civitai_json = gr.JSON(value={}, visible=False)
+                search_civitai_desc = gr.Markdown(value="", visible=False)
+        dl_url = gr.Textbox(label="Download URL", placeholder="https://civitai.com/api/download/models/28907", value="", lines=1)
+        civitai_key = gr.Textbox(label="Your Civitai Key", value="", max_lines=1)
+        with gr.Row():
+            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).")
+        with gr.Row():
+            newrepo_id = gr.Textbox(label="Upload repo ID", placeholder="yourid/yourrepo", value="", max_lines=1)
+            newrepo_type = gr.Radio(label="Upload repo type", choices=["model", "dataset"], value="model")
+            is_private = gr.Checkbox(label="Create private repo", value=True)
+        uploaded_urls = gr.CheckboxGroup(visible=False, choices=[], value=None) # hidden
+        run_button = gr.Button(value="Download and Upload")
+        urls_md = gr.Markdown()
+    gr.DuplicateButton(value="Duplicate Space")
+
+    gr.on(
+        triggers=[run_button.click],
+        fn=download_civitai,
+        inputs=[dl_url, civitai_key, hf_token, uploaded_urls, newrepo_id, newrepo_type, is_private],
+        outputs=[uploaded_urls, urls_md],
+    )
+    gr.on(
+        triggers=[search_civitai_submit.click, search_civitai_query.submit, search_civitai_tag.submit],
+        fn=search_civitai,
+        inputs=[search_civitai_query, search_civitai_type, search_civitai_basemodel, search_civitai_sort, search_civitai_period, search_civitai_tag],
+        outputs=[search_civitai_result, search_civitai_desc, search_civitai_submit, search_civitai_query],
+        queue=True,
+        show_api=False,
+    )
+    search_civitai_result.change(select_civitai_item, [search_civitai_result], [dl_url, search_civitai_desc], queue=False, show_api=False)
+
+demo.queue()
+demo.launch()

civitai_to_hf.py

@@ -0,0 +1,127 @@
+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 utils import get_token, set_token, is_repo_exists, get_user_agent, get_download_file
+
+
+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="Start uploading...")
+        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
+    return url
+
+
+def download_file(dl_url, civitai_key, progress=gr.Progress(track_tqdm=True)):
+    download_dir = "."
+    progress(0, desc="Start downloading...")
+    output_filename = get_download_file(download_dir, dl_url, civitai_key)
+    return output_filename
+
+
+def download_civitai(dl_url, civitai_key, hf_token, urls, 
+                     newrepo_id, repo_type="model", is_private=True, progress=gr.Progress(track_tqdm=True)):
+    if hf_token: set_token(hf_token)
+    else: set_token(os.environ.get("HF_TOKEN"))
+    if not civitai_key: civitai_key = os.environ.get("CIVITAI_API_KEY")
+    if not hf_token or not civitai_key: raise gr.Error("HF write token and Civitai API key is required.")
+    file = download_file(dl_url, civitai_key)
+    if not urls: urls = []
+    url = upload_safetensors_to_repo(file, newrepo_id, repo_type, is_private)
+    progress(1, desc="Processing...")
+    if url: urls.append(url)
+    Path(file).unlink()
+    md = ""
+    for u in urls:
+        md += f"[Uploaded {str(u)}]({str(u)})<br>"
+    gc.collect()
+    return gr.update(value=urls, choices=urls), gr.update(value=md)
+
+
+CIVITAI_TYPE = ["Checkpoint", "TextualInversion", "Hypernetwork", "AestheticGradient", "LORA", "Controlnet", "Poses"]
+CIVITAI_BASEMODEL = ["Pony", "SD 1.5", "SDXL 1.0", "Flux.1 D", "Flux.1 S"]
+CIVITAI_SORT = ["Highest Rated", "Most Downloaded", "Newest"]
+CIVITAI_PERIOD = ["AllTime", "Year", "Month", "Week", "Day"]
+
+
+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_agent = get_user_agent()
+    headers = {'User-Agent': user_agent, 'content-type': 'application/json'}
+    base_url = 'https://civitai.com/api/v1/models'
+    params = {'sort': sort, 'period': period, 'limit': limit, 'nsfw': 'true'}
+    if len(types) != 0: params["types"] = types
+    if query: params["query"] = query
+    if tag: params["tag"] = tag
+    session = requests.Session()
+    retries = Retry(total=5, backoff_factor=1, status_forcelist=[500, 502, 503, 504])
+    session.mount("https://", HTTPAdapter(max_retries=retries))
+    try:
+        r = session.get(base_url, params=params, headers=headers, stream=True, timeout=(3.0, 30))
+    except Exception as e:
+        print(e)
+        return None
+    else:
+        if not r.ok: return None
+        json = r.json()
+        if 'items' not in json: return None
+        items = []
+        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']
+                item['tags'] = j['tags']
+                item['model_name'] = model['name']
+                item['base_model'] = model['baseModel']
+                item['dl_url'] = model['downloadUrl']
+                item['md'] = f'<img src="{model["images"][0]["url"]}" alt="thumbnail" width="150" height="240"><br>[LoRA Model URL](https://civitai.com/models/{j["id"]})'
+                items.append(item)
+        return items
+
+
+civitai_last_results = {}
+
+
+def search_civitai(query, types, base_model=[], sort=CIVITAI_SORT[0], period=CIVITAI_PERIOD[0], tag=""):
+    global civitai_last_results
+    items = search_on_civitai(query, types, base_model, 100, sort, period, tag)
+    if not items: return gr.update(choices=[("", "")], value="", visible=False),\
+          gr.update(value="", visible=False), gr.update(visible=True), gr.update(visible=True)
+    civitai_last_results = {}
+    choices = []
+    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']} / Tags: {', '.join(item['tags'])})"
+        value = item['dl_url']
+        choices.append((name, value))
+        civitai_last_results[value] = item
+    if not choices: return gr.update(choices=[("", "")], value="", visible=False),\
+          gr.update(value="", visible=False), gr.update(visible=True), gr.update(visible=True)
+    result = civitai_last_results.get(choices[0][1], "None")
+    md = result['md'] if result else ""
+    return gr.update(choices=choices, value=choices[0][1], visible=True), gr.update(value=md, visible=True),\
+          gr.update(visible=True), gr.update(visible=True)
+
+
+def select_civitai_item(search_result):
+    if not "http" in search_result: return gr.update(value=""), gr.update(value="None", visible=True)
+    result = civitai_last_results.get(search_result, "None")
+    md = result['md'] if result else ""
+    return gr.update(value=search_result), gr.update(value=md, visible=True)

packages.txt

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

requirements.txt

@@ -0,0 +1,2 @@
+huggingface-hub
+gdown

utils.py

@@ -0,0 +1,161 @@
+import gradio as gr
+from huggingface_hub import HfApi, HfFolder, hf_hub_download
+import os
+from pathlib import Path
+import shutil
+import gc
+import re
+import urllib.parse
+
+
+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_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_sub(a, b):
+    return [e for e in a if e not in b]
+
+
+def is_repo_name(s):
+    return re.fullmatch(r'^[^/,\s\"\']+/[^/,\s\"\']+$', s)
+
+
+def split_hf_url(url: str):
+    try:
+        s = list(re.findall(r'^(?:https?://huggingface.co/)(?:(datasets)/)?(.+?/.+?)/\w+?/.+?/(?:(.+)/)?(.+?.safetensors)(?:\?download=true)?$', url)[0])
+        if len(s) < 4: return "", "", "", ""
+        repo_id = s[1]
+        repo_type = "dataset" if s[0] == "datasets" else "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:
+        if subfolder is not None: hf_hub_download(repo_id=repo_id, filename=filename, subfolder=subfolder, repo_type=repo_type, local_dir=directory, token=hf_token)
+        else: hf_hub_download(repo_id=repo_id, filename=filename, repo_type=repo_type, local_dir=directory, token=hf_token)
+    except Exception as e:
+        print(f"Failed to download: {e}")
+
+
+def download_thing(directory, url, civitai_api_key="", progress=gr.Progress(track_tqdm=True)): # requires aria2, gdown
+    hf_token = get_token()
+    url = url.strip()
+    if "drive.google.com" in url:
+        original_dir = os.getcwd()
+        os.chdir(directory)
+        os.system(f"gdown --fuzzy {url}")
+        os.chdir(original_dir)
+    elif "huggingface.co" in url:
+        url = url.replace("?download=true", "")
+        if "/blob/" in url:
+            url = url.replace("/blob/", "/resolve/")
+        #user_header = f'"Authorization: Bearer {hf_token}"'
+        if hf_token:
+            download_hf_file(directory, url)
+            #os.system(f"aria2c --console-log-level=error --summary-interval=10 --header={user_header} -c -x 16 -k 1M -s 16 {url} -d {directory}  -o {url.split('/')[-1]}")
+        else:
+            os.system(f"aria2c --optimize-concurrent-downloads --console-log-level=error --summary-interval=10 -c -x 16 -k 1M -s 16 {url} -d {directory}  -o {url.split('/')[-1]}")
+    elif "civitai.com" in url:
+        if "?" in url:
+            url = url.split("?")[0]
+        if civitai_api_key:
+            url = url + f"?token={civitai_api_key}"
+            os.system(f"aria2c --console-log-level=error --summary-interval=10 -c -x 16 -k 1M -s 16 -d {directory} {url}")
+        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}")
+
+
+def get_local_model_list(dir_path):
+    model_list = []
+    valid_extensions = ('.safetensors')
+    for file in Path(dir_path).glob("**/*.*"):
+        if file.is_file() and file.suffix in valid_extensions:
+            file_path = str(file)
+            model_list.append(file_path)
+    return model_list
+
+
+def get_download_file(temp_dir, url, civitai_key, progress=gr.Progress(track_tqdm=True)):
+    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}")
+        before = get_local_model_list(temp_dir)
+        try:
+            download_thing(temp_dir, url.strip(), civitai_key)
+        except Exception:
+            print(f"Download failed: {url}")
+            return ""
+        after = get_local_model_list(temp_dir)
+        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