383 lines
11 KiB
Python
383 lines
11 KiB
Python
# 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 _cos_public_domain() -> str:
|
||
"""COS 公网域名（无尾斜杠）。"""
|
||
return (getattr(settings, 'COS_DOMAIN', '') or '').rstrip('/')
|
||
|
||
|
||
def to_full_cos_url(url_or_key) -> str:
|
||
"""
|
||
相对 COS Key / 协议相对 URL → 完整 https URL。
|
||
已是 http(s) 则原样返回。用于交付图、退款证据图入库，避免后台当完整 URL 用时裂图。
|
||
"""
|
||
s = str(url_or_key or '').strip()
|
||
if not s:
|
||
return ''
|
||
if s.startswith('//'):
|
||
return f'https:{s}'
|
||
if s.startswith('http://') or s.startswith('https://'):
|
||
return s
|
||
domain = _cos_public_domain()
|
||
key = s.lstrip('/')
|
||
if not domain:
|
||
return key
|
||
return f'{domain}/{key}'
|
||
|
||
|
||
def cos_url_variants(url_or_key) -> list:
|
||
"""同一张图在库里可能存相对或绝对，删除匹配时两边都试。"""
|
||
s = str(url_or_key or '').strip()
|
||
if not s:
|
||
return []
|
||
full = to_full_cos_url(s)
|
||
key = _url_to_oss_key(s) or s.lstrip('/')
|
||
out = []
|
||
for x in (s, full, key, f'/{key}' if key else ''):
|
||
if x and x not in out:
|
||
out.append(x)
|
||
return out
|
||
|
||
|
||
def _url_to_oss_key(file_url):
|
||
"""从完整 URL 或相对路径提取 OSS 对象 Key"""
|
||
if not file_url:
|
||
return None
|
||
file_url = str(file_url).strip()
|
||
if not file_url:
|
||
return None
|
||
|
||
oss_type = getattr(settings, 'OSS_TYPE', 'tencent')
|
||
if oss_type == 'tencent':
|
||
domain = getattr(settings, 'COS_DOMAIN', '')
|
||
elif oss_type == 'aliyun':
|
||
domain = getattr(settings, 'ALIYUN_OSS_DOMAIN', '')
|
||
else:
|
||
return None
|
||
|
||
if file_url.startswith('http'):
|
||
if domain and domain not in file_url:
|
||
return None
|
||
prefix = f"{domain.rstrip('/')}/"
|
||
if file_url.startswith(prefix):
|
||
return file_url[len(prefix):]
|
||
return file_url.split('://', 1)[-1].split('/', 1)[-1]
|
||
|
||
return file_url.lstrip('/')
|
||
|
||
|
||
def copy_in_oss(src_key_or_url, dest_key):
|
||
"""
|
||
在 OSS 内复制对象（服务端复制，不经过本地）
|
||
src_key_or_url: 源对象 Key 或完整 URL
|
||
dest_key: 目标对象 Key（相对路径）
|
||
返回: 成功 True，失败 False
|
||
"""
|
||
try:
|
||
src_key = _url_to_oss_key(src_key_or_url)
|
||
if not src_key or not dest_key:
|
||
return False
|
||
|
||
dest_key = dest_key.lstrip('/')
|
||
oss_type = getattr(settings, 'OSS_TYPE', 'tencent')
|
||
|
||
if oss_type == 'tencent':
|
||
return _copy_in_tencent_cos(src_key, dest_key)
|
||
elif oss_type == 'aliyun':
|
||
return _copy_in_aliyun_oss(src_key, dest_key)
|
||
return False
|
||
except Exception as e:
|
||
print(f"OSS复制失败: {e}")
|
||
return False
|
||
|
||
|
||
def _copy_in_tencent_cos(src_key, dest_key):
|
||
client = get_oss_client()
|
||
bucket = getattr(settings, 'COS_BUCKET', '')
|
||
region = getattr(settings, 'COS_REGION', 'ap-shanghai')
|
||
try:
|
||
client.copy_object(
|
||
Bucket=bucket,
|
||
Key=dest_key,
|
||
CopySource={
|
||
'Bucket': bucket,
|
||
'Region': region,
|
||
'Key': src_key,
|
||
},
|
||
)
|
||
return True
|
||
except Exception as e:
|
||
print(f"腾讯云COS复制失败: {e}")
|
||
return False
|
||
|
||
|
||
def _copy_in_aliyun_oss(src_key, dest_key):
|
||
bucket = get_oss_client()
|
||
try:
|
||
result = bucket.copy_object(src_key, dest_key)
|
||
return result.status == 200
|
||
except Exception as e:
|
||
print(f"阿里云OSS复制失败: {e}")
|
||
return False
|
||
|
||
|
||
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
|