127 lines
3.9 KiB
Python
127 lines
3.9 KiB
Python
|
# 阿里云OSS服务
|
|||
|
|
import os
|
|||
|
|
import uuid
|
|||
|
|
import datetime
|
|||
|
|
from typing import Optional
|
|||
|
|
import oss2
|
|||
|
|
from PIL import Image
|
|||
|
|
import io
|
|||
|
|
|
|||
|
|
# OSS配置
|
|||
|
|
OSS_CONFIG = {
|
|||
|
|
"access_key_id": os.getenv("OSS_ACCESS_KEY_ID", "LTAI5t6HUnpFBLEK9194kPVG"),
|
|||
|
|
"access_key_secret": os.getenv("OSS_ACCESS_KEY_SECRET", "LEr4Q8yRxb8D5b24cKfCwlt4MMoke1"),
|
|||
|
|
"bucket_name": "jiachenlong-oss",
|
|||
|
|
"endpoint": "oss-cn-hangzhou.aliyuncs.com",
|
|||
|
|
"public_url": "https://jiachenlong-oss.oss-cn-hangzhou.aliyuncs.com"
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
# 图片压缩配置
|
|||
|
|
IMAGE_CONFIG = {
|
|||
|
|
"max_size": 1024 * 1024, # 1MB
|
|||
|
|
"max_width": 2048,
|
|||
|
|
"max_height": 2048,
|
|||
|
|
"quality": 85,
|
|||
|
|
"format": "JPEG"
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
# 初始化OSS
|
|||
|
|
auth = oss2.Auth(OSS_CONFIG["access_key_id"], OSS_CONFIG["access_key_secret"])
|
|||
|
|
bucket = oss2.Bucket(auth, OSS_CONFIG["endpoint"], OSS_CONFIG["bucket_name"])
|
|||
|
|
|
|||
|
|
|
|||
|
|
def get_oss_path(prefix: str, user_id: str = None, filename: str = None) -> str:
|
|||
|
|
"""生成OSS存储路径"""
|
|||
|
|
now = datetime.datetime.now()
|
|||
|
|
year = now.strftime("%Y")
|
|||
|
|
month = now.strftime("%m")
|
|||
|
|
day = now.strftime("%d")
|
|||
|
|
|
|||
|
|
if filename:
|
|||
|
|
ext = filename.split('.')[-1] if '.' in filename else 'jpg'
|
|||
|
|
unique_name = f"{uuid.uuid4().hex}.{ext}"
|
|||
|
|
else:
|
|||
|
|
unique_name = f"{uuid.uuid4().hex}.jpg"
|
|||
|
|
|
|||
|
|
if user_id:
|
|||
|
|
path = f"{prefix}/{user_id}/{year}/{month}/{unique_name}"
|
|||
|
|
else:
|
|||
|
|
path = f"{prefix}/{year}/{month}/{day}/{unique_name}"
|
|||
|
|
|
|||
|
|
return path, unique_name
|
|||
|
|
|
|||
|
|
|
|||
|
|
def upload_to_oss(file_data: bytes, oss_key: str, compress: bool = True) -> str:
|
|||
|
|
"""上传文件到OSS，返回公网URL"""
|
|||
|
|
try:
|
|||
|
|
# 如果是图片，进行压缩
|
|||
|
|
if compress and any(oss_key.lower().endswith(ext) for ext in ['.jpg', '.jpeg', '.png', '.webp']):
|
|||
|
|
file_data = compress_image(file_data)
|
|||
|
|
|
|||
|
|
# 上传文件
|
|||
|
|
result = bucket.put_object(oss_key, file_data)
|
|||
|
|
|
|||
|
|
if result.status == 200:
|
|||
|
|
# 返回公网URL
|
|||
|
|
return f"{OSS_CONFIG['public_url']}/{oss_key}"
|
|||
|
|
else:
|
|||
|
|
raise Exception(f"OSS上传失败: {result.status}")
|
|||
|
|
|
|||
|
|
except Exception as e:
|
|||
|
|
raise Exception(f"OSS上传失败: {str(e)}")
|
|||
|
|
|
|||
|
|
|
|||
|
|
def delete_from_oss(oss_key: str) -> bool:
|
|||
|
|
"""从OSS删除文件"""
|
|||
|
|
try:
|
|||
|
|
result = bucket.delete_object(oss_key)
|
|||
|
|
return result.status == 204
|
|||
|
|
except Exception as e:
|
|||
|
|
print(f"OSS删除失败: {str(e)}")
|
|||
|
|
return False
|
|||
|
|
|
|||
|
|
|
|||
|
|
def get_public_url(oss_key: str) -> str:
|
|||
|
|
"""获取公网URL"""
|
|||
|
|
return f"{OSS_CONFIG['public_url']}/{oss_key}"
|
|||
|
|
|
|||
|
|
|
|||
|
|
def compress_image(image_data: bytes, max_size: int = None) -> bytes:
|
|||
|
|
"""压缩图片到指定大小以内"""
|
|||
|
|
if max_size is None:
|
|||
|
|
max_size = IMAGE_CONFIG["max_size"]
|
|||
|
|
|
|||
|
|
# 如果已经小于限制，直接返回
|
|||
|
|
if len(image_data) <= max_size:
|
|||
|
|
return image_data
|
|||
|
|
|
|||
|
|
# 打开图片
|
|||
|
|
img = Image.open(io.BytesIO(image_data))
|
|||
|
|
|
|||
|
|
# 如果是PNG且有透明通道，转换为RGB
|
|||
|
|
if img.mode in ('RGBA', 'P'):
|
|||
|
|
img = img.convert('RGB')
|
|||
|
|
|
|||
|
|
# 逐步降低质量直到达到目标大小
|
|||
|
|
quality = 95
|
|||
|
|
compressed_data = image_data
|
|||
|
|
|
|||
|
|
while quality > 30 and len(compressed_data) > max_size:
|
|||
|
|
output = io.BytesIO()
|
|||
|
|
img.save(output, format=IMAGE_CONFIG["format"], quality=quality, optimize=True)
|
|||
|
|
compressed_data = output.getvalue()
|
|||
|
|
quality -= 10
|
|||
|
|
|
|||
|
|
# 如果还是太大，缩小尺寸
|
|||
|
|
if len(compressed_data) > max_size:
|
|||
|
|
width, height = img.size
|
|||
|
|
while len(compressed_data) > max_size and width > 400:
|
|||
|
|
width = int(width * 0.8)
|
|||
|
|
height = int(height * 0.8)
|
|||
|
|
img_resized = img.resize((width, height), Image.Resampling.LANCZOS)
|
|||
|
|
output = io.BytesIO()
|
|||
|
|
img_resized.save(output, format=IMAGE_CONFIG["format"], quality=80, optimize=True)
|
|||
|
|
compressed_data = output.getvalue()
|
|||
|
|
|
|||
|
|
return compressed_data
|