takahe/core/files.py

import io

import blurhash
import httpx
from django.conf import settings
from django.core.files import File
from django.core.files.base import ContentFile
from PIL import Image, ImageOps


def resize_image(
    image: File,
    *,
    size: tuple[int, int],
    cover=True,
    keep_format=False,
) -> File:
    """
    Resizes an image to fit insize the given size (cropping one dimension
    to fit if needed)
    """
    with Image.open(image) as img:
        if cover:
            resized_image = ImageOps.fit(img, size)
        else:
            resized_image = ImageOps.contain(img, size)
        new_image_bytes = io.BytesIO()
        if keep_format:
            resized_image.save(new_image_bytes, format=image.format)
            file = File(new_image_bytes)
        else:
            resized_image.save(new_image_bytes, format="webp")
            file = File(new_image_bytes, name="image.webp")
        file.image = resized_image
        return file


def blurhash_image(file) -> str:
    """
    Returns the blurhash for an image
    """
    return blurhash.encode(file, 4, 4)


async def get_remote_file(
    url: str,
    *,
    timeout: float = settings.SETUP.REMOTE_TIMEOUT,
    max_size: int | None = None,
) -> tuple[File | None, str | None]:
    """
    Download a URL and return the File and content-type.
    """
    async with httpx.AsyncClient() as client:
        async with client.stream("GET", url, timeout=timeout) as stream:
            allow_download = max_size is None
            if max_size:
                try:
                    content_length = int(stream.headers["content-length"])
                    allow_download = content_length <= max_size
                except TypeError:
                    pass
            if allow_download:
                file = ContentFile(await stream.aread(), name=url)
                return file, stream.headers["content-type"]

    return None, None