93 lines
2.6 KiB
Python
93 lines
2.6 KiB
Python
# 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]
|