共计 2592 个字符,预计需要花费 7 分钟才能阅读完成。
运行环境
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
执行效果
1、执行前COS控制台
2、配置需要上传的文件路径,开始上传
3、控制台数据
正文完