Upload app.py

app.py

@@ -6,7 +6,7 @@ import os
 import tempfile
 import shutil
 import urllib
-from huggingface_hub import whoami, HfApi, hf_hub_download
+from huggingface_hub import whoami, HfApi, hf_hub_download, RepoCard
 from huggingface_hub.utils import build_hf_headers, hf_raise_for_status
 from gradio_huggingfacehub_search import HuggingfaceHubSearch
 
@@ -15,10 +15,23 @@ ENDPOINT = "https://huggingface.co"
 
 REPO_TYPES = ["model", "dataset", "space"]
 HF_REPO = os.environ.get("HF_REPO") if os.environ.get("HF_REPO") else "" # set your default repo
+HF_REPO_PREFIX = os.environ.get("HF_REPO_PREFIX") if os.environ.get("HF_REPO_PREFIX") else "" # set your default repo prefix
+HF_REPO_SUFFIX = os.environ.get("HF_REPO_SUFFIX") if os.environ.get("HF_REPO_SUFFIX") else "" # set your default repo suffix
+HF_USER = os.environ.get("HF_USER") if os.environ.get("HF_USER") else "" # set your username
 REGEX_HF_REPO = r'^[\w_\-\.]+/[\w_\-\.]+$'
 
-def duplicate(source_repo, dst_repo, repo_type, private, overwrite, auto_dir, oauth_token: gr.OAuthToken | None, progress=gr.Progress(track_tqdm=True)):
-    print(oauth_token.token)
+def remove_repo_tags(repo_id: str, tags: list[str], repo_type: str, hf_token: str):
+    try:
+        card = RepoCard.load(repo_id, repo_type=repo_type, token=hf_token)
+        orig_content = card.content
+        for tag in tags:
+            if 'tags' in card.data and tag in card.data['tags']: card.data['tags'].remove(tag)
+        if card.content == orig_content: return
+        card.push_to_hub(repo_id=repo_id, repo_type=repo_type, token=hf_token)
+    except Exception as e:
+        print(f"Failed to remove tags from repocard. {e}")
+
+def duplicate(source_repo, dst_repo, repo_type, private, overwrite, auto_dir, remove_tag, oauth_token: gr.OAuthToken | None, progress=gr.Progress(track_tqdm=True)):
     hf_token = oauth_token.token
     api = HfApi(token=hf_token)
     try:
