将单个文件上传到 ModelScope

到写下这篇博客时，ModelScope Api 依然不支持上传单文件（HuggingFace Api 在这方面就做得很好），想要上传单文件只能通过 Git 进行，所以自己手搓了 ipynb 脚本实现这个功能。

下面是实现功能的代码（使用前需要使用 Pip 安装 huggingface_hub、modelscope，还有 Git LFS）。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
# 获取文件夹中所有的文件的绝对路径
def get_all_file(directory):
    import os
    file_list = []
    for dirname, _, filenames in os.walk(directory):
        for filename in filenames:
            file_list.append(os.path.join(dirname, filename))
    return file_list


# 这种方法繁琐一点
# def get_all_file(directory):
#     import os
#     file_list = os.listdir(directory)
#     file_lists = []
#     for each_file in file_list:
#         sub_file = os.path.join(directory, each_file)
#         if os.path.isdir(sub_file):
#             file_lists.extend(get_all_file(sub_file))
#         else:
#             file_lists.append(sub_file)
#     return file_lists


# 获取相对路径(弃用)
# def get_rel_path(folder_path ,path_list):
#     import os
#     rel_path_list = []
#     for path in path_list:
#         rel_path = os.path.relpath(path, folder_path)
#         rel_path_list.append(rel_path)
#     return rel_path_list

# ModelScope

# 检测是否存在于仓库中
def is_file_exists_in_ms_repo(upload_file, work_path, repo):
    import os
    repo_file = os.path.join(work_path, repo.split("/").pop(), upload_file)
    if os.path.exists(repo_file):
        return True
    else:
        return False


# 克隆仓库
def clone_modelscope_without_lfs(ms_access_token, repo, work_path):
    import os
    # 禁用 Git LFS
    os.environ["GIT_LFS_SKIP_SMUDGE"] = "1"
    !git lfs uninstall

    # 本地存在仓库时进行删除
    repo_name = repo.split("/").pop()
    path = os.path.join(work_path, repo_name)
    if os.path.exists(path):
        !rm -rf "{path}"

    # 下载仓库并启用 Git LFS
    repo_url = f"https://oauth2:{ms_access_token}@www.modelscope.cn/{repo}.git"
    !git clone "{repo_url}" "{path}"
    os.environ["GIT_LFS_SKIP_SMUDGE"] = "0"
    !git lfs install


# 上传文件至 ModelScope
def push_file_to_modelscope(ms_access_token, repo, work_path, upload_path):
    import os
    from modelscope.hub.api import HubApi
    if repo.split("/").pop() == upload_path.split("/").pop():
        raise Exception("本地要上传的仓库名称与要上传到的仓库的名称相同, 这将导致本地要上传的仓库被自动删除")

    api = HubApi()
    try:
        ms_access_token = api.login(ms_access_token)[0] # 将 ModelScope Token 转为 ModelScope Git Token
        print(":: ModelScope Token 验证成功")
    except Exception as e:
        print(":: ModelScope Token 验证失败: ", e)
        return

    os.chdir(work_path)
    count = 0
    upload_file_lists = get_all_file(upload_path) # 原文件的路径列表

    print(f"将文件上传至 ModelScope:: {upload_path} -> {os.path.join(work_path, repo.split('/').pop())}")
    count = 0
    sum = len(upload_file_lists)
    for upload_file in upload_file_lists:
        count += 1

        print(f"[{count}/{sum}]:: 克隆仓库到 {work_path}")
        clone_modelscope_without_lfs(ms_access_token, repo, work_path)
        rel_upload_file = os.path.relpath(upload_file, upload_path) # 原文件相对路径
        repo_path = os.path.join(work_path, repo.split("/").pop()) # 仓库的绝对路径
        print(f"[{count}/{sum}]:: 要上传的文件的相对路径: {rel_upload_file}")
        print(f"[{count}/{sum}]:: 绝对路径: {upload_file}")
        print(f"[{count}/{sum}]:: 仓库地址: {repo_path}")

        # 检测文件是否存在于仓库中
        if is_file_exists_in_ms_repo(rel_upload_file, work_path, repo):
            !rm -rf "{repo_path}"
            print(f"[{count}/{sum}]:: {os.path.basename(upload_file)} 已存在于仓库中")
        else:
            # 为仓库创建对应的文件夹
            p_path = os.path.dirname(os.path.join(work_path, repo.split("/").pop(), rel_upload_file)) # 原文件到仓库中后对应的父文件夹
            if not os.path.exists(p_path):
                print(f"[{count}/{sum}]:: 创建文件对应的父文件夹: {p_path}")
                os.makedirs(p_path, exist_ok=True)

            print(f"[{count}/{sum}]:: {upload_file} -> {repo_path}")
            !cp "{upload_file}" "{p_path}"

            os.chdir(repo_path)
            file_name = rel_upload_file.split("/").pop() # 文件名
            print(f"[{count}/{sum}]:: 添加文件： {rel_upload_file}")
            !git add "{rel_upload_file}"
            print(f"[{count}/{sum}]:: 提交信息: \"upload {file_name}\"")
            !git commit -m "upload {file_name}"
            !git config lfs.https://oauth2:{ms_access_token}@www.modelscope.cn/{repo}.git/info/lfs.locksverify true
            print(f"[{count}/{sum}]:: 上传 {file_name} 到 {repo} 中")
            !git push

            os.chdir(work_path)
            !rm -rf "{repo_path}"
            print(f"[{count}/{sum}]:: 上传 {file_name} 完成")

    print(f"[{count}/{sum}]:: {repo} 仓库上传完成")


