import io import uuid from PIL import Image from PIL.ExifTags import TAGS from django.http import JsonResponse, FileResponse from rest_framework.views import APIView from rest_framework.parsers import MultiPartParser, FormParser from rest_framework import status from rest_framework.decorators import permission_classes from rest_framework.permissions import AllowAny from ..serializers import ImageCompressRequestSerializer @permission_classes([AllowAny]) class ImageCompressView(APIView): parser_classes = [MultiPartParser, FormParser] def post(self, request): serializer = ImageCompressRequestSerializer(data=request.data) if not serializer.is_valid(): return JsonResponse( {'error': '参数错误', 'details': serializer.errors}, status=status.HTTP_400_BAD_REQUEST ) file = serializer.validated_data['file'] mode = serializer.validated_data['mode'] quality = serializer.validated_data['quality'] output_format = serializer.validated_data['format'] keep_exif = serializer.validated_data['keep_exif'] width = serializer.validated_data.get('width') height = serializer.validated_data.get('height') maintain_aspect_ratio = serializer.validated_data.get('maintain_aspect_ratio', True) try: img = Image.open(file) img.verify() file.seek(0) img = Image.open(file) except Exception: return JsonResponse( {'error': '无效的图片文件'}, status=status.HTTP_400_BAD_REQUEST ) try: original_size = file.size original_format = img.format.lower() if img.format else 'unknown' if output_format == 'original': output_format = original_format if width or height: img = self._resize_image(img, width, height, maintain_aspect_ratio) if mode == 'lossless': compressed_data = self._compress_lossless(img, output_format, keep_exif) else: compressed_data = self._compress_lossy(img, output_format, quality, keep_exif) compressed_size = len(compressed_data) compression_ratio = (1 - compressed_size / original_size) * 100 response = FileResponse( io.BytesIO(compressed_data), content_type=f'image/{output_format}' ) response['Content-Disposition'] = f'attachment; filename="compressed_{file.name}"' response['X-Original-Size'] = str(original_size) response['X-Compressed-Size'] = str(compressed_size) response['X-Compression-Ratio'] = f'{compression_ratio:.1f}' response['X-Original-Format'] = original_format response['X-Output-Format'] = output_format response['X-Compression-Mode'] = mode response['X-Quality'] = str(quality) return response except Exception as e: return JsonResponse( {'error': f'图片处理失败: {str(e)}'}, status=status.HTTP_500_INTERNAL_SERVER_ERROR ) @staticmethod def _resize_image(img, width, height, maintain_aspect_ratio): if maintain_aspect_ratio: if width and height: ratio = min(width / img.width, height / img.height) new_size = (int(img.width * ratio), int(img.height * ratio)) elif width: ratio = width / img.width new_size = (width, int(img.height * ratio)) else: ratio = height / img.height new_size = (int(img.width * ratio), height) else: new_size = (width or img.width, height or img.height) return img.resize(new_size, Image.Resampling.LANCZOS) @staticmethod def _compress_lossless(img, output_format, keep_exif): output = io.BytesIO() exif_data = img.info.get('exif') if keep_exif else None if output_format == 'jpeg': if img.mode in ('RGBA', 'P'): img = img.convert('RGB') kwargs = {'quality': 100, 'optimize': True, 'subsampling': 0} if exif_data: kwargs['exif'] = exif_data img.save(output, 'JPEG', **kwargs) elif output_format == 'png': img.save(output, 'PNG', optimize=True) elif output_format == 'webp': kwargs = {'lossless': True} if exif_data: kwargs['exif'] = exif_data img.save(output, 'WebP', **kwargs) elif output_format == 'avif': img.save(output, 'AVIF', quality=100, speed=6) else: img.save(output, output_format.upper(), optimize=True) return output.getvalue() @staticmethod def _compress_lossy(img, output_format, quality, keep_exif): output = io.BytesIO() exif_data = img.info.get('exif') if keep_exif else None if output_format == 'jpeg': if img.mode in ('RGBA', 'P'): img = img.convert('RGB') kwargs = {'quality': quality, 'optimize': True} if exif_data: kwargs['exif'] = exif_data img.save(output, 'JPEG', **kwargs) elif output_format == 'png': img.save(output, 'PNG', optimize=True, compress_level=9) elif output_format == 'webp': img.save(output, 'WebP', quality=quality, method=6) elif output_format == 'avif': img.save(output, 'AVIF', quality=quality, speed=6) else: img.save(output, output_format.upper(), quality=quality, optimize=True) return output.getvalue() @permission_classes([AllowAny]) class EXIFExtractView(APIView): parser_classes = [MultiPartParser, FormParser] def post(self, request): file = request.FILES.get('file') if not file: return JsonResponse( {'error': '请上传图片文件'}, status=status.HTTP_400_BAD_REQUEST ) try: img = Image.open(file) exif_data = self._extract_exif(img) return JsonResponse({ 'success': True, 'exif': exif_data }) except Exception as e: return JsonResponse( {'error': f'EXIF 提取失败: {str(e)}'}, status=status.HTTP_500_INTERNAL_SERVER_ERROR ) @staticmethod def _extract_exif(img): exif_data = {} if hasattr(img, '_getexif'): exif = img._getexif() if exif: for tag_id, value in exif.items(): tag = TAGS.get(tag_id, tag_id) try: exif_data[tag] = str(value) except (TypeError, ValueError): exif_data[tag] = repr(value) return exif_data