腾讯云cos本地和云端同步python工具类

cos同步工具类

# -*- coding=utf-8
from qcloud_cos import CosConfig
from qcloud_cos import CosS3Client
from qcloud_cos import CosServiceError
from qcloud_cos.cos_threadpool import SimpleThreadPool
import os
#import sys
#import logging
import hashlib
from logging_init import logger#logging.basicConfig(level=logging.INFO, stream=sys.stdout)class Cos_helper:secret_id = ""secret_key = ""region = ""token = Noneconfig = Noneclient = Nonebucket = Nonefile_path = Nonecos_path = ''def __init__(self, secret_id, secret_key, region, bucket, domain):# 腾讯云COSV5Python SDK, 目前可以支持Python2.6与Python2.7以及Python3.x# pip安装指南:pip install -U cos-python-sdk-v5# cos最新可用地域,参照https://www.qcloud.com/document/product/436/6224# 设置用户属性, 包括secret_id, secret_key, region# appid已在配置中移除,请在参数Bucket中带上appid。Bucket由bucketname-appid组成self.secret_id = secret_id  # 替换为用户的secret_idself.secret_key = secret_key  # 替换为用户的secret_keyself.region = region  # 替换为用户的regionself.bucket = bucket # 存储桶名self.token = None  # 使用临时密钥需要传入Token，默认为空,可不填self.config = CosConfig(Region=self.region, SecretId=self.secret_id, SecretKey=self.secret_key, Token=self.token)  # 获取配置对象self.client = CosS3Client(self.config)self.cos_path = 'https://'+bucket+'.cos.'+region+'.'+domain+'/'def reset(self, secret_id, secret_key, region, bucket, domain):self.secret_id = secret_id  # 替换为用户的secret_idself.secret_key = secret_key  # 替换为用户的secret_keyself.region = region  # 替换为用户的regionself.bucket = bucket # 存储桶名self.token = None  # 使用临时密钥需要传入Token，默认为空,可不填self.config = CosConfig(Region=self.region, SecretId=self.secret_id, SecretKey=self.secret_key, Token=self.token)  # 获取配置对象self.client = CosS3Client(self.config)self.cos_path = 'https://'+bucket+'.cos.'+region+'.'+domain+'/'def get_cos_files_list(self, marker_key, maxsize, cos_path):logger.debug('get_cos_files_list path: '+ cos_path + '\n')marker = 0maxKeys = 1000if marker_key != None:marker = marker_keyif maxsize != None:maxKeys = maxsizeresponse = self.client.list_objects(Bucket = self.bucket,Prefix = cos_path,Marker = marker,MaxKeys = maxKeys,)#logger.debug(response)if 'Contents' not in response:return Nonelogger.debug(response['Contents']) return response['Contents']def get_local_files_list(self, files_path):logger.debug('get_local_files_list\n')file_name_list = os.listdir(files_path)return file_name_listdef get_local_file_md5(self, file_name):md5_str = Nonefile_object = open(file_name, 'rb')file_content = file_object.read()file_object.close()md5 = hashlib.md5()md5.update(file_content)md5_str = md5.hexdigest()logger.debug('get_local_file_md5 file: %s, md5: %s\n' % (file_name, md5_str))return md5_strdef delete_file_from_cos(self, cos_path):logger.debug('delete_file_from_cos file: %s\n' % cos_path)response = self.client.delete_object(Bucket=self.bucket,Key=cos_path)logger.debug(response)def push_file_to_cos(self, local_path, cos_path):logger.debug('push_file_to_cos file: %s\n' % (cos_path))md5_str = self.get_local_file_md5(local_path)# 高级上传接口(推荐)response = self.client.upload_file(Bucket=self.bucket,LocalFilePath=local_path,Key=cos_path,PartSize=10,MAXThread=10,Metadata={'x-cos-meta-md5' : md5_str      #设置自定义参数，设置为 MD5 校验值})logger.debug(response['ETag'])def pull_file_from_cos(self, local_path, cos_path):logger.debug('pull_file_from_cos file: %s\n' % cos_path)# 文件下载 获取文件到本地response = self.client.download_file(Bucket=self.bucket,Key=cos_path,DestFilePath=local_path,PartSize=10,MAXThread=10)def sync_file_to_cos(self, local_path, cos_path):file_exist = Falsemaxsize = 100total_cos_file_list = Noneif not os.path.exists(local_path):os.makedirs(local_path)local_file_name_list = self.get_local_files_list(local_path)logger.debug(local_file_name_list)# 获取cos上指定路径下的所有文件cos_file_list = self.get_cos_files_list("", maxsize, cos_path)total_cos_file_list = cos_file_listwhile cos_file_list and len(cos_file_list) == maxsize:print("key: " + total_cos_file_list[-1]['Key'])cos_file_list = self.get_cos_files_list(total_cos_file_list[-1]['Key'], maxsize, cos_path)if cos_file_list == None:breaktotal_cos_file_list += cos_file_listif total_cos_file_list != None:logger.debug(total_cos_file_list)# cos没有该文件，直接上传，有则比对MD5是否不一样，再上传for local_file_name in local_file_name_list:file_exist = Falseif total_cos_file_list != None:for cos_file in total_cos_file_list:if (cos_path + local_file_name) == cos_file['Key']:file_exist = Truemd5_str = self.get_local_file_md5(local_path + local_file_name)logger.debug(cos_file['ETag'])logger.debug(md5_str != cos_file['ETag'][1:-1])if md5_str != cos_file['ETag'][1:-1]:logger.debug(local_path + local_file_name)self.push_file_to_cos(local_path + local_file_name, cos_file['Key'])breakif file_exist == False:md5_str = self.get_local_file_md5(local_path + local_file_name)self.push_file_to_cos(local_path + local_file_name, cos_path + local_file_name)# 删除cos上本地没有的文件for i in range(len(local_file_name_list)):local_file_name_list[i] = cos_path + local_file_name_list[i]if total_cos_file_list != None:for cos_file in total_cos_file_list:if cos_file['Key'] == local_path:continueif cos_file['Key'] not in local_file_name_list:self.delete_file_from_cos(cos_file['Key'])def sync_file_from_cos(self, local_path, cos_path):file_exist = Falsemaxsize = 100total_cos_file_list = Nonetotal_cos_file_path_list = []if not os.path.exists(local_path):os.makedirs(local_path)local_file_name_list = self.get_local_files_list(local_path)logger.debug(local_file_name_list)# 获取cos上指定路径下的所有文件cos_file_list = self.get_cos_files_list("", maxsize, cos_path)total_cos_file_list = cos_file_listwhile cos_file_list and len(cos_file_list) == maxsize:print("key: " + total_cos_file_list[-1]['Key'])cos_file_list = self.get_cos_files_list(total_cos_file_list[-1]['Key'], maxsize, cos_path)if cos_file_list == None:breaktotal_cos_file_list += cos_file_listif total_cos_file_list != None:logger.debug(total_cos_file_list)else:logger.error("cos path:%s is not exist\n", cos_path)return# 本地没有该文件，直接下载，有则比对MD5是否不一样，再下载for cos_file in total_cos_file_list:if cos_file['Key'] == local_path:continuetotal_cos_file_path_list.append(cos_file['Key'])file_exist = Falseif local_file_name_list != None and len(local_file_name_list):for local_file_name in local_file_name_list:if (cos_path + local_file_name) == cos_file['Key']:file_exist = Truemd5_str = self.get_local_file_md5(local_path + local_file_name)logger.debug(cos_file['ETag'])logger.debug(md5_str != cos_file['ETag'][1:-1])if md5_str != cos_file['ETag'][1:-1]:logger.debug(local_path + local_file_name)self.pull_file_from_cos(local_path + local_file_name, cos_file['Key'])breakif file_exist == False:self.pull_file_from_cos(local_path + cos_file['Key'][len(cos_path):], cos_file['Key'])# 删除cos上没有的本地文件if local_file_name_list != None and len(local_file_name_list):for local_file in local_file_name_list:if total_cos_file_path_list != None:if cos_path + local_file not in total_cos_file_path_list:os.remove(local_path + local_file)