# HuggingFace

# 获取 HuggingFace 仓库中的文件列表
def get_hf_repo_file_list(hf_access_token, repo, repo_type):
    from huggingface_hub import HfApi
    api = HfApi()
    model_list = api.list_repo_files(
        repo_id = repo,
        repo_type = repo_type,
        token = hf_access_token
    )
    return model_list


# 上传文件到 HuggingFace
def push_file_to_huggingface(hf_access_token, repo, repo_type, work_path, upload_path):
    import os
    from pathlib import Path
    from huggingface_hub import HfApi, CommitOperationAdd
    api = HfApi()

    try:
        api.whoami(token = hf_access_token)
        print(":: HuggingFace Token 验证成功")
    except Exception as e:
        print(":: HuggingFace Token 验证失败: ", e)

    os.chdir(work_path)
    count = 0
    upload_file_lists = get_all_file(upload_path) # 原文件的路径列表

    print(f"将文件上传至 HuggingFace:: {upload_path} -> {os.path.join(work_path, repo.split('/').pop())}")
    count = 0
    sum = len(upload_file_lists)
    hf_repo_flie_list = get_hf_repo_file_list(hf_access_token, repo, repo_type) # 获取仓库中的文件列表
    for upload_file in upload_file_lists:
        count += 1

        print(f"[{count}/{sum}]:: 克隆仓库到 {work_path}")
        rel_upload_file = os.path.relpath(upload_file, upload_path) # 原文件相对路径
        repo_path = os.path.join(work_path, repo.split("/").pop()) # 仓库的绝对路径
        print(f"[{count}/{sum}]:: 要上传的文件的相对路径: {rel_upload_file}")
        print(f"[{count}/{sum}]:: 绝对路径: {upload_file}")
        print(f"[{count}/{sum}]:: 仓库地址: {repo_path}")

        # 检测文件是否存在于仓库中
        if rel_upload_file in hf_repo_flie_list:
            print(f"[{count}/{sum}]:: {os.path.basename(upload_file)} 已存在于仓库中")
        else:
            file_name = rel_upload_file.split("/").pop() # 文件名
            hf_path_in_repo = Path(rel_upload_file).as_posix()
            local_file_obj = Path(upload_file).as_posix()
            print(f"[{count}/{sum}]:: {local_file_obj} -> {repo}/{hf_path_in_repo}")
            operations = [ CommitOperationAdd(path_in_repo = hf_path_in_repo, path_or_fileobj = local_file_obj) ]
            print(f"[{count}/{sum}]:: 上传 {file_name} 到 {repo} 中")
            print(f"repo: {repo}, repo_type: {repo_type}")
            api.create_commit(
                repo_id = repo,
                operations = operations,
                commit_message = f"Upload {file_name}",
                repo_type = repo_type,
                token = hf_access_token
            )
            print(f"[{count}/{sum}]:: 上传 {file_name} 完成")

    print(f"[{count}/{sum}]:: {repo} 仓库上传完成")

# 使用方法
push_file_to_modelscope(
    ms_access_token = "xxxxxxxxxxxxx", # Modelscope Token
    repo = "licyks/test", # Modelscope 的仓库地址
    work_path = "D:/Downloads", # 脚本工作目录, 仓库将克隆至该文件夹中
    upload_path = "D:/Downloads/BaiduNetdiskWorkspace" # 要上传文件的目录
)

