114 lines
3.3 KiB
Python
114 lines
3.3 KiB
Python
import hashlib
|
|
import io
|
|
from dataclasses import dataclass
|
|
|
|
from django.conf import settings
|
|
from minio import Minio
|
|
from PIL import Image, UnidentifiedImageError
|
|
|
|
|
|
class DraftImageError(ValueError):
|
|
def __init__(self, status_code: int, message: str):
|
|
super().__init__(message)
|
|
self.status_code = status_code
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class PreparedImage:
|
|
data: bytes
|
|
file_name: str
|
|
mime_type: str
|
|
extension: str
|
|
sha256: str
|
|
|
|
|
|
FORMATS = {
|
|
"JPEG": ("image/jpeg", "jpg"),
|
|
"PNG": ("image/png", "png"),
|
|
"WEBP": ("image/webp", "webp"),
|
|
}
|
|
|
|
|
|
def prepare_image(raw: bytes, original_name: str) -> PreparedImage:
|
|
max_bytes = settings.TICKET_MATERIAL_MAX_FILE_BYTES
|
|
if len(raw) > max_bytes:
|
|
raise DraftImageError(413, f"单个材料不能超过 {max_bytes // 1024 // 1024} MB")
|
|
try:
|
|
with Image.open(io.BytesIO(raw)) as image:
|
|
image.verify()
|
|
with Image.open(io.BytesIO(raw)) as image:
|
|
if image.format not in FORMATS or getattr(image, "is_animated", False):
|
|
raise DraftImageError(415, "仅支持 JPEG、PNG 和 WebP 静态图片")
|
|
mime_type, extension = FORMATS[image.format]
|
|
if image.mode not in {"RGB", "RGBA", "L"}:
|
|
image = image.convert("RGBA" if "A" in image.getbands() else "RGB")
|
|
output = io.BytesIO()
|
|
save_format = "JPEG" if extension == "jpg" else image.format
|
|
if save_format == "JPEG" and image.mode != "RGB":
|
|
image = image.convert("RGB")
|
|
image.save(output, format=save_format, quality=90, optimize=True)
|
|
except DraftImageError:
|
|
raise
|
|
except (UnidentifiedImageError, OSError, ValueError) as error:
|
|
raise DraftImageError(415, "图片内容无法识别或格式不受支持") from error
|
|
data = output.getvalue()
|
|
stem = original_name.rsplit(".", 1)[0].strip() or "image"
|
|
return PreparedImage(
|
|
data=data,
|
|
file_name=f"{stem}.{extension}",
|
|
mime_type=mime_type,
|
|
extension=extension,
|
|
sha256=hashlib.sha256(data).hexdigest(),
|
|
)
|
|
|
|
|
|
def client() -> Minio:
|
|
return Minio(
|
|
settings.MINIO_ENDPOINT,
|
|
access_key=settings.MINIO_ACCESS_KEY,
|
|
secret_key=settings.MINIO_SECRET_KEY,
|
|
secure=settings.MINIO_SECURE,
|
|
)
|
|
|
|
|
|
def ensure_bucket(storage: Minio) -> None:
|
|
if not storage.bucket_exists(settings.MINIO_BUCKET):
|
|
storage.make_bucket(settings.MINIO_BUCKET)
|
|
|
|
|
|
def put_object(path: str, data: bytes, mime_type: str) -> None:
|
|
storage = client()
|
|
ensure_bucket(storage)
|
|
storage.put_object(
|
|
settings.MINIO_BUCKET,
|
|
path,
|
|
io.BytesIO(data),
|
|
len(data),
|
|
content_type=mime_type,
|
|
)
|
|
|
|
|
|
def get_object(path: str) -> bytes:
|
|
response = client().get_object(settings.MINIO_BUCKET, path)
|
|
try:
|
|
return response.read()
|
|
finally:
|
|
response.close()
|
|
response.release_conn()
|
|
|
|
|
|
def delete_object(path: str) -> None:
|
|
client().remove_object(settings.MINIO_BUCKET, path)
|
|
|
|
|
|
def copy_object(source_path: str, target_path: str) -> None:
|
|
from minio.commonconfig import CopySource
|
|
|
|
storage = client()
|
|
ensure_bucket(storage)
|
|
storage.copy_object(
|
|
settings.MINIO_BUCKET,
|
|
target_path,
|
|
CopySource(settings.MINIO_BUCKET, source_path),
|
|
)
|