feat: add trusted ticket intake backend

backend/apps/tickets/storage.py

@@ -0,0 +1,112 @@
+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:
+    if len(raw) > 10 * 1024 * 1024:
+        raise DraftImageError(413, "单张图片不能超过 10 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),
+    )