# 这个方法也可以用在 HuggingFace 上
push_file_to_huggingface(
    hf_access_token = "xxxxxxxxxxxx", # HuggingFace Token
    repo = "licyk/test", # HuggingFace 仓库地址
    repo_type = "model", # HuggingFace 仓库种类
    work_path = "D:/Downloads", # 脚本工作目录, 仓库将克隆至该文件夹中
    upload_path = "D:/Downloads/BaiduNetdiskWorkspace" # 要上传文件的目录
)

脚本的大概做了一下几件事：

  1. 遍历将要上传文件的文件夹中的所有文件，并记录到一个待上传的文件列表中。
  2. 禁用 Git LFS 后将仓库克隆到本地，再启用 Git LFS，这时候仓库中的 LFS 文件以指针的形式存在，大大减小占用空间。
  3. 遍历待上传的文件列表，检测仓库中是否已存在这个文件，如果不存在，则把文件复制到本地仓库中，并推送至远端仓库中。
  4. 删除本地的仓库，再重复2和3步骤，直至待上传的文件列表遍历完成。

这个方法虽然有点繁琐，但解决了无法通过 ModelScope Api 上传单文件的问题。

把代码整合到 Colab 上可实现做仓库镜像的功能。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
#@title 👇 环境配置

# 文件下载
def aria2(url, path = None, filename = None):
    import os
    if path is None:
        path = os.getcwd()
    if filename is None:
        filename = url.split("/").pop()
    if not os.path.exists(path + "/" + filename):
        print(f":: 开始下载 {filename} ，路径: {path}/{filename}")
        !aria2c --console-log-level=error -c -x 16 -s 16 "{url}" -d "{path}" -o "{filename}"
        if os.path.exists(path + "/" + filename) and not os.path.exists(path + "/" + filename + ".aria2"):
            print(f":: {filename} 下载完成")
        else:
            print(f":: {filename} 下载中断")
    else:
        if os.path.exists(path + "/" + filename + ".aria2"):
            print(f":: 开始下载 {filename} ，路径: {path}/{filename}")
            !aria2c --console-log-level=error -c -x 16 -s 16 "{url}" -d "{path}" -o "{filename}"
            if os.path.exists(path + "/" + filename) and not os.path.exists(path + "/" + filename + ".aria2"):
                print(f":: {filename} 下载完成")
            else:
                print(f":: {filename} 下载中断")
        else:
            print(f":: {filename} 文件已存在，路径: {path}/{filename}")

###############################################
# 上传单文件

# 获取文件夹中所有的文件的绝对路径
def get_all_file(directory):
    import os
    file_list = []
    for dirname, _, filenames in os.walk(directory):
        for filename in filenames:
            file_list.append(os.path.join(dirname, filename))
    return file_list


# 这种方法繁琐一点
# def get_all_file(directory):
#     import os
#     file_list = os.listdir(directory)
#     file_lists = []
#     for each_file in file_list:
#         sub_file = os.path.join(directory, each_file)
#         if os.path.isdir(sub_file):
#             file_lists.extend(get_all_file(sub_file))
#         else:
#             file_lists.append(sub_file)
#     return file_lists


# 获取相对路径(弃用)
# def get_rel_path(folder_path ,path_list):
#     import os
#     rel_path_list = []
#     for path in path_list:
#         rel_path = os.path.relpath(path, folder_path)
#         rel_path_list.append(rel_path)
#     return rel_path_list

# ModelScope

# 检测是否存在于仓库中
def is_file_exists_in_ms_repo(upload_file, work_path, repo):
    import os
    repo_file = os.path.join(work_path, repo.split("/").pop(), upload_file)
    if os.path.exists(repo_file):
        return True
    else:
        return False


# 克隆仓库
def clone_modelscope_without_lfs(ms_access_token, repo, work_path):
    import os
    # 禁用 Git LFS
    os.environ["GIT_LFS_SKIP_SMUDGE"] = "1"
    !git lfs uninstall

    # 本地存在仓库时进行删除
    repo_name = repo.split("/").pop()
    path = os.path.join(work_path, repo_name)
    if os.path.exists(path):
        !rm -rf "{path}"

    # 下载仓库并启用 Git LFS
    repo_url = f"https://oauth2:{ms_access_token}@www.modelscope.cn/{repo}.git"
    !git clone "{repo_url}" "{path}"
    os.environ["GIT_LFS_SKIP_SMUDGE"] = "0"
    !git lfs install


