ad/molodetz/attachments.py

# retoor <retoor@molodetz.nl>
from pathlib import Path

from molodetz import config
from molodetz.database import db, shard_path

IMAGE_EXTENSIONS = frozenset({"jpg", "jpeg", "png", "gif", "webp", "avif"})
DOCUMENT_EXTENSIONS = frozenset({"pdf", "md", "txt"})
INLINE_EXTENSIONS = IMAGE_EXTENSIONS | frozenset({"mp4", "webm", "mp3", "ogg", "wav", "svg"})


def allowed_extensions():
    return IMAGE_EXTENSIONS | DOCUMENT_EXTENSIONS


def extension_of(name):
    return Path(name).suffix.lower().lstrip(".")


def is_image(name):
    return extension_of(name) in IMAGE_EXTENSIONS


def format_size(size):
    size = float(size or 0)
    for unit in ("B", "KB", "MB", "GB"):
        if size < 1024 or unit == "GB":
            return f"{size:.0f} {unit}" if unit == "B" else f"{size:.1f} {unit}"
        size /= 1024
    return f"{size:.1f} GB"


def icon_for(name):
    ext = extension_of(name)
    if ext in IMAGE_EXTENSIONS:
        return "image"
    if ext == "pdf":
        return "pdf"
    return "file"


def blob_location(uid, extension):
    directory = f"attachments/{shard_path(uid)}"
    stored_name = f"{uid.replace('-', '')}.{extension}"
    return directory, stored_name


def blob_path(directory, stored_name):
    root = config.DATA_PATHS["uploads"].resolve()
    path = (root / directory / stored_name).resolve()
    if root not in path.parents:
        raise ValueError("blob path escapes the uploads root")
    return path


def blob_url(directory, stored_name):
    return f"/static/uploads/{directory}/{stored_name}"


def write_blob(uid, extension, data):
    if extension not in allowed_extensions():
        raise ValueError(f"extension {extension} not allowed")
    directory, stored_name = blob_location(uid, extension)
    path = blob_path(directory, stored_name)
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_bytes(data)
    return directory, stored_name


def unlink_blob(directory, stored_name):
    try:
        blob_path(directory, stored_name).unlink(missing_ok=True)
        return True
    except (OSError, ValueError):
        return False


def referenced_blob_paths():
    paths = set()
    if "media_items" in db.tables:
        for row in db["media_items"].find(deleted_at=None):
            if row.get("thumb_directory") and row.get("thumb_name"):
                paths.add(blob_path(row["thumb_directory"], row["thumb_name"]))
    return paths


def orphan_blobs():
    root = config.DATA_PATHS["attachments"]
    if not root.exists():
        return []
    referenced = referenced_blob_paths()
    return [path for path in root.rglob("*") if path.is_file() and path.resolve() not in referenced]