152 lines
5.5 KiB
Python
152 lines
5.5 KiB
Python
# -*- coding=utf-8
|
||
from qcloud_cos import CosConfig
|
||
from qcloud_cos import CosS3Client
|
||
from qcloud_cos.cos_exception import CosServiceError
|
||
import sys
|
||
import os
|
||
import logging
|
||
|
||
# 正常情况日志级别使用 INFO，需要定位时可以修改为 DEBUG，此时 SDK 会打印和服务端的通信信息
|
||
# logging.basicConfig(level=logging.INFO, stream=sys.stdout)
|
||
|
||
g_cos_root = 'https://xinglu-1324629296.cos.ap-beijing.myqcloud.com'
|
||
g_cos_bucket = 'xinglu-1324629296'
|
||
|
||
def get_client():
|
||
# 1. 设置用户属性, 包括 secret_id, secret_key, region等。Appid 已在 CosConfig 中移除，请在参数 Bucket 中带上 Appid。Bucket 由 BucketName-Appid 组成
|
||
# secret_id = os.environ['COS_SECRET_ID'] # 用户的 SecretId，建议使用子账号密钥，授权遵循最小权限指引，降低使用风险。子账号密钥获取可参见 https://cloud.tencent.com/document/product/598/37140
|
||
# secret_key = os.environ['COS_SECRET_KEY'] # 用户的 SecretKey，建议使用子账号密钥，授权遵循最小权限指引，降低使用风险。子账号密钥获取可参见 https://cloud.tencent.com/document/product/598/37140
|
||
"""
|
||
SecretId:AKIDFLFvOfjbFi171mLnEXp370rHZozymN2c
|
||
SecretKey:a6RYUDl6EAtQmurb8To4AFRCoDOM6g3D
|
||
"""
|
||
secret_id = 'AKIDFLFvOfjbFi171mLnEXp370rHZozymN2c'
|
||
secret_key = 'a6RYUDl6EAtQmurb8To4AFRCoDOM6g3D'
|
||
region = 'ap-beijing' # 替换为用户的 region，已创建桶归属的 region 可以在控制台查看，https://console.cloud.tencent.com/cos5/bucket
|
||
# COS 支持的所有 region 列表参见https://cloud.tencent.com/document/product/436/6224
|
||
token = None # 如果使用永久密钥不需要填入 token，如果使用临时密钥需要填入，临时密钥生成和使用指引参见 https://cloud.tencent.com/document/product/436/14048
|
||
scheme = 'https' # 指定使用 http/https 协议来访问 COS，默认为 https，可不填
|
||
|
||
config = CosConfig(Region=region, SecretId=secret_id, SecretKey=secret_key, Token=token, Scheme=scheme)
|
||
client = CosS3Client(config)
|
||
return client
|
||
|
||
|
||
def upload_file_to_cos(local_filename: str, cos_path: str):
|
||
client = get_client()
|
||
# 本地路径 简单上传
|
||
response = client.put_object_from_local_file(
|
||
Bucket=g_cos_bucket,
|
||
LocalFilePath=local_filename,
|
||
Key=cos_path
|
||
)
|
||
print(response['ETag'])
|
||
return response
|
||
|
||
|
||
def get_file_from_cos(cos_path: str, local_filename: str):
|
||
"""
|
||
从cos上下载文件
|
||
:param cos_path:
|
||
:param local_filename:
|
||
:return:
|
||
"""
|
||
client = get_client()
|
||
response = client.get_object(
|
||
Bucket=g_cos_bucket,
|
||
Key=cos_path
|
||
)
|
||
response['Body'].get_stream_to_file(local_filename)
|
||
|
||
|
||
|
||
def upload_order_images_to_cos(idx_to_localfile: dict, orderid: str):
|
||
"""
|
||
将一组本地图片，上传到对应的cos的订单目录下
|
||
:param idx_to_localfile:
|
||
:param orderid:
|
||
:return: idx_to_cosfile
|
||
"""
|
||
xlsyn_cos_host = 'https://xinglu-1324629296.cos.ap-beijing.myqcloud.com'
|
||
cos_order_path = '/user/wave_survey/%s' % orderid
|
||
idx_to_cosfile = {}
|
||
for idx, local_filepath in idx_to_localfile.items():
|
||
image_filename = local_filepath.split('/')[-1]
|
||
cos_path = '%s/%s' % (cos_order_path, image_filename)
|
||
upload_file_to_cos(local_filepath, cos_path)
|
||
cos_file = '%s/%s' % (xlsyn_cos_host, cos_path)
|
||
idx_to_cosfile[idx] = cos_file
|
||
return idx_to_cosfile
|
||
|
||
|
||
class CosFolderManager:
|
||
def __init__(self, client, bucket):
|
||
self.client = client
|
||
self.bucket = bucket
|
||
|
||
def _normalize_folder(self, folder_path):
|
||
"""确保文件夹路径以 / 结尾"""
|
||
if not folder_path:
|
||
return ''
|
||
# 去除开头的 /
|
||
if folder_path.startswith('/'):
|
||
folder_path = folder_path[1:]
|
||
# 确保结尾有 /
|
||
if not folder_path.endswith('/'):
|
||
folder_path += '/'
|
||
return folder_path
|
||
|
||
def folder_exists(self, folder_path):
|
||
"""
|
||
判断文件夹是否存在
|
||
:param folder_path: 文件夹路径，如 'data/logs/' 或 'data/logs'
|
||
:return: True/False
|
||
"""
|
||
key = self._normalize_folder(folder_path)
|
||
if not key:
|
||
return True # 根目录默认存在
|
||
|
||
try:
|
||
self.client.head_object(Bucket=self.bucket, Key=key)
|
||
return True
|
||
except CosServiceError as e:
|
||
if e.get_status_code() == 404:
|
||
return False
|
||
raise # 其他错误向上抛出
|
||
|
||
def create_folder(self, folder_path):
|
||
"""
|
||
创建文件夹（上传一个以 / 结尾的空对象）
|
||
:param folder_path: 文件夹路径
|
||
:return: None
|
||
"""
|
||
key = self._normalize_folder(folder_path)
|
||
if not key:
|
||
return # 根目录无需创建
|
||
|
||
self.client.put_object(
|
||
Bucket=self.bucket,
|
||
Key=key,
|
||
Body=b'' # 空内容
|
||
)
|
||
|
||
def ensure_folder(self, folder_path):
|
||
"""
|
||
确保文件夹存在，不存在则创建
|
||
:param folder_path: 文件夹路径
|
||
:return: True(已存在) / False(刚创建)
|
||
"""
|
||
if self.folder_exists(folder_path):
|
||
return True
|
||
else:
|
||
self.create_folder(folder_path)
|
||
return False
|
||
|
||
|
||
if __name__ == '__main__':
|
||
idx_to_localfile = {0:'D:/slgwork/slgcode/wave_survey/1739430848000.jpg',
|
||
2:'D:/slgwork/slgcode/wave_survey/1739430851000.jpg'}
|
||
idx_to_cosfile = upload_order_images_to_cos(idx_to_localfile, 'slg_orderid')
|
||
print(idx_to_cosfile)
|
||
|