Browse Source
feat(watermark): add automatic responsive watermarking on image uploads
feat(watermark): add automatic responsive watermarking on image uploads
- Create reusable watermark utility in utils/watermark.py supporting PIL, UploadedFile, bytes, and DRF serializer fields - Default position set to bottom-left with responsive 32% width scaling and alpha blending - Auto-resolve SVG watermark to high-resolution PNG - Integrate automatic watermarking into Article thumbnails, Hadith/Book Reference images, Corrections, Original Texts, and Interpretationsmaster
7 changed files with 403 additions and 5 deletions
-
15apps/article/serializers_admin.py
-
15apps/hadis/models/hadis.py
-
5apps/hadis/models/reference.py
-
5apps/hadis/models/transmitter.py
-
19apps/hadis/serializers/serializers_admin.py
-
BINstatic/images/watermark.png
-
349utils/watermark.py
|
After Width: 292 | Height: 83 | Size: 5.2 KiB |
@ -0,0 +1,349 @@ |
|||
""" |
|||
Watermark utility for applying logo/watermark overlays onto uploaded images. |
|||
Supports Django UploadedFile, file paths, raw bytes, and PIL Image instances. |
|||
Default position is bottom-left. |
|||
""" |
|||
import io |
|||
import os |
|||
import logging |
|||
from pathlib import Path |
|||
from typing import Optional, Union, Tuple |
|||
from PIL import Image, ImageEnhance |
|||
from django.conf import settings |
|||
from django.core.files.uploadedfile import SimpleUploadedFile, UploadedFile |
|||
|
|||
logger = logging.getLogger(__name__) |
|||
|
|||
# Default watermark search paths |
|||
DEFAULT_WATERMARK_PATHS = [ |
|||
getattr(settings, "WATERMARK_IMAGE_PATH", None), |
|||
os.path.join(settings.BASE_DIR, "static", "images", "watermark.png"), |
|||
os.path.join(settings.BASE_DIR, "static", "images", "watermark.svg"), |
|||
os.path.join(settings.BASE_DIR, "static", "images", "Frame.svg"), |
|||
] |
|||
|
|||
|
|||
def _get_default_watermark_path() -> Optional[str]: |
|||
for p in DEFAULT_WATERMARK_PATHS: |
|||
if p and os.path.exists(p): |
|||
return p |
|||
return None |
|||
|
|||
|
|||
def is_watermarkable_image(filename: str, content_type: Optional[str] = None) -> bool: |
|||
""" |
|||
Check if a file is an image format that supports watermarking. |
|||
Excludes SVG and GIF animations. |
|||
""" |
|||
name = (filename or "").lower() |
|||
ctype = (content_type or "").lower() |
|||
|
|||
if name.endswith(".gif") or name.endswith(".svg"): |
|||
return False |
|||
if ctype in {"image/gif", "image/svg+xml"}: |
|||
return False |
|||
|
|||
valid_exts = (".jpg", ".jpeg", ".png", ".webp", ".bmp", ".tif", ".tiff", ".avif") |
|||
return name.endswith(valid_exts) or ctype.startswith("image/") |
|||
|
|||
|
|||
def apply_watermark_to_image( |
|||
base_img: Image.Image, |
|||
watermark_source: Optional[Union[str, Path, Image.Image]] = None, |
|||
position: str = "bottom_left", |
|||
opacity: float = 0.85, |
|||
scale_ratio: float = 0.32, |
|||
padding: int = 24, |
|||
) -> Image.Image: |
|||
""" |
|||
Apply a watermark image onto a base PIL Image. |
|||
|
|||
Args: |
|||
base_img: PIL Image to watermark. |
|||
watermark_source: Path to watermark image or PIL Image object. |
|||
position: 'bottom_left' (default) | 'bottom_right' | 'top_right' | 'top_left' | 'center'. |
|||
opacity: Float 0.0 to 1.0. |
|||
scale_ratio: Watermark width relative to base image width (default 0.32). |
|||
padding: Pixel padding from borders. |
|||
|
|||
Returns: |
|||
New PIL Image with watermark blended. |
|||
""" |
|||
if watermark_source is None: |
|||
watermark_source = _get_default_watermark_path() |
|||
|
|||
if not watermark_source: |
|||
logger.debug("[Watermark] No watermark image found; skipping watermarking.") |
|||
return base_img |
|||
|
|||
try: |
|||
if isinstance(watermark_source, (str, Path)): |
|||
src_str = str(watermark_source) |
|||
if not os.path.exists(src_str): |
|||
logger.warning("[Watermark] File not found: %s", src_str) |
|||
return base_img |
|||
|
|||
# Pillow cannot open SVG directly; resolve to raster PNG if SVG given |
|||
if src_str.lower().endswith(".svg"): |
|||
png_candidate = os.path.splitext(src_str)[0] + ".png" |
|||
if os.path.exists(png_candidate): |
|||
src_str = png_candidate |
|||
else: |
|||
default_png = os.path.join(settings.BASE_DIR, "static", "images", "watermark.png") |
|||
if os.path.exists(default_png): |
|||
src_str = default_png |
|||
else: |
|||
logger.warning("[Watermark] Cannot open SVG without raster version: %s", src_str) |
|||
return base_img |
|||
|
|||
wm_img = Image.open(src_str).convert("RGBA") |
|||
elif isinstance(watermark_source, Image.Image): |
|||
wm_img = watermark_source.convert("RGBA") |
|||
else: |
|||
return base_img |
|||
|
|||
# Ensure base image is in RGBA for clean alpha blending |
|||
original_mode = base_img.mode |
|||
original_format = getattr(base_img, "format", None) |
|||
base_rgba = base_img.convert("RGBA") |
|||
base_w, base_h = base_rgba.size |
|||
|
|||
# Compute responsive dimensions for watermark (clear and prominent) |
|||
target_wm_w = max(int(base_w * scale_ratio), 120) |
|||
wm_aspect = wm_img.height / wm_img.width |
|||
target_wm_h = max(int(target_wm_w * wm_aspect), 35) |
|||
|
|||
# Scale down if height exceeds 45% of base image height |
|||
if target_wm_h > int(base_h * 0.45): |
|||
target_wm_h = int(base_h * 0.45) |
|||
target_wm_w = int(target_wm_h / wm_aspect) |
|||
|
|||
wm_resized = wm_img.resize((target_wm_w, target_wm_h), Image.Resampling.LANCZOS) |
|||
|
|||
# Adjust watermark opacity |
|||
if opacity < 1.0: |
|||
alpha = wm_resized.split()[3] |
|||
alpha = ImageEnhance.Brightness(alpha).enhance(max(0.0, min(1.0, opacity))) |
|||
wm_resized.putalpha(alpha) |
|||
|
|||
# Determine coordinates |
|||
pos = position.lower() |
|||
if pos == "bottom_left": |
|||
x = padding |
|||
y = base_h - target_wm_h - padding |
|||
elif pos == "bottom_right": |
|||
x = base_w - target_wm_w - padding |
|||
y = base_h - target_wm_h - padding |
|||
elif pos == "top_right": |
|||
x = base_w - target_wm_w - padding |
|||
y = padding |
|||
elif pos == "top_left": |
|||
x = padding |
|||
y = padding |
|||
elif pos == "center": |
|||
x = (base_w - target_wm_w) // 2 |
|||
y = (base_h - target_wm_h) // 2 |
|||
else: |
|||
x = padding |
|||
y = base_h - target_wm_h - padding |
|||
|
|||
# Keep inside image boundaries |
|||
x = max(0, min(x, base_w - target_wm_w)) |
|||
y = max(0, min(y, base_h - target_wm_h)) |
|||
|
|||
# Composite watermark onto base image |
|||
transparent_layer = Image.new("RGBA", (base_w, base_h), (0, 0, 0, 0)) |
|||
transparent_layer.paste(wm_resized, (x, y), mask=wm_resized) |
|||
watermarked_rgba = Image.alpha_composite(base_rgba, transparent_layer) |
|||
|
|||
# Restore original color mode if needed (e.g. RGB for JPEG) |
|||
if original_mode in ("RGB", "L"): |
|||
result = watermarked_rgba.convert(original_mode) |
|||
else: |
|||
result = watermarked_rgba |
|||
|
|||
result.format = original_format or "PNG" |
|||
return result |
|||
|
|||
except Exception as exc: |
|||
logger.error("[Watermark] Failed to apply watermark: %s", exc) |
|||
return base_img |
|||
|
|||
|
|||
def apply_watermark_to_bytes( |
|||
image_bytes: bytes, |
|||
filename: str = "image.png", |
|||
watermark_source: Optional[Union[str, Path, Image.Image]] = None, |
|||
position: str = "bottom_left", |
|||
opacity: float = 0.85, |
|||
scale_ratio: float = 0.32, |
|||
padding: int = 24, |
|||
quality: int = 90, |
|||
) -> bytes: |
|||
""" |
|||
Apply watermark to raw image bytes and return watermarked image bytes. |
|||
""" |
|||
if not image_bytes: |
|||
return image_bytes |
|||
|
|||
try: |
|||
with Image.open(io.BytesIO(image_bytes)) as img: |
|||
format_name = img.format or "JPEG" |
|||
watermarked_img = apply_watermark_to_image( |
|||
base_img=img, |
|||
watermark_source=watermark_source, |
|||
position=position, |
|||
opacity=opacity, |
|||
scale_ratio=scale_ratio, |
|||
padding=padding, |
|||
) |
|||
|
|||
out_io = io.BytesIO() |
|||
if format_name.upper() in ("JPEG", "JPG"): |
|||
if watermarked_img.mode != "RGB": |
|||
watermarked_img = watermarked_img.convert("RGB") |
|||
watermarked_img.save(out_io, format="JPEG", quality=quality, optimize=True) |
|||
elif format_name.upper() == "WEBP": |
|||
watermarked_img.save(out_io, format="WEBP", quality=quality, method=6) |
|||
else: |
|||
watermarked_img.save(out_io, format="PNG", optimize=True) |
|||
|
|||
return out_io.getvalue() |
|||
|
|||
except Exception as exc: |
|||
logger.error("[Watermark] Error processing image bytes: %s", exc) |
|||
return image_bytes |
|||
|
|||
|
|||
def apply_watermark_to_upload( |
|||
file_obj, |
|||
watermark_source: Optional[Union[str, Path, Image.Image]] = None, |
|||
position: str = "bottom_left", |
|||
opacity: float = 0.85, |
|||
scale_ratio: float = 0.32, |
|||
padding: int = 24, |
|||
quality: int = 90, |
|||
) -> Optional[SimpleUploadedFile]: |
|||
""" |
|||
Takes an uploaded file object (e.g. from request.FILES), applies watermark, |
|||
and returns a new SimpleUploadedFile ready to be saved to an ImageField. |
|||
""" |
|||
if not file_obj: |
|||
return file_obj |
|||
|
|||
filename = getattr(file_obj, "name", "upload.jpg") |
|||
content_type = getattr(file_obj, "content_type", "image/jpeg") |
|||
|
|||
if not is_watermarkable_image(filename, content_type): |
|||
return file_obj |
|||
|
|||
try: |
|||
raw_bytes = file_obj.read() |
|||
if hasattr(file_obj, "seek"): |
|||
file_obj.seek(0) |
|||
|
|||
if not raw_bytes: |
|||
return file_obj |
|||
|
|||
watermarked_bytes = apply_watermark_to_bytes( |
|||
image_bytes=raw_bytes, |
|||
filename=filename, |
|||
watermark_source=watermark_source, |
|||
position=position, |
|||
opacity=opacity, |
|||
scale_ratio=scale_ratio, |
|||
padding=padding, |
|||
quality=quality, |
|||
) |
|||
|
|||
return SimpleUploadedFile( |
|||
name=filename, |
|||
content=watermarked_bytes, |
|||
content_type=content_type, |
|||
) |
|||
|
|||
except Exception as exc: |
|||
logger.error("[Watermark] Failed to process upload %s: %s", filename, exc) |
|||
if hasattr(file_obj, "seek"): |
|||
file_obj.seek(0) |
|||
return file_obj |
|||
|
|||
|
|||
def apply_watermark_to_field_file( |
|||
field_file, |
|||
position: str = "bottom_left", |
|||
opacity: float = 0.85, |
|||
scale_ratio: float = 0.32, |
|||
padding: int = 24, |
|||
quality: int = 90, |
|||
): |
|||
""" |
|||
Applies watermark to a Django FieldFile instance inside a model's save() method |
|||
if a new or uncommitted file is present. |
|||
""" |
|||
if not field_file or not hasattr(field_file, "file"): |
|||
return |
|||
|
|||
try: |
|||
file_obj = field_file.file |
|||
# Check if it's an UploadedFile instance or not yet saved to storage |
|||
if isinstance(file_obj, UploadedFile) or not getattr(field_file, "_committed", True): |
|||
watermarked = apply_watermark_to_upload( |
|||
file_obj=file_obj, |
|||
position=position, |
|||
opacity=opacity, |
|||
scale_ratio=scale_ratio, |
|||
padding=padding, |
|||
quality=quality, |
|||
) |
|||
if watermarked: |
|||
field_file.save(field_file.name, watermarked, save=False) |
|||
except Exception as exc: |
|||
logger.error("[Watermark] Failed to watermark field file: %s", exc) |
|||
|
|||
|
|||
# DRF Serializer Field Integration |
|||
try: |
|||
from rest_framework import serializers |
|||
|
|||
class WatermarkedImageField(serializers.ImageField): |
|||
""" |
|||
DRF ImageField that automatically applies a watermark to uploaded image files. |
|||
""" |
|||
def __init__( |
|||
self, |
|||
*args, |
|||
watermark_source=None, |
|||
position="bottom_left", |
|||
opacity=0.85, |
|||
scale_ratio=0.32, |
|||
padding=24, |
|||
quality=90, |
|||
**kwargs, |
|||
): |
|||
self.watermark_source = watermark_source |
|||
self.position = position |
|||
self.opacity = opacity |
|||
self.scale_ratio = scale_ratio |
|||
self.padding = padding |
|||
self.quality = quality |
|||
super().__init__(*args, **kwargs) |
|||
|
|||
def to_internal_value(self, data): |
|||
uploaded = super().to_internal_value(data) |
|||
if not uploaded: |
|||
return uploaded |
|||
|
|||
watermarked = apply_watermark_to_upload( |
|||
file_obj=uploaded, |
|||
watermark_source=self.watermark_source, |
|||
position=self.position, |
|||
opacity=self.opacity, |
|||
scale_ratio=self.scale_ratio, |
|||
padding=self.padding, |
|||
quality=self.quality, |
|||
) |
|||
return watermarked or uploaded |
|||
|
|||
except ImportError: |
|||
pass |
|||
Write
Preview
Loading…
Cancel
Save
Reference in new issue