# 上传文件至 ModelScope
def push_file_to_modelscope(ms_access_token, repo, work_path, upload_path):
    import os
    from modelscope.hub.api import HubApi
    if repo.split("/").pop() == upload_path.split("/").pop():
        raise Exception("本地要上传的仓库名称与要上传到的仓库的名称相同, 这将导致本地要上传的仓库被自动删除")

    api = HubApi()
    try:
        ms_access_token = api.login(ms_access_token)[0] # 将 ModelScope Token 转为 ModelScope Git Token
        print(":: ModelScope Token 验证成功")
    except Exception as e:
        print(":: ModelScope Token 验证失败: ", e)

    os.chdir(work_path)
    count = 0
    upload_file_lists = get_all_file(upload_path) # 原文件的路径列表

    print(f"将文件上传至 ModelScope:: {upload_path} -> {os.path.join(work_path, repo.split('/').pop())}")
    count = 0
    sum = len(upload_file_lists)
    for upload_file in upload_file_lists:
        count += 1

        print(f"[{count}/{sum}]:: 克隆仓库到 {work_path}")
        clone_modelscope_without_lfs(ms_access_token, repo, work_path)
        rel_upload_file = os.path.relpath(upload_file, upload_path) # 原文件相对路径
        repo_path = os.path.join(work_path, repo.split("/").pop()) # 仓库的绝对路径
        print(f"[{count}/{sum}]:: 要上传的文件的相对路径: {rel_upload_file}")
        print(f"[{count}/{sum}]:: 绝对路径: {upload_file}")
        print(f"[{count}/{sum}]:: 仓库地址: {repo_path}")

        # 检测文件是否存在于仓库中
        if is_file_exists_in_ms_repo(rel_upload_file, work_path, repo):
            !rm -rf "{repo_path}"
            print(f"[{count}/{sum}]:: {os.path.basename(upload_file)} 已存在于仓库中")
        else:
            # 为仓库创建对应的文件夹
            p_path = os.path.dirname(os.path.join(work_path, repo.split("/").pop(), rel_upload_file)) # 原文件到仓库中后对应的父文件夹
            if not os.path.exists(p_path):
                print(f"[{count}/{sum}]:: 创建文件对应的父文件夹: {p_path}")
                os.makedirs(p_path, exist_ok=True)

            print(f"[{count}/{sum}]:: {upload_file} -> {repo_path}")
            !cp "{upload_file}" "{p_path}"

            os.chdir(repo_path)
            file_name = rel_upload_file.split("/").pop() # 文件名
            print(f"[{count}/{sum}]:: 添加文件： {rel_upload_file}")
            !git add "{rel_upload_file}"
            print(f"[{count}/{sum}]:: 提交信息: \"upload {file_name}\"")
            !git commit -m "upload {file_name}"
            !git config lfs.https://oauth2:{ms_access_token}@www.modelscope.cn/{repo}.git/info/lfs.locksverify true
            print(f"[{count}/{sum}]:: 上传 {file_name} 到 {repo} 中")
            !git push

            os.chdir(work_path)
            !rm -rf "{repo_path}"
            print(f"[{count}/{sum}]:: 上传 {file_name} 完成")

    print(f"[{count}/{sum}]:: {repo} 仓库上传完成")


# HuggingFace

# 获取 HuggingFace 仓库中的文件列表
def get_hf_repo_file_list(hf_access_token, repo, repo_type):
    from huggingface_hub import HfApi
    api = HfApi()
    model_list = api.list_repo_files(
        repo_id = repo,
        repo_type = repo_type,
        token = hf_access_token
    )
    return model_list


