import gradio as gr import os from git_commd import GitCommandWrapper from typing import List, Optional import logging HF_TOKEN = os.environ["HF_TOKEN"] if "HF_TOKEN" in os.environ else "" WiseModel_TOKEN = os.environ["WM_TOKEN"] if "WM_TOKEN" in os.environ else "" GIT_USER = os.environ["GIT_USER"] if "GIT_USER" in os.environ else "" GIT_EMAIL = os.environ["GIT_EMAIL"] if "GIT_EMAIL" in os.environ else "" def get_cache_dir(): from random_word import RandomWords r = RandomWords() return r.get_random_word() def check_disk(): import os return os.system("df -h /") def init_git_user(username: str, email: str): if not username or not email: raise gr.Error( "Please config your git username and email in the environment variables." ) os.system(f"git config --global user.email {email}") os.system(f"git config --global user.name {username}") def pull_from_wisemodel( token: str, url: str, repo_name: str, cache_dir, branch: Optional[str] = None ): print("pull_from_wisemodel start") print(cache_dir) os.makedirs(cache_dir, exist_ok=True) # os.system("cd "+cache_dir) gitCmd = GitCommandWrapper() isLfs = gitCmd.is_lfs_installed() if not isLfs: gitCmd.git_lfs_install() gitCmd.clone(cache_dir, token, url, repo_name, branch) print("pull_from_wisemodel end") return f"Pulled {branch} to temp folder {cache_dir}: {url}" def push_to_wiseModel( token: str, url: str, repo_name: str, cache_dir: str, hf_repo_id: str, branch: Optional[str] = None, ): print("push_to_wiseModel start") gitCmd = GitCommandWrapper() isLfs = gitCmd.is_lfs_installed() if not isLfs: gitCmd.git_lfs_install() gitCmd.clone(cache_dir, token, url, repo_name, branch) repo_dir = f"./{cache_dir}/{repo_name}" os.makedirs(repo_dir, exist_ok=True) source_dir = f"./{cache_dir}/hf/{hf_repo_id}" move_file(source_dir, repo_dir) #, excludes=[".git", ".gitattributes"]) gitCmd.add(repo_dir, all_files=True) gitCmd.commit(repo_dir, message="commit from hf to wisemodel") gitCmd.push(repo_dir, token, url, branch, branch) print("push_to_wiseModel end") return f"Pushed {branch} to {url}" def move_file(source: str, destination: str, excludes: list[str] = []): import shutil try: # move all files in the source directory to the destination directory # list all files in the source directory files = os.listdir(source) for file in files: if file in excludes: continue # if file already exists in the destination directory, remove it if os.path.exists(f"{destination}/{file}"): logging.info(f"Removing {file} from {destination}") # incase is a directory, remove it recursively if os.path.isdir(f"{destination}/{file}"): # force remove the directory and all its contents shutil.rmtree(f"{destination}/{file}", ignore_errors=True) else: os.remove(f"{destination}/{file}") logging.info(f"Moving {file} to {destination}") # move each file to destination Directory, if it already exists, it will be replaced shutil.move(f"{source}/{file}", destination) except Exception as e: logging.exception(e) return f"Error moving files from {source} to {destination}, {e.args[0]}" return "file moved" def remove_file(cache_dir, repo_name): import os try: os.remove(f"{cache_dir}/{repo_name}") except: return "" return "README.md file removed" def push_to_hf(cache_dir, WiseModel_repo_name, hf_repo_id): from huggingface_hub import HfApi if not HF_TOKEN: raise gr.Error("Please enter your HF_TOKEN") print("push_to_hf start") api = HfApi(token=HF_TOKEN) # Token is not persisted on the machine. output = api.upload_folder( folder_path=f"{cache_dir}/{WiseModel_repo_name}", repo_id=hf_repo_id, repo_type="model", ) print("push_to_hf end") return f"Pushed to {hf_repo_id}" def pull_from_hf(cache_dir, hf_repo_id): from huggingface_hub import HfApi # if not HF_TOKEN: # raise gr.Error("Please enter your HF_TOKEN") print("pull_from_hf start") api = HfApi(token=HF_TOKEN) # Token is not persisted on the machine. output = api.snapshot_download( repo_id=hf_repo_id, repo_type="model", local_dir=cache_dir + "/hf/" + hf_repo_id, local_dir_use_symlinks=False, ) print(f"pull_from_hf end, output: {output}") return f"Pulled from {hf_repo_id}" def handle(wisemodel_link, hf_repo_id): cache_dir = get_cache_dir() wiseModel_repo_url = ( wisemodel_link.replace(".git", "") .replace("git", "") .replace("clone", "") .replace(" ", "") ) wiseModel_repo_info = ( wisemodel_link.replace(".git", "") .replace("git", "") .replace("clone", "") .replace(" ", "") .split("/") ) print(wiseModel_repo_info) wisemodel_repo_name = wiseModel_repo_info[-1] stages = [ (check_disk, (), {}), # # Run all the sanity checks on README.md # (pull_from_wisemodel, (WiseModel_TOKEN,wiseModel_repo_url,wisemodel_repo_name, cache_dir,"main"), {}), # (remove_file, (wisemodel_repo_name, cache_dir), {}), # (check_disk, (), {}), # (push_to_hf, (cache_dir, wisemodel_repo_name, hf_repo_id), {}), # (check_disk, (), {}), (init_git_user, (GIT_USER, GIT_EMAIL), {}), (pull_from_hf, (cache_dir, hf_repo_id), {}), (remove_file, (hf_repo_id, cache_dir), {}), (check_disk, (), {}), ( push_to_wiseModel, ( WiseModel_TOKEN, wiseModel_repo_url, wisemodel_repo_name, cache_dir, hf_repo_id, "main", ), {}, ), (check_disk, (), {}), ] results = [] errors = [] for func, args, kwargs in stages: try: results.append(str(func(*args, **kwargs))) except Exception as e: logging.exception(e) errors.append(str(e)) if errors: break return "\n\n".join(results), "\n\n".join(errors) with gr.Blocks() as demo: gr.Markdown( """ # HF-to-wisemodel/从Huggingface上拉取模型、数据集等到wisemodel # 这是一个示例Space,实际使用请参考下面说明，先Duplicate一个私有的space - 这个space可以把已经发布在Huggingface上的模型拉取到wisemodel上。 - This space uploads model from Huggingface to wisemodel. - **请确认您是repo的拥有者或者有权限操作！** - **Please make sure that you're the owner of the repo or have permission from the owner to do so!** # 如何使用这个空间？ # How to use this Space? - 点击右上角settings后面的“…”按钮，选择“Duplicate this Space”创建一个私有的space，同时输入wisemodel的token(必填)、wisemodel的用户名（GIT_USER 必填）、wisemodel的注册邮箱（GIT_EMAIL 必填），确保有相应repo写入的权限。你还需要输入你的git用户名和邮箱。 - Duplicate this Space and providing WiseModel token (mandatory) and your read/write HF token (mandatory).you also need to provide your git username and email to push the model to WiseModel. - 在wiseModel上创建一个空的repo，这一步需要手动完成，Space不会为您创建一个空的repo。 - Create your target model repo on WiseModel. This step is not automated. - 在刚刚自己创建的私有space里填写相应的信息，wisemodel的git clone链接，以及Huggingface的repo名称。 - In your own private Space, fill in information below. - 点击submit按钮，然后可以通过logs按钮查看进度。 - Click submit then watch for output in container log for progress. """ ) wisemodel_link = gr.Textbox( label="Copy the git download link from the model detail page of wisemodel（从wisemodel上获取该模型的完整git clone链接） " ) hf_repo_id = gr.Textbox( label="Source HF Model Repo ID (case sensitive). \nPlease make sure that this model has already been created" ) with gr.Row(): button = gr.Button("Submit", variant="primary") clear = gr.Button("Clear") error = gr.Textbox(label="Error") output = gr.Textbox(label="Output") button.click(handle, [wisemodel_link, hf_repo_id], [output, error]) if __name__ == "__main__": demo.launch(debug=True)