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
|