69 lines
1.9 KiB
Python
69 lines
1.9 KiB
Python
from qcloud_cos import CosConfig, CosS3Client
|
||
from app.core.config import settings
|
||
import sys
|
||
import logging
|
||
|
||
# 正常情况日志级别使用INFO，需要定位时可以修改为DEBUG，此时SDK会打印和服务端的通信信息
|
||
logging.basicConfig(level=logging.INFO, stream=sys.stdout)
|
||
|
||
class COSManager:
|
||
def __init__(self):
|
||
config = CosConfig(
|
||
Region=settings.COS_REGION,
|
||
SecretId=settings.TENCENT_SECRET_ID,
|
||
SecretKey=settings.TENCENT_SECRET_KEY
|
||
)
|
||
self.client = CosS3Client(config)
|
||
|
||
def get_upload_url(self, key: str, expires: int = 3600) -> str:
|
||
"""
|
||
获取预签名上传URL
|
||
|
||
Args:
|
||
key: 对象键
|
||
expires: 签名有效期（秒）
|
||
|
||
Returns:
|
||
预签名URL
|
||
"""
|
||
try:
|
||
url = self.client.get_presigned_url(
|
||
Method='PUT',
|
||
Bucket=settings.COS_BUCKET,
|
||
Key=key,
|
||
Expired=expires
|
||
)
|
||
return url
|
||
except Exception as e:
|
||
logging.error(f"获取上传URL失败: {str(e)}")
|
||
raise
|
||
|
||
def get_download_url(self, key: str, expires: int = 3600) -> str:
|
||
"""
|
||
获取预签名下载URL
|
||
|
||
Args:
|
||
key: 对象键
|
||
expires: 签名有效期（秒）
|
||
|
||
Returns:
|
||
预签名URL
|
||
"""
|
||
try:
|
||
url = self.client.get_presigned_url(
|
||
Method='GET',
|
||
Bucket=settings.COS_BUCKET,
|
||
Key=key,
|
||
Expired=expires
|
||
)
|
||
return url
|
||
except Exception as e:
|
||
logging.error(f"获取下载URL失败: {str(e)}")
|
||
raise
|
||
|
||
def put_object(self, **kwargs):
|
||
"""上传对象的封装方法"""
|
||
return self.client.put_object(**kwargs)
|
||
|
||
# 创建全局 COS 客户端实例
|
||
cos_manager = COSManager() |