@@ -41,7 +54,9 @@ def duplicate(source_repo, dst_repo, repo_type, private, overwrite, auto_dir, oa
             subfolder = subfolder.removesuffix("/")
         if auto_dir: subfolder = source_repo
 
-        if overwrite and api.repo_exists(repo_id=dst_repo, repo_type=repo_type, token=hf_token) or subfolder:
+        if not overwrite and api.repo_exists(repo_id=dst_repo, repo_type=repo_type, token=hf_token): raise gr.Error(f"Repo already exists {dst_repo}")
+
+        if overwrite or subfolder:
             temp_dir = tempfile.mkdtemp()
             api.create_repo(repo_id=dst_repo, repo_type=repo_type, private=private, exist_ok=True, token=hf_token)
             for path in api.list_repo_files(repo_id=source_repo, repo_type=repo_type, token=hf_token):
@@ -67,6 +82,8 @@ def duplicate(source_repo, dst_repo, repo_type, private, overwrite, auto_dir, oa
 
             repo_url = r.json().get("url")
 
+        if remove_tag: remove_repo_tags(dst_repo, ["not-for-all-audiences"], repo_type, hf_token)
+
         return (
             f'Find your repo <a href=\'{repo_url}\' target="_blank" style="text-decoration:underline">here</a>',
             "sp.jpg",
@@ -76,41 +93,207 @@ def duplicate(source_repo, dst_repo, repo_type, private, overwrite, auto_dir, oa
         print(e)
         raise gr.Error(f"Error occured: {e}")
 
-interface = gr.Interface(
-    fn=duplicate,
-    inputs=[
-        HuggingfaceHubSearch(
-            placeholder="Source repository (e.g. osanseviero/src)",
-            search_type=["model", "dataset", "space"],
-            sumbit_on_select=False,
-        ),
-        gr.Textbox(placeholder="Destination repository (e.g. osanseviero/dst)", value=HF_REPO),
-        gr.Dropdown(choices=REPO_TYPES, value="model"),
-        gr.Checkbox(label="Make new repo private?", value=True),
-        gr.Checkbox(label="Overwrite existing repo?", value=True),
-        gr.Checkbox(label="Create subdirectories automatically?", value=True),
-    ],
-    outputs=[
-        gr.Markdown(label="output"),
-        gr.Image(show_label=False),
-    ],
-    title="Duplicate your repo!",
-    description="Duplicate a Hugging Face repository! This Space is a an experimental demo.",
-    allow_flagging="never",
-    live=False
-)
+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 duplicate_m2o(source_repos_str, dst_repo, repo_type, private, overwrite, oauth_token: gr.OAuthToken | None, progress=gr.Progress(track_tqdm=True)):
+    hf_token = oauth_token.token
+    api = HfApi(token=hf_token)
+    try:
+        if not repo_type in REPO_TYPES:
+            raise ValueError("need to select valid repo type")
+        _ = whoami(oauth_token.token)
+        # ^ this will throw if token is invalid
+    except Exception as e:
+        raise gr.Error(f"""Oops, you forgot to login. Please use the loggin button on the top left to migrate your repo {e}""")
+    
+    try:
+        if re.fullmatch(REGEX_HF_REPO, dst_repo): subfolder_prefix = ""
+        else:
+            dst_repo, subfolder_prefix = re.findall(r'^([\w_\-\.]+/[\w_\-\.]+)/?(.+)?$', dst_repo)[0]
+            subfolder_prefix = subfolder.removesuffix("/")
+        if not overwrite and api.repo_exists(repo_id=dst_repo, repo_type=repo_type, token=hf_token): raise gr.Error(f"Repo already exists {dst_repo}")
+        source_repos = parse_repos(source_repos_str)
+        for source_repo in source_repos:
+            if re.fullmatch(REGEX_HF_REPO, source_repo): target = ""
+            else:
+                source_repo, target = re.findall(r'^(?:http.+\.co/)?(?:datasets)?(?:spaces)?([\w_\-\.]+/[\w_\-\.]+)/?(?:blob/main/)?(?:resolve/main/)?(.+)?$', source_repo)[0]
+                target = urllib.parse.unquote(target.removesuffix("/"))
+
+            subfolder = subfolder_prefix + "/" + source_repo if subfolder_prefix else source_repo
+
+            temp_dir = tempfile.mkdtemp()
+            api.create_repo(repo_id=dst_repo, repo_type=repo_type, private=private, exist_ok=True, token=hf_token)
+            for path in api.list_repo_files(repo_id=source_repo, repo_type=repo_type, token=hf_token):
+                if target and target not in path: continue
+                file = hf_hub_download(repo_id=source_repo, filename=path, repo_type=repo_type, local_dir=temp_dir, 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=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=repo_type, token=hf_token)
+                if Path(file).exists(): Path(file).unlink()
+            if repo_type == "dataset": repo_url = f"https://huggingface.co/datasets/{dst_repo}"
+            elif repo_type == "space": repo_url = f"https://huggingface.co/spaces/{dst_repo}"
+            else: repo_url = f"https://huggingface.co/{dst_repo}"
+            shutil.rmtree(temp_dir)
+
+        return (
+            f'Find your repo <a href=\'{repo_url}\' target="_blank" style="text-decoration:underline">here</a>',
+            "sp.jpg",
+        )
+
+    except Exception as e:
+        print(e)
+        raise gr.Error(f"Error occured: {e}")
+
+def duplicate_m2m(source_repos_str, hf_user, repo_type, private, overwrite, remove_tag, repo_prefix, repo_suffix, oauth_token: gr.OAuthToken | None, progress=gr.Progress(track_tqdm=True)):
+    hf_token = oauth_token.token
+    api = HfApi(token=hf_token)
+    try:
+        if not repo_type in REPO_TYPES:
+            raise ValueError("need to select valid repo type")
+        _ = whoami(oauth_token.token)
+        # ^ this will throw if token is invalid
+    except Exception as e:
+        raise gr.Error(f"""Oops, you forgot to login. Please use the loggin button on the top left to migrate your repo {e}""")
+    
+    try:
+        source_repos = parse_repos(source_repos_str)
+        repo_url_result = 'Find your repo '
+        for source_repo in source_repos:
+            if not re.fullmatch(REGEX_HF_REPO, source_repo) or not api.repo_exists(repo_id=source_repo, repo_type=repo_type, token=hf_token): continue
+            dst_repo = hf_user + "/" + repo_prefix + source_repo.split("/")[-1] + repo_suffix
+            if not re.fullmatch(REGEX_HF_REPO, dst_repo): continue
+            if not overwrite and api.repo_exists(repo_id=dst_repo, repo_type=repo_type, token=hf_token):
+                gr.Info(f"Repo already exists {dst_repo}")
+                continue
+
+            r = requests.post(
+                f"{ENDPOINT}/api/{repo_type}s/{source_repo}/duplicate",
+                headers=build_hf_headers(token=oauth_token.token),
+                json={"repository": dst_repo, "private": private},
+            )
+            hf_raise_for_status(r)
+
+            repo_url = r.json().get("url")
+            repo_url_result += f'<a href=\'{repo_url}\' target="_blank" style="text-decoration:underline">{dst_repo}</a><br>\n'
+
+            if remove_tag: remove_repo_tags(dst_repo, ["not-for-all-audiences"], repo_type, hf_token)
+
+        return (
+            repo_url_result,
+            "sp.jpg",
+        )
+
+    except Exception as e:
+        print(e)
+        raise gr.Error(f"Error occured: {e}")
+
+def add_repo_text(repo_id: str, source_repos: str):
+    return source_repos + "\n" + repo_id if source_repos else repo_id
 
 def swap_visibilty(profile: gr.OAuthProfile | None):
     return gr.update(elem_classes=["main_ui_logged_in"]) if profile else gr.update(elem_classes=["main_ui_logged_out"])
-        
+
 css = '''
 .main_ui_logged_out{opacity: 0.3; pointer-events: none}
+.title {text-align: center; align-items: center}
 '''
 with gr.Blocks(css=css) as demo:
     gr.LoginButton()
     with gr.Column(elem_classes="main_ui_logged_out") as main_ui:
-        interface.render()
+        gr.Markdown("# Duplicate your repo!", elem_classes="title")
+        gr.Markdown("Duplicate a Hugging Face repository! This Space is a an experimental demo.")
+        with gr.Tab("One to One"):
+            with gr.Row():
+                with gr.Column():
+                    search = HuggingfaceHubSearch(
+                        label="source_repo",
+                        placeholder="Source repository (e.g. osanseviero/src)",
+                        search_type=["model", "dataset", "space"],
+                        sumbit_on_select=False,
+                    )
+                    with gr.Group():
+                        dst_repo = gr.Textbox(label="dst_repo", placeholder="Destination repository (e.g. osanseviero/dst)", value=HF_REPO)
+                        repo_type = gr.Dropdown(label="repo_type", choices=REPO_TYPES, value="model")
+                        with gr.Row():
+                            is_private = gr.Checkbox(label="Make new repo private?", value=True)
+                            is_overwrite = gr.Checkbox(label="Overwrite existing repo?", value=True)
+                            is_subdir = gr.Checkbox(label="Create subdirectories automatically?", value=True)
+                            is_remtag = gr.Checkbox(label="Remove NFAA tag?", value=True)
+                    with gr.Row():
+                        submit_button = gr.Button("Submit", variant="primary")
+                        clear_button = gr.Button("Clear", variant="secondary")
+                with gr.Column():
+                    output_md = gr.Markdown(label="output")
+                    output_image = gr.Image(show_label=False)
+        with gr.Tab("Multi to One"):
+            with gr.Row():
+                with gr.Column():
+                    m2o_search = HuggingfaceHubSearch(
+                        label="source_repo",
+                        placeholder="Source repository (e.g. osanseviero/src)",
+                        search_type=["model", "dataset", "space"],
+                        sumbit_on_select=True,
+                    )
+                    m2o_source_repos = gr.Textbox(label="source_repos", placeholder="Source repositories (e.g. osanseviero/src)\n...", value="", lines=10)
+                    with gr.Group():
+                        m2o_dst_repo = gr.Textbox(label="dst_repo", placeholder="Destination repository (e.g. osanseviero/dst)", value=HF_REPO)
+                        m2o_repo_type = gr.Dropdown(label="repo_type", choices=REPO_TYPES, value="model")
+                        with gr.Row():
+                            m2o_is_private = gr.Checkbox(label="Make new repo private?", value=True)
+                            m2o_is_overwrite = gr.Checkbox(label="Overwrite existing repo?", value=True)
+                    with gr.Row():
+                        m2o_submit_button = gr.Button("Submit", variant="primary")
+                        m2o_clear_button = gr.Button("Clear", variant="secondary")
+                with gr.Column():
+                    m2o_output_md = gr.Markdown(label="output")
+                    m2o_output_image = gr.Image(show_label=False)
+        with gr.Tab("Multi to Multi"):
+            with gr.Row():
+                with gr.Column():
+                    m2m_search = HuggingfaceHubSearch(
+                        label="source_repo",
+                        placeholder="Source repository (e.g. osanseviero/src)",
+                        search_type=["model", "dataset", "space"],
+                        sumbit_on_select=True,
+                    )
+                    m2m_source_repos = gr.Textbox(label="source_repos", placeholder="Source repositories (e.g. osanseviero/src)\n...", value="", lines=10)
+                    with gr.Group():
+                        with gr.Row():
+                            m2m_user = gr.Textbox(label="hf_user", placeholder="Your HF username", value=HF_USER)
+                            m2m_prefix = gr.Textbox(label="repo_prefix", value=HF_REPO_PREFIX)
+                            m2m_suffix = gr.Textbox(label="repo_suffix", value=HF_REPO_SUFFIX)
+                        m2m_repo_type = gr.Dropdown(label="repo_type", choices=REPO_TYPES, value="model")
+                        with gr.Row():
+                            m2m_is_private = gr.Checkbox(label="Make new repo private?", value=True)
+                            m2m_is_overwrite = gr.Checkbox(label="Overwrite existing repo?", value=False)
+                            m2m_is_remtag = gr.Checkbox(label="Remove NFAA tag?", value=True)
+                    with gr.Row():
+                        m2m_submit_button = gr.Button("Submit", variant="primary")
+                        m2m_clear_button = gr.Button("Clear", variant="secondary")
+                with gr.Column():
+                    m2m_output_md = gr.Markdown(label="output")
+                    m2m_output_image = gr.Image(show_label=False)
     demo.load(fn=swap_visibilty, outputs=main_ui)
-    
+    submit_button.click(duplicate, [search, dst_repo, repo_type, is_private, is_overwrite, is_subdir, is_remtag], [output_md, output_image])
+    clear_button.click(lambda: ("", HF_REPO, "model", True, True, True, True), None, [search, dst_repo, repo_type, is_private, is_overwrite, is_subdir, is_remtag], queue=False)
+    m2o_search.submit(add_repo_text, [m2o_search, m2o_source_repos], [m2o_source_repos], queue=False)
+    m2o_submit_button.click(duplicate_m2o, [m2o_source_repos, m2o_dst_repo, m2o_repo_type, m2o_is_private, m2o_is_overwrite], [m2o_output_md, m2o_output_image])
+    m2o_clear_button.click(lambda: ("", HF_REPO, "model", True, True, ""), None,
+                           [m2o_search, m2o_dst_repo, m2o_repo_type, m2o_is_private, m2o_is_overwrite, m2o_source_repos], queue=False)
+    m2m_search.submit(add_repo_text, [m2m_search, m2m_source_repos], [m2m_source_repos], queue=False)
+    m2m_submit_button.click(duplicate_m2m, [m2m_source_repos, m2m_user, m2m_repo_type, m2m_is_private, m2m_is_overwrite, m2m_is_remtag, m2m_prefix, m2m_suffix],
+                            [m2m_output_md, m2m_output_image])
+    m2m_clear_button.click(lambda: ("", HF_USER, "model", True, False, True, "", HF_REPO_PREFIX, HF_REPO_SUFFIX), None,
+                           [m2m_search, m2m_user, m2m_repo_type, m2m_is_private, m2m_is_overwrite, m2m_is_remtag, m2m_source_repos, m2m_prefix, m2m_suffix], queue=False)
+
 demo.queue()
 demo.launch()