103 lines
3.2 KiB
Python
103 lines
3.2 KiB
Python
import logging
|
||
import os
|
||
import uuid
|
||
from datetime import datetime
|
||
from qcloud_cos import CosConfig, CosS3Client
|
||
from app.core.config import settings
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# 腾讯云COS配置
|
||
config = CosConfig(
|
||
Region=settings.COS_REGION,
|
||
SecretId=settings.COS_SECRET_ID,
|
||
SecretKey=settings.COS_SECRET_KEY
|
||
)
|
||
|
||
# 创建客户端
|
||
cos_client = CosS3Client(config)
|
||
|
||
def generate_file_path(directory: str, file_extension: str) -> str:
|
||
"""生成文件路径"""
|
||
today = datetime.now().strftime("%Y%m%d")
|
||
filename = f"{uuid.uuid4().hex}{file_extension}"
|
||
return f"{directory}/{today}/{filename}"
|
||
|
||
async def upload_file(file_content: bytes, file_extension: str, directory: str = "uploads") -> str:
|
||
"""上传文件到腾讯云COS"""
|
||
try:
|
||
# 生成唯一文件路径
|
||
file_path = generate_file_path(directory, file_extension)
|
||
|
||
# 上传到腾讯云COS
|
||
cos_client.put_object(
|
||
Bucket=settings.COS_BUCKET,
|
||
Body=file_content,
|
||
Key=file_path
|
||
)
|
||
|
||
# 返回可访问的URL
|
||
url = f"https://{settings.COS_BUCKET}.cos.{settings.COS_REGION}.myqcloud.com/{file_path}"
|
||
logger.info(f"文件上传成功: {url}")
|
||
return url
|
||
except Exception as e:
|
||
logger.error(f"文件上传失败: {str(e)}")
|
||
raise
|
||
|
||
async def upload_file_from_url(url: str, directory: str = "uploads") -> str:
|
||
"""从URL下载文件并上传到腾讯云COS"""
|
||
try:
|
||
import requests
|
||
|
||
# 下载文件
|
||
response = requests.get(url, timeout=10)
|
||
response.raise_for_status()
|
||
|
||
# 获取文件扩展名
|
||
file_extension = os.path.splitext(url)[1]
|
||
if not file_extension:
|
||
# 如果URL没有扩展名，根据内容类型判断
|
||
content_type = response.headers.get("Content-Type", "")
|
||
if "jpeg" in content_type or "jpg" in content_type:
|
||
file_extension = ".jpg"
|
||
elif "png" in content_type:
|
||
file_extension = ".png"
|
||
elif "gif" in content_type:
|
||
file_extension = ".gif"
|
||
else:
|
||
file_extension = ".bin" # 默认二进制文件
|
||
|
||
# 上传到腾讯云COS
|
||
return await upload_file(response.content, file_extension, directory)
|
||
except Exception as e:
|
||
logger.error(f"从URL上传文件失败: {str(e)}")
|
||
raise
|
||
|
||
async def get_presigned_url(key: str, expires: int = 3600) -> str:
|
||
"""获取预签名URL"""
|
||
try:
|
||
# 生成预签名URL
|
||
url = cos_client.get_presigned_url(
|
||
Method='GET',
|
||
Bucket=settings.COS_BUCKET,
|
||
Key=key,
|
||
Expired=expires
|
||
)
|
||
return url
|
||
except Exception as e:
|
||
logger.error(f"获取预签名URL失败: {str(e)}")
|
||
raise
|
||
|
||
async def delete_file(key: str) -> bool:
|
||
"""删除文件"""
|
||
try:
|
||
# 删除文件
|
||
cos_client.delete_object(
|
||
Bucket=settings.COS_BUCKET,
|
||
Key=key
|
||
)
|
||
logger.info(f"文件删除成功: {key}")
|
||
return True
|
||
except Exception as e:
|
||
logger.error(f"文件删除失败: {str(e)}")
|
||
return False |