环境准备

七牛云

申请 AccessKey 和 SecretKey

https://www.qiniu.com/products/kodo#docs

安装 Python 的 SDK

文档地址

https://developer.qiniu.com/kodo/1242/python#4

1
pip3 install qiniu

安装 Python 运行环境

https://www.runoob.com/python3/python3-install.html

编写代码

新建upload_help.py文件

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
# -*- coding: utf-8 -*-
# flake8: noqa

from qiniu import Auth, put_file

# 需要填写你的 Access Key 和 Secret Key
access_key = '申请到的AccessKey'
secret_key = '申请到的SecretKey'
global_token = ''

def get_token():
    """ 获取上传文件的token """

    if global_token != '':
        return global_token

    # 构建鉴权对象
    q = Auth(access_key, secret_key)

    # 要上传的空间
    bucket_name = '在七牛云添加的bucket空间'

    # 生成上传 Token，可以指定过期时间等
    token = q.upload_token(bucket_name, None, 3600)

    return token


def upload_file(local_file, key):
    try:
        """ 上传文件到七牛云存储 """
        token = get_token()
        ret, info = put_file(token, key, local_file, version='v2')
        if info.status_code == 200:
            return True
        return False
    except:
        return False

新建main.py文件

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
# -*- coding: utf-8 -*-
import os
import upload_help


def dir_name(file_dir):
    total_count = 0
    total_succeed_count = 0
    total_fail_list = []
    print('开始上传文件...')
    for root, dirs, files in os.walk(file_dir):
        for dir in dirs:
            if dir != "html" and dir != "preview":
                count, succeed_count, fail_list = file_name(
                    '%s/%s' % (root, dir))
                total_count += count
                total_succeed_count += succeed_count
                total_fail_list.extend(fail_list)
    print('上传完成...')
    print('%s本次共处理%s个文件' % (file_dir, total_count))
    print('%s本次上传成功%s个文件,失败%s个文件' %
          (file_dir, total_succeed_count, total_count - total_succeed_count))
    open('%s/fail_list.txt' % file_dir, 'w').write(str(total_fail_list))


def file_name(file_dir):
    total_count = 0
    total_succeed_count = 0
    fail_list = []

    for root, dirs, files in os.walk(file_dir):
        count = 0
        succeed_count = 0
        for file in files:
            fname = '%s/%s' % (root, file)
            # 由于业务需要，过滤到下面那两个文件夹，可以删除
            previewIndex = fname.find('/preview/')
            htmlIndex = fname.find('/html/')
            if previewIndex == -1 and htmlIndex == -1:
                count += 1
                succeed = upload_help.upload_file(fname, file)
                if succeed:
                    succeed_count += 1
                if succeed != True:
                    fail_list.append(fname)

        total_count += count
        total_succeed_count += succeed_count
    return total_count, total_succeed_count, fail_list


if __name__ == '__main__':
    # 文件夹路径
    dir_name(r'/Users/vanzheng/xx/dirs')

运行

1
python3 main.py