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

2018年8月13日11:49:17 发表评论 2,307 views

本文最后更新于2018年8月13日，已超过 1 年没有更新，如果文章内容或图片资源失效，请留言反馈，我们会及时处理，谢谢！

运行环境

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

# -*- 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()

# -*- 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