# 上传文件到 HuggingFace
def push_file_to_huggingface(hf_access_token, repo, repo_type, work_path, upload_path):
    import os
    from pathlib import Path
    from huggingface_hub import HfApi, CommitOperationAdd
    api = HfApi()

    try:
        api.whoami(token = hf_access_token)
        print(":: HuggingFace Token 验证成功")
    except Exception as e:
        print(":: HuggingFace Token 验证失败: ", e)

    os.chdir(work_path)
    count = 0
    upload_file_lists = get_all_file(upload_path) # 原文件的路径列表

    print(f"将文件上传至 HuggingFace:: {upload_path} -> {os.path.join(work_path, repo.split('/').pop())}")
    count = 0
    sum = len(upload_file_lists)
    hf_repo_flie_list = get_hf_repo_file_list(hf_access_token, repo, repo_type) # 获取仓库中的文件列表
    for upload_file in upload_file_lists:
        count += 1

        print(f"[{count}/{sum}]:: 克隆仓库到 {work_path}")
        rel_upload_file = os.path.relpath(upload_file, upload_path) # 原文件相对路径
        repo_path = os.path.join(work_path, repo.split("/").pop()) # 仓库的绝对路径
        print(f"[{count}/{sum}]:: 要上传的文件的相对路径: {rel_upload_file}")
        print(f"[{count}/{sum}]:: 绝对路径: {upload_file}")
        print(f"[{count}/{sum}]:: 仓库地址: {repo_path}")

        # 检测文件是否存在于仓库中
        if rel_upload_file in hf_repo_flie_list:
            print(f"[{count}/{sum}]:: {os.path.basename(upload_file)} 已存在于仓库中")
        else:
            file_name = rel_upload_file.split("/").pop() # 文件名
            hf_path_in_repo = Path(rel_upload_file).as_posix()
            local_file_obj = Path(upload_file).as_posix()
            print(f"[{count}/{sum}]:: {local_file_obj} -> {repo}/{hf_path_in_repo}")
            operations = [ CommitOperationAdd(path_in_repo = hf_path_in_repo, path_or_fileobj = local_file_obj) ]
            print(f"[{count}/{sum}]:: 上传 {file_name} 到 {repo} 中")
            print(f"repo: {repo}, repo_type: {repo_type}")
            api.create_commit(
                repo_id = repo,
                operations = operations,
                commit_message = f"Upload {file_name}",
                repo_type = repo_type,
                token = hf_access_token
            )
            print(f"[{count}/{sum}]:: 上传 {file_name} 完成")

    print(f"[{count}/{sum}]:: {repo} 仓库上传完成")

# 使用方法
# push_file_to_modelscope(
#     ms_access_token = "xxxxxxxxxxxxx", # Modelscope Token
#     repo = "licyks/test", # Modelscope 的仓库地址
#     work_path = "D:/Downloads", # 脚本工作目录, 仓库将克隆至该文件夹中
#     upload_path = "D:/Downloads/BaiduNetdiskWorkspace" # 要上传文件的目录
# )

# push_file_to_huggingface(
#     hf_access_token = "xxxxxxxxxxxx", # HuggingFace Token
#     repo = "licyk/test", # HuggingFace 仓库地址
#     repo_type = "model", # HuggingFace 仓库种类
#     work_path = "D:/Downloads", # 脚本工作目录, 仓库将克隆至该文件夹中
#     upload_path = "D:/Downloads/BaiduNetdiskWorkspace" # 要上传文件的目录
# )

###############################################
# HuggingFace / ModelScope Token 验证

def verify_huggingface_token(hf_token):
    from huggingface_hub import HfApi
    api = HfApi()
    try:
        api.whoami(hf_token)
        print(":: HuggingFace Token 验证成功")
    except Exception as e:
        print(":: HuggingFace Token 验证失败: ", e)


def verify_modelscope_token(ms_token):
    from modelscope.hub.api import HubApi
    api = HubApi()
    try:
        api.login(ms_token)
        print(":: ModelScope Token 验证成功")
    except Exception as e:
        print(":: ModelScope Token 验证失败: ", e)

def get_modelscope_git_token(ms_token):
    from modelscope.hub.api import HubApi
    api = HubApi()
    try:
        token = api.login(ms_token)[0]
        return token
    except:
        return None

###############################################
# 配置 Git 信息
def set_git_config(email, username):
    print(":: 配置 Git 信息中")
    !git config --global user.email "{email}"
    !git config --global user.name "{username}"

###############################################
print(":: 配置环境中")
!pip install huggingface_hub modelscope
!apt update
!apt install aria2 wget -y
!apt clean
!pip cache purge
from IPython.display import clear_output
clear_output(wait=False)
###############################################

HF_TOKEN = "" #@param {type:"string"}
MS_TOKEN = "" #@param {type:"string"}
MS_GIT_TOKEN = get_modelscope_git_token(MS_TOKEN)
GIT_USER_EMAIL = "" #@param {type:"string"}
GIT_USER_NAME = "" #@param {type:"string"}

###############################################
print(":: 验证 Token 中")
verify_huggingface_token(HF_TOKEN)
verify_modelscope_token(MS_TOKEN)
set_git_config(GIT_USER_EMAIL, GIT_USER_NAME)
print(":: 环境配置完成")

真希望 ModelScope Api 能更加完善。

经验
#linux
将单个文件上传到 ModelScope
http://licyk.github.io/2024/06/14/upload-single-file-to-modelscope/
作者
licyk
发布于
2024年6月14日
许可协议