import hashlib from shutil import COPY_BUFSIZE # type: ignore import blurhash # type: ignore from fastapi import UploadFile from loguru import logger from PIL import Image from PIL import ImageOps from sqlalchemy import select from app import activitypub as ap from app import models from app.config import BASE_URL from app.config import ROOT_DIR from app.database import AsyncSession UPLOAD_DIR = ROOT_DIR / "data" / "uploads" async def save_upload(db_session: AsyncSession, f: UploadFile) -> models.Upload: # Compute the hash h = hashlib.blake2b(digest_size=32) while True: buf = f.file.read(COPY_BUFSIZE) if not buf: break h.update(buf) content_hash = h.hexdigest() f.file.seek(0) existing_upload = ( await db_session.execute( select(models.Upload).where(models.Upload.content_hash == content_hash) ) ).scalar_one_or_none() if existing_upload: logger.info(f"Upload with {content_hash=} already exists") return existing_upload logger.info(f"Creating new Upload with {content_hash=}") dest_filename = UPLOAD_DIR / content_hash has_thumbnail = False image_blurhash = None width = None height = None if f.content_type.startswith("image") and not f.content_type == "image/gif": with Image.open(f.file) as _original_image: # Fix image orientation (as we will remove the info from the EXIF # metadata) original_image = ImageOps.exif_transpose(_original_image) # Re-creating the image drop the EXIF metadata destination_image = Image.new( original_image.mode, original_image.size, ) destination_image.putdata(original_image.getdata()) destination_image.save( dest_filename, format=_original_image.format, ) with open(dest_filename, "rb") as dest_f: image_blurhash = blurhash.encode(dest_f, x_components=4, y_components=3) try: width, height = destination_image.size destination_image.thumbnail((740, 740)) destination_image.save( UPLOAD_DIR / f"{content_hash}_resized", format="webp", ) except Exception: logger.exception( f"Failed to created thumbnail for {f.filename}/{content_hash}" ) else: has_thumbnail = True logger.info("Thumbnail generated") else: with open(dest_filename, "wb") as dest: while True: buf = f.file.read(COPY_BUFSIZE) if not buf: break dest.write(buf) new_upload = models.Upload( content_type=f.content_type, content_hash=content_hash, has_thumbnail=has_thumbnail, blurhash=image_blurhash, width=width, height=height, ) db_session.add(new_upload) await db_session.commit() return new_upload def upload_to_attachment( upload: models.Upload, filename: str, alt_text: str | None, ) -> ap.RawObject: extra_attachment_fields = {} if upload.blurhash: extra_attachment_fields.update( { "blurhash": upload.blurhash, "height": upload.height, "width": upload.width, } ) return { "type": "Document", "mediaType": upload.content_type, "name": alt_text or filename, "url": BASE_URL + f"/attachments/{upload.content_hash}/{filename}", **extra_attachment_fields, }