You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
379 lines
13 KiB
379 lines
13 KiB
"""
|
|
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 math
|
|
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__)
|
|
|
|
def _get_default_watermark_path() -> Optional[str]:
|
|
base_dir = getattr(settings, "BASE_DIR", None) or Path(__file__).resolve().parent.parent
|
|
base_dir_str = str(base_dir)
|
|
|
|
search_paths = [
|
|
getattr(settings, "WATERMARK_IMAGE_PATH", None),
|
|
os.path.join(base_dir_str, "static", "images", "watermark.png"),
|
|
os.path.join(base_dir_str, "static", "images", "watermark.svg"),
|
|
os.path.join(base_dir_str, "static", "images", "Frame.svg"),
|
|
]
|
|
|
|
for p in search_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.25,
|
|
padding: Optional[int] = None,
|
|
) -> 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 scale relative to image geometric mean area (default 0.25).
|
|
padding: Pixel padding from borders. If None, dynamically calculated based on image size.
|
|
|
|
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
|
|
|
|
# 1. Compute responsive dimensions using Geometric Mean Area
|
|
# This keeps watermark visual area consistent across wide, square, and portrait images
|
|
ref_dim = math.sqrt(base_w * base_h)
|
|
target_wm_w = int(ref_dim * scale_ratio)
|
|
wm_aspect = wm_img.height / wm_img.width
|
|
target_wm_h = int(target_wm_w * wm_aspect)
|
|
|
|
# 2. Dynamic aspect-ratio-aware clamping:
|
|
# - Width constraint: between 14% and 38% of base width
|
|
# - Height constraint: between 4% and 16% of base height
|
|
max_w = max(int(base_w * 0.38), 60)
|
|
min_w = max(int(base_w * 0.14), 40)
|
|
max_h = max(int(base_h * 0.16), 20)
|
|
min_h = max(int(base_h * 0.04), 12)
|
|
|
|
if target_wm_w > max_w:
|
|
target_wm_w = max_w
|
|
target_wm_h = int(target_wm_w * wm_aspect)
|
|
if target_wm_h > max_h:
|
|
target_wm_h = max_h
|
|
target_wm_w = int(target_wm_h / wm_aspect)
|
|
if target_wm_w < min_w:
|
|
target_wm_w = min_w
|
|
target_wm_h = int(target_wm_w * wm_aspect)
|
|
if target_wm_h < min_h:
|
|
target_wm_h = min_h
|
|
target_wm_w = int(target_wm_h / wm_aspect)
|
|
|
|
# Safe clamp to stay within physical bounds
|
|
target_wm_w = max(1, min(target_wm_w, base_w - 4))
|
|
target_wm_h = max(1, min(target_wm_h, base_h - 4))
|
|
|
|
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)
|
|
|
|
# Dynamic responsive padding if not explicitly overridden
|
|
if padding is None:
|
|
actual_padding = max(8, min(int(min(base_w, base_h) * 0.025), 48))
|
|
else:
|
|
actual_padding = padding
|
|
|
|
# Determine coordinates
|
|
pos = position.lower()
|
|
if pos == "bottom_left":
|
|
x = actual_padding
|
|
y = base_h - target_wm_h - actual_padding
|
|
elif pos == "bottom_right":
|
|
x = base_w - target_wm_w - actual_padding
|
|
y = base_h - target_wm_h - actual_padding
|
|
elif pos == "top_right":
|
|
x = base_w - target_wm_w - actual_padding
|
|
y = actual_padding
|
|
elif pos == "top_left":
|
|
x = actual_padding
|
|
y = actual_padding
|
|
elif pos == "center":
|
|
x = (base_w - target_wm_w) // 2
|
|
y = (base_h - target_wm_h) // 2
|
|
else:
|
|
x = actual_padding
|
|
y = base_h - target_wm_h - actual_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.25,
|
|
padding: Optional[int] = None,
|
|
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.25,
|
|
padding: Optional[int] = None,
|
|
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.25,
|
|
padding: Optional[int] = None,
|
|
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.25,
|
|
padding=None,
|
|
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
|