Private
Public Access
ksmith
8ca3e2e23d
Users can set a display name (shown instead of username in the message list, room member list, TopBar, and admin Users tab) and upload a real avatar, replacing the generated color-initial avatars everywhere a user appears. Avatars are square-cropped and downscaled to 512px, reusing app/storage.py's upload primitives from image uploads with a new square option. Two deliberate divergences from message-image handling, documented in backend/README.md: the previous avatar file is deleted on replace/remove (safe since it's strictly one file per user), and avatar serving is not room-gated and uses a short cache (identity-addressed and mutable, unlike a message image's permanent content-addressed URL). Frontend: new ProfileModal reachable from the TopBar account menu; AuthContext gains updateUser() so a profile change reflects instantly everywhere without a refetch.
103 lines
3.2 KiB
Python
103 lines
3.2 KiB
Python
import io
|
|
import pathlib
|
|
import uuid
|
|
|
|
from PIL import Image, UnidentifiedImageError
|
|
|
|
# backend/app/storage.py -> backend/ -> repo root -- same
|
|
# resolve-relative-to-file convention FRONTEND_DIST uses in app/main.py, so
|
|
# this lands in the right place in both local dev and the /srv/chatapp
|
|
# production layout with zero new config.
|
|
UPLOADS_DIR = pathlib.Path(__file__).resolve().parent.parent.parent / "uploads"
|
|
|
|
MAX_IMAGE_BYTES = 8 * 1024 * 1024
|
|
_READ_CHUNK_BYTES = 1024 * 1024
|
|
_MAX_DIMENSION = 2000
|
|
|
|
# (storage extension, Pillow format name)
|
|
ALLOWED_IMAGE_CONTENT_TYPES: dict[str, tuple[str, str]] = {
|
|
"image/jpeg": (".jpg", "JPEG"),
|
|
"image/png": (".png", "PNG"),
|
|
"image/gif": (".gif", "GIF"),
|
|
"image/webp": (".webp", "WEBP"),
|
|
}
|
|
|
|
|
|
class ImageTooLargeError(Exception):
|
|
pass
|
|
|
|
|
|
class InvalidImageError(Exception):
|
|
pass
|
|
|
|
|
|
async def read_capped(file, cap: int = MAX_IMAGE_BYTES) -> bytes:
|
|
"""Reads an UploadFile-like object in chunks, raising as soon as `cap`
|
|
is exceeded rather than after buffering the whole (potentially huge)
|
|
body first."""
|
|
chunks = []
|
|
total = 0
|
|
while True:
|
|
chunk = await file.read(_READ_CHUNK_BYTES)
|
|
if not chunk:
|
|
break
|
|
total += len(chunk)
|
|
if total > cap:
|
|
raise ImageTooLargeError()
|
|
chunks.append(chunk)
|
|
return b"".join(chunks)
|
|
|
|
|
|
def process_image(
|
|
data: bytes,
|
|
content_type: str,
|
|
*,
|
|
square: bool = False,
|
|
max_dimension: int | None = None,
|
|
) -> tuple[bytes, str]:
|
|
"""Confirms `data` is a genuinely decodable image (not just a spoofed
|
|
Content-Type header) and downscales it so its longer side is
|
|
<=max_dimension (default 2000px) -- except GIF, left untouched so
|
|
animation isn't collapsed to a single frame. When `square` is set
|
|
(avatars), center-crops to the shorter side first. Returns
|
|
(final_bytes, storage_extension)."""
|
|
ext, pillow_format = ALLOWED_IMAGE_CONTENT_TYPES[content_type]
|
|
dimension_cap = max_dimension or _MAX_DIMENSION
|
|
|
|
try:
|
|
with Image.open(io.BytesIO(data)) as probe:
|
|
probe.verify()
|
|
except (UnidentifiedImageError, OSError, ValueError) as exc:
|
|
raise InvalidImageError() from exc
|
|
|
|
if content_type == "image/gif":
|
|
return data, ext
|
|
|
|
# verify() leaves the image unusable for further processing, so reopen.
|
|
image = Image.open(io.BytesIO(data))
|
|
image.load()
|
|
if pillow_format == "JPEG" and image.mode in ("RGBA", "P"):
|
|
image = image.convert("RGB")
|
|
if square:
|
|
side = min(image.width, image.height)
|
|
left = (image.width - side) // 2
|
|
top = (image.height - side) // 2
|
|
image = image.crop((left, top, left + side, top + side))
|
|
image.thumbnail((dimension_cap, dimension_cap))
|
|
out = io.BytesIO()
|
|
image.save(out, format=pillow_format)
|
|
return out.getvalue(), ext
|
|
|
|
|
|
def save_image(data: bytes, ext: str) -> str:
|
|
UPLOADS_DIR.mkdir(parents=True, exist_ok=True)
|
|
storage_filename = f"{uuid.uuid4()}{ext}"
|
|
(UPLOADS_DIR / storage_filename).write_bytes(data)
|
|
return storage_filename
|
|
|
|
|
|
def delete_image(storage_filename: str) -> None:
|
|
"""Best-effort delete -- a missing file (already gone, or never
|
|
written) is not an error."""
|
|
(UPLOADS_DIR / storage_filename).unlink(missing_ok=True)
|