Python 使用cos-python-sdk-v5 上传文件到腾讯云COS对象存储

651次阅读

运行环境

Python：python 3

系统： Windows/Linux

记录日志代码

# -*- coding:utf-8 -*-
__author__ = 'wx'
import logging
import os
import time
import configparser
class Log(object):
    def __init__(self):
        configoper = configparser.ConfigParser()
        configoper.read('config.txt')
        self.logname = configoper['logconfig']['logname']

    def createDir(self):
        ttime = time.strftime('%Y-%m-%d', time.gmtime()) + '-'
        log_name = ttime + self.logname
        log_dir = os.path.join(os.path.dirname(__file__),'Logs')
        logfilename = os.path.join(log_dir,log_name)
        if os.path.isdir(log_dir) == False:
            os.mkdir(log_dir)
        return logfilename

    def createLoger(self,logfilename):
        logger = logging.getLogger()
        logger.setLevel(logging.INFO)
        hander = logging.FileHandler(logfilename)
        hander.setLevel(logging.INFO)
        formatter = logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s")
        hander.setFormatter(formatter)
        logger.addHandler(hander)
        return logger

上传至COS代码

# -*- coding:utf-8 -*-
__author__ = 'wx'

from qcloud_cos import CosConfig
from qcloud_cos import CosS3Client
import configparser
import time
import os
from get_log import *

class CosOper(Log):
    def __init__(self,logger):
        config = configparser.ConfigParser()
        config.read('config.txt')
        self.secret_id = config['common']['secret_id']
        self.secret_key = config['common']['secret_key']
        self.cos_region = config['common']['region']
        self.bucket_name = config['common']['bucket_name']
        self.local_path = config['common']['local_path']
        self.logger = logger
    def cos_client(self):
        config = CosConfig(Secret_id=self.secret_id, Secret_key=self.secret_key, Region=self.cos_region)
        cos_client = CosS3Client(config)
        return cos_client

    def file_name(self):
        for files in os.walk(self.local_path):
            return files[2]
    def upload(self, filedir ,cos_client, partsize=10, maxthread=5):
        dir = time.strftime("%Y-%m-%d")
        self.logger.info('----------------------------------------------------------------------------------------------------')
        self.logger.info('Start Upload')
        for filename in filedir:
            try:
                response = cos_client.upload_file(
                    Bucket = self.bucket_name,
                    LocalFilePath = self.local_path + "/" + filename,
                    Key = "HW/" + str(dir) + "/" + filename,
                    PartSize = partsize,
                    MAXThread = maxthread
                    )
                print(response['ETag'])

            except Exception:
                return None
        self.logger.info('Upload completed.')

def main():
    logoper = Log()
    logname = logoper.createDir()
    logger = logoper.createLoger(logname)

    cosoper = CosOper(logger)
    client = cosoper.cos_client()
    filename =cosoper.file_name()
    cosoper.upload(filename,client)

if __name__ == '__main__':
    main()

config配置文件

[common]
# secret_id
secret_id = AKIDMdjegcmoGYiol*****************

# secret_key
secret_key = d5MRL4Voxyvl*****************

# COS Region
region = ap-chengdu

# COS bucketname
bucket_name = ***ctest-1*****

local_path = D:/test/DBbak

[logconfig]
#日志文件名称
logname = cosupload.log