Delete test4

test4/huggingface.py

@@ -1,434 +0,0 @@
-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, s3.us-east-1.amazonaws.com,  lfs.huggingface.co'
-
-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()