# retoor 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]