七牛云数据存储Demo

利用七牛云的Python SDK实现文件上传、下载等操作。

import os
import requests
import qiniu
from qiniu import BucketManager
from qiniu import Auth
from qiniu import utils

# 你自己的accessKey
accessKey = 'DjZp39jT_GN7yr0i_vf5Zs3vBnzYjzkxd7kt6tuK'
# 你自己的secretKey
secretKey = 'HouhuF7Essfu-VveQ3GYXTcEddgP1qrwtblWuMNa'
# 你自己的bucket_domain
bucket_domain = '7xjsrw.com3.w1.blg.clouddn.com'


class Kpan():
    def __init__(self, base_dir, ak=accessKey, sk=secretKey):
        """初始化Kpan类
            base_dir：工作路径
            ak：accessKey
            sk： secretKey
        """
        self.auth = Auth(ak, sk)
        self.bucket_manager = BucketManager(self.auth)
        self.kpan_bucket = 'kpan'
        self.base_dir = os.path.abspath(base_dir)

    def upload_file(self, key):
        filePath = os.path.join(self.base_dir, key)
        # 计算当前文件的hash值，如果服务器上存在相同的副本取消上传
        file_hash = utils.etag(filePath)

        if file_hash == self.is_file_exist(key):
            print('服务器已存在相同的副本，上传取消...')
            return

        upToken = self.auth.upload_token(self.kpan_bucket, key)
        ret, info = qiniu.put_file(upToken, key, filePath)
        print('%s上传成功...' % key)

    def download_file(self, key, toPath=None):
        '''下载单个文件
           key:要下载的文件
           toPath：存储路径，默认为base_dir
        '''
        if not toPath:
            toPath = self.base_dir
        base_url = 'http://%s/%s' % (bucket_domain, key)
        # 设置token过期时间
        private_url = self.auth.private_download_url(base_url, expires=3600)
        r = requests.get(private_url)

        path = os.path.join(toPath, key)
        dir, _ = os.path.split(path)
        if not os.path.exists(dir):
            os.makedirs(dir)
        with open(path, 'wb') as f:
            f.write(r.content)
            print('%s 文件保存成功' % key)

    def get_all_files(self):
        items = self.list_all_file()
        for item in items:
            self.download_file(item['key'])

    def delete_file(self, key):
        if not self.is_file_exist(key):
            print('%s 文件不存在...' % key)
            return

        ret, info = self.bucket_manager.delete(self.kpan_bucket, key)
        if not ret:
            print('%s 文件删除成功...' % key)
        else:
            print(ret)

    def list_all_file(self):
        ret, eof, info = self.bucket_manager.list(self.kpan_bucket)
        items = ret['items']
        # 返回所有文件列表
        return items

    def rename(self, key, tokey):
        if not self.is_file_exist(key) or self.is_file_exist(tokey):
            print('重命名失败,“%s”可能已存在' % tokey)
            return

        ret, info = self.bucket_manager.rename(self.kpan_bucket, key, tokey)
        if not ret:
            print('将“%s”重命名为“%s”成功...' % (key, tokey))
        else:
            print('“%s”重命名失败...' % key)

    def is_file_exist(self, key):
        # 检查服务器上文件是否存在，如存在返回hash值
        ret, info = self.bucket_manager.stat(self.kpan_bucket, key)
        if not ret:
            return False
        else:
            return ret['hash']


    def upload_all_files(self):
        for fpathe, dirs, fs in os.walk(self.base_dir):
            for f in fs:
                _path = os.path.join(fpathe, f)
                key = os.path.relpath(_path, self.base_dir)
                if os.name == 'nt':
                    key = key.replace('\', '/')
                pan.upload_file(key)
        print('所有文件上次成功...')


if __name__ == '__main__':
    # 测试文件上传
    pan = Kpan('D:/Kpan')
    # pan.get_all_files()
    pan.upload_all_files()