xjyplayer uploaded at 2024-03-24 17:30:21

test4/huggingface.py

@@ -0,0 +1,434 @@
+import subprocess
+import json
+from huggingface_hub import HfApi, list_models
+from huggingface_hub.utils import HfHubHTTPError, RepositoryNotFoundError, RevisionNotFoundError
+from huggingface_hub.hf_api import RepoFile, RepoFolder
+import os
+os.environ['NO_PROXY'] = 'localhost, 127.0.0.1, ::1, ip.cn, chinaz.com, 192.168.0.0/16,  mirrors.cloud.aliyuncs.com, hf-mirror.com'
+
+import re
+import json
+from datetime import datetime
+
+# Specify the mirror URL
+mirror_url = "https://hf-mirror.com"
+
+raw_url = "https://huggingface.co"
+user=""
+user_id=""
+api=None
+model_list = {}
+commit_message = ""
+
+def save_access_token(access_token):
+    """保存 AccessToken 到文件中."""
+    with open("access_token.json", "w") as f:
+        json.dump({"access_token": access_token}, f)
+
+def load_access_token():
+    """加载保存的 AccessToken."""
+    try:
+        with open("access_token.json", "r") as f:
+            return json.load(f)["access_token"]
+    except FileNotFoundError:
+        return None
+
+def login(init=False):
+
+    """登录到 Hugging Face 账户. 只有初始化的时候才可以略过登录，直接读取本地文件，其他情况代表手动调用"""
+    access_token = load_access_token()
+    if not (access_token and init):
+        access_token = input("请输入您的 AccessToken: ")
+        save_access_token(access_token)
+    os.environ['HF_ENDPOINT'] = mirror_url
+    subprocess.run(["huggingface-cli", "login", "--token", access_token], capture_output=True, text=True)
+
+    os.environ['HF_ENDPOINT'] = ''
+
+    return access_token
+
+def makesure_login(init=False):
+    global user, api, user_id
+
+    if init:
+        login(init)
+
+    """确保登录成功."""
+    while True:
+        status, res = check_login()
+        if status:
+            print(f"登录成功，用户名为: {res}")
+            if api == None:
+                api = HfApi(endpoint=mirror_url, token=load_access_token())
+                user=res
+                user_id = user
+            break
+        else:
+            print(f"登录失败，请重新登录, {res}")
+            login(init)
+
+
+def check_login():
+    os.environ['HF_ENDPOINT'] = mirror_url
+    result = subprocess.run(["huggingface-cli", "whoami"], capture_output=True, text=True)
+    os.environ['HF_ENDPOINT'] = ""
+
+    if result.returncode == 0:
+        return True, result.stdout.split("\n")[0]
+    else:
+        return False, result.stderr
+
+
+def list_projects():
+    global user, api, model_list
+    """列出用户的项目列表."""
+    makesure_login()
+    # List models
+    models = api.list_models(author=user, token=load_access_token())
+    # Print table header
+    print("\nProjects:")
+    print("| {:<5} | {:<40} | {:<30} | {:<10} |".format("ID", "Model ID", "Created At", "Private"))
+    print("| {:<5} | {:<40} | {:<30} | {:<10} |".format("-"*5, "-"*40, "-"*30, "-"*10))
+    start_num = 99
+    end_num = 99
+    for i, model in enumerate(models, start=100):
+        end_num += 1
+        model_list[i] = model.id
+        print(f"| {i:<5} | {model.id.ljust(40)} | {str(model.created_at).ljust(30)} | {str(model.private).ljust(10)} |")
+    # # List datasets
+    # print("\nDatasets:")
+    # datasets = api.list_datasets()
+    # # List spaces
+    # print("\nSpaces:")
+    # spaces = api.list_spaces()
+
+    while True:
+        print("\n请选择操作:")
+        print("1. 添加一个新项目")
+        print("2. 下载某个项目")
+        print("3. 选择一个项目为当前项目")
+        print("4. 返回上一级")
+        choice = input("请输入操作编号: ")
+        if choice == "1":
+            repo_name = input("请输入存储库名称, 直接回车取消: ")
+            if repo_name == "":
+                continue
+            private = input("是否为私有库， 请输入 y 或 n: ") == "y"
+            repo_id = create_repository(repo_name, private= private)
+
+        elif choice == "2":
+            download_project()
+        elif choice == "3":
+            select_current_project(start_num + 1, end_num)
+        elif choice == "4":
+            return  # Return to the previous level
+        else:
+            print("无效的操作编号，请重新输入")
+
+
+def select_current_project(start_num, end_num):
+    """选择一个项目为当前项目."""
+    global current_project, model_list
+    while True:
+        selected_id = input(f"请在 {str(start_num)} - {str(end_num)} 之中选择一个项目编号，或直接输入repo_id作为当前选择的项目: ")
+        if selected_id.isdigit() and start_num <= int(selected_id) <= end_num:
+            current_project = model_list[int(selected_id)]
+            print(f"当前项目: {current_project}")
+            show_project_details()
+            break
+        elif "/" in selected_id:
+            current_project = selected_id
+            print(f"当前项目: {current_project}")
+            show_project_details()
+            break
+        else:
+            print("无效的项目编号，请重新选���。")
+
+def show_project_details():
+    """显示项目的具体内容，并提供下载和上传文件的选项."""
+    global current_project, api
+    makesure_login()
+    if current_project:
+        try:
+            # Get project details
+            info = api.repo_info(current_project)
+            print(f"项目信息:")
+            print(f"ID: {info.id}")
+            print(f"Author: {info.author}")
+            print(f"Created At: {info.created_at}")
+            print(f"Last Modified: {info.last_modified}")
+            print(f"Private: {info.private}")
+            
+            print_tree(current_project)
+            # Display project content using tree structure
+            print("\n项目内容:")
+
+            # Provide options for downloading or uploading files
+            while True:
+                print("\n请选择操作:")
+                print("1. 下载一个或多个项目文件")
+                print("2. 上传一个或多个项目文件")
+                print("3. 返回上一级")
+                choice = input("请输入操作编号: ")
+                if choice == "1":
+                    download_files(current_project)
+                elif choice == "2":
+                    upload_files(current_project)
+                elif choice == "3":
+                    return  # Return to the previous level
+                else:
+                    print("无效的操作编号，请重新输入。")
+        except HfHubHTTPError as e:
+            print(f"获取项目详情失败: {e}")
+    else:
+        print("未选择任何项目。")
+
+
+
+def format_size(size):
+    """将文件大小格式化为更易读的形式."""
+    power = 2**10
+    n = 0
+    power_labels = {0: '', 1: 'K', 2: 'M', 3: 'G', 4: 'T'}
+    while size > power:
+        size /= power
+        n += 1
+    return f"{size:.2f} {power_labels[n]}B"
+
+def print_tree(repo_id, path=None, expand=False, revision=None, repo_type=None, indent=""):
+    """以树状形式打印项目文件结构."""
+    global api
+    if path is None:
+        path = ''
+    repo_tree = api.list_repo_tree(repo_id, path)
+    for item in repo_tree:
+        if isinstance(item, RepoFile):
+            file_size = format_size(item.size) if item.size else "0 B"
+            print(indent + f"📄 {item.path} (Size: {file_size})")
+        elif isinstance(item, RepoFolder):
+            print(indent + f"📁 {item.path}/")
+            print_tree(repo_id, path=f"{item.path}", expand=expand, indent=indent + "   ")
+
+def save_last_folder(folder_path):
+    """
+    Save the last used folder path to a configuration file.
+    If the folder does not exist, create it.
+    
+    Args:
+    - folder_path (str): The folder path to save.
+    """
+    if not os.path.exists(folder_path):
+        os.makedirs(folder_path, exist_ok=True)
+        print(f"文件夹 '{folder_path}' 不存在，已创建。")
+
+    config = {"last_folder": folder_path}
+    with open("config.json", "w") as f:
+        json.dump(config, f)
+
+def get_default_folder():
+    default_folder = os.getcwd()  # 默认文件夹为当前工作目录
+    config_file = "config.json"
+    if os.path.exists(config_file):
+        with open(config_file, "r") as f:
+            config = json.load(f)
+            if "last_folder" in config and os.path.exists(config["last_folder"]):
+                default_folder = config["last_folder"]
+    return default_folder
+
+def download_files(repo_id):
+    total_files = 0
+    successful_downloads = 0
+    failed_downloads = 0
+
+    default_folder = get_default_folder()
+
+    while True:
+        print("\n请输入要下载的文件路径，多个文件请使用逗号分隔, 回车返回上一级")
+        file_paths_input = input("请输入文件路径: ")
+        if len(file_paths_input) == 0:
+            return
+        
+        folder_path = input(f"请输入文件夹路径，按回车使用{default_folder}：")
+
+        if folder_path == "":
+            folder_path = default_folder
+
+        save_last_folder(folder_path)
+
+        file_paths = [file_path.strip() for file_path in file_paths_input.split(",")]
+        total_files += len(file_paths)
+
+        # Prompt user to place downloaded files in corresponding folders
+        user_choice = input("是否把下载的文件放到对应的文件夹？(Y/n)：")
+        if user_choice.lower() != 'n':
+            dir_obj = True
+
+        for file_path in file_paths:
+            # Extract filename and subfolder from file_path
+            if "/" in file_path:
+                parts = file_path.rsplit("/", 1)
+                filename = parts[-1]
+                subfolder = parts[0]
+            else:
+                filename = file_path
+                subfolder = ""
+
+            try:
+                path_input = folder_path
+                if dir_obj:
+                    path_input = folder_path + "/" + subfolder
+                print(f'downloading {filename} to {path_input}...')
+                api.hf_hub_download(repo_id=repo_id, filename=filename, subfolder=subfolder, local_dir=path_input, local_dir_use_symlinks=False)
+                successful_downloads += 1
+            except Exception as e:
+                print(f"下载文件 '{filename}' 时出错：{e}")
+                failed_downloads += 1
+
+        print(f"\n总共下载文件数: {total_files}")
+        print(f"成功下载文件数: {successful_downloads}")
+        print(f"失败下载文件数: {failed_downloads}")
+
+import os
+
+def upload_files(repo_id):
+    """上传一个或多个项目文件."""
+    global api
+    makesure_login()
+    if not is_own_project(repo_id):
+        print("检测到当前项目不是本人的项目，故而没有上传权限")
+        return
+    while True:
+        file_paths_input = input("请输入要上传的文件路径(多个文件请用逗号分隔), 回车返回上一级: ")
+        if len(file_paths_input) == 0:
+            return
+        file_paths = [path.strip() for path in file_paths_input.split(",")]
+
+        relative_paths_input = input("请输入要上传的文件相对路径, 没有会自动创建，不填全部上传到根目录(多个路径请用逗号分隔): ")
+        if len(relative_paths_input) == 0:
+            relative_paths = [""]
+        else:
+            relative_paths = [path.strip() for path in relative_paths_input.split(",")]
+
+        if len(relative_paths) < len(file_paths):
+            fill_choice = input("文件路径数目大于相对路径数目，请选择填充方式：\n1. 使用根目录填充\n2. 根据最后一个路径填充\n请选择(1/2): ")
+            if fill_choice == "1":
+                relative_paths += ["" for _ in range(len(file_paths) - len(relative_paths))]
+            elif fill_choice == "2":
+                last_path = relative_paths[-1] if relative_paths else ""
+                relative_paths += [last_path for _ in range(len(file_paths) - len(relative_paths))]
+        elif len(relative_paths) > len(file_paths):
+            ignored_paths = relative_paths[len(file_paths):]
+            print(f"文件路径数目小于相对路径数目,以下相对路径将被忽略: {', '.join(ignored_paths)}")
+            relative_paths = relative_paths[:len(file_paths)]
+
+        # 获取当前时间
+        now = datetime.now()
+        current_time = now.strftime("%Y-%m-%d %H:%M:%S")
+        default_message = f"{user_id} uploaded at {current_time}"
+        commit_message = input(f"请输入本次提交的目的，(回车默认填写:{default_message}):")
+
+        for file_path, relative_path in zip(file_paths, relative_paths):
+            # Convert relative path to absolute path
+            absolute_path = os.path.abspath(file_path)
+            if not os.path.exists(file_path):
+                print(f"文件 '{file_path}' 不存在，跳过上传。")
+                continue
+
+            try:
+
+                # 构造 commit_message
+                if len(commit_message) == 0: 
+                    commit_message = default_message
+                else: 
+                    commit_message = f"{user_id}: " + commit_message
+                file_name = os.path.basename(file_path)
+                print(f'uploading {absolute_path} to {repo_id}/{relative_path}/{file_name}...')
+
+                api.upload_file(repo_id=repo_id, path_or_fileobj=absolute_path, path_in_repo=relative_path + "/"  + file_name, commit_message=commit_message)
+                print(f"文件 {file_path} 上传成功。")
+            except HfHubHTTPError as e:
+                print(f"文件 {file_path} 上传失败: {e}")
+            except ValueError as e:
+                print(f"文件 {file_path} 上传失败: {e}")
+            except RepositoryNotFoundError as e:
+                print(f"文件 {file_path} 上传失败: {e}")
+            except RevisionNotFoundError as e:
+                print(f"文件 {file_path} 上传失败: {e}")
+
+        break
+
+def is_own_project(repo_id):
+    """检查给定的 repo_id 是否为当前用户的项目."""
+    return repo_id in model_list.values()
+
+def create_repository(repo_name, private = False):
+    global user, api
+    makesure_login()
+    try:
+        # 创建存储库
+        api.create_repo(repo_name, private=private)
+    except HfHubHTTPError as e:
+        error_code = extract_first_three_digits(str(e))
+        if error_code == 403:
+            print("HTTPError 403: You don't have write permissions to the repository.")
+        elif error_code == 404:
+            print("HTTPError 404: The repository or project doesn't exist on the Hub.")
+        elif error_code == 409:
+            print("HTTPError 409: The project already exists in the collection and _ok=False.")
+        else:
+            print(f"HTTPError {e.server_message}")
+
+    except ValueError as e:
+        """bug"""
+        print("warning:"  + str(e))
+    
+    repo_id = user + '/' + repo_name
+    if not private:
+        url = mirror_url + "/" + repo_id
+    else:
+        url = "https://huggingface.co" + + "/" + repo_id
+    # 检查是否成功创建存储库
+    if url:
+        print(f"存储库 {repo_id} 创建成功！地址为：" + url)
+    else:
+        print("存储库创建失败，请检查权限或存储库是否已存在。")
+    return repo_id
+
+def extract_first_three_digits(s):
+    match = re.search(r'\d{3}', s)
+    if match:
+        return int(match.group())
+    return 0
+
+def getrepoid_from_url(url):
+    parts = url.split('/')
+    repo_id = parts[-2] + "/" + parts[-1]
+    return repo_id
+
+def download_project():
+    """下载某个项目."""
+    # Your code to download a project goes here
+    pass
+
+def switch_account():
+    """切换账户."""
+    access_token = input("请输入您的新 AccessToken: ")
+    save_access_token(access_token)
+    makesure_login()  # 切换账户后立即登录
+
+def main():
+    makesure_login(True)
+    while True:
+        print("\n请选择操作:")
+        print("1. 查看项目列表")
+        print("2. 切换账户")
+        choice = input("请输入操作编号: ")
+        if choice == "1":
+            list_projects()
+        elif choice == "2":
+            switch_account()
+        else:
+            print("无效的操作编号，请重新输入")
+
+if __name__ == "__main__":
+    main()