测试

from cos_helper import Cos_helperCOS_ACCESS_KEY_ID = "key id"
COS_SECRET_ACCESS_KEY ="key"
COS_REGION = "地区"
COS_BASE_DOMAIN = "域名"
COS_BUCKET = "桶名"cos_helper = Cos_helper(COS_ACCESS_KEY_ID, COS_SECRET_ACCESS_KEY, COS_REGION,COS_BUCKET,COS_BASE_DOMAIN)cos_file_list = cos_helper.get_cos_files_list(0, 100, "cos目录")
print(cos_file_list)
if cos_file_list:print('aaaa')

腾讯云cos本地和云端同步python工具类相关推荐

tp6腾讯云、七牛云对象存储的工具类以及异步上传
composer安装 composer require lorine/oss-utils 代码: use Lorine\OssUtils\OssService;$config = ['ak' => ...
一个集成阿里云、腾讯云、七牛云对象存储的工具类
1.安装composer扩展 composer require china-lishuo/oss-utils 2.这是图片上传到云存储+cdn /*** @param Request $request ...
mysql备份至cos_宝塔面板网站文件/数据库定时同步备份至腾讯云COS设置
本来老蒋这篇文章是要分享张戈同学关于利用腾讯云COS备份网站和数据库脚本工具的整理的,但是翻看之前的博文发现我们能用到的面板和工具包大部分都自带第三方云存储接口快速备份的.所以这篇文章延期到后面再去分 ...
Java springboot项目引入腾讯云COS实现上传
Java springboot项目引入腾讯云COS实现上传 pom.xml 配置类CosConfig.java 上传工具类CosClientUtil.java pom.xml <!--腾讯云上传 ...
阿里巴巴 OSS与AWS（亚马逊） S3 和腾讯云cos 存储服务介绍篇
前言对象存储服务,简单来说,可以把它当成一个"网盘",可以上传下载数据,也可以直接在这个"网盘"中对文件进行某些操作. 1.定时或者基于某种条件自动地,每天从 ...
腾讯云cos html,腾讯云对象存储（COS）插件安装设置图文教程
腾讯云对象存储(COS)插件是腾讯云官方开发的针对在 wordpress中使用腾讯云对象存储cos的用户开发的辅助插件.通过在 wordpress程序中安装插件的方式,很方便的把WordPress静态 ...
华为云实战之对象存储的使用以及与腾讯云COS对比
一.OBS是什么 OBS即对象存储服务(Object Storage Service),是一个基于对象的海量存储服务,为客户提供海量.安全.高可靠.低成本的数据存储能力,包括:创建.修改.删除桶,上传 ...
（Ⅲ）使用七牛云作为图床获取外链方式总结（已更换为使用PicGO+腾讯云COS）
1. 图床的选择 (1) 什么是图床? 很多技术人写作都有在用 Markdown 轻量级标记语言进行博客写作,这种写作让我们不用像使用 Word 那么麻烦调整排版和格式,而只需专心写作照样完成排版的一 ...
（Ⅲ）使用七牛云作为图床获取外链方式总结（已更换为使用PicGO+腾讯云COS）...
1. 图床的选择 (1) 什么是图床? 很多技术人写作都有在用 Markdown 轻量级标记语言进行博客写作,这种写作让我们不用像使用 Word 那么麻烦调整排版和格式,而只需专心写作照样完成排版的一 ...

腾讯云cos本地和云端同步python工具类

cos同步工具类

测试

腾讯云cos本地和云端同步python工具类相关推荐

最新文章

热门文章