第一次提交：小程序前端最新版本
This commit is contained in:
XingQue
512
utils/oss_utils.py
Normal file
512
utils/oss_utils.py
Normal file
@@ -0,0 +1,512 @@
|
||||
# utils/oss_utils.py
|
||||
import os
|
||||
import uuid
|
||||
from django.conf import settings
|
||||
from PIL import Image, UnidentifiedImageError
|
||||
import io
|
||||
|
||||
'''try:
|
||||
from qcloud_cos import CosConfig, CosS3Client
|
||||
|
||||
TENCENT_COS_AVAILABLE = True
|
||||
except ImportError:
|
||||
TENCENT_COS_AVAILABLE = False
|
||||
|
||||
try:
|
||||
import oss2
|
||||
|
||||
ALIYUN_OSS_AVAILABLE = True
|
||||
except ImportError:
|
||||
ALIYUN_OSS_AVAILABLE = False
|
||||
|
||||
|
||||
def validate_image(file):
|
||||
"""
|
||||
验证图片文件
|
||||
使用Pillow检查图片是否有效
|
||||
"""
|
||||
try:
|
||||
# 打开图片但不保存到磁盘
|
||||
img = Image.open(file)
|
||||
img.verify() # 验证文件完整性
|
||||
|
||||
# 重置文件指针（因为PIL读取后会移动指针）
|
||||
file.seek(0)
|
||||
|
||||
# 检查图片格式
|
||||
allowed_formats = ['JPEG', 'PNG', 'GIF', 'WEBP', 'BMP']
|
||||
if img.format not in allowed_formats:
|
||||
return False, f"不支持{img.format}格式，请使用JPEG、PNG、GIF、WEBP或BMP格式"
|
||||
|
||||
# 检查图片大小（最大5MB）
|
||||
file_size = len(file.read())
|
||||
file.seek(0)
|
||||
if file_size > 5 * 1024 * 1024:
|
||||
return False, "图片大小不能超过5MB"
|
||||
|
||||
# 检查图片尺寸（最大2000x2000）
|
||||
img = Image.open(file)
|
||||
file.seek(0)
|
||||
if img.width > 3000 or img.height > 3000:
|
||||
return False, "图片尺寸不能超过2000x2000像素"
|
||||
|
||||
return True, "图片验证通过"
|
||||
|
||||
except UnidentifiedImageError:
|
||||
return False, "文件不是有效的图片"
|
||||
except Exception as e:
|
||||
return False, f"图片验证失败: {str(e)}"
|
||||
|
||||
|
||||
def get_oss_client():
|
||||
"""
|
||||
获取OSS客户端
|
||||
根据配置自动选择云厂商
|
||||
"""
|
||||
oss_type = getattr(settings, 'OSS_TYPE', 'tencent') # 默认为腾讯云
|
||||
|
||||
if oss_type == 'tencent' and TENCENT_COS_AVAILABLE:
|
||||
# 腾讯云COS
|
||||
config = CosConfig(
|
||||
Region=getattr(settings, 'COS_REGION', 'ap-shanghai'),
|
||||
SecretId=getattr(settings, 'COS_SECRET_ID', ''),
|
||||
SecretKey=getattr(settings, 'COS_SECRET_KEY', ''),
|
||||
Scheme=getattr(settings, 'COS_SCHEME', 'https')
|
||||
)
|
||||
return CosS3Client(config)
|
||||
|
||||
elif oss_type == 'aliyun' and ALIYUN_OSS_AVAILABLE:
|
||||
# 阿里云OSS
|
||||
auth = oss2.Auth(
|
||||
getattr(settings, 'ALIYUN_ACCESS_KEY_ID', ''),
|
||||
getattr(settings, 'ALIYUN_ACCESS_KEY_SECRET', '')
|
||||
)
|
||||
endpoint = getattr(settings, 'ALIYUN_OSS_ENDPOINT', '')
|
||||
bucket_name = getattr(settings, 'ALIYUN_OSS_BUCKET', '')
|
||||
return oss2.Bucket(auth, endpoint, bucket_name)
|
||||
|
||||
else:
|
||||
raise ImportError(f"不支持{oss_type}云存储或相关SDK未安装")
|
||||
|
||||
|
||||
def upload_to_oss(file_obj, file_path):
|
||||
"""
|
||||
通用上传文件到OSS
|
||||
file_obj: Django的UploadedFile对象或类文件对象
|
||||
file_path: 在OSS中的相对路径，如 'avatar/123/abc.jpg'
|
||||
返回: 文件访问URL
|
||||
"""
|
||||
try:
|
||||
oss_type = getattr(settings, 'OSS_TYPE', 'tencent')
|
||||
|
||||
if oss_type == 'tencent':
|
||||
return _upload_to_tencent_cos(file_obj, file_path)
|
||||
elif oss_type == 'aliyun':
|
||||
return _upload_to_aliyun_oss(file_obj, file_path)
|
||||
else:
|
||||
raise ValueError(f"不支持的OSS类型: {oss_type}")
|
||||
|
||||
except Exception as e:
|
||||
print(f"OSS上传失败: {e}")
|
||||
return None
|
||||
|
||||
|
||||
def _upload_to_tencent_cos(file_obj, file_path):
|
||||
"""
|
||||
上传到腾讯云COS
|
||||
"""
|
||||
client = get_oss_client()
|
||||
|
||||
# 获取文件扩展名
|
||||
ext = file_path.split('.')[-1].lower() if '.' in file_path else ''
|
||||
content_type_map = {
|
||||
'jpg': 'image/jpeg', 'jpeg': 'image/jpeg',
|
||||
'png': 'image/png', 'gif': 'image/gif',
|
||||
'webp': 'image/webp', 'bmp': 'image/bmp'
|
||||
}
|
||||
content_type = content_type_map.get(ext, 'image/jpeg')
|
||||
|
||||
# 如果是Django的UploadedFile，需要读取内容
|
||||
if hasattr(file_obj, 'read'):
|
||||
file_content = file_obj.read()
|
||||
else:
|
||||
file_content = file_obj
|
||||
|
||||
# 上传文件
|
||||
response = client.put_object(
|
||||
Bucket=getattr(settings, 'COS_BUCKET', ''),
|
||||
Body=file_content,
|
||||
Key=file_path,
|
||||
ContentType=content_type,
|
||||
ContentDisposition='inline'
|
||||
)
|
||||
|
||||
# 返回完整URL
|
||||
domain = getattr(settings, 'COS_DOMAIN', '')
|
||||
return f"{domain}/{file_path}"
|
||||
|
||||
|
||||
def _upload_to_aliyun_oss(file_obj, file_path):
|
||||
"""
|
||||
上传到阿里云OSS
|
||||
"""
|
||||
bucket = get_oss_client()
|
||||
|
||||
# 如果是Django的UploadedFile，需要读取内容
|
||||
if hasattr(file_obj, 'read'):
|
||||
file_content = file_obj.read()
|
||||
else:
|
||||
file_content = file_obj
|
||||
|
||||
# 上传文件
|
||||
result = bucket.put_object(file_path, file_content)
|
||||
|
||||
if result.status == 200:
|
||||
# 阿里云OSS返回的URL
|
||||
domain = getattr(settings, 'ALIYUN_OSS_DOMAIN', '')
|
||||
return f"{domain}/{file_path}"
|
||||
else:
|
||||
return None
|
||||
|
||||
|
||||
def delete_from_oss(file_url):
|
||||
"""
|
||||
从OSS删除文件
|
||||
file_url: 可以是完整URL或相对路径
|
||||
"""
|
||||
try:
|
||||
if not file_url:
|
||||
return False
|
||||
|
||||
oss_type = getattr(settings, 'OSS_TYPE', 'tencent')
|
||||
|
||||
# 从URL中提取文件路径
|
||||
if oss_type == 'tencent':
|
||||
domain = getattr(settings, 'COS_DOMAIN', '')
|
||||
|
||||
# 判断是完整URL还是相对路径
|
||||
if file_url.startswith('http'):
|
||||
# 完整URL
|
||||
if domain not in file_url:
|
||||
return False # URL不属于我们的域名
|
||||
file_key = file_url.replace(f"{domain}/", "")
|
||||
else:
|
||||
# 相对路径，直接使用
|
||||
file_key = file_url
|
||||
|
||||
return _delete_from_tencent_cos(file_key)
|
||||
|
||||
elif oss_type == 'aliyun':
|
||||
domain = getattr(settings, 'ALIYUN_OSS_DOMAIN', '')
|
||||
|
||||
if file_url.startswith('http'):
|
||||
if domain not in file_url:
|
||||
return False
|
||||
file_key = file_url.replace(f"{domain}/", "")
|
||||
else:
|
||||
file_key = file_url
|
||||
|
||||
return _delete_from_aliyun_oss(file_key)
|
||||
|
||||
else:
|
||||
return False
|
||||
|
||||
except Exception as e:
|
||||
print(f"OSS删除失败: {e}")
|
||||
return False
|
||||
|
||||
|
||||
def _delete_from_tencent_cos(file_key):
|
||||
"""
|
||||
从腾讯云COS删除文件
|
||||
"""
|
||||
client = get_oss_client()
|
||||
|
||||
try:
|
||||
client.delete_object(
|
||||
Bucket=getattr(settings, 'COS_BUCKET', ''),
|
||||
Key=file_key
|
||||
)
|
||||
return True
|
||||
except Exception as e:
|
||||
print(f"腾讯云COS删除失败: {e}")
|
||||
return False
|
||||
|
||||
|
||||
def _delete_from_aliyun_oss(file_key):
|
||||
"""
|
||||
从阿里云OSS删除文件
|
||||
"""
|
||||
bucket = get_oss_client()
|
||||
|
||||
try:
|
||||
bucket.delete_object(file_key)
|
||||
return True
|
||||
except Exception as e:
|
||||
print(f"阿里云OSS删除失败: {e}")
|
||||
return False'''
|
||||
|
||||
|
||||
|
||||
|
||||
# utils/oss_utils.py
|
||||
import os
|
||||
import uuid
|
||||
from django.conf import settings
|
||||
from PIL import Image, UnidentifiedImageError
|
||||
import io
|
||||
|
||||
try:
|
||||
from qcloud_cos import CosConfig, CosS3Client
|
||||
|
||||
TENCENT_COS_AVAILABLE = True
|
||||
except ImportError:
|
||||
TENCENT_COS_AVAILABLE = False
|
||||
|
||||
try:
|
||||
import oss2
|
||||
|
||||
ALIYUN_OSS_AVAILABLE = True
|
||||
except ImportError:
|
||||
ALIYUN_OSS_AVAILABLE = False
|
||||
|
||||
# ========== 通用 MIME 类型映射（扩展支持非图片文件） ==========
|
||||
MIME_TYPES = {
|
||||
# 图片
|
||||
'jpg': 'image/jpeg', 'jpeg': 'image/jpeg', 'png': 'image/png',
|
||||
'gif': 'image/gif', 'webp': 'image/webp', 'bmp': 'image/bmp',
|
||||
# JSON / 文本
|
||||
'json': 'application/json', 'txt': 'text/plain',
|
||||
# 音频（小程序录音常见格式）
|
||||
'mp3': 'audio/mpeg', 'aac': 'audio/aac', 'wav': 'audio/wav',
|
||||
'm4a': 'audio/mp4', 'ogg': 'audio/ogg',
|
||||
# 视频
|
||||
'mp4': 'video/mp4', 'webm': 'video/webm',
|
||||
# 二进制
|
||||
'bin': 'application/octet-stream',
|
||||
}
|
||||
|
||||
# ====================================================================
|
||||
|
||||
def validate_image(file):
|
||||
"""
|
||||
验证图片文件
|
||||
使用Pillow检查图片是否有效
|
||||
"""
|
||||
try:
|
||||
# 打开图片但不保存到磁盘
|
||||
img = Image.open(file)
|
||||
img.verify() # 验证文件完整性
|
||||
|
||||
# 重置文件指针（因为PIL读取后会移动指针）
|
||||
file.seek(0)
|
||||
|
||||
# 检查图片格式
|
||||
allowed_formats = ['JPEG', 'PNG', 'GIF', 'WEBP', 'BMP']
|
||||
if img.format not in allowed_formats:
|
||||
return False, f"不支持{img.format}格式，请使用JPEG、PNG、GIF、WEBP或BMP格式"
|
||||
|
||||
# 检查图片大小（最大5MB）
|
||||
file_size = len(file.read())
|
||||
file.seek(0)
|
||||
if file_size > 5 * 1024 * 1024:
|
||||
return False, "图片大小不能超过5MB"
|
||||
|
||||
# 检查图片尺寸（最大2000x2000）
|
||||
img = Image.open(file)
|
||||
file.seek(0)
|
||||
if img.width > 3000 or img.height > 3000:
|
||||
return False, "图片尺寸不能超过2000x2000像素"
|
||||
|
||||
return True, "图片验证通过"
|
||||
|
||||
except UnidentifiedImageError:
|
||||
return False, "文件不是有效的图片"
|
||||
except Exception as e:
|
||||
return False, f"图片验证失败: {str(e)}"
|
||||
|
||||
|
||||
def get_oss_client():
|
||||
"""
|
||||
获取OSS客户端
|
||||
根据配置自动选择云厂商
|
||||
"""
|
||||
oss_type = getattr(settings, 'OSS_TYPE', 'tencent') # 默认为腾讯云
|
||||
|
||||
if oss_type == 'tencent' and TENCENT_COS_AVAILABLE:
|
||||
# 腾讯云COS
|
||||
config = CosConfig(
|
||||
Region=getattr(settings, 'COS_REGION', 'ap-shanghai'),
|
||||
SecretId=getattr(settings, 'COS_SECRET_ID', ''),
|
||||
SecretKey=getattr(settings, 'COS_SECRET_KEY', ''),
|
||||
Scheme=getattr(settings, 'COS_SCHEME', 'https')
|
||||
)
|
||||
return CosS3Client(config)
|
||||
|
||||
elif oss_type == 'aliyun' and ALIYUN_OSS_AVAILABLE:
|
||||
# 阿里云OSS
|
||||
auth = oss2.Auth(
|
||||
getattr(settings, 'ALIYUN_ACCESS_KEY_ID', ''),
|
||||
getattr(settings, 'ALIYUN_ACCESS_KEY_SECRET', '')
|
||||
)
|
||||
endpoint = getattr(settings, 'ALIYUN_OSS_ENDPOINT', '')
|
||||
bucket_name = getattr(settings, 'ALIYUN_OSS_BUCKET', '')
|
||||
return oss2.Bucket(auth, endpoint, bucket_name)
|
||||
|
||||
else:
|
||||
raise ImportError(f"不支持{oss_type}云存储或相关SDK未安装")
|
||||
|
||||
|
||||
def upload_to_oss(file_obj, file_path, content_type=None):
|
||||
"""
|
||||
通用上传文件到OSS
|
||||
file_obj: Django的UploadedFile对象或类文件对象
|
||||
file_path: 在OSS中的相对路径，如 'avatar/123/abc.jpg'
|
||||
content_type: 可选，MIME类型；若不传则根据文件扩展名自动推断
|
||||
返回: 文件访问URL
|
||||
"""
|
||||
try:
|
||||
oss_type = getattr(settings, 'OSS_TYPE', 'tencent')
|
||||
|
||||
if oss_type == 'tencent':
|
||||
return _upload_to_tencent_cos(file_obj, file_path, content_type)
|
||||
elif oss_type == 'aliyun':
|
||||
return _upload_to_aliyun_oss(file_obj, file_path)
|
||||
else:
|
||||
raise ValueError(f"不支持的OSS类型: {oss_type}")
|
||||
|
||||
except Exception as e:
|
||||
print(f"OSS上传失败: {e}")
|
||||
return None
|
||||
|
||||
|
||||
def _upload_to_tencent_cos(file_obj, file_path, content_type=None):
|
||||
"""
|
||||
上传到腾讯云COS（扩展支持任意文件类型）
|
||||
"""
|
||||
client = get_oss_client()
|
||||
|
||||
# 若调用方未指定 Content-Type，则根据扩展名自动推断
|
||||
if not content_type:
|
||||
ext = file_path.rsplit('.', 1)[-1].lower() if '.' in file_path else ''
|
||||
content_type = MIME_TYPES.get(ext, 'application/octet-stream')
|
||||
|
||||
# 读取文件内容（兼容 Django UploadedFile 和普通 bytes）
|
||||
if hasattr(file_obj, 'read'):
|
||||
file_content = file_obj.read()
|
||||
else:
|
||||
file_content = file_obj
|
||||
|
||||
# 上传文件
|
||||
response = client.put_object(
|
||||
Bucket=getattr(settings, 'COS_BUCKET', ''),
|
||||
Body=file_content,
|
||||
Key=file_path,
|
||||
ContentType=content_type,
|
||||
ContentDisposition='inline'
|
||||
)
|
||||
|
||||
# 返回完整URL
|
||||
domain = getattr(settings, 'COS_DOMAIN', '')
|
||||
return f"{domain.rstrip('/')}/{file_path.lstrip('/')}"
|
||||
|
||||
|
||||
def _upload_to_aliyun_oss(file_obj, file_path):
|
||||
"""
|
||||
上传到阿里云OSS（保持原样，未修改）
|
||||
"""
|
||||
bucket = get_oss_client()
|
||||
|
||||
# 如果是Django的UploadedFile，需要读取内容
|
||||
if hasattr(file_obj, 'read'):
|
||||
file_content = file_obj.read()
|
||||
else:
|
||||
file_content = file_obj
|
||||
|
||||
# 上传文件
|
||||
result = bucket.put_object(file_path, file_content)
|
||||
|
||||
if result.status == 200:
|
||||
# 阿里云OSS返回的URL
|
||||
domain = getattr(settings, 'ALIYUN_OSS_DOMAIN', '')
|
||||
return f"{domain}/{file_path}"
|
||||
else:
|
||||
return None
|
||||
|
||||
|
||||
def delete_from_oss(file_url):
|
||||
"""
|
||||
从OSS删除文件
|
||||
file_url: 可以是完整URL或相对路径
|
||||
"""
|
||||
try:
|
||||
if not file_url:
|
||||
return False
|
||||
|
||||
oss_type = getattr(settings, 'OSS_TYPE', 'tencent')
|
||||
|
||||
# 从URL中提取文件路径
|
||||
if oss_type == 'tencent':
|
||||
domain = getattr(settings, 'COS_DOMAIN', '')
|
||||
|
||||
# 判断是完整URL还是相对路径
|
||||
if file_url.startswith('http'):
|
||||
# 完整URL
|
||||
if domain not in file_url:
|
||||
return False # URL不属于我们的域名
|
||||
file_key = file_url.replace(f"{domain}/", "")
|
||||
else:
|
||||
# 相对路径，直接使用
|
||||
file_key = file_url
|
||||
|
||||
return _delete_from_tencent_cos(file_key)
|
||||
|
||||
elif oss_type == 'aliyun':
|
||||
domain = getattr(settings, 'ALIYUN_OSS_DOMAIN', '')
|
||||
|
||||
if file_url.startswith('http'):
|
||||
if domain not in file_url:
|
||||
return False
|
||||
file_key = file_url.replace(f"{domain}/", "")
|
||||
else:
|
||||
file_key = file_url
|
||||
|
||||
return _delete_from_aliyun_oss(file_key)
|
||||
|
||||
else:
|
||||
return False
|
||||
|
||||
except Exception as e:
|
||||
print(f"OSS删除失败: {e}")
|
||||
return False
|
||||
|
||||
|
||||
def _delete_from_tencent_cos(file_key):
|
||||
"""
|
||||
从腾讯云COS删除文件
|
||||
"""
|
||||
client = get_oss_client()
|
||||
|
||||
try:
|
||||
client.delete_object(
|
||||
Bucket=getattr(settings, 'COS_BUCKET', ''),
|
||||
Key=file_key
|
||||
)
|
||||
return True
|
||||
except Exception as e:
|
||||
print(f"腾讯云COS删除失败: {e}")
|
||||
return False
|
||||
|
||||
|
||||
def _delete_from_aliyun_oss(file_key):
|
||||
"""
|
||||
从阿里云OSS删除文件
|
||||
"""
|
||||
bucket = get_oss_client()
|
||||
|
||||
try:
|
||||
bucket.delete_object(file_key)
|
||||
return True
|
||||
except Exception as e:
|
||||
print(f"阿里云OSS删除失败: {e}")
|
||||
return False
|
||||
Reference in New Issue
Block a user