Upload 7 files

app.py

@@ -21,7 +21,7 @@ with gr.Blocks(theme="NoCrypt/miku@>=1.2.2", fill_width=True, css=css, delete_ca
                 search_civitai_period = gr.Radio(label="Period", choices=CIVITAI_PERIOD, value="Month")
                 search_civitai_limit = gr.Slider(label="Limit", minimum=1, maximum=100, step=1, value=100)
                 search_civitai_page = gr.Slider(label="Page", minimum=1, maximum=100, step=1, value=1)
-            with gr.Row():
+            with gr.Row(equal_height=True):
                 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)
@@ -30,7 +30,7 @@ with gr.Blocks(theme="NoCrypt/miku@>=1.2.2", fill_width=True, css=css, delete_ca
                 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():
+                with gr.Row(equal_height=True):
                     with gr.Column(scale=9):
                         with gr.Accordion("Select from Gallery", open=False):
                             search_civitai_gallery = gr.Gallery([], label="Results", allow_preview=False, columns=5, elem_id="gallery", show_share_button=False, interactive=False)

civitai_to_hf.py

@@ -6,14 +6,15 @@ 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, list_uniq
+from utils import get_token, set_token, is_repo_name, is_repo_exists, get_user_agent, get_download_file, list_uniq, duplicate_hf_repo, HF_SUBFOLDER_NAME
 import re
 from PIL import Image
 import json
 import pandas as pd
+import tempfile
 
 
-TEMP_DIR = "."
+TEMP_DIR = tempfile.mkdtemp()
 
 
 def parse_urls(s):
@@ -25,12 +26,22 @@ def parse_urls(s):
         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)))
+    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)):
@@ -107,8 +118,8 @@ def download_civitai(dl_url, civitai_key, hf_token, urls,
     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.")
-    dl_urls = parse_urls(dl_url)
     if not urls: urls = []
+    dl_urls = parse_urls(dl_url)
     for u in dl_urls:
         file = download_file(u, civitai_key)
         if not Path(file).exists() or not Path(file).is_file(): continue
@@ -116,6 +127,10 @@ def download_civitai(dl_url, civitai_key, hf_token, urls,
         if url:
             urls.append(url)
             if is_info: upload_info_to_repo(u, file, newrepo_id, repo_type, is_private)
+    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)
     progress(1, desc="Processing...")
     md = f'### Your repo: [{newrepo_id}]({"https://huggingface.co/datasets/" if repo_type == "dataset" else "https://huggingface.co/"}{newrepo_id})\n'
     for u in urls:
@@ -131,7 +146,7 @@ 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: str = "", page: int = 1):
+                      sort: str = "Highest Rated", period: str = "AllTime", tag: str = "", user: str = "", page: int = 1):
     user_agent = get_user_agent()
     headers = {'User-Agent': user_agent, 'content-type': 'application/json'}
     base_url = 'https://civitai.com/api/v1/models'
@@ -179,6 +194,7 @@ civitai_last_results = {}
 civitai_last_choices = [("", "")]
 civitai_last_gallery = []
 
+
 def search_civitai(query, types, base_model=[], sort=CIVITAI_SORT[0], period=CIVITAI_PERIOD[0], tag="", user="", limit=100, page=1, gallery=[]):
     global civitai_last_results, civitai_last_choices, civitai_last_gallery
     civitai_last_choices = [("", "")]

utils.py

@@ -72,15 +72,24 @@ def list_sub(a, b):
 
 
 def is_repo_name(s):
-    return re.fullmatch(r'^[^/,\s\"\']+/[^/,\s\"\']+$', 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)/)?(.+?/.+?)/\w+?/.+?/(?:(.+)/)?(.+?.\w+)(?:\?download=true)?$', url)[0])
+        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]
-        repo_type = "dataset" if s[0] == "datasets" else "model"
+        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
@@ -160,8 +169,7 @@ def get_download_file(temp_dir, url, civitai_key, progress=gr.Progress(track_tqd
     return new_file
 
 
-# https://huggingface.co/docs/huggingface_hub/v0.25.1/en/package_reference/file_download#huggingface_hub.snapshot_download
-def download_repo(repo_id, dir_path, progress=gr.Progress(track_tqdm=True)):
+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"],
@@ -173,20 +181,49 @@ def download_repo(repo_id, dir_path, progress=gr.Progress(track_tqdm=True)):
         return False
 
 
-def upload_repo(new_repo_id, dir_path, is_private, progress=gr.Progress(track_tqdm=True)):
+def upload_repo(repo_id: str, dir_path: str, is_private: bool, 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=new_repo_id, token=hf_token, private=is_private, exist_ok=True)
+        api.create_repo(repo_id=repo_id, token=hf_token, private=is_private, exist_ok=True)
         for path in Path(dir_path).glob("*"):
             if path.is_dir():
-                api.upload_folder(repo_id=new_repo_id, folder_path=str(path), path_in_repo=path.name, token=hf_token)
+                api.upload_folder(repo_id=repo_id, folder_path=str(path), path_in_repo=path.name, token=hf_token)
             elif path.is_file():
-                api.upload_file(repo_id=new_repo_id, path_or_fileobj=str(path), path_in_repo=path.name, token=hf_token)
+                api.upload_file(repo_id=repo_id, path_or_fileobj=str(path), path_in_repo=path.name, token=hf_token)
         progress(1, desc="Uploaded.")
-        url = f"https://huggingface.co/{new_repo_id}"
+        return get_hf_url(repo_id, "model")
     except Exception as e:
-        print(f"Error: Failed to upload to {new_repo_id}. {e}")
+        print(f"Error: Failed to upload to {repo_id}. {e}")
         return ""
-    return url
+
+
+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 ""
+