148 lines
5.1 KiB
Python
148 lines
5.1 KiB
Python
import base64
|
||
import binascii
|
||
import logging
|
||
from io import BytesIO
|
||
|
||
from django.core.files.base import ContentFile
|
||
from PIL import Image
|
||
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 Exception:
|
||
return self.default_ext, "image/jpeg"
|
||
|
||
def to_internal_value(self, data):
|
||
# Принимаем None/пустое
|
||
if data in (None, "", b""):
|
||
return super().to_internal_value(data)
|
||
|
||
if isinstance(data, str):
|
||
# Срезаем префикс data:image/...;base64, если есть
|
||
if data.startswith("data:image"):
|
||
try:
|
||
_, data = data.split(";base64,", 1)
|
||
except ValueError:
|
||
logger.warning(
|
||
"Base64 image validation failed",
|
||
extra={
|
||
"event": "image.base64.validation_failed",
|
||
"reason": "invalid_data_uri",
|
||
},
|
||
)
|
||
raise serializers.ValidationError("Некорректный формат data URI.")
|
||
|
||
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.")
|
||
|
||
ext, _ = self._detect_ext_mime(decoded)
|
||
file = ContentFile(decoded, name=f"upload.{ext}")
|
||
result = super().to_internal_value(file)
|
||
logger.debug(
|
||
"Base64 image decoded",
|
||
extra={
|
||
"event": "image.base64.decoded",
|
||
"image_format": ext,
|
||
"image_bytes": len(decoded),
|
||
},
|
||
)
|
||
return result
|
||
|
||
# КРИТИЧНО: проверяем, не пустой ли файл
|
||
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 hasattr(data, "read"):
|
||
# Проверяем содержимое файла
|
||
try:
|
||
content = data.read()
|
||
if hasattr(data, "seek"):
|
||
data.seek(0)
|
||
if not content or len(content) == 0:
|
||
logger.warning(
|
||
"Image upload validation failed",
|
||
extra={
|
||
"event": "image.upload.validation_failed",
|
||
"reason": "empty_file",
|
||
},
|
||
)
|
||
return None
|
||
except Exception as error:
|
||
logger.warning(
|
||
"Image upload inspection failed",
|
||
extra={
|
||
"event": "image.upload.inspection_failed",
|
||
"error_type": type(error).__name__,
|
||
},
|
||
exc_info=True,
|
||
)
|
||
|
||
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 Exception:
|
||
try:
|
||
with open(value.path, "rb") as f:
|
||
content = f.read()
|
||
except Exception:
|
||
return ""
|
||
|
||
b64 = base64.b64encode(content).decode("ascii")
|
||
_, mime = self._detect_ext_mime(content)
|
||
return f"data:{mime};base64,{b64}"
|