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}"