156 lines
5.2 KiB
Python
156 lines
5.2 KiB
Python
import base64
|
||
import binascii
|
||
import logging
|
||
from io import BytesIO
|
||
|
||
from django.core.files.base import ContentFile
|
||
from PIL import Image, UnidentifiedImageError
|
||
from rest_framework import serializers
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class Base64ImageField(serializers.ImageField):
|
||
"""
|
||
Двухстороннее поле:
|
||
- input: принимает строку base64 или data URI (data:image/...;base64,xxxx)
|
||
- output: возвращает data URI (data:image/...;base64,xxxx)
|
||
Совместимо с Python 3.13 (без imghdr).
|
||
"""
|
||
|
||
default_ext = "jpg"
|
||
|
||
def _detect_ext_mime(self, content: bytes) -> tuple[str, str]:
|
||
"""Определяем расширение и MIME через Pillow."""
|
||
try:
|
||
im = Image.open(BytesIO(content))
|
||
fmt = (im.format or "").upper()
|
||
ext_map = {
|
||
"JPEG": "jpg",
|
||
"JPG": "jpg",
|
||
"PNG": "png",
|
||
"WEBP": "webp",
|
||
"GIF": "gif",
|
||
"BMP": "bmp",
|
||
"TIFF": "tiff",
|
||
"ICO": "ico",
|
||
}
|
||
ext = ext_map.get(fmt, self.default_ext)
|
||
mime = Image.MIME.get(fmt, "image/jpeg")
|
||
return ext, mime
|
||
except (OSError, UnidentifiedImageError, ValueError):
|
||
return self.default_ext, "image/jpeg"
|
||
|
||
def _decode_base64_image(self, data: str):
|
||
if data.startswith("data:image"):
|
||
try:
|
||
_, data = data.split(";base64,", 1)
|
||
except ValueError as exc:
|
||
logger.warning(
|
||
"Base64 image validation failed",
|
||
extra={
|
||
"event": "image.base64.validation_failed",
|
||
"reason": "invalid_data_uri",
|
||
},
|
||
)
|
||
raise serializers.ValidationError(
|
||
"Некорректный формат data URI."
|
||
) from exc
|
||
|
||
try:
|
||
decoded = base64.b64decode(data)
|
||
except (TypeError, ValueError, binascii.Error) as error:
|
||
logger.warning(
|
||
"Base64 image validation failed",
|
||
extra={
|
||
"event": "image.base64.validation_failed",
|
||
"reason": "invalid_base64",
|
||
"error_type": type(error).__name__,
|
||
},
|
||
)
|
||
raise serializers.ValidationError("Невалидный base64.") from error
|
||
|
||
extension, _ = self._detect_ext_mime(decoded)
|
||
file = ContentFile(decoded, name=f"upload.{extension}")
|
||
result = super().to_internal_value(file)
|
||
logger.debug(
|
||
"Base64 image decoded",
|
||
extra={
|
||
"event": "image.base64.decoded",
|
||
"image_format": extension,
|
||
"image_bytes": len(decoded),
|
||
},
|
||
)
|
||
return result
|
||
|
||
@staticmethod
|
||
def _uploaded_file_is_empty(data) -> bool:
|
||
if not hasattr(data, "read"):
|
||
return False
|
||
try:
|
||
content = data.read()
|
||
if hasattr(data, "seek"):
|
||
data.seek(0)
|
||
return not content
|
||
except (OSError, ValueError) as error:
|
||
logger.warning(
|
||
"Image upload inspection failed",
|
||
extra={
|
||
"event": "image.upload.inspection_failed",
|
||
"error_type": type(error).__name__,
|
||
},
|
||
exc_info=True,
|
||
)
|
||
return False
|
||
|
||
def to_internal_value(self, data):
|
||
# Принимаем None/пустое
|
||
if data in (None, "", b""):
|
||
return super().to_internal_value(data)
|
||
|
||
if isinstance(data, str):
|
||
return self._decode_base64_image(data)
|
||
|
||
# КРИТИЧНО: проверяем, не пустой ли файл
|
||
if hasattr(data, "size") and data.size == 0:
|
||
logger.warning(
|
||
"Image upload validation failed",
|
||
extra={
|
||
"event": "image.upload.validation_failed",
|
||
"reason": "empty_file",
|
||
},
|
||
)
|
||
return None
|
||
|
||
if self._uploaded_file_is_empty(data):
|
||
logger.warning(
|
||
"Image upload validation failed",
|
||
extra={
|
||
"event": "image.upload.validation_failed",
|
||
"reason": "empty_file",
|
||
},
|
||
)
|
||
return None
|
||
|
||
return super().to_internal_value(data)
|
||
|
||
def to_representation(self, value):
|
||
if not value:
|
||
return ""
|
||
# Получаем байты
|
||
try:
|
||
file_obj = getattr(value, "file", None) or value
|
||
content = file_obj.read()
|
||
if hasattr(file_obj, "seek"):
|
||
file_obj.seek(0)
|
||
except (AttributeError, OSError, ValueError):
|
||
try:
|
||
with open(value.path, "rb") as f:
|
||
content = f.read()
|
||
except (AttributeError, OSError, ValueError):
|
||
return ""
|
||
|
||
b64 = base64.b64encode(content).decode("ascii")
|
||
_, mime = self._detect_ext_mime(content)
|
||
return f"data:{mime};base64,{b64}"
|