public_git/bambuddy
1
0
Fork 0
mirror of https://github.com/maziggy/bambuddy.git synced 2026-10-02 20:22:15 +02:00
Code Issues Packages Projects Releases Wiki Activity
Files
dev
bambuddy/backend/app/api/routes/library.py
T

6086 lines
267 KiB
Python
Raw Permalink Blame History

"""API routes for File Manager (Library) functionality."""
import asyncio
import base64
import binascii
import contextlib
import hashlib
import json
import logging
import os
import re
import shutil
import uuid
import zipfile
from datetime import datetime, timezone
from pathlib import Path
from fastapi import APIRouter, Depends, File, HTTPException, Query, Response, UploadFile
from fastapi.responses import FileResponse as FastAPIFileResponse
from sqlalchemy import distinct, func, select, update
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import selectinload
from backend.app.api.routes.cloud import resolve_api_key_cloud_owner
from backend.app.api.routes.library_variants import normalize_model_name, resolve_variant_model
from backend.app.api.routes.print_queue import _extract_filament_types_from_3mf
from backend.app.core.auth import (
require_media_token_ownership,
require_ownership_permission,
require_permission_if_auth_enabled,
)
from backend.app.core.config import settings as app_settings
from backend.app.core.database import async_session, get_db
from backend.app.core.permissions import Permission
from backend.app.core.tasks import spawn_background_task
from backend.app.models.archive import PrintArchive
from backend.app.models.library import LibraryFile, LibraryFileTag, LibraryFolder
from backend.app.models.print_queue import PrintQueueItem
from backend.app.models.printer import Printer
from backend.app.models.project import Project
from backend.app.models.user import User
from backend.app.schemas.library import (
AddToQueueError,
AddToQueueRequest,
AddToQueueResponse,
AddToQueueResult,
BatchThumbnailRequest,
BatchThumbnailResponse,
BatchThumbnailResult,
BulkDeleteRequest,
BulkDeleteResponse,
ClientThumbnailResponse,
CombineFilesRequest,
ExternalFolderCreate,
FileDuplicate,
FileListResponse,
FileMoveRequest,
FileResponse as FileResponseSchema,
FileUpdate,
FileUploadResponse,
FolderCreate,
FolderReadmeResponse,
FolderResponse,
FolderTreeItem,
FolderUpdate,
TagSummary,
ZipExtractError,
ZipExtractResponse,
ZipExtractResult,
)
from backend.app.schemas.slicer import SliceRequest, SliceResponse
from backend.app.services.archive import ThreeMFParser
from backend.app.services.design_settings import (
DesignOverride,
apply_design_overrides,
extract_design_process_overrides,
overrides_from_config,
)
from backend.app.services.filament_requirements import annotate_rack_groups
from backend.app.services.pdf_thumbnail import generate_pdf_thumbnail
from backend.app.services.plate_thumbnail import inject_plate_thumbnails_if_missing
from backend.app.services.print_confirmation import confirm_outcome_for_new_queue_item
from backend.app.services.process_overrides import apply_process_overrides
from backend.app.services.slice_output_check import (
missing_start_gcode_message,
start_gcode_is_missing,
unresolved_filament_message,
unresolved_filament_slots,
)
from backend.app.services.stl_thumbnail import MIN_USABLE_STL_BYTES, generate_stl_thumbnail
from backend.app.utils.filename import (
MAX_FILENAME_BYTES,
InvalidFilenameError,
safe_path_component,
validate_print_filename,
)
from backend.app.utils.library_paths import library_photos_dir, remove_library_photos_dir
from backend.app.utils.printer_models import is_gcode_compatible
from backend.app.utils.safe_path import PathTraversalError, assert_under, safe_join_under
from backend.app.utils.threemf_tools import (
carries_gcode,
default_plate_gcode_name,
expand_to_project_slots,
extract_embedded_presets_from_3mf,
extract_nozzle_mapping_from_3mf,
extract_project_filaments_from_3mf,
sanitize_project_settings_sentinels,
select_plate_gcode_name,
supports_enabled_in_config,
)
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/library", tags=["library"])
# Path of the embedded slicer config inside a BambuStudio/OrcaSlicer 3MF.
_PROJECT_SETTINGS_PATH = "Metadata/project_settings.config"
def _ensure_library_file_visible(
library_file: LibraryFile | None,
user: User | None,
can_read_all: bool,
) -> LibraryFile:
"""Per-file visibility gate for ownership-scoped LIBRARY reads (#1726-adjacent).
Mirrors archives.py::_ensure_archive_visible — single enforcement point so a
less-guarded sibling route can't accidentally leak a row. Same shape:
- Missing / soft-deleted → 404 (not 403, to avoid id-enumeration leaks).
- ``can_read_all`` true (LIBRARY_READ_ALL or auth disabled) → file returned.
- ``can_read_all`` false and ``created_by_id != user.id`` → 404.
- Ownerless files (``created_by_id is None``) require ALL — fail-closed.
"""
if library_file is None or getattr(library_file, "deleted_at", None) is not None:
raise HTTPException(404, "File not found")
if can_read_all:
return library_file
if user is None:
raise HTTPException(404, "File not found")
if library_file.created_by_id is None or library_file.created_by_id != user.id:
raise HTTPException(404, "File not found")
return library_file
def get_library_dir() -> Path:
"""Get the library storage directory."""
base_dir = Path(app_settings.archive_dir)
library_dir = base_dir / "library"
library_dir.mkdir(parents=True, exist_ok=True)
return library_dir
def get_library_files_dir() -> Path:
"""Get the directory for library files."""
files_dir = get_library_dir() / "files"
files_dir.mkdir(parents=True, exist_ok=True)
return files_dir
def classify_file_type(filename: str, file_path: Path | str | None = None) -> str:
"""Return the canonical ``LibraryFile.file_type`` for *filename*.
Compound extensions are preserved — a `.gcode.3mf` file (a sliced
output, still a 3MF zip on disk) is classified ``gcode.3mf`` rather
than ``3mf``. Pre-#1600 this was only done in the external-scan
path; the upload / ZIP-extract / in-process paths all stripped to
the trailing extension and stored ``3mf``, so the FE had to accept
both. Unified here so every ingest path stores the same value and
downstream gates (gcode download, file-type filter, thumbnail
extraction) only need to handle one canonical name per file family.
Files with no extension classify as ``unknown``.
Pass ``file_path`` and a ``.3mf`` is judged on what the zip actually holds
rather than on its name (#2993). The name is not evidence: a plate exported
from Studio or a print dispatched through the cloud reaches the archive as
``Foo.3mf``, G-code and all, and downloading that and re-importing it used
to land a fully printable file in the library as a source-only project. The
file is not opened when the name already settles it, so the common case
still costs nothing.
"""
lower = filename.lower()
if lower.endswith(".gcode.3mf"):
return "gcode.3mf"
ext = os.path.splitext(lower)[1]
file_type = ext[1:] if ext else "unknown"
if file_type == "3mf" and file_path is not None and carries_gcode(file_path):
return "gcode.3mf"
return file_type
def get_library_thumbnails_dir() -> Path:
"""Get the directory for library thumbnails."""
thumbnails_dir = get_library_dir() / "thumbnails"
thumbnails_dir.mkdir(parents=True, exist_ok=True)
return thumbnails_dir
def to_relative_path(absolute_path: Path | str) -> str:
"""Convert an absolute path to a path relative to base_dir for storage."""
if not absolute_path:
return ""
abs_path = Path(absolute_path)
base_dir = Path(app_settings.base_dir)
try:
return str(abs_path.relative_to(base_dir))
except ValueError:
# Path is not under base_dir, return as-is (shouldn't happen normally)
return str(abs_path)
def to_absolute_path(relative_path: str | None) -> Path | None:
"""Convert a relative path (from database) to an absolute path for file operations."""
if not relative_path:
return None
path = Path(relative_path)
# Handle already-absolute paths verbatim (backwards compatibility during migration).
# Legacy DB rows may store absolute paths that predate the base_dir layout; the
# traversal guard below only applies to relative paths coming from user input.
if path.is_absolute():
return path.resolve()
base = Path(app_settings.base_dir).resolve()
resolved = (base / relative_path).resolve()
# Guard against path traversal — resolved path must stay inside base_dir.
# Use is_relative_to() to avoid the /data/app vs /data/app_evil prefix confusion
# that a plain startswith(str(base)) check would miss.
if not resolved.is_relative_to(base):
raise ValueError(f"Path escapes base directory: {relative_path!r}")
return resolved
def calculate_file_hash(file_path: Path) -> str:
"""Calculate SHA256 hash of a file."""
sha256_hash = hashlib.sha256()
with open(file_path, "rb") as f:
for byte_block in iter(lambda: f.read(4096), b""):
sha256_hash.update(byte_block)
return sha256_hash.hexdigest()
def validate_print_file_upload(filename: str, content: bytes) -> None:
"""Reject obviously-unprintable uploads early so the printer doesn't see them (#1401).
Bambu printers in network mode only parse ``.gcode.3mf`` zip containers
— raw ``.gcode`` and corrupt/non-zip ``.3mf`` uploads cascade into a
confusing "Printing stopped because the printer was unable to parse the
3mf file" rejection 30 seconds after the user clicks Print. The
the queue dispatch path appends ``.3mf`` to a raw-gcode filename when
constructing the FTP destination, which is how the printer ends up with a
file named ``.gcode.3mf`` whose body is raw gcode — exactly the shape that
triggers the firmware parse failure. Catching both classes here gives an
actionable error at the
upload itself.
Compares the filename suffix rather than ``os.path.splitext`` because
compound extensions like ``.gcode.3mf`` show up as just ``.3mf`` after
``splitext`` — same content validation needs to fire for both
single-``.3mf`` and ``.gcode.3mf`` uploads.
Raises ``HTTPException(400, ...)`` with a human-readable message on
rejection; returns ``None`` for valid (or irrelevant — e.g. STL,
image) uploads.
"""
lower_filename = filename.lower()
is_3mf_upload = lower_filename.endswith(".3mf")
is_raw_gcode_upload = lower_filename.endswith(".gcode") and not lower_filename.endswith(".gcode.3mf")
if is_raw_gcode_upload:
raise HTTPException(
status_code=400,
detail=(
"Raw .gcode files can't be printed on Bambu printers in network mode — "
"they need a .gcode.3mf zip container (gcode plus metadata). Re-export from "
"your slicer and make sure the file ends in '.gcode.3mf', not just '.gcode'. "
"If your OS hides extensions, double-check the file with the extension visible."
),
)
if is_3mf_upload and not content.startswith(b"PK\x03\x04"):
raise HTTPException(
status_code=400,
detail=(
"This .3mf file isn't a valid ZIP container. 3MF files are ZIP archives — "
"either the file is corrupted or it's raw gcode renamed to .3mf. Re-export "
"from your slicer using its 'Export Plate Sliced File' action."
),
)
def _resolve_upload_destination(target_folder: LibraryFolder | None, filename: str) -> tuple[Path, bool]:
"""Resolve the on-disk destination for an uploaded file.
Non-external target: returns ``(<library_files_dir>/<uuid><ext>, False)``.
Writable external target: writes to ``<external_path>/<filename>``
(preserves the real filename so the file is recognisable on the mount);
returns ``(dest, True)``. Raises ``HTTPException`` for read-only external
folders (403), missing/inaccessible/non-writable external paths (400), and
filename collisions on the external mount (409). See #1112 — previously
uploads to writable external folders were silently misrouted to the
internal library dir.
"""
if target_folder is not None and target_folder.is_external:
if target_folder.external_readonly:
raise HTTPException(status_code=403, detail="Cannot upload to a read-only external folder")
if not target_folder.external_path:
raise HTTPException(status_code=400, detail="External folder has no configured path")
ext_dir = Path(target_folder.external_path)
if not ext_dir.exists() or not ext_dir.is_dir():
raise HTTPException(
status_code=400,
detail=f"External path is not accessible: {target_folder.external_path}",
)
if not os.access(ext_dir, os.W_OK):
raise HTTPException(
status_code=400,
detail=f"External path is not writable: {target_folder.external_path}",
)
# Guard against path-traversal via a pathological filename — join then
# verify the resolved destination is still inside the external dir.
dest = (ext_dir / filename).resolve() # SEC-PATH-OK: resolve + relative_to containment check on next line
try:
dest.relative_to(ext_dir.resolve())
except ValueError:
raise HTTPException(status_code=400, detail="Invalid filename")
if dest.exists():
raise HTTPException(
status_code=409,
detail=f"A file named {filename!r} already exists in the external folder",
)
return dest, True
ext = os.path.splitext(filename)[1].lower()
return get_library_files_dir() / f"{uuid.uuid4().hex}{ext}", False
def _unique_external_name(ext_dir: Path, filename: str) -> str:
"""Return ``filename``, or the first free ``<stem> (n)<suffix>`` variant.
Splits on the *compound* extension so re-slicing ``Bidoof.3mf`` yields
``Bidoof (2).gcode.3mf`` rather than ``Bidoof.gcode (2).3mf``.
Uploads answer a name collision with a 409, which is right for a file the
user just chose to send. A slice is not that: re-slicing the same source
with different settings is routine, and the second run has already spent
minutes of CPU by the time the name is known -- refusing to store it would
throw that away. Overwriting is worse still, since the target is somebody's
NAS and the file being replaced may not even be ours.
"""
stem = filename[: -len(".gcode.3mf")] if filename.endswith(".gcode.3mf") else Path(filename).stem
suffix = ".gcode.3mf" if filename.endswith(".gcode.3mf") else Path(filename).suffix
candidate = filename
counter = 2
# Bounded: a directory holding 999 re-slices of one model is pathological,
# and an unbounded loop here would hang the request on a mount that lies
# about exists() (some SMB shares do under contention).
#
# safe_join_under rather than `ext_dir / candidate`: `filename` derives
# from a name read out of a 3MF, so the very first probe must not be able
# to stat its way outside the mount. It raises PathTraversalError, which
# the caller turns into a managed-storage fallback.
while safe_join_under(ext_dir, candidate, http=False).exists() and counter < 1000:
candidate = f"{stem} ({counter}){suffix}"
counter += 1
return candidate
def _resolve_slice_destination(target_folder: LibraryFolder | None, out_filename: str) -> tuple[Path, bool, str | None]:
"""Resolve where a slice result should be written.
Returns ``(path, is_external, fallback_reason)``. ``fallback_reason`` is
``None`` on the normal paths and otherwise names why an external folder
could not receive the file, so the caller can tell the user instead of
quietly filing it elsewhere.
Slicing a file that lives on an external mount used to store the output in
the managed library dir unconditionally, while giving the new row the
external folder's ``folder_id`` (#2810). The file therefore appeared in the
right folder in the UI and never arrived on the share, which is the one
place the user was looking -- and made it un-reproducible from the web UI
alone. Uploads learned this in #1112 (``_resolve_upload_destination``) and
moves in its follow-up (``_move_file_bytes``); slicing was the last write
path still assuming managed storage.
Unlike uploads, a failure here does not raise. The bytes exist and cost
real time to produce, so an unwritable target falls back to managed storage
with a reason attached rather than discarding the slice.
"""
if target_folder is None or not target_folder.is_external:
return get_library_files_dir() / f"{uuid.uuid4().hex}.gcode.3mf", False, None
if target_folder.external_readonly:
return get_library_files_dir() / f"{uuid.uuid4().hex}.gcode.3mf", False, "external_readonly"
if not target_folder.external_path:
return get_library_files_dir() / f"{uuid.uuid4().hex}.gcode.3mf", False, "external_no_path"
ext_dir = Path(target_folder.external_path)
if not ext_dir.exists() or not ext_dir.is_dir():
return get_library_files_dir() / f"{uuid.uuid4().hex}.gcode.3mf", False, "external_unreachable"
if not os.access(ext_dir, os.W_OK):
return get_library_files_dir() / f"{uuid.uuid4().hex}.gcode.3mf", False, "external_not_writable"
try:
dest = safe_join_under(ext_dir, _unique_external_name(ext_dir, out_filename), http=False)
except PathTraversalError:
# The source filename reached us from a 3MF on disk, so this is
# defensive rather than expected -- but a name that escapes the mount
# must land in managed storage, never outside it.
return get_library_files_dir() / f"{uuid.uuid4().hex}.gcode.3mf", False, "external_invalid_name"
return dest, True, None
async def _folder_tree_file_ids(db: AsyncSession, folder_id: int) -> list[int]:
"""Every ``LibraryFile`` id under ``folder_id``, at any depth.
Deleting a folder cascades to its whole subtree, so anything that has to be
released before that delete (queue items, cross-model candidates) needs the
subtree, not just the folder's own files.
Trashed rows are included deliberately: they are still real rows and the
cascade takes them too.
"""
file_ids: list[int] = []
pending = [folder_id]
# The API refuses to make a folder its own ancestor, so a loop here would
# mean the table is already corrupt -- but this walk runs inside a delete
# request, and hanging one is worse than the cost of a set.
seen: set[int] = set()
while pending:
current = pending.pop()
if current in seen:
continue
seen.add(current)
file_ids.extend(
(await db.execute(select(LibraryFile.id).where(LibraryFile.folder_id == current))).scalars().all()
)
pending.extend(
(await db.execute(select(LibraryFolder.id).where(LibraryFolder.parent_id == current))).scalars().all()
)
return file_ids
def _stored_file_path(abs_path: Path, is_external: bool) -> str:
"""Produce the value to persist in ``LibraryFile.file_path``.
External files store the absolute mount path directly (same as scan does),
so ``to_absolute_path`` round-trips through its ``is_absolute()`` fast
path. Managed files store a path relative to ``base_dir`` for portability.
"""
return str(abs_path) if is_external else to_relative_path(abs_path)
class _MoveSkip(Exception):
"""Signalled by ``_move_file_bytes`` to skip a file with a user-visible reason.
Carries an optional `code` for machine-friendly grouping (the
front-end can localise it) and a fallback English `reason` for logs.
"""
def __init__(self, code: str, reason: str):
super().__init__(reason)
self.code = code
self.reason = reason
def _resolve_source_disk_path(file: LibraryFile) -> Path | None:
"""Return the absolute on-disk path for an existing LibraryFile, or None
if it can't be located (legacy DB row, deleted file, etc.)."""
if file.is_external:
return Path(file.file_path) if file.file_path else None
return to_absolute_path(file.file_path)
def _move_file_bytes(file: LibraryFile, target_folder: LibraryFolder | None) -> str:
"""Physically relocate `file`'s bytes to match `target_folder`.
Used by the move endpoint when source/target straddle the
managed↔external boundary (#1112 follow-up — the prior implementation
updated the DB row's ``folder_id`` but never moved the bytes, so a
file moved to an external SMB folder showed up in Bambuddy's UI but
not on the NAS).
Returns the new ``file_path`` value to persist (relative for managed
targets, absolute for external targets — matches the upload + scan
paths). Raises ``_MoveSkip`` for any condition that would make the
move unsafe (target unwritable, filename collision, source missing).
The copy-then-unlink ordering means a partial copy followed by a
failed unlink leaves both the source and the dest on disk — better
than the symmetric "rename or move" which would lose the source if
the target write didn't complete on a flaky mount. The DB row stays
pointed at the source until the caller commits the new ``file_path``.
"""
src = _resolve_source_disk_path(file)
if not src or not src.exists():
raise _MoveSkip("source_missing", "source file missing on disk")
target_is_external = target_folder is not None and target_folder.is_external
if target_is_external:
if target_folder.external_readonly:
# Already blocked at top level, but defence-in-depth.
raise _MoveSkip("target_readonly", "target external folder is read-only")
if not target_folder.external_path:
raise _MoveSkip("target_misconfigured", "target external folder has no path")
ext_dir = Path(target_folder.external_path)
if not ext_dir.exists() or not ext_dir.is_dir():
raise _MoveSkip("target_inaccessible", f"target path not accessible: {ext_dir}")
if not os.access(ext_dir, os.W_OK):
raise _MoveSkip("target_unwritable", f"target path not writable: {ext_dir}")
dest = (ext_dir / file.filename).resolve() # SEC-PATH-OK: resolve + relative_to containment check on next line
try:
dest.relative_to(ext_dir.resolve())
except ValueError:
raise _MoveSkip("invalid_filename", f"unsafe filename: {file.filename!r}") from None
if dest.exists():
raise _MoveSkip("name_collision", f"a file named {file.filename!r} already exists in target")
try:
shutil.copy2(src, dest)
except OSError as e:
# Clean up partial dest so a retry can succeed.
with contextlib.suppress(OSError):
dest.unlink(missing_ok=True)
raise _MoveSkip("copy_failed", f"copy failed: {e}") from e
else:
# → managed (root or non-external folder): generate a fresh UUID
# filename in the internal store so we don't collide with another
# file that happens to share `filename`.
ext = src.suffix.lower()
dest = get_library_files_dir() / f"{uuid.uuid4().hex}{ext}"
try:
shutil.copy2(src, dest)
except OSError as e:
with contextlib.suppress(OSError):
dest.unlink(missing_ok=True)
raise _MoveSkip("copy_failed", f"copy failed: {e}") from e
# Copy succeeded — unlink the original. A failure here leaves an
# orphan on disk but the DB row is consistent against the new dest.
try:
src.unlink(missing_ok=True)
except OSError as e:
logger.warning(
"Move: copied %s → %s but couldn't remove source: %s",
src,
dest,
e,
)
return _stored_file_path(dest, is_external=target_is_external)
def _clean_3mf_metadata(obj):
"""Strip bytes and thumbnail-carrier keys so the payload is JSON-storable.
Shared by ``upload_file`` and :func:`save_3mf_bytes_to_library` — the
``ThreeMFParser`` output embeds the thumbnail bytes under
``_thumbnail_data``/``_thumbnail_ext`` and may also include raw bytes in
other fields, none of which can be JSON-encoded.
"""
if isinstance(obj, dict):
return {
k: _clean_3mf_metadata(v)
for k, v in obj.items()
if not isinstance(v, bytes) and k not in ("_thumbnail_data", "_thumbnail_ext")
}
if isinstance(obj, list):
return [_clean_3mf_metadata(i) for i in obj if not isinstance(i, bytes)]
if isinstance(obj, bytes):
return None
return obj
def _read_3mf_entry(zip_path: Path, entry: str) -> bytes | None:
"""Return the raw bytes of an entry inside a 3MF (ZIP), or ``None`` when
the file isn't a parseable zip / doesn't contain that entry / any IO
error. Used to lift the source archive's per-plate render onto a
re-sliced archive (#1493 follow-up) — the slicer CLI often doesn't
emit a fresh ``Metadata/plate_N.png`` and the project-wide cover-art
fallback in :class:`ThreeMFParser` looks unrelated to the actual slice.
"""
try:
with zipfile.ZipFile(zip_path, "r") as zf:
if entry not in zf.namelist():
return None
return zf.read(entry)
except (zipfile.BadZipFile, OSError, KeyError):
return None
def _without_print_name(metadata: dict | None) -> dict | None:
"""Drop the embedded 3MF Title (``print_name``) from library-file metadata.
The 3MF ``<metadata name="Title">`` holds the in-app project title — the
generic ``"Exported 3D Model"`` for a Bambu Studio "Save As", a marketing
title for a MakerWorld download — never the filename the user saved as.
The FileManager keys its display name, search and sort off ``print_name``,
so storing it makes every card show the wrong name (#1489). A library
file's display name is its filename; only ``PrintArchive`` carries a real
``print_name``. Returns the input unchanged when there's nothing to strip;
otherwise a new dict (never mutates the argument).
"""
if not metadata or "print_name" not in metadata:
return metadata
return {k: v for k, v in metadata.items() if k != "print_name"}
async def save_3mf_bytes_to_library(
db: AsyncSession,
*,
file_bytes: bytes,
filename: str,
folder_id: int | None = None,
source_type: str | None = None,
source_url: str | None = None,
owner_id: int | None = None,
) -> tuple[LibraryFile, bool]:
"""Save a 3MF blob into the library and return ``(library_file, was_existing)``.
Used by routes that receive a 3MF in-process rather than as a multipart
upload (currently: MakerWorld import; reusable for any future source that
fetches bytes server-side). Deduplicates by ``source_url`` when provided —
if a LibraryFile with the same source_url already exists, the existing
row is returned and the bytes are NOT re-saved (MakerWorld signed URLs
change each download, so hash-based dedupe alone would miss re-imports).
Parses 3MF metadata + thumbnail the same way the multipart upload route
does, via :class:`ThreeMFParser`. Paths are stored as relative so the
library is portable across installs.
"""
# Source-URL-based dedupe: return the existing row untouched.
if source_url:
existing = await db.execute(LibraryFile.active().where(LibraryFile.source_url == source_url).limit(1))
existing_row = existing.scalar_one_or_none()
if existing_row is not None:
return existing_row, True
# Resolve target folder so writable-external destinations land on the
# mount with the real filename, instead of being silently misrouted to
# the internal library dir with a UUID name (#1645). Mirrors what the
# multipart-upload path has done since #1112. ``_resolve_upload_destination``
# also enforces the 403 read-only / 400 unwritable / 409 collision
# rejections — the makerworld route layer already pre-checks read-only,
# but the helper's checks remain as defence-in-depth for any future
# caller that skips that route gate.
target_folder: LibraryFolder | None = None
if folder_id is not None:
folder_q = await db.execute(select(LibraryFolder).where(LibraryFolder.id == folder_id))
target_folder = folder_q.scalar_one_or_none()
file_path, is_external = _resolve_upload_destination(target_folder, filename)
ext = file_path.suffix.lower() or ".3mf"
with open(file_path, "wb") as fh:
fh.write(file_bytes)
file_hash = calculate_file_hash(file_path)
# Extract metadata + thumbnail from the 3MF.
metadata: dict | None = None
thumbnail_path: str | None = None
if ext == ".3mf":
try:
# Off the event loop: the parser still decompresses the whole model
# entry, which for a combined plate is hundreds of MB (#3162).
parser = ThreeMFParser(str(file_path))
raw_metadata = await asyncio.to_thread(parser.parse)
thumb_data = raw_metadata.get("_thumbnail_data")
thumb_ext = raw_metadata.get("_thumbnail_ext", ".png")
if thumb_data:
thumbs_dir = get_library_thumbnails_dir()
thumb_filename = f"{uuid.uuid4().hex}{thumb_ext}"
thumb_path = thumbs_dir / thumb_filename # SEC-PATH-OK: thumb_filename = uuid.uuid4().hex + thumb_ext
with open(thumb_path, "wb") as fh:
fh.write(thumb_data)
thumbnail_path = str(thumb_path)
metadata = _clean_3mf_metadata(raw_metadata) or None
except Exception as exc:
# Matches the multipart upload route's behaviour — a bad 3MF should
# still land in the library so the user can see / delete it rather
# than failing the whole request.
logger.warning("Failed to parse 3MF %s: %s", filename, exc)
library_file = LibraryFile(
folder_id=folder_id,
is_external=is_external,
filename=filename,
file_path=_stored_file_path(file_path, is_external),
file_type=classify_file_type(filename, file_path),
file_size=len(file_bytes),
file_hash=file_hash,
thumbnail_path=to_relative_path(thumbnail_path) if thumbnail_path else None,
file_metadata=_without_print_name(metadata),
source_type=source_type,
source_url=source_url,
created_by_id=owner_id,
)
db.add(library_file)
await db.commit()
await db.refresh(library_file)
return library_file, False
def extract_gcode_thumbnail(file_path: Path) -> bytes | None:
"""Extract embedded thumbnail from gcode file.
Supports PrusaSlicer/BambuStudio format:
; thumbnail begin WxH SIZE
; base64data...
; thumbnail end
"""
try:
thumbnail_data = None
in_thumbnail = False
thumbnail_lines = []
best_size = 0
with open(file_path, errors="ignore") as f:
# Only read first 50KB for performance (thumbnails are at the start)
content = f.read(50000)
for line in content.split("\n"):
line = line.strip()
# Check for thumbnail start
if line.startswith("; thumbnail begin"):
in_thumbnail = True
thumbnail_lines = []
# Parse dimensions: "; thumbnail begin 300x300 12345"
match = re.search(r"(\d+)x(\d+)", line)
if match:
width = int(match.group(1))
# Prefer larger thumbnails (up to 300px)
if width > best_size and width <= 300:
best_size = width
continue
# Check for thumbnail end
if line.startswith("; thumbnail end"):
if in_thumbnail and thumbnail_lines:
try:
# Decode the base64 data
b64_data = "".join(thumbnail_lines)
decoded = base64.b64decode(b64_data)
# Only keep if this is the best size or first valid thumbnail
if thumbnail_data is None or best_size > 0:
thumbnail_data = decoded
except (binascii.Error, ValueError):
pass # Skip thumbnail with invalid base64 data
in_thumbnail = False
thumbnail_lines = []
continue
# Collect thumbnail data
if in_thumbnail and line.startswith(";"):
# Remove the leading "; " or ";"
data_line = line[1:].strip()
if data_line:
thumbnail_lines.append(data_line)
return thumbnail_data
except Exception as e:
logger.warning("Failed to extract gcode thumbnail: %s", e)
return None
def create_image_thumbnail(file_path: Path, thumbnails_dir: Path, max_size: int = 256) -> str | None:
"""Create a thumbnail from an image file.
For small images, copies directly. For larger images, resizes.
Returns the thumbnail path or None on failure.
"""
try:
from PIL import Image
thumb_filename = f"{uuid.uuid4().hex}.png"
thumb_path = thumbnails_dir / thumb_filename # SEC-PATH-OK: thumb_filename = uuid.uuid4().hex + ".png"
with Image.open(file_path) as img:
# Convert to RGB if necessary (for PNG with transparency, etc.)
if img.mode in ("RGBA", "LA", "P"):
# Create white background for transparency
background = Image.new("RGB", img.size, (255, 255, 255))
if img.mode == "P":
img = img.convert("RGBA")
background.paste(img, mask=img.split()[-1] if img.mode == "RGBA" else None)
img = background
elif img.mode != "RGB":
img = img.convert("RGB")
# Resize if larger than max_size
if img.width > max_size or img.height > max_size:
img.thumbnail((max_size, max_size), Image.Resampling.LANCZOS)
img.save(thumb_path, "PNG", optimize=True)
return str(thumb_path)
except ImportError:
# PIL not installed, just copy the file if it's small enough
logger.warning("PIL not installed, copying image as thumbnail")
try:
file_size = file_path.stat().st_size
if file_size < 500000: # Less than 500KB
thumb_filename = f"{uuid.uuid4().hex}{file_path.suffix}"
thumb_path = (
thumbnails_dir / thumb_filename
) # SEC-PATH-OK: thumb_filename = uuid.uuid4().hex + file_path.suffix
shutil.copy2(file_path, thumb_path)
return str(thumb_path)
except OSError:
pass # File inaccessible; fall through to return None
return None
except Exception as e:
logger.warning("Failed to create image thumbnail: %s", e)
return None
# Supported image extensions for thumbnails
IMAGE_EXTENSIONS = {".png", ".jpg", ".jpeg", ".gif", ".webp", ".bmp", ".tiff", ".tif"}
# File types whose thumbnails are rendered client-side and uploaded back
# (#2976). The server has no renderer for STEP (that would need OpenCascade)
# or the spreadsheet types, so the browser posts its first preview render to
# POST /files/{id}/preview-thumbnail instead. PDF is rendered server-side with
# PDFium when it lands and stays here for a PDF that renderer cannot read.
# Kept to exactly these types so the endpoint can never overwrite a
# server-generated STL/3MF/G-code/image thumbnail.
CLIENT_THUMBNAIL_TYPES = {"step", "stp", "pdf", "csv", "xlsx", "ods"}
# Photos of the printed result (#3077): same allowlist and naming as the
# archive photo routes. 10 MB is ample for a phone camera JPEG.
PHOTO_EXTENSIONS = (".jpg", ".jpeg", ".png", ".webp")
PHOTO_MEDIA_TYPES = {
".jpg": "image/jpeg",
".jpeg": "image/jpeg",
".png": "image/png",
".webp": "image/webp",
}
MAX_PHOTO_BYTES = 10 * 1024 * 1024
# Upper bound for an uploaded client-rendered thumbnail. The FE sends a
# 256px PNG (a few tens of KB); anything near this limit is not a thumbnail.
MAX_CLIENT_THUMBNAIL_BYTES = 2 * 1024 * 1024
# Upper bound on the *decoded* size, checked against the header before any
# pixels are allocated: a few-KB PNG can declare 12000x7000 and still be under
# PIL's own decompression-bomb limit, which would be ~340 MB of RGBA.
MAX_CLIENT_THUMBNAIL_EDGE = 2048
# What the endpoint stores. The grid renders at ~256px, so anything larger is
# downscaled rather than kept.
STORED_CLIENT_THUMBNAIL_EDGE = 512
async def _backfill_external_thumbnails(folder_ids: list[int]) -> None:
"""Generate STL and PDF thumbnails for an external folder tree in the background.
Spawned via ``asyncio.create_task`` from ``scan_external_folder`` so the
HTTP request can return as soon as the filesystem walk + folder/file rows
are committed. Thumbnails for thousands of STL files would otherwise hold
the request open for many minutes (each file triggers a ``trimesh.load``
+ matplotlib render, ~1-5s each) and the FE modal times out before the
final ``db.commit()`` runs — causing the original symptom in #1299 where
subdirectories never showed up because nothing got committed. PDFs are
faster (a PDFium page render) but a share holding hundreds of them would
still hold the request open, so they are rendered here too.
Opens its own session because the request session is closed by the time
this task starts running. Commits per-file so a worker restart mid-run
only loses the in-flight file. Caps STL load to a single file at a time
to avoid memory pressure on systems with many huge STLs.
"""
if not folder_ids:
return
thumbnails_dir = get_library_thumbnails_dir()
async with async_session() as db:
result = await db.execute(
LibraryFile.active().where(
LibraryFile.folder_id.in_(folder_ids),
LibraryFile.file_type.in_(("stl", "pdf")),
LibraryFile.thumbnail_path.is_(None),
)
)
target_files = result.scalars().all()
if not target_files:
return
logger.info(
"Backfilling STL/PDF thumbnails: %d file(s) across %d folder(s)",
len(target_files),
len(folder_ids),
)
for target_file in target_files:
abs_path = to_absolute_path(target_file.file_path)
if not abs_path or not abs_path.exists():
continue
if target_file.file_type == "pdf":
# generate_pdf_thumbnail never raises; an unreadable PDF
# returns None and keeps the browser-preview fallback.
thumb_path = await asyncio.to_thread(generate_pdf_thumbnail, abs_path, thumbnails_dir)
if thumb_path:
target_file.thumbnail_path = to_relative_path(Path(thumb_path))
await db.commit()
continue
# Pre-skip files too small to contain even a single triangle.
# Bulk-uploaded ZIPs of stub STLs would otherwise trigger one
# trimesh.load() call + one debug log line per stub.
try:
if abs_path.stat().st_size < MIN_USABLE_STL_BYTES:
continue
except OSError:
continue
try:
thumb_path = generate_stl_thumbnail(abs_path, thumbnails_dir)
except Exception as exc: # noqa: BLE001 — never let one bad STL kill the rest
logger.debug("STL thumbnail backfill skipped %s: %s", abs_path, exc)
continue
if thumb_path:
target_file.thumbnail_path = to_relative_path(Path(thumb_path))
await db.commit()
# ============ Folder Endpoints ============
@router.get("/folders", response_model=list[FolderTreeItem])
@router.get("/folders/", response_model=list[FolderTreeItem])
async def list_folders(
response: Response,
db: AsyncSession = Depends(get_db),
_: tuple[User | None, bool] = Depends(
require_ownership_permission(
Permission.LIBRARY_READ_ALL,
Permission.LIBRARY_READ_OWN,
)
),
):
"""Get all folders as a tree structure."""
# Prevent browser caching of folder list
response.headers["Cache-Control"] = "no-cache, no-store, must-revalidate"
# Get all folders with project and archive joins
result = await db.execute(
select(LibraryFolder, Project.name, PrintArchive.print_name)
.outerjoin(Project, LibraryFolder.project_id == Project.id)
.outerjoin(PrintArchive, LibraryFolder.archive_id == PrintArchive.id)
.order_by(LibraryFolder.name)
)
rows = result.all()
# Get file counts per folder
file_counts_result = await db.execute(
select(LibraryFile.folder_id, func.count(LibraryFile.id))
.where(LibraryFile.folder_id.isnot(None), LibraryFile.deleted_at.is_(None))
.group_by(LibraryFile.folder_id)
)
file_counts = dict(file_counts_result.all())
# Latest immediate-child file activity per folder (#1770/#2680). Real on-disk
# mtime when we have it (external scans populate ``fs_modified_at``), else the
# DB ``updated_at`` — COALESCE so external rows scanned before this field
# existed, and internal uploads, still contribute a signal. This is the
# per-folder *leaf* value; subtree descent is aggregated recursively below.
latest_file_activity_result = await db.execute(
select(
LibraryFile.folder_id,
func.max(func.coalesce(LibraryFile.fs_modified_at, LibraryFile.updated_at)),
)
.where(LibraryFile.folder_id.isnot(None), LibraryFile.deleted_at.is_(None))
.group_by(LibraryFile.folder_id)
)
latest_file_activity = dict(latest_file_activity_result.all())
# Build tree structure. Each folder's initial ``latest_activity_at`` is its own
# leaf activity: the newer of its real directory mtime (fallback updated_at)
# and its immediate files' mtime. The recursive bubble below then rolls each
# subtree's newest descendant up to its ancestors (#2680 — sorting must match
# ``ls -t`` recursively, so a freshly-added deep file lifts every parent).
folder_map = {}
root_folders = []
for folder, project_name, archive_name in rows:
own_activity = folder.fs_modified_at or folder.updated_at
latest_file = latest_file_activity.get(folder.id)
if latest_file is not None and latest_file > own_activity:
own_activity = latest_file
folder_item = FolderTreeItem(
id=folder.id,
name=folder.name,
parent_id=folder.parent_id,
project_id=folder.project_id,
archive_id=folder.archive_id,
project_name=project_name,
archive_name=archive_name,
is_external=folder.is_external,
external_path=folder.external_path,
external_readonly=folder.external_readonly,
file_count=file_counts.get(folder.id, 0),
latest_activity_at=own_activity,
children=[],
)
folder_map[folder.id] = folder_item
# Link children to parents
for folder, _, _ in rows:
folder_item = folder_map[folder.id]
if folder.parent_id is None:
root_folders.append(folder_item)
elif folder.parent_id in folder_map:
folder_map[folder.parent_id].children.append(folder_item)
# Recursive newest-descendant bubble (#2680). Post-order: a folder's activity
# becomes the max of its own leaf activity and every descendant's, so sorting
# the tree by ``latest_activity_at`` surfaces the branch with the most recent
# activity anywhere inside it. Iterative stack keeps deep external mounts off
# Python's recursion limit.
def _bubble(root: FolderTreeItem) -> None:
order: list[FolderTreeItem] = []
stack = [root]
while stack:
node = stack.pop()
order.append(node)
stack.extend(node.children)
for node in reversed(order): # deepest first
for child in node.children:
if child.latest_activity_at is not None and (
node.latest_activity_at is None or child.latest_activity_at > node.latest_activity_at
):
node.latest_activity_at = child.latest_activity_at
for root in root_folders:
_bubble(root)
return root_folders
@router.get("/folders/by-project/{project_id}", response_model=list[FolderResponse])
async def get_folders_by_project(
project_id: int,
db: AsyncSession = Depends(get_db),
_: tuple[User | None, bool] = Depends(
require_ownership_permission(
Permission.LIBRARY_READ_ALL,
Permission.LIBRARY_READ_OWN,
)
),
):
"""Get all folders linked to a specific project."""
result = await db.execute(
select(LibraryFolder, Project.name)
.outerjoin(Project, LibraryFolder.project_id == Project.id)
.where(LibraryFolder.project_id == project_id)
.order_by(LibraryFolder.name)
)
rows = result.all()
folders = []
for folder, project_name in rows:
# Get file count + latest file activity (#1770/#2680) in one trip. Prefer
# the real on-disk mtime (external scans), fall back to the DB updated_at.
agg_result = await db.execute(
select(
func.count(LibraryFile.id),
func.max(func.coalesce(LibraryFile.fs_modified_at, LibraryFile.updated_at)),
).where(
LibraryFile.folder_id == folder.id,
LibraryFile.deleted_at.is_(None),
)
)
file_count, latest_file = agg_result.one()
file_count = file_count or 0
own_activity = folder.fs_modified_at or folder.updated_at
latest_activity_at = max(own_activity, latest_file) if latest_file is not None else own_activity
folders.append(
FolderResponse(
id=folder.id,
name=folder.name,
parent_id=folder.parent_id,
project_id=folder.project_id,
archive_id=folder.archive_id,
project_name=project_name,
archive_name=None,
is_external=folder.is_external,
external_path=folder.external_path,
external_readonly=folder.external_readonly,
external_show_hidden=folder.external_show_hidden,
file_count=file_count,
latest_activity_at=latest_activity_at,
created_at=folder.created_at,
updated_at=folder.updated_at,
)
)
return folders
@router.get("/folders/by-archive/{archive_id}", response_model=list[FolderResponse])
async def get_folders_by_archive(
archive_id: int,
db: AsyncSession = Depends(get_db),
_: tuple[User | None, bool] = Depends(
require_ownership_permission(
Permission.LIBRARY_READ_ALL,
Permission.LIBRARY_READ_OWN,
)
),
):
"""Get all folders linked to a specific archive."""
result = await db.execute(
select(LibraryFolder, PrintArchive.print_name)
.outerjoin(PrintArchive, LibraryFolder.archive_id == PrintArchive.id)
.where(LibraryFolder.archive_id == archive_id)
.order_by(LibraryFolder.name)
)
rows = result.all()
folders = []
for folder, archive_name in rows:
# Get file count + latest file activity (#1770/#2680) in one trip. Prefer
# the real on-disk mtime (external scans), fall back to the DB updated_at.
agg_result = await db.execute(
select(
func.count(LibraryFile.id),
func.max(func.coalesce(LibraryFile.fs_modified_at, LibraryFile.updated_at)),
).where(
LibraryFile.folder_id == folder.id,
LibraryFile.deleted_at.is_(None),
)
)
file_count, latest_file = agg_result.one()
file_count = file_count or 0
own_activity = folder.fs_modified_at or folder.updated_at
latest_activity_at = max(own_activity, latest_file) if latest_file is not None else own_activity
folders.append(
FolderResponse(
id=folder.id,
name=folder.name,
parent_id=folder.parent_id,
project_id=folder.project_id,
archive_id=folder.archive_id,
project_name=None,
archive_name=archive_name,
is_external=folder.is_external,
external_path=folder.external_path,
external_readonly=folder.external_readonly,
external_show_hidden=folder.external_show_hidden,
file_count=file_count,
latest_activity_at=latest_activity_at,
created_at=folder.created_at,
updated_at=folder.updated_at,
)
)
return folders
@router.post("/folders", response_model=FolderResponse)
@router.post("/folders/", response_model=FolderResponse)
async def create_folder(
data: FolderCreate,
db: AsyncSession = Depends(get_db),
_: User | None = Depends(require_permission_if_auth_enabled(Permission.LIBRARY_UPLOAD)),
):
"""Create a new folder."""
# Verify parent exists if specified
if data.parent_id is not None:
parent_result = await db.execute(select(LibraryFolder).where(LibraryFolder.id == data.parent_id))
if not parent_result.scalar_one_or_none():
raise HTTPException(status_code=404, detail="Parent folder not found")
# Verify project exists if specified
project_name = None
if data.project_id is not None:
project_result = await db.execute(select(Project).where(Project.id == data.project_id))
project = project_result.scalar_one_or_none()
if not project:
raise HTTPException(status_code=404, detail="Project not found")
project_name = project.name
# Verify archive exists if specified
archive_name = None
if data.archive_id is not None:
archive_result = await db.execute(select(PrintArchive).where(PrintArchive.id == data.archive_id))
archive = archive_result.scalar_one_or_none()
if not archive:
raise HTTPException(status_code=404, detail="Archive not found")
archive_name = archive.print_name
folder = LibraryFolder(
name=data.name,
parent_id=data.parent_id,
project_id=data.project_id,
archive_id=data.archive_id,
)
db.add(folder)
await db.commit()
await db.refresh(folder)
return FolderResponse(
id=folder.id,
name=folder.name,
parent_id=folder.parent_id,
project_id=folder.project_id,
archive_id=folder.archive_id,
project_name=project_name,
archive_name=archive_name,
is_external=folder.is_external,
external_path=folder.external_path,
external_readonly=folder.external_readonly,
external_show_hidden=folder.external_show_hidden,
file_count=0,
# New folder has no files yet — fall back to the folder's own
# updated_at so this matches the list-route semantics (#1770).
latest_activity_at=folder.updated_at,
created_at=folder.created_at,
updated_at=folder.updated_at,
)
@router.get("/folders/{folder_id}", response_model=FolderResponse)
async def get_folder(
folder_id: int,
db: AsyncSession = Depends(get_db),
_: tuple[User | None, bool] = Depends(
require_ownership_permission(
Permission.LIBRARY_READ_ALL,
Permission.LIBRARY_READ_OWN,
)
),
):
"""Get a folder by ID."""
result = await db.execute(
select(LibraryFolder, Project.name, PrintArchive.print_name)
.outerjoin(Project, LibraryFolder.project_id == Project.id)
.outerjoin(PrintArchive, LibraryFolder.archive_id == PrintArchive.id)
.where(LibraryFolder.id == folder_id)
)
row = result.one_or_none()
if not row:
raise HTTPException(status_code=404, detail="Folder not found")
folder, project_name, archive_name = row
# Get file count + latest file activity (#1770) in one trip
agg_result = await db.execute(
select(
func.count(LibraryFile.id),
func.max(LibraryFile.updated_at),
).where(
LibraryFile.folder_id == folder_id,
LibraryFile.deleted_at.is_(None),
)
)
file_count, latest_file = agg_result.one()
file_count = file_count or 0
latest_activity_at = max(folder.updated_at, latest_file) if latest_file is not None else folder.updated_at
return FolderResponse(
id=folder.id,
name=folder.name,
parent_id=folder.parent_id,
project_id=folder.project_id,
archive_id=folder.archive_id,
project_name=project_name,
archive_name=archive_name,
is_external=folder.is_external,
external_path=folder.external_path,
external_readonly=folder.external_readonly,
external_show_hidden=folder.external_show_hidden,
file_count=file_count,
latest_activity_at=latest_activity_at,
created_at=folder.created_at,
updated_at=folder.updated_at,
)
_README_BYTES_CAP = 512 * 1024 # 512 KiB — model descriptions don't need more
_README_PREFERRED_STEMS = ("readme", "description")
@router.get("/folders/{folder_id}/readme", response_model=FolderReadmeResponse)
async def get_folder_readme(
folder_id: int,
db: AsyncSession = Depends(get_db),
auth_result: tuple[User | None, bool] = Depends(
require_ownership_permission(
Permission.LIBRARY_READ_ALL,
Permission.LIBRARY_READ_OWN,
)
),
):
"""Return the first markdown description file for a folder (#1268).
Picks ``README.md`` / ``readme.md`` / ``description.md`` first (any case),
otherwise the alphabetically-first ``*.md`` in the folder. 404 when no
markdown file is present so the FE can hide the side panel.
"""
user, can_read_all = auth_result
folder_row = await db.execute(select(LibraryFolder.id).where(LibraryFolder.id == folder_id))
if folder_row.scalar_one_or_none() is None:
raise HTTPException(status_code=404, detail="Folder not found")
query = LibraryFile.active().where(
LibraryFile.folder_id == folder_id,
func.lower(LibraryFile.filename).like("%.md"),
)
if user is not None and not can_read_all:
query = query.where(LibraryFile.created_by_id == user.id)
result = await db.execute(query)
candidates = result.scalars().all()
if not candidates:
raise HTTPException(status_code=404, detail="No markdown description in folder")
def sort_key(f: LibraryFile) -> tuple[int, str]:
stem = os.path.splitext(f.filename.lower())[0]
try:
return (_README_PREFERRED_STEMS.index(stem), f.filename.lower())
except ValueError:
return (len(_README_PREFERRED_STEMS), f.filename.lower())
pick = sorted(candidates, key=sort_key)[0]
abs_path = to_absolute_path(pick.file_path)
if not abs_path or not abs_path.exists():
raise HTTPException(status_code=404, detail="Markdown file missing on disk")
try:
raw = abs_path.read_bytes()
except OSError as e:
logger.warning("Folder readme read failed for %s: %s", abs_path, e)
raise HTTPException(status_code=500, detail="Could not read markdown file") from None
truncated = len(raw) > _README_BYTES_CAP
if truncated:
raw = raw[:_README_BYTES_CAP]
# `errors="replace"` so a single bad byte never blanks the panel.
content = raw.decode("utf-8", errors="replace")
return FolderReadmeResponse(filename=pick.filename, content=content, truncated=truncated)
@router.put("/folders/{folder_id}", response_model=FolderResponse)
async def update_folder(
folder_id: int,
data: FolderUpdate,
db: AsyncSession = Depends(get_db),
_: User | None = Depends(require_permission_if_auth_enabled(Permission.LIBRARY_UPDATE_ALL)),
):
"""Update a folder.
Note: Folders require library:update_all permission since they don't have
ownership tracking.
"""
result = await db.execute(select(LibraryFolder).where(LibraryFolder.id == folder_id))
folder = result.scalar_one_or_none()
if not folder:
raise HTTPException(status_code=404, detail="Folder not found")
if data.name is not None:
folder.name = data.name
if data.parent_id is not None:
# Prevent circular reference
if data.parent_id == folder_id:
raise HTTPException(status_code=400, detail="Folder cannot be its own parent")
# Check for circular reference in ancestors
if data.parent_id != 0: # 0 means move to root
current_id = data.parent_id
while current_id is not None:
if current_id == folder_id:
raise HTTPException(status_code=400, detail="Cannot move folder into its own subtree")
parent_result = await db.execute(select(LibraryFolder.parent_id).where(LibraryFolder.id == current_id))
current_id = parent_result.scalar()
folder.parent_id = data.parent_id
else:
folder.parent_id = None
# Update project_id (0 to unlink)
if data.project_id is not None:
if data.project_id == 0:
folder.project_id = None
else:
# Verify project exists
project_result = await db.execute(select(Project).where(Project.id == data.project_id))
if not project_result.scalar_one_or_none():
raise HTTPException(status_code=404, detail="Project not found")
folder.project_id = data.project_id
# Update archive_id (0 to unlink)
if data.archive_id is not None:
if data.archive_id == 0:
folder.archive_id = None
else:
# Verify archive exists
archive_result = await db.execute(select(PrintArchive).where(PrintArchive.id == data.archive_id))
if not archive_result.scalar_one_or_none():
raise HTTPException(status_code=404, detail="Archive not found")
folder.archive_id = data.archive_id
await db.commit()
await db.refresh(folder)
# Get file count + latest file activity (#1770) and names
agg_result = await db.execute(
select(
func.count(LibraryFile.id),
func.max(LibraryFile.updated_at),
).where(
LibraryFile.folder_id == folder_id,
LibraryFile.deleted_at.is_(None),
)
)
file_count, latest_file = agg_result.one()
file_count = file_count or 0
latest_activity_at = max(folder.updated_at, latest_file) if latest_file is not None else folder.updated_at
# Get project and archive names
project_name = None
archive_name = None
if folder.project_id:
project_result = await db.execute(select(Project.name).where(Project.id == folder.project_id))
project_name = project_result.scalar()
if folder.archive_id:
archive_result = await db.execute(select(PrintArchive.print_name).where(PrintArchive.id == folder.archive_id))
archive_name = archive_result.scalar()
return FolderResponse(
id=folder.id,
name=folder.name,
parent_id=folder.parent_id,
project_id=folder.project_id,
archive_id=folder.archive_id,
project_name=project_name,
archive_name=archive_name,
is_external=folder.is_external,
external_path=folder.external_path,
external_readonly=folder.external_readonly,
external_show_hidden=folder.external_show_hidden,
file_count=file_count,
latest_activity_at=latest_activity_at,
created_at=folder.created_at,
updated_at=folder.updated_at,
)
async def _restricted_folder_delete_blocker(db: AsyncSession, folder: LibraryFolder) -> str | None:
"""Why a library:delete_own user may NOT delete this folder, or None if they may.
Folders have no ownership tracking, so users without library:delete_all may
only delete folders that are truly empty — an empty folder contains nobody's
data (#1781). "Empty" must include trashed files: LibraryFile.folder_id
cascades on folder delete, so a folder holding another user's trashed file
would silently break trash restore.
"""
if folder.is_external:
return "External folders can only be deleted by users with library:delete_all"
if folder.project_id is not None or folder.archive_id is not None:
return "Folders linked to a project or archive can only be deleted by users with library:delete_all"
child_result = await db.execute(select(func.count(LibraryFolder.id)).where(LibraryFolder.parent_id == folder.id))
if (child_result.scalar() or 0) > 0:
return "Only empty folders can be deleted without library:delete_all"
# Includes trashed files (no deleted_at filter) — see docstring.
file_result = await db.execute(select(func.count(LibraryFile.id)).where(LibraryFile.folder_id == folder.id))
if (file_result.scalar() or 0) > 0:
return "Only empty folders can be deleted without library:delete_all (the folder may contain trashed files)"
return None
@router.delete("/folders/{folder_id}")
async def delete_folder(
folder_id: int,
db: AsyncSession = Depends(get_db),
auth_result: tuple[User | None, bool] = Depends(
require_ownership_permission(
Permission.LIBRARY_DELETE_ALL,
Permission.LIBRARY_DELETE_OWN,
)
),
):
"""Delete a folder and all its contents (cascade).
Folders have no ownership tracking, so cascade deletion requires
library:delete_all. Users with only library:delete_own may delete empty,
non-external, non-linked folders (#1781).
"""
_, can_modify_all = auth_result
result = await db.execute(select(LibraryFolder).where(LibraryFolder.id == folder_id))
folder = result.scalar_one_or_none()
if not folder:
raise HTTPException(status_code=404, detail="Folder not found")
if not can_modify_all:
blocker = await _restricted_folder_delete_blocker(db, folder)
if blocker:
raise HTTPException(status_code=403, detail=blocker)
# External folders: only remove DB records, never delete files from external path
is_ext = folder.is_external
# Get all files in this folder and subfolders to delete from disk
async def get_all_file_ids(fid: int) -> list[int]:
"""Recursively get all file IDs in a folder tree."""
file_ids = []
# Get files in this folder
files_result = await db.execute(
select(LibraryFile.id, LibraryFile.file_path, LibraryFile.thumbnail_path, LibraryFile.is_external).where(
LibraryFile.folder_id == fid
)
)
for fid_val, file_path, thumb_path, file_is_ext in files_result.all():
file_ids.append(fid_val)
# Only delete non-external files from disk
if not is_ext and not file_is_ext:
try:
if file_path and os.path.exists(file_path):
os.remove(file_path)
if thumb_path and os.path.exists(thumb_path):
os.remove(thumb_path)
except OSError as e:
logger.warning("Failed to delete file: %s", e)
# Get child folders and recurse
children_result = await db.execute(select(LibraryFolder.id).where(LibraryFolder.parent_id == fid))
for (child_id,) in children_result.all():
file_ids.extend(await get_all_file_ids(child_id))
return file_ids
doomed_file_ids = await get_all_file_ids(folder_id)
# The folder cascade hard-deletes every file row under it, so the queue has
# to be taken off them first — same as the single-file delete below (#2819).
# The return value used to be discarded here, which is why this never
# happened for a folder delete.
from backend.app.services.library_trash import delete_dependent_variants, release_queue_references
await delete_dependent_variants(db, doomed_file_ids)
await release_queue_references(db, doomed_file_ids)
for doomed_id in doomed_file_ids:
remove_library_photos_dir(doomed_id)
# Delete folder (cascade will handle files and subfolders)
await db.delete(folder)
await db.commit()
return {"status": "success", "message": "Folder deleted"}
# ============ External Folder Endpoints ============
# GHSA-r2qv follow-up (audit finding I1): external-folder mount path uses an
# allowlist of operator-opted-in roots rather than the original denylist of
# system directories. The denylist shape was fail-open-on-growth — anything
# not enumerated (``/data`` containing other users' archives, ``/root``,
# arbitrary NFS/SMB mounts, the Bambuddy ``LOG_DIR``) could be mounted by any
# user with ``LIBRARY_UPLOAD``. The allowlist defaults to empty and is
# extended via the ``BAMBUDDY_EXTERNAL_ROOTS`` env var (colon-separated
# absolute paths). The route is additionally gated on ``SETTINGS_UPDATE``
# (admin scope) rather than ``LIBRARY_UPLOAD`` because mounting host paths
# is an operator-level capability that crosses user boundaries.
# Bambuddy-owned data directories. Hardcode-rejected even if the operator
# tries to add them to ``BAMBUDDY_EXTERNAL_ROOTS`` — mounting these would
# allow reading other users' archives, log files, or the static assets path.
def _bambuddy_reserved_roots() -> tuple[Path, ...]:
"""Resolved Bambuddy-owned directories that may NEVER be mounted as an
external folder regardless of the operator's allowlist.
Resolved at call time because tests patch ``settings.base_dir`` /
``settings.log_dir`` to a temp dir; resolving lazily picks up the
patched values rather than module-import-time values.
"""
from backend.app.core.config import settings as app_settings
reserved = [app_settings.base_dir, app_settings.log_dir, app_settings.static_dir, app_settings.archive_dir]
return tuple(Path(p).resolve() for p in reserved if p is not None)
def _allowed_external_roots() -> tuple[Path, ...]:
"""Parse ``BAMBUDDY_EXTERNAL_ROOTS`` into resolved allowed roots.
Empty env var (the default) means external folders are disabled.
Operators opt in explicitly: ``BAMBUDDY_EXTERNAL_ROOTS=/mnt/library:/srv/3d``
Returns a tuple of resolved ``Path`` objects; entries that don't
resolve to absolute paths are silently dropped (operator error, not
a security boundary). Resolved lazily so tests can monkeypatch.
"""
raw = os.environ.get("BAMBUDDY_EXTERNAL_ROOTS", "")
roots: list[Path] = []
for entry in raw.split(":"):
entry = entry.strip()
if not entry:
continue
try:
resolved = Path(entry).resolve()
except (OSError, RuntimeError): # noqa: BLE001 — operator config error, not a security boundary
continue
if resolved.is_absolute():
roots.append(resolved)
return tuple(roots)
def _path_within(child: Path, parent: Path) -> bool:
"""Return True if ``child`` is ``parent`` or any descendant.
Uses ``Path.relative_to`` semantics (raises ``ValueError`` on miss)
instead of string ``startswith``, which would falsely match
``/data-other`` against ``/data``. ``Path.is_relative_to`` is the
sanctioned form on Python 3.9+; both are available here.
"""
try:
child.relative_to(parent)
except ValueError:
return False
return True
# Supported file extensions for external folder scanning
_SCANNABLE_EXTENSIONS = {
".3mf",
".gcode",
".gcode.3mf",
".stl",
".obj",
".step",
".stp",
".png",
".jpg",
".jpeg",
".gif",
".webp",
".svg",
".md",
# Documents that ship alongside a job folder and now have in-app
# previews (#2976): drawings/datasheets and part lists.
".pdf",
".csv",
".xlsx",
".ods",
}
def _validate_external_path(path_str: str) -> Path:
"""Validate an external path is safe to mount.
Allowlist semantics:
1. Path must be absolute and resolve cleanly (symlink-escape rejected
implicitly by the resolved-startswith check below).
2. Path must fall under one of the roots enumerated in
``BAMBUDDY_EXTERNAL_ROOTS``; empty allowlist (the default)
means external folders are not available on this deployment.
3. Path must NOT fall under any Bambuddy-owned directory (``base_dir``,
``log_dir``, ``static_dir``, ``archive_dir``) — the reserved set
takes precedence over the allowlist, so an operator who accidentally
sets ``BAMBUDDY_EXTERNAL_ROOTS=/`` does not expose ``/data``.
4. Existence + directory-type + readability gates remain.
"""
path = Path(path_str).resolve()
if not path.is_absolute():
raise HTTPException(status_code=400, detail="Path must be absolute")
allowed_roots = _allowed_external_roots()
if not allowed_roots:
raise HTTPException(
status_code=400,
detail=(
"External folders are not enabled on this deployment. Ask the "
"operator to set BAMBUDDY_EXTERNAL_ROOTS=<colon-separated paths>."
),
)
# Reserved (Bambuddy-owned) paths are rejected before the allowlist check
# so an over-broad allowlist (e.g. operator set "/" for testing) cannot
# expose Bambuddy's own data dir or log dir.
for reserved in _bambuddy_reserved_roots():
if _path_within(path, reserved):
raise HTTPException(
status_code=400,
detail=f"Cannot mount Bambuddy-managed directory: {reserved}",
)
if not any(_path_within(path, root) for root in allowed_roots):
raise HTTPException(
status_code=400,
detail=(
f"Path '{path}' is not within an allowed external root. "
f"Allowed roots: {', '.join(str(r) for r in allowed_roots)}"
),
)
if not path.exists():
raise HTTPException(status_code=400, detail=f"Path does not exist: {path}")
if not path.is_dir():
raise HTTPException(status_code=400, detail=f"Path is not a directory: {path}")
# Check readability
if not os.access(path, os.R_OK):
raise HTTPException(status_code=400, detail=f"Path is not readable: {path}")
return path
@router.post("/folders/external", response_model=FolderResponse)
async def create_external_folder(
data: ExternalFolderCreate,
db: AsyncSession = Depends(get_db),
# GHSA-r2qv follow-up (I1): elevated from LIBRARY_UPLOAD to SETTINGS_UPDATE.
# Registering a host filesystem path as a Bambuddy library folder is an
# operator-level capability that crosses user boundaries (one user's
# registered external folder is visible to every other user via
# /api/v1/library/folders). LIBRARY_UPLOAD was always the wrong scope —
# SETTINGS_UPDATE is the admin-class gate that already protects every
# other host-affecting setting (SMTP, LDAP, cloud, smart plugs).
_: User | None = Depends(require_permission_if_auth_enabled(Permission.SETTINGS_UPDATE)),
):
"""Create an external folder that points to a host directory."""
resolved = _validate_external_path(data.external_path)
# Check no other external folder already points to this path
existing = await db.execute(
select(LibraryFolder).where(
LibraryFolder.is_external.is_(True),
LibraryFolder.external_path == str(resolved),
)
)
if existing.scalar_one_or_none():
raise HTTPException(status_code=409, detail="An external folder already exists for this path")
# Verify parent exists if specified
if data.parent_id is not None:
parent_result = await db.execute(select(LibraryFolder).where(LibraryFolder.id == data.parent_id))
if not parent_result.scalar_one_or_none():
raise HTTPException(status_code=404, detail="Parent folder not found")
folder = LibraryFolder(
name=data.name,
parent_id=data.parent_id,
is_external=True,
external_path=str(resolved),
external_readonly=data.readonly,
external_show_hidden=data.show_hidden,
)
db.add(folder)
await db.commit()
await db.refresh(folder)
return FolderResponse(
id=folder.id,
name=folder.name,
parent_id=folder.parent_id,
project_id=None,
archive_id=None,
is_external=True,
external_path=folder.external_path,
external_readonly=folder.external_readonly,
external_show_hidden=folder.external_show_hidden,
file_count=0,
# Newly-created external folder hasn't been scanned yet — fall back
# to the folder's own updated_at (#1770).
latest_activity_at=folder.updated_at,
created_at=folder.created_at,
updated_at=folder.updated_at,
)
def _mtime_to_datetime(mtime: float) -> datetime:
"""Convert an ``os.stat().st_mtime`` epoch value to a naive-UTC datetime (#2680).
Naive UTC to match the other library timestamp columns (``created_at`` /
``updated_at`` are naive ``func.now()``), so activity comparisons never mix
naive and aware values on either dialect.
"""
return datetime.fromtimestamp(mtime, tz=timezone.utc).replace(tzinfo=None)
@router.post("/folders/{folder_id}/scan")
async def scan_external_folder(
folder_id: int,
db: AsyncSession = Depends(get_db),
_: User | None = Depends(require_permission_if_auth_enabled(Permission.LIBRARY_UPLOAD)),
):
"""Scan an external folder and sync files to the database.
Discovers new files, removes DB entries for deleted files.
Does not copy files — stores the external path directly.
"""
result = await db.execute(select(LibraryFolder).where(LibraryFolder.id == folder_id))
folder = result.scalar_one_or_none()
if not folder:
raise HTTPException(status_code=404, detail="Folder not found")
if not folder.is_external or not folder.external_path:
raise HTTPException(status_code=400, detail="Not an external folder")
ext_path = Path(folder.external_path)
if not ext_path.exists() or not ext_path.is_dir():
raise HTTPException(status_code=400, detail=f"External path is not accessible: {folder.external_path}")
# Collect all existing child external subfolder IDs (single query)
all_folder_ids = [folder_id]
child_result = await db.execute(
select(LibraryFolder).where(
LibraryFolder.is_external.is_(True),
LibraryFolder.parent_id.isnot(None),
)
)
all_child_folders = child_result.scalars().all()
# Walk the parent chain to find all descendants of folder_id
parent_to_children: dict[int, list] = {}
for cf in all_child_folders:
parent_to_children.setdefault(cf.parent_id, []).append(cf)
queue = [folder_id]
while queue:
pid = queue.pop()
for child in parent_to_children.get(pid, []):
all_folder_ids.append(child.id)
queue.append(child.id)
# Get existing DB files across root and all subfolders
existing_result = await db.execute(
LibraryFile.active().where(
LibraryFile.folder_id.in_(all_folder_ids),
LibraryFile.is_external.is_(True),
)
)
existing_files = {f.file_path: f for f in existing_result.scalars().all()}
# Build folder cache: relative path -> folder_id (for resolving subfolders)
# Pre-populate with existing child folders keyed by their external_path
folder_cache: dict[str, int] = {"": folder_id}
for fid in all_folder_ids:
if fid == folder_id:
continue
# Find the child folder object
for cf in all_child_folders:
if cf.id == fid and cf.external_path:
try:
rel = str(Path(cf.external_path).relative_to(ext_path))
if rel != ".":
folder_cache[rel] = cf.id
except ValueError:
pass
# Scan the directory
added = 0
removed = 0
found_paths: set[str] = set()
seen_rel_dirs: set[str] = set()
# Real on-disk mtime per visited folder id (#2680), applied after the walk.
folder_mtimes: dict[int, datetime] = {}
for dirpath, dirnames, filenames in os.walk(ext_path):
# Filter hidden directories unless configured
if not folder.external_show_hidden:
dirnames[:] = [d for d in dirnames if not d.startswith(".")]
rel_dir = str(Path(dirpath).relative_to(ext_path))
if rel_dir == ".":
rel_dir = ""
seen_rel_dirs.add(rel_dir)
# Resolve or create subfolder chain for this directory
if rel_dir and rel_dir not in folder_cache:
parts = Path(rel_dir).parts
current_path = ""
current_parent = folder_id
for part in parts:
current_path = f"{current_path}/{part}".lstrip("/")
if current_path in folder_cache:
current_parent = folder_cache[current_path]
else:
existing_sub = await db.execute(
select(LibraryFolder).where(
LibraryFolder.name == part,
LibraryFolder.parent_id == current_parent,
LibraryFolder.is_external.is_(True),
)
)
existing_folder = existing_sub.scalar_one_or_none()
if existing_folder:
current_parent = existing_folder.id
else:
new_folder = LibraryFolder(
name=part,
parent_id=current_parent,
is_external=True,
external_path=str(
ext_path / current_path
), # SEC-PATH-OK: current_path built from Path(rel_dir).parts of an os.walk descent under ext_path
external_readonly=folder.external_readonly,
external_show_hidden=folder.external_show_hidden,
)
db.add(new_folder)
await db.flush()
current_parent = new_folder.id
folder_cache[current_path] = current_parent
target_folder_id = folder_cache.get(rel_dir, folder_id)
# Record this directory's own mtime (#2680). os.walk visits every
# directory once, so this covers the root external folder and every
# subfolder (existing or just created). Applied to the folder rows
# after the walk completes.
try:
folder_mtimes[target_folder_id] = _mtime_to_datetime(os.stat(dirpath).st_mtime)
except OSError:
pass
for filename in filenames:
# Skip hidden files unless configured
if not folder.external_show_hidden and filename.startswith("."):
continue
filepath = (
Path(dirpath) / filename
) # SEC-PATH-OK: dirpath + filename from os.walk(ext_path); filesystem-discovered, not user input
ext = filepath.suffix.lower()
# Check for compound extensions like .gcode.3mf
if ext not in _SCANNABLE_EXTENSIONS:
# Check compound
compound = "".join(filepath.suffixes[-2:]).lower() if len(filepath.suffixes) >= 2 else ""
if compound not in _SCANNABLE_EXTENSIONS:
continue
# Resolve symlinks and ensure still under external_path
try:
real_path = filepath.resolve()
real_path.relative_to(ext_path.resolve())
except (ValueError, OSError):
continue # Symlink escapes the external dir
file_path_str = str(filepath)
found_paths.add(file_path_str)
if file_path_str in existing_files:
# Already tracked — refresh its on-disk mtime (#2680) so a file
# edited/replaced over the mount (samba, etc.) re-sorts correctly
# and old rows scanned before this field existed get backfilled.
tracked = existing_files[file_path_str]
try:
fs_mtime = _mtime_to_datetime(filepath.stat().st_mtime)
except OSError:
fs_mtime = None
if fs_mtime is not None and tracked.fs_modified_at != fs_mtime:
tracked.fs_modified_at = fs_mtime
continue
# Get file info
try:
stat = filepath.stat()
except OSError:
continue
# The zip is opened for the thumbnail immediately below either way,
# so judging a `.3mf` on its contents rather than its name (#2993)
# costs this scan nothing.
file_type = classify_file_type(filename, filepath)
# Extract thumbnail for 3mf files (including .gcode.3mf sliced
# outputs — those are 3MF zips on disk and carry the same
# thumbnail Metadata/plate_1.png the parser reads). Pre-#1600
# the gate was `file_type == "3mf"` alone, so .gcode.3mf files
# in external folders silently got no thumbnail.
thumbnail_path = None
file_metadata = None
if file_type in ("3mf", "gcode.3mf"):
try:
parser = ThreeMFParser(str(filepath))
raw_metadata = parser.parse()
if raw_metadata:
# Extract thumbnail before cleaning metadata
thumb_data = raw_metadata.get("_thumbnail_data")
thumbnail_ext = raw_metadata.get("_thumbnail_ext", ".png")
if thumb_data:
thumb_dir = get_library_thumbnails_dir()
thumb_filename = f"{uuid.uuid4().hex}{thumbnail_ext}"
thumb_full = (
thumb_dir / thumb_filename
) # SEC-PATH-OK: thumb_filename = uuid.uuid4().hex + thumbnail_ext
thumb_full.write_bytes(thumb_data)
thumbnail_path = to_relative_path(thumb_full)
# Clean metadata - remove non-JSON-serializable data (bytes, etc.)
def clean_metadata(obj):
if isinstance(obj, dict):
return {
k: clean_metadata(v)
for k, v in obj.items()
if not isinstance(v, bytes) and k not in ("_thumbnail_data", "_thumbnail_ext")
}
elif isinstance(obj, list):
return [clean_metadata(i) for i in obj if not isinstance(i, bytes)]
elif isinstance(obj, bytes):
return None
return obj
file_metadata = clean_metadata(raw_metadata)
except Exception as e:
logger.debug("Failed to extract metadata from external 3mf %s: %s", filepath, e)
# STL and PDF thumbnails are deferred to a background task spawned
# after the scan's db.commit() — see _backfill_external_thumbnails.
# Doing them inline would block the HTTP request for minutes on a
# large NAS mount (#1299).
# Extract gcode thumbnail
if file_type == "gcode" and thumbnail_path is None:
thumb_data = extract_gcode_thumbnail(filepath)
if thumb_data:
thumb_dir = get_library_thumbnails_dir()
thumb_filename = f"{uuid.uuid4().hex}.png"
thumb_full = thumb_dir / thumb_filename # SEC-PATH-OK: thumb_filename = uuid.uuid4().hex + ".png"
thumb_full.write_bytes(thumb_data)
thumbnail_path = to_relative_path(thumb_full)
# Create thumbnail for image files
if ext.lower() in IMAGE_EXTENSIONS and thumbnail_path is None:
thumbnail_path_str = create_image_thumbnail(filepath, get_library_thumbnails_dir())
if thumbnail_path_str:
thumbnail_path = to_relative_path(Path(thumbnail_path_str))
db_file = LibraryFile(
folder_id=target_folder_id,
is_external=True,
filename=filename,
file_path=file_path_str,
file_type=file_type,
file_size=stat.st_size,
file_hash=None, # Skip hashing external files for performance
thumbnail_path=thumbnail_path,
file_metadata=_without_print_name(file_metadata),
fs_modified_at=_mtime_to_datetime(stat.st_mtime), # #2680: real on-disk mtime
)
db.add(db_file)
added += 1
# Remove DB entries for files that no longer exist on disk.
#
# Gate on actual disk presence, NOT merely absence from found_paths:
# found_paths only collects extensions in _SCANNABLE_EXTENSIONS, so a
# record for any other file the upload path admitted (e.g. a .md README,
# #2520) would otherwise be treated as "deleted from disk" and purged on
# every scan even though the file is still there. os.path.exists keeps
# such records; genuinely-deleted files (absent from disk) are still
# cleaned up. External file_path is the absolute on-disk path.
for path_str, db_file in existing_files.items():
if path_str not in found_paths and not os.path.exists(path_str):
# Clean up thumbnail if we generated one
if db_file.thumbnail_path:
try:
abs_thumb = to_absolute_path(db_file.thumbnail_path)
if abs_thumb and abs_thumb.exists():
abs_thumb.unlink()
except OSError:
pass
# The row is gone for good — external files skip the trash — so
# its photos go with it rather than being orphaned under an id
# nothing points at any more (#3077).
remove_library_photos_dir(db_file.id)
await db.delete(db_file)
removed += 1
# Remove empty subfolders whose directories no longer exist on disk
# Process deepest-first by sorting on path depth (descending)
subfolder_entries = [(rel, fid) for rel, fid in folder_cache.items() if rel and fid != folder_id]
subfolder_entries.sort(key=lambda x: x[0].count("/"), reverse=True)
for rel_path, sub_fid in subfolder_entries:
if rel_path in seen_rel_dirs:
continue # Directory still exists on disk
# Check if subfolder has any remaining files
file_count_result = await db.execute(
select(func.count(LibraryFile.id)).where(
LibraryFile.folder_id == sub_fid,
LibraryFile.deleted_at.is_(None),
)
)
if (file_count_result.scalar() or 0) == 0:
# Check if it has any remaining child folders
child_count_result = await db.execute(
select(func.count(LibraryFolder.id)).where(LibraryFolder.parent_id == sub_fid)
)
if (child_count_result.scalar() or 0) == 0:
sub_folder_result = await db.execute(select(LibraryFolder).where(LibraryFolder.id == sub_fid))
sub_folder_obj = sub_folder_result.scalar_one_or_none()
if sub_folder_obj:
await db.delete(sub_folder_obj)
folder_mtimes.pop(sub_fid, None)
# Persist each visited folder's real directory mtime (#2680). Fetched in one
# trip; folders deleted by the cleanup above were dropped from folder_mtimes.
if folder_mtimes:
folders_result = await db.execute(select(LibraryFolder).where(LibraryFolder.id.in_(list(folder_mtimes.keys()))))
for folder_obj in folders_result.scalars().all():
new_mtime = folder_mtimes.get(folder_obj.id)
if new_mtime is not None and folder_obj.fs_modified_at != new_mtime:
folder_obj.fs_modified_at = new_mtime
await db.commit()
# Spawn STL/PDF thumbnail backfill in the background — the scan endpoint
# returns immediately so the FE modal closes and subdirectories are
# visible right away; thumbnails fill in over the following seconds /
# minutes as the task processes each file. Survives FE refresh —
# the task lives in the FastAPI event loop, not the request scope.
# folder_cache.values() covers the root + every pre-existing subfolder
# + every subfolder created during this scan. all_folder_ids on its own
# would miss the newly-created ones (it's snapshotted before the walk).
spawn_background_task(
_backfill_external_thumbnails(list(set(folder_cache.values()))),
name=f"thumbnail-backfill-folder-{folder_id}",
)
return {"status": "success", "added": added, "removed": removed}
# ============ File Endpoints ============
@router.get("/files", response_model=list[FileListResponse])
@router.get("/files/", response_model=list[FileListResponse])
async def list_files(
response: Response,
folder_id: int | None = None,
project_id: int | None = None,
include_root: bool = True,
internal_only: bool = False,
external_only: bool = False,
recursive: bool = False,
tag_ids: list[int] = Query(default_factory=list),
db: AsyncSession = Depends(get_db),
auth_result: tuple[User | None, bool] = Depends(
require_ownership_permission(
Permission.LIBRARY_READ_ALL,
Permission.LIBRARY_READ_OWN,
)
),
):
"""List files, optionally filtered by folder or project.
Args:
folder_id: Filter by folder ID. If None and include_root=True, returns root files.
project_id: Return all files across folders linked to this project (bulk fetch, avoids N+1).
include_root: If True and folder_id is None, returns files at root level.
If False and folder_id is None, returns all files.
internal_only: Restrict the result to files in managed storage (`is_external=False`).
Used by the File Manager's "All Files" sidebar entry so a linked NAS
with hundreds of files doesn't drown the user's own uploads (#1621).
external_only: Restrict the result to files under external folders
(`is_external=True`) — the symmetric combined view for users with
multiple linked external sources (#1621).
recursive: When combined with ``folder_id``, also include files in every
descendant subfolder (#1268). Implemented via a recursive CTE
that walks ``library_folders.parent_id``. Default off so
existing callers (folder browsing, etc.) keep their narrow
single-folder semantics.
tag_ids: Restrict the listing to files carrying ALL of these tags
(AND semantics, #1268). When non-empty the folder filter is
intentionally bypassed — tags are cross-cutting and the user
wants "every file with this tag" regardless of where it lives.
``recursive`` becomes irrelevant in that case.
"""
if internal_only and external_only:
raise HTTPException(
status_code=400,
detail="internal_only and external_only are mutually exclusive",
)
user, can_read_all = auth_result
query = LibraryFile.active().options(
selectinload(LibraryFile.created_by),
selectinload(LibraryFile.tags),
)
if user is not None and not can_read_all:
query = query.where(LibraryFile.created_by_id == user.id)
if tag_ids:
# Cross-cutting filter — every requested tag must be present on the
# file. JOIN + GROUP BY + HAVING COUNT(DISTINCT) is portable across
# SQLite and Postgres without dialect tricks. We deliberately skip
# the folder / project / include_root scoping below so the result
# is the global "all files carrying these tags".
unique_tag_ids = list(dict.fromkeys(tag_ids))
query = (
query.join(LibraryFileTag, LibraryFileTag.file_id == LibraryFile.id)
.where(LibraryFileTag.tag_id.in_(unique_tag_ids))
.group_by(LibraryFile.id)
.having(func.count(distinct(LibraryFileTag.tag_id)) == len(unique_tag_ids))
)
elif folder_id is not None and recursive:
# Walk the subtree starting at folder_id and collect every descendant
# id. Recursive CTE works on both SQLite (>=3.8.3, shipped 2014) and
# Postgres without dialect branching.
roots = (
select(LibraryFolder.id).where(LibraryFolder.id == folder_id).cte(name="folder_descendants", recursive=True)
)
descendants = roots.union_all(select(LibraryFolder.id).join(roots, LibraryFolder.parent_id == roots.c.id))
query = query.where(LibraryFile.folder_id.in_(select(descendants.c.id)))
elif folder_id is not None:
query = query.where(LibraryFile.folder_id == folder_id)
elif project_id is not None:
# Single join instead of one query per folder (avoids N+1 pattern)
query = query.join(LibraryFolder, LibraryFile.folder_id == LibraryFolder.id)
query = query.where(LibraryFolder.project_id == project_id)
elif include_root:
query = query.where(LibraryFile.folder_id.is_(None))
if internal_only:
query = query.where(LibraryFile.is_external.is_(False))
elif external_only:
query = query.where(LibraryFile.is_external.is_(True))
query = query.order_by(LibraryFile.filename)
result = await db.execute(query)
files = result.scalars().unique().all() if tag_ids else result.scalars().all()
# Get duplicate counts
hash_counts = {}
if files:
hashes = [f.file_hash for f in files if f.file_hash]
if hashes:
dup_result = await db.execute(
select(LibraryFile.file_hash, func.count(LibraryFile.id))
.where(LibraryFile.file_hash.in_(hashes), LibraryFile.deleted_at.is_(None))
.group_by(LibraryFile.file_hash)
)
hash_counts = {h: c - 1 for h, c in dup_result.all()} # -1 to exclude self
# Variant group sizes (#671 / #2570). Counted across the whole group rather
# than the rows on screen — members can sit in different folders, so counting
# the listing would under-report and the "2 versions" badge would blink in
# and out as the user navigated.
variant_counts: dict[int, int] = {}
group_ids = {f.variant_group_id for f in files if f.variant_group_id}
if group_ids:
count_result = await db.execute(
select(LibraryFile.variant_group_id, func.count(LibraryFile.id))
.where(LibraryFile.variant_group_id.in_(group_ids), LibraryFile.deleted_at.is_(None))
.group_by(LibraryFile.variant_group_id)
)
variant_counts = dict(count_result.all())
# Prevent browser caching of file list
response.headers["Cache-Control"] = "no-cache, no-store, must-revalidate"
file_list = []
for f in files:
# Extract key metadata for display
print_name = None
print_time = None
filament_grams = None
sliced_for_model = None
if f.file_metadata:
print_name = f.file_metadata.get("print_name")
print_time = f.file_metadata.get("print_time_seconds")
filament_grams = f.file_metadata.get("filament_used_grams")
sliced_for_model = f.file_metadata.get("sliced_for_model")
file_list.append(
FileListResponse(
id=f.id,
folder_id=f.folder_id,
is_external=f.is_external,
filename=f.filename,
file_type=f.file_type,
file_size=f.file_size,
thumbnail_path=f.thumbnail_path,
print_count=f.print_count,
duplicate_count=hash_counts.get(f.file_hash, 0) if f.file_hash else 0,
created_by_id=f.created_by_id,
created_by_username=f.created_by.username if f.created_by else None,
created_at=f.created_at,
fs_modified_at=f.fs_modified_at,
print_name=print_name,
print_time_seconds=print_time,
filament_used_grams=filament_grams,
sliced_for_model=sliced_for_model,
tags=[TagSummary(id=t.id, name=t.name) for t in f.tags],
variant_group_id=f.variant_group_id,
variant_count=variant_counts.get(f.variant_group_id, 0) if f.variant_group_id else 0,
external_url=f.external_url,
has_notes=bool(f.notes),
photo_count=len(f.photos or []),
)
)
return file_list
@router.post("/files", response_model=FileUploadResponse)
@router.post("/files/", response_model=FileUploadResponse)
async def upload_file(
file: UploadFile = File(...),
folder_id: int | None = None,
generate_stl_thumbnails: bool = Query(default=True),
db: AsyncSession = Depends(get_db),
current_user: User | None = Depends(require_permission_if_auth_enabled(Permission.LIBRARY_UPLOAD)),
):
"""Upload a file to the library."""
try:
if not file.filename:
raise HTTPException(status_code=400, detail="Filename is required")
filename = file.filename
# Reject FAT32/exFAT-incompatible filenames up front (#1540).
try:
validate_print_filename(filename)
except InvalidFilenameError as e:
raise HTTPException(status_code=400, detail=str(e)) from e
# `ext` stays the trailing extension because the on-disk filename uses
# it directly and the 3MF-parse branch below still gates on
# `ext == ".3mf"`, which is correct for both `.3mf` and `.gcode.3mf`.
# `file_type` is compound-aware and is decided further down, once the
# bytes are on disk to be read.
ext = os.path.splitext(filename)[1].lower()
# Verify folder exists if specified
target_folder = None
if folder_id is not None:
folder_result = await db.execute(select(LibraryFolder).where(LibraryFolder.id == folder_id))
target_folder = folder_result.scalar_one_or_none()
if not target_folder:
raise HTTPException(status_code=404, detail="Folder not found")
# Writable external folders write through to the mount so the file is
# visible outside Bambuddy (#1112); everything else lands under the
# internal library dir with a UUID-scoped filename. Resolved BEFORE
# the content validation below so folder-permission rejections
# (403 read-only, 400 missing path, 409 collision) still surface
# before any "bad file format" 400 — preserves existing error
# ordering / tests.
file_path, is_external_upload = _resolve_upload_destination(target_folder, filename)
# Read upload now so the validation can sniff magic bytes; the file
# is written to disk only after the checks. #1401.
content = await file.read()
validate_print_file_upload(filename, content)
# Save file
with open(file_path, "wb") as f:
f.write(content)
# Now that the bytes are on disk the zip can settle what the name only
# guessed at: a sliced 3MF uploaded as `Foo.3mf` is a sliced 3MF (#2993).
file_type = classify_file_type(filename, file_path)
# Calculate hash
file_hash = calculate_file_hash(file_path)
# Check for duplicates
dup_result = await db.execute(
select(LibraryFile.id).where(LibraryFile.file_hash == file_hash, LibraryFile.deleted_at.is_(None)).limit(1)
)
duplicate_of = dup_result.scalar()
# Extract metadata and thumbnail
metadata = {}
thumbnail_path = None
thumbnails_dir = get_library_thumbnails_dir()
if ext == ".3mf":
try:
parser = ThreeMFParser(str(file_path))
raw_metadata = parser.parse()
# Extract thumbnail before cleaning metadata
thumbnail_data = raw_metadata.get("_thumbnail_data")
thumbnail_ext = raw_metadata.get("_thumbnail_ext", ".png")
# Save thumbnail if extracted
if thumbnail_data:
thumb_filename = f"{uuid.uuid4().hex}{thumbnail_ext}"
thumb_path = (
thumbnails_dir / thumb_filename
) # SEC-PATH-OK: thumb_filename = uuid.uuid4().hex + thumbnail_ext
with open(thumb_path, "wb") as f:
f.write(thumbnail_data)
thumbnail_path = str(thumb_path)
# Clean metadata - remove non-JSON-serializable data (bytes, etc.)
def clean_metadata(obj):
if isinstance(obj, dict):
return {
k: clean_metadata(v)
for k, v in obj.items()
if not isinstance(v, bytes) and k not in ("_thumbnail_data", "_thumbnail_ext")
}
elif isinstance(obj, list):
return [clean_metadata(i) for i in obj if not isinstance(i, bytes)]
elif isinstance(obj, bytes):
return None
return obj
metadata = clean_metadata(raw_metadata)
except Exception as e:
logger.warning("Failed to parse 3MF: %s", e)
elif ext == ".gcode":
# Extract embedded thumbnail from gcode
try:
thumbnail_data = extract_gcode_thumbnail(file_path)
if thumbnail_data:
thumb_filename = f"{uuid.uuid4().hex}.png"
thumb_path = (
thumbnails_dir / thumb_filename
) # SEC-PATH-OK: thumb_filename = uuid.uuid4().hex + ".png"
with open(thumb_path, "wb") as f:
f.write(thumbnail_data)
thumbnail_path = str(thumb_path)
except Exception as e:
logger.warning("Failed to extract gcode thumbnail: %s", e)
elif ext.lower() in IMAGE_EXTENSIONS:
# For image files, create a thumbnail from the image itself
thumbnail_path = create_image_thumbnail(file_path, thumbnails_dir)
elif ext.lower() == ".pdf":
# Page one, rendered server-side; the browser preview's own
# render (POST /preview-thumbnail) remains the fallback.
thumbnail_path = await asyncio.to_thread(generate_pdf_thumbnail, file_path, thumbnails_dir)
elif ext == ".stl":
# Generate STL thumbnail if enabled. Same MIN_USABLE_STL_BYTES
# pre-skip as extract_zip_file — stubs / placeholders below this
# size can't contain a triangle so trimesh would return an empty
# mesh anyway.
if generate_stl_thumbnails:
try:
if file_path.stat().st_size >= MIN_USABLE_STL_BYTES:
thumbnail_path = generate_stl_thumbnail(file_path, thumbnails_dir)
except OSError:
pass
# Create database entry (managed files store relative paths for portability;
# external files store the absolute mount path — same shape as scan produces)
library_file = LibraryFile(
folder_id=folder_id,
is_external=is_external_upload,
filename=filename,
file_path=_stored_file_path(file_path, is_external_upload),
file_type=file_type,
file_size=len(content),
file_hash=file_hash,
thumbnail_path=to_relative_path(thumbnail_path) if thumbnail_path else None,
file_metadata=_without_print_name(metadata) if metadata else None,
created_by_id=current_user.id if current_user else None,
)
db.add(library_file)
await db.commit()
await db.refresh(library_file)
return FileUploadResponse(
id=library_file.id,
filename=library_file.filename,
file_type=library_file.file_type,
file_size=library_file.file_size,
thumbnail_path=library_file.thumbnail_path,
duplicate_of=duplicate_of,
metadata=library_file.file_metadata,
)
except HTTPException:
raise
except Exception as e:
logger.error("Upload failed for %s: %s", file.filename, e, exc_info=True)
raise HTTPException(status_code=500, detail=f"Upload failed: {str(e)}")
@router.post("/files/extract-zip", response_model=ZipExtractResponse)
async def extract_zip_file(
file: UploadFile = File(...),
folder_id: int | None = Query(default=None),
preserve_structure: bool = Query(default=True),
create_folder_from_zip: bool = Query(default=False),
generate_stl_thumbnails: bool = Query(default=True),
db: AsyncSession = Depends(get_db),
current_user: User | None = Depends(require_permission_if_auth_enabled(Permission.LIBRARY_UPLOAD)),
):
"""Upload and extract a ZIP file to the library.
Args:
file: The ZIP file to extract
folder_id: Target folder ID (None = root)
preserve_structure: If True, recreate folder structure from ZIP; if False, extract all files flat
create_folder_from_zip: If True, create a folder named after the ZIP file and extract into it
generate_stl_thumbnails: If True, generate thumbnails for STL files
"""
import tempfile
if not file.filename or not file.filename.lower().endswith(".zip"):
raise HTTPException(status_code=400, detail="Only ZIP files are supported")
# Verify target folder exists if specified
if folder_id is not None:
folder_result = await db.execute(select(LibraryFolder).where(LibraryFolder.id == folder_id))
target_folder = folder_result.scalar_one_or_none()
if not target_folder:
raise HTTPException(status_code=404, detail="Target folder not found")
if target_folder.is_external and target_folder.external_readonly:
raise HTTPException(status_code=403, detail="Cannot extract ZIP to a read-only external folder")
if target_folder.is_external:
# Writable external folders aren't supported by extract-zip because the
# nested-subfolder creation path would need to mkdir on the mount and
# create matching is_external=True LibraryFolder rows — a separate
# design. Direct the user at Scan, which already handles that shape
# (#1112).
raise HTTPException(
status_code=400,
detail=(
"Cannot extract ZIP directly into an external folder. "
"Extract the ZIP on the external mount and run 'Scan External Folder' instead."
),
)
# Save ZIP to temp file
try:
with tempfile.NamedTemporaryFile(delete=False, suffix=".zip") as tmp:
content = await file.read()
tmp.write(content)
tmp_path = tmp.name
except Exception as e:
raise HTTPException(status_code=500, detail=f"Failed to save ZIP file: {str(e)}")
extracted_files: list[ZipExtractResult] = []
errors: list[ZipExtractError] = []
folders_created = 0
folder_cache: dict[str, int] = {} # path -> folder_id
# If create_folder_from_zip is True, create a folder named after the ZIP file
zip_folder_id = folder_id
logger.info(
f"ZIP extraction: create_folder_from_zip={create_folder_from_zip}, folder_id={folder_id}, filename={file.filename}"
)
if create_folder_from_zip and file.filename:
# Remove .zip extension to get folder name
zip_folder_name = file.filename[:-4] if file.filename.lower().endswith(".zip") else file.filename
# Check if folder already exists
existing = await db.execute(
select(LibraryFolder).where(
LibraryFolder.name == zip_folder_name,
LibraryFolder.parent_id == folder_id if folder_id else LibraryFolder.parent_id.is_(None),
)
)
existing_folder = existing.scalar_one_or_none()
if existing_folder:
zip_folder_id = existing_folder.id
logger.info("Reusing existing folder '%s' with id=%s", zip_folder_name, zip_folder_id)
else:
# Create folder
new_folder = LibraryFolder(name=zip_folder_name, parent_id=folder_id)
db.add(new_folder)
await db.flush()
await db.commit() # Commit folder creation immediately
zip_folder_id = new_folder.id
folders_created += 1
logger.info("Created new folder '%s' with id=%s", zip_folder_name, zip_folder_id)
try:
with zipfile.ZipFile(tmp_path, "r") as zf:
# Filter out directories and hidden/system files
file_list = [
name
for name in zf.namelist()
if not name.endswith("/")
and not name.startswith("__MACOSX")
and not os.path.basename(name).startswith(".")
]
for zip_path in file_list:
try:
# Determine target folder (use zip_folder_id as base if create_folder_from_zip was used)
target_folder_id = zip_folder_id
if preserve_structure:
# Get directory path from ZIP
dir_path = os.path.dirname(zip_path)
if dir_path:
# Create folder structure
parts = dir_path.split("/")
current_parent = zip_folder_id
current_path = ""
for part in parts:
if not part:
continue
current_path = f"{current_path}/{part}" if current_path else part
if current_path in folder_cache:
current_parent = folder_cache[current_path]
else:
# Check if folder exists
existing = await db.execute(
select(LibraryFolder).where(
LibraryFolder.name == part,
LibraryFolder.parent_id == current_parent
if current_parent
else LibraryFolder.parent_id.is_(None),
)
)
existing_folder = existing.scalar_one_or_none()
if existing_folder:
current_parent = existing_folder.id
else:
# Create folder
new_folder = LibraryFolder(name=part, parent_id=current_parent)
db.add(new_folder)
await db.flush()
current_parent = new_folder.id
folders_created += 1
folder_cache[current_path] = current_parent
target_folder_id = current_parent
# Extract file
filename = os.path.basename(zip_path)
ext = os.path.splitext(filename)[1].lower()
# Generate unique filename for storage
unique_filename = f"{uuid.uuid4().hex}{ext}"
file_path = (
get_library_files_dir() / unique_filename
) # SEC-PATH-OK: unique_filename = uuid.uuid4().hex + ext
# Extract and save file
file_content = zf.read(zip_path)
with open(file_path, "wb") as f:
f.write(file_content)
# Classified once the bytes are on disk so a sliced 3MF
# named `Foo.3mf` inside the zip is recognised as sliced
# (#2993) rather than trusted to say so in its name.
file_type = classify_file_type(filename, file_path)
# Calculate hash
file_hash = calculate_file_hash(file_path)
# Extract metadata and thumbnail for 3MF files
metadata = {}
thumbnail_path = None
thumbnails_dir = get_library_thumbnails_dir()
if ext == ".3mf":
try:
parser = ThreeMFParser(str(file_path))
raw_metadata = parser.parse()
thumbnail_data = raw_metadata.get("_thumbnail_data")
thumbnail_ext = raw_metadata.get("_thumbnail_ext", ".png")
if thumbnail_data:
thumb_filename = f"{uuid.uuid4().hex}{thumbnail_ext}"
thumb_path = (
thumbnails_dir / thumb_filename
) # SEC-PATH-OK: thumb_filename = uuid.uuid4().hex + thumbnail_ext
with open(thumb_path, "wb") as f:
f.write(thumbnail_data)
thumbnail_path = str(thumb_path)
def clean_metadata(obj):
if isinstance(obj, dict):
return {
k: clean_metadata(v)
for k, v in obj.items()
if not isinstance(v, bytes) and k not in ("_thumbnail_data", "_thumbnail_ext")
}
elif isinstance(obj, list):
return [clean_metadata(i) for i in obj if not isinstance(i, bytes)]
elif isinstance(obj, bytes):
return None
return obj
metadata = clean_metadata(raw_metadata)
except Exception as e:
logger.warning("Failed to parse 3MF from ZIP: %s", e)
elif ext == ".gcode":
try:
thumbnail_data = extract_gcode_thumbnail(file_path)
if thumbnail_data:
thumb_filename = f"{uuid.uuid4().hex}.png"
thumb_path = (
thumbnails_dir / thumb_filename
) # SEC-PATH-OK: thumb_filename = uuid.uuid4().hex + ".png"
with open(thumb_path, "wb") as f:
f.write(thumbnail_data)
thumbnail_path = str(thumb_path)
except Exception as e:
logger.warning("Failed to extract gcode thumbnail from ZIP: %s", e)
elif ext.lower() in IMAGE_EXTENSIONS:
thumbnail_path = create_image_thumbnail(file_path, thumbnails_dir)
elif ext.lower() == ".pdf":
thumbnail_path = await asyncio.to_thread(generate_pdf_thumbnail, file_path, thumbnails_dir)
elif ext == ".stl":
# Generate STL thumbnail if enabled. Pre-skip files
# below MIN_USABLE_STL_BYTES — they can't contain
# even a single triangle, and bulk-uploaded ZIPs of
# stub STLs would otherwise log one debug line per
# file via the empty-mesh branch in trimesh.load.
if generate_stl_thumbnails and len(file_content) >= MIN_USABLE_STL_BYTES:
thumbnail_path = generate_stl_thumbnail(file_path, thumbnails_dir)
# Create database entry (store relative paths for portability)
library_file = LibraryFile(
folder_id=target_folder_id,
filename=filename,
file_path=to_relative_path(file_path),
file_type=file_type,
file_size=len(file_content),
file_hash=file_hash,
thumbnail_path=to_relative_path(thumbnail_path) if thumbnail_path else None,
file_metadata=_without_print_name(metadata) if metadata else None,
created_by_id=current_user.id if current_user else None,
)
db.add(library_file)
await db.flush()
await db.refresh(library_file)
extracted_files.append(
ZipExtractResult(
filename=filename,
file_id=library_file.id,
folder_id=target_folder_id,
)
)
# Commit after each file to release database lock
# This prevents long-running transactions from blocking other requests
await db.commit()
except Exception as e:
logger.error("Failed to extract %s: %s", zip_path, e)
errors.append(ZipExtractError(filename=os.path.basename(zip_path), error=str(e)))
# Rollback the failed file but continue with others
await db.rollback()
return ZipExtractResponse(
extracted=len(extracted_files),
folders_created=folders_created,
files=extracted_files,
errors=errors,
)
except zipfile.BadZipFile:
raise HTTPException(status_code=400, detail="Invalid or corrupted ZIP file")
except Exception as e:
logger.error("ZIP extraction failed: %s", e, exc_info=True)
raise HTTPException(status_code=500, detail=f"ZIP extraction failed: {str(e)}")
finally:
# Clean up temp file
try:
os.unlink(tmp_path)
except OSError:
pass # Best-effort temp file cleanup; ignore if already removed
# ============ STL Thumbnail Batch Generation ============
@router.post("/generate-stl-thumbnails", response_model=BatchThumbnailResponse)
async def batch_generate_stl_thumbnails(
request: BatchThumbnailRequest,
db: AsyncSession = Depends(get_db),
auth_result: tuple[User | None, bool] = Depends(
require_ownership_permission(
Permission.LIBRARY_UPDATE_ALL,
Permission.LIBRARY_UPDATE_OWN,
)
),
):
"""Generate thumbnails for STL and PDF files in batch.
With library:update_all this covers every matching file; with only
library:update_own it is narrowed to the caller's own files, the same
rule as update_file. The File Manager offers the toolbar button and the
per-file "Generate Thumbnail" entry to update_own users, and both land
here.
PDFs are included so the ones added before server-side PDF thumbnails
existed can be backfilled without opening each preview. The route keeps
its name for API compatibility.
Can generate thumbnails for:
- Specific file IDs (file_ids)
- All STL/PDF files in a folder (folder_id)
- All STL/PDF files missing thumbnails (all_missing=True)
"""
thumbnails_dir = get_library_thumbnails_dir()
results: list[BatchThumbnailResult] = []
# Build query based on request
query = LibraryFile.active().where(LibraryFile.file_type.in_(("stl", "pdf")))
user, can_modify_all = auth_result
if not can_modify_all:
query = query.where(LibraryFile.created_by_id == user.id)
if request.file_ids:
# Specific files
query = query.where(LibraryFile.id.in_(request.file_ids))
elif request.folder_id is not None:
# All STL/PDF files in a specific folder
query = query.where(LibraryFile.folder_id == request.folder_id)
if not request.all_missing:
# If not specifically asking for missing thumbnails, get all
pass
else:
query = query.where(LibraryFile.thumbnail_path.is_(None))
elif request.all_missing:
# All STL/PDF files without thumbnails
query = query.where(LibraryFile.thumbnail_path.is_(None))
else:
# No criteria specified - return empty
return BatchThumbnailResponse(
processed=0,
succeeded=0,
failed=0,
results=[],
)
result = await db.execute(query)
target_files = result.scalars().all()
succeeded = 0
failed = 0
for target_file in target_files:
file_path = to_absolute_path(target_file.file_path)
if not file_path or not file_path.exists():
results.append(
BatchThumbnailResult(
file_id=target_file.id,
filename=target_file.filename,
success=False,
error="File not found on disk",
)
)
failed += 1
continue
try:
if target_file.file_type == "pdf":
thumbnail_path = await asyncio.to_thread(generate_pdf_thumbnail, file_path, thumbnails_dir)
else:
thumbnail_path = generate_stl_thumbnail(file_path, thumbnails_dir)
if thumbnail_path:
# Update database with relative path
target_file.thumbnail_path = to_relative_path(thumbnail_path)
await db.flush()
results.append(
BatchThumbnailResult(
file_id=target_file.id,
filename=target_file.filename,
success=True,
)
)
succeeded += 1
else:
results.append(
BatchThumbnailResult(
file_id=target_file.id,
filename=target_file.filename,
success=False,
error="Thumbnail generation failed",
)
)
failed += 1
except Exception as e:
logger.error("Failed to generate thumbnail for %s: %s", target_file.filename, e)
results.append(
BatchThumbnailResult(
file_id=target_file.id,
filename=target_file.filename,
success=False,
error=str(e),
)
)
failed += 1
await db.commit()
return BatchThumbnailResponse(
processed=len(target_files),
succeeded=succeeded,
failed=failed,
results=results,
)
# ============ Queue Operations ============
# NOTE: These routes must be defined BEFORE /files/{file_id} to avoid path parameter conflicts
def is_sliced_file(filename: str) -> bool:
"""Check if a file is a sliced (printable) file.
Sliced files are:
- .gcode files
- .3mf files that contain '.gcode.' in the name (e.g., filename.gcode.3mf)
"""
lower = filename.lower()
return lower.endswith(".gcode") or ".gcode." in lower
@router.post("/files/combine", response_model=FileUploadResponse)
async def combine_files(
request: CombineFilesRequest,
db: AsyncSession = Depends(get_db),
current_user: User | None = Depends(require_permission_if_auth_enabled(Permission.LIBRARY_UPLOAD)),
):
"""Combine STL library files into one multi-object 3MF.
The slicer sidecar takes one model per slice, so putting several separate
STLs (or several copies of one, #2999) on one plate means building that
file first. The result is a new library file that slices like any other
3MF; with auto-arrange on, the slicer lays the objects out on the bed.
The sources are left untouched.
"""
from backend.app.services.mesh_combine import CombinePart, MeshCombineError, combine_parts_to_3mf
filename = request.filename.strip()
if not filename.lower().endswith(".3mf"):
filename = f"{filename}.3mf"
try:
validate_print_filename(filename)
except InvalidFilenameError as e:
raise HTTPException(status_code=400, detail=str(e)) from e
if request.folder_id is not None:
folder = (
await db.execute(select(LibraryFolder).where(LibraryFolder.id == request.folder_id))
).scalar_one_or_none()
if folder is None:
raise HTTPException(status_code=404, detail="Folder not found")
# Same per-row visibility the slice route applies: a READ_OWN caller must
# not be able to pull another user's model into their own file by raw id.
can_read_all = current_user is None or current_user.has_permission(Permission.LIBRARY_READ_ALL.value)
# The same file listed twice is one object with the copies added up, so
# its mesh is loaded and stored once. Order follows first appearance.
copies_by_id: dict[int, int] = {}
for item in request.items:
copies_by_id[item.file_id] = copies_by_id.get(item.file_id, 0) + item.copies
rows = (await db.execute(LibraryFile.active().where(LibraryFile.id.in_(copies_by_id)))).scalars().all()
by_id = {row.id: row for row in rows}
# Gate every source before touching any of them on disk, so the answer for
# a file the caller can't see is the same 404 whatever else is in the list.
sources = [_ensure_library_file_visible(by_id.get(file_id), current_user, can_read_all) for file_id in copies_by_id]
parts: list[CombinePart] = []
for lib_file in sources:
if not lib_file.filename.lower().endswith(".stl"):
raise HTTPException(status_code=400, detail=f"Only STL files can be combined: {lib_file.filename}")
src_path = _resolve_source_disk_path(lib_file)
if src_path is None or not src_path.exists():
raise HTTPException(status_code=404, detail=f"Source file missing on disk: {lib_file.filename}")
parts.append(CombinePart(name=lib_file.filename, path=src_path, copies=copies_by_id[lib_file.id]))
try:
content = await asyncio.to_thread(combine_parts_to_3mf, parts)
except MeshCombineError as e:
raise HTTPException(status_code=400, detail=str(e)) from e
# The preview is embedded as Metadata/thumbnail.png by combine_parts_to_3mf,
# so ThreeMFParser picks it up here like any other 3MF's thumbnail. Loading
# the combined file back to render one would expand every copy.
library_file, _ = await save_3mf_bytes_to_library(
db,
file_bytes=content,
filename=filename,
folder_id=request.folder_id,
source_type="combined",
owner_id=current_user.id if current_user else None,
)
return FileUploadResponse(
id=library_file.id,
filename=library_file.filename,
file_type=library_file.file_type,
file_size=library_file.file_size,
thumbnail_path=library_file.thumbnail_path,
metadata=library_file.file_metadata,
)
@router.post("/files/add-to-queue", response_model=AddToQueueResponse)
async def add_files_to_queue(
request: AddToQueueRequest,
db: AsyncSession = Depends(get_db),
current_user: User | None = Depends(require_permission_if_auth_enabled(Permission.QUEUE_CREATE)),
):
"""Add library files to the print queue.
Only sliced files (.gcode or .gcode.3mf) can be added to the queue.
The archive will be created automatically when the print starts.
A caller may name a printer or a target model for the whole batch; with
neither, each file is aimed at the model it says it was sliced for. The
gates are the ones ``POST /queue/`` applies to a single item, because an
item that reaches the scheduler through this route has to be as printable
as one that reaches it through that one (#3112).
"""
added: list[AddToQueueResult] = []
errors: list[AddToQueueError] = []
# Batch-level targeting. Rejected outright rather than per file: the whole
# request names one destination, so a bad one is not a property of any
# single file and reporting it fourteen times would say nothing extra.
target_model_norm = normalize_model_name(request.target_model)
if request.printer_id is not None and target_model_norm:
raise HTTPException(400, "Cannot specify both printer_id and target_model")
if request.printer_id is not None:
printer_row = (await db.execute(select(Printer).where(Printer.id == request.printer_id))).scalar_one_or_none()
if not printer_row:
raise HTTPException(400, "Printer not found")
# Active printers of every model, read once, and only when the batch has no
# printer of its own -- with one named, neither the check below nor the
# inference in the loop consults it. The explicit target is validated for
# the same reason POST /queue/ validates: a model nobody owns is a queue
# item that waits forever. The inferred target reads the same set and
# silently declines when it finds nothing, because there, owning no such
# printer is the user's situation rather than their mistake -- the file
# still queues, as the unassigned row it has always been.
active_models: set[str] = set()
if request.printer_id is None:
active_models = {
model
for (model,) in (
await db.execute(select(Printer.model).where(Printer.is_active == True).distinct()) # noqa: E712
).all()
if model
}
if target_model_norm and target_model_norm not in active_models:
raise HTTPException(400, f"No active printers for model: {target_model_norm}")
# Get all requested files
result = await db.execute(LibraryFile.active().where(LibraryFile.id.in_(request.file_ids)))
files = {f.id: f for f in result.scalars().all()}
# Ownership-scoped reads apply here as everywhere else in this module: a
# file the caller may not read is a file they may not print. Dropped from
# the map rather than refused by name, so the per-file error below is the
# same "File not found" an unknown id gets and the response says nothing
# about which ids exist. Ownerless rows need LIBRARY_READ_ALL, matching
# _ensure_library_file_visible.
if current_user is not None and not current_user.has_permission(Permission.LIBRARY_READ_ALL.value):
files = {fid: f for fid, f in files.items() if f.created_by_id == current_user.id}
# Project attribution (#1897): a file queued from a project-linked folder
# inherits that project, so the resulting archive counts toward the
# project's progress. A file's own project link wins over its folder's.
folder_ids = {f.folder_id for f in files.values() if f.folder_id is not None}
folder_projects: dict[int, int | None] = {}
if folder_ids:
folder_result = await db.execute(
select(LibraryFolder.id, LibraryFolder.project_id).where(LibraryFolder.id.in_(folder_ids))
)
folder_projects = dict(folder_result.all())
# Get max position for queue ordering
pos_result = await db.execute(select(func.coalesce(func.max(PrintQueueItem.position), 0)))
max_position = pos_result.scalar() or 0
# There is no per-job ask-for-outcome toggle on a bulk add, so the rows take
# the same default the print dialog seeds itself from (#1898).
confirm_outcome = await confirm_outcome_for_new_queue_item(db)
for file_id in request.file_ids:
lib_file = files.get(file_id)
if not lib_file:
errors.append(AddToQueueError(file_id=file_id, filename="(not found)", error="File not found"))
continue
# Validate file is sliced
if not is_sliced_file(lib_file.filename):
errors.append(
AddToQueueError(
file_id=file_id,
filename=lib_file.filename,
error="Not a sliced file. Only .gcode or .gcode.3mf files can be printed.",
)
)
continue
try:
# Verify file exists on disk
file_path = Path(app_settings.base_dir) / lib_file.file_path
if not file_path.exists():
errors.append(
AddToQueueError(file_id=file_id, filename=lib_file.filename, error="File not found on disk")
)
continue
# The Bambu SD card is FAT32/exFAT, so an illegal character 553s at
# upload time. POST /queue/ rejects those at queue time (#1540) and
# this route did not, which turned a nameable mistake into a print
# that failed hours later.
try:
validate_print_filename(lib_file.filename)
except InvalidFilenameError as e:
errors.append(AddToQueueError(file_id=file_id, filename=lib_file.filename, error=str(e)))
continue
# Where this file is aimed. An explicit printer wins; an explicit
# model applies to every file and has to be one this file can
# legally run on; with neither, the file's own declaration is used
# when some active printer answers to it.
item_printer_id = request.printer_id
item_target_model: str | None = None
if item_printer_id is None:
if target_model_norm:
sliced_for = (lib_file.file_metadata or {}).get("sliced_for_model")
if not is_gcode_compatible(sliced_for, target_model_norm):
errors.append(
AddToQueueError(
file_id=file_id,
filename=lib_file.filename,
error=(
f"File was sliced for {sliced_for} and cannot be dispatched to "
f"{target_model_norm} printers"
),
)
)
continue
item_target_model = target_model_norm
else:
inferred = resolve_variant_model(lib_file)
item_target_model = inferred if inferred in active_models else None
# Filament the scheduler must match before handing a model-based
# item to hardware. Without it the item goes to whichever printer
# of that model is idle, whatever is loaded in it.
required_filament_types = None
if item_target_model:
# POST /queue/'s own extractor, borrowed rather than
# reimplemented: a second copy of this rule is a second thing
# to keep in step.
#
# Off the loop, unlike there: that route parses one 3MF per
# request and this one parses every file in the batch, so on a
# bulk add of a few hundred -- especially from an external
# folder on a NAS -- the zip reads add up to a stall the whole
# event loop takes, status ingest included.
filament_types = await asyncio.to_thread(_extract_filament_types_from_3mf, file_path)
if filament_types:
required_filament_types = json.dumps(filament_types)
# Create queue item referencing library file (archive created at print start)
max_position += 1
queue_item = PrintQueueItem(
printer_id=item_printer_id,
target_model=item_target_model,
required_filament_types=required_filament_types,
library_file_id=file_id,
project_id=lib_file.project_id
or (folder_projects.get(lib_file.folder_id) if lib_file.folder_id is not None else None),
position=max_position,
status="pending",
confirm_outcome=confirm_outcome,
# Without this the row is ownerless, and `queue:read_own` filters
# on `created_by_id` — so the user who queued the file could not
# see it in their own queue.
created_by_id=current_user.id if current_user else None,
)
db.add(queue_item)
await db.flush() # Get queue_item.id
added.append(
AddToQueueResult(
file_id=file_id,
filename=lib_file.filename,
queue_item_id=queue_item.id,
)
)
except Exception as e:
logger.exception("Error adding file %s to queue", file_id)
errors.append(AddToQueueError(file_id=file_id, filename=lib_file.filename, error=str(e)))
# Nothing queued and something to say about why. Returning 200 here is what
# made this look like a working call that quietly did nothing: a client that
# checks the status code sees success, and the reasons sit in a body it had
# no cause to read (#3112). Partial success stays 200 -- items really were
# created, and the per-file errors belong with them.
if not added and errors:
raise HTTPException(
400,
detail={
"message": "No files could be added to the queue.",
"errors": [e.model_dump() for e in errors],
},
)
await db.commit()
return AddToQueueResponse(added=added, errors=errors)
@router.get("/files/{file_id}/plates")
async def get_library_file_plates(
file_id: int,
db: AsyncSession = Depends(get_db),
auth_result: tuple[User | None, bool] = Depends(
require_ownership_permission(
Permission.LIBRARY_READ_ALL,
Permission.LIBRARY_READ_OWN,
)
),
):
"""Get available plates from a multi-plate 3MF library file.
Returns a list of plates with their index, name, thumbnail availability,
and filament requirements. For single-plate exports, returns a single plate.
"""
import json
import defusedxml.ElementTree as ET
user, can_read_all = auth_result
# Get the library file
result = await db.execute(LibraryFile.active().where(LibraryFile.id == file_id))
lib_file = _ensure_library_file_visible(result.scalar_one_or_none(), user, can_read_all)
if not lib_file:
raise HTTPException(status_code=404, detail="File not found")
file_path = Path(app_settings.base_dir) / lib_file.file_path
if not file_path.exists():
raise HTTPException(status_code=404, detail="File not found on disk")
# Only 3MF files have plates
if not lib_file.filename.lower().endswith(".3mf"):
return {"file_id": file_id, "filename": lib_file.filename, "plates": [], "is_multi_plate": False}
plates = []
# Printer / process preset names the 3MF was prepared with — used by the
# SliceModal to default its dropdowns (#1325). Initialised here so the
# final return never raises NameError when the file isn't a valid zip.
embedded_presets: dict[str, str | None] = {"printer": None, "process": None}
# Process settings the designer changed away from the stock preset (#2622).
# Offered in the SliceModal so a cross-printer re-slice can carry them
# instead of silently losing them to the picked process profile.
design_overrides: list[dict] = []
try:
with zipfile.ZipFile(file_path, "r") as zf:
namelist = zf.namelist()
embedded_presets = extract_embedded_presets_from_3mf(zf)
if _PROJECT_SETTINGS_PATH in namelist:
try:
design_overrides = [
o._asdict()
for o in overrides_from_config(json.loads(zf.read(_PROJECT_SETTINGS_PATH).decode("utf-8")))
]
except (ValueError, OSError, KeyError):
design_overrides = []
# Find all plate gcode files to determine available plates
gcode_files = [n for n in namelist if n.startswith("Metadata/plate_") and n.endswith(".gcode")]
# If no gcode is present (source-only or unsliced), fall back to plate JSON/PNG
plate_indices: list[int] = []
if gcode_files:
# Extract plate indices from gcode filenames
for gf in gcode_files:
try:
plate_str = gf[15:-6] # Remove "Metadata/plate_" and ".gcode"
plate_indices.append(int(plate_str))
except ValueError:
pass # Skip gcode file with non-numeric plate index
else:
plate_json_files = [n for n in namelist if n.startswith("Metadata/plate_") and n.endswith(".json")]
plate_png_files = [
n
for n in namelist
if n.startswith("Metadata/plate_")
and n.endswith(".png")
and "_small" not in n
and "no_light" not in n
]
plate_name_candidates = plate_json_files + plate_png_files
plate_re = re.compile(r"^Metadata/plate_(\d+)\.(json|png)$")
seen_indices: set[int] = set()
for name in plate_name_candidates:
match = plate_re.match(name)
if match:
try:
index = int(match.group(1))
except ValueError:
continue
if index in seen_indices:
continue
seen_indices.add(index)
plate_indices.append(index)
if not plate_indices:
# No plate metadata found
return {"file_id": file_id, "filename": lib_file.filename, "plates": [], "is_multi_plate": False}
plate_indices.sort()
# Parse model_settings.config for plate names + object assignments
plate_names = {}
plate_object_ids: dict[int, list[str]] = {}
object_names_by_id: dict[str, str] = {}
if "Metadata/model_settings.config" in namelist:
try:
model_content = zf.read("Metadata/model_settings.config").decode()
model_root = ET.fromstring(model_content)
for obj_elem in model_root.findall(".//object"):
obj_id = obj_elem.get("id")
if not obj_id:
continue
name_meta = obj_elem.find("metadata[@key='name']")
obj_name = name_meta.get("value") if name_meta is not None else None
if obj_name:
object_names_by_id[obj_id] = obj_name
for plate_elem in model_root.findall(".//plate"):
plater_id = None
plater_name = None
for meta in plate_elem.findall("metadata"):
key = meta.get("key")
value = meta.get("value")
if key == "plater_id" and value:
try:
plater_id = int(value)
except ValueError:
pass # Ignore plate with non-numeric plater_id
elif key == "plater_name" and value:
plater_name = value.strip()
if plater_id is not None and plater_name:
plate_names[plater_id] = plater_name
if plater_id is not None:
for instance_elem in plate_elem.findall("model_instance"):
for inst_meta in instance_elem.findall("metadata"):
if inst_meta.get("key") == "object_id":
obj_id = inst_meta.get("value")
if not obj_id:
continue
plate_object_ids.setdefault(plater_id, [])
if obj_id not in plate_object_ids[plater_id]:
plate_object_ids[plater_id].append(obj_id)
except Exception:
pass # model_settings.config is optional; skip if missing or malformed
# Parse slice_info.config for plate metadata
plate_metadata = {}
if "Metadata/slice_info.config" in namelist:
content = zf.read("Metadata/slice_info.config").decode()
root = ET.fromstring(content)
for plate_elem in root.findall(".//plate"):
plate_info = {"filaments": [], "prediction": None, "weight": None, "name": None, "objects": []}
plate_index = None
for meta in plate_elem.findall("metadata"):
key = meta.get("key")
value = meta.get("value")
if key == "index" and value:
try:
plate_index = int(value)
except ValueError:
pass # Ignore plate with non-numeric index
elif key == "prediction" and value:
try:
plate_info["prediction"] = int(value)
except ValueError:
pass # Leave prediction as None if not a valid integer
elif key == "weight" and value:
try:
plate_info["weight"] = float(value)
except ValueError:
pass # Leave weight as None if not a valid number
# Get filaments used in this plate
for filament_elem in plate_elem.findall("filament"):
filament_id = filament_elem.get("id")
filament_type = filament_elem.get("type", "")
filament_color = filament_elem.get("color", "")
used_g = filament_elem.get("used_g", "0")
used_m = filament_elem.get("used_m", "0")
try:
used_grams = float(used_g)
except (ValueError, TypeError):
used_grams = 0
if used_grams > 0 and filament_id:
plate_info["filaments"].append(
{
"slot_id": int(filament_id),
"type": filament_type,
"color": filament_color,
"used_grams": round(used_grams, 1),
"used_meters": float(used_m) if used_m else 0,
}
)
plate_info["filaments"].sort(key=lambda x: x["slot_id"])
# Collect object names
for obj_elem in plate_elem.findall("object"):
obj_name = obj_elem.get("name")
if obj_name and obj_name not in plate_info["objects"]:
plate_info["objects"].append(obj_name)
# Set plate name
if plate_index is not None:
custom_name = plate_names.get(plate_index)
if custom_name:
plate_info["name"] = custom_name
elif plate_info["objects"]:
plate_info["name"] = plate_info["objects"][0]
plate_metadata[plate_index] = plate_info
# Parse plate_*.json for object lists when slice_info is missing
plate_json_objects: dict[int, list[str]] = {}
for name in namelist:
match = re.match(r"^Metadata/plate_(\d+)\.json$", name)
if not match:
continue
try:
plate_index = int(match.group(1))
except ValueError:
continue
try:
payload = json.loads(zf.read(name).decode())
bbox_objects = payload.get("bbox_objects", [])
names: list[str] = []
for obj in bbox_objects:
obj_name = obj.get("name") if isinstance(obj, dict) else None
if obj_name and obj_name not in names:
names.append(obj_name)
if names:
plate_json_objects[plate_index] = names
except Exception:
continue
# Build plate list
for idx in plate_indices:
meta = plate_metadata.get(idx, {})
has_thumbnail = f"Metadata/plate_{idx}.png" in namelist
objects = meta.get("objects", [])
if not objects:
objects = plate_json_objects.get(idx, [])
if not objects and plate_object_ids.get(idx):
objects = [
object_names_by_id.get(obj_id, f"Object {obj_id}") for obj_id in plate_object_ids.get(idx, [])
]
plate_name = meta.get("name")
if not plate_name:
plate_name = plate_names.get(idx)
if not plate_name and objects:
plate_name = objects[0]
plates.append(
{
"index": idx,
"name": plate_name,
"objects": objects,
"object_count": len(objects),
"has_thumbnail": has_thumbnail,
"thumbnail_url": f"/api/v1/library/files/{file_id}/plate-thumbnail/{idx}"
if has_thumbnail
else None,
"print_time_seconds": meta.get("prediction"),
"filament_used_grams": meta.get("weight"),
"filaments": meta.get("filaments", []),
}
)
except Exception as e:
logger.warning("Failed to parse plates from library file %s: %s", file_id, e)
return {
"file_id": file_id,
"filename": lib_file.filename,
"plates": plates,
"is_multi_plate": len(plates) > 1,
"embedded_printer": embedded_presets["printer"],
"embedded_process": embedded_presets["process"],
"design_overrides": design_overrides,
}
@router.get("/files/{file_id}/plate-thumbnail/{plate_index}")
async def get_library_file_plate_thumbnail(
file_id: int,
plate_index: int,
db: AsyncSession = Depends(get_db),
auth_result: tuple[User | None, bool] = Depends(
require_media_token_ownership(
Permission.LIBRARY_READ_ALL,
Permission.LIBRARY_READ_OWN,
)
),
):
"""Get the thumbnail image for a specific plate from a library file.
Ownership-gated on the same terms as the file itself (#3025).
"""
from starlette.responses import Response
user, can_read_all = auth_result
result = await db.execute(LibraryFile.active().where(LibraryFile.id == file_id))
lib_file = _ensure_library_file_visible(result.scalar_one_or_none(), user, can_read_all)
file_path = Path(app_settings.base_dir) / lib_file.file_path
if not file_path.exists():
raise HTTPException(status_code=404, detail="File not found on disk")
try:
with zipfile.ZipFile(file_path, "r") as zf:
thumb_path = f"Metadata/plate_{plate_index}.png"
if thumb_path in zf.namelist():
data = zf.read(thumb_path)
return Response(content=data, media_type="image/png")
except Exception:
pass # Archive unreadable or thumbnail missing; fall through to 404
raise HTTPException(status_code=404, detail=f"Thumbnail for plate {plate_index} not found")
async def _try_preview_slice_filaments(
db: AsyncSession,
*,
kind: str,
source_id: int,
plate_id: int,
file_path: Path,
request_id: str | None = None,
) -> list[dict] | None:
"""Run a preview slice via the user's configured sidecar. Same shape as
the matching helper in archives.py — see that module for rationale.
``request_id``: when supplied, forwarded to the sidecar so the
SliceModal's inline spinner + toast can poll the matching progress
endpoint and show "Generating G-code (45%)" for the preview as well.
"""
from backend.app.api.routes.settings import get_setting
from backend.app.services.slice_preview import get_preview_filaments
from backend.app.services.slicer_api import get_stall_timeout_seconds
preferred = (await get_setting(db, "preferred_slicer")) or "bambu_studio"
if preferred == "orcaslicer":
configured = await get_setting(db, "orcaslicer_api_url")
api_url = (configured or app_settings.slicer_api_url).strip()
elif preferred == "bambu_studio":
configured = await get_setting(db, "bambu_studio_api_url")
api_url = (configured or app_settings.bambu_studio_api_url).strip()
else:
return None
if not api_url:
return None
try:
file_bytes = file_path.read_bytes()
except OSError:
return None
return await get_preview_filaments(
kind=kind,
source_id=source_id,
plate_id=plate_id,
file_bytes=file_bytes,
file_name=file_path.name,
api_url=api_url,
request_id=request_id,
timeout_seconds=await get_stall_timeout_seconds(db),
)
@router.get("/files/{file_id}/filament-requirements")
async def get_library_file_filament_requirements(
file_id: int,
plate_id: int | None = None,
request_id: str | None = None,
full_slots: bool = False,
db: AsyncSession = Depends(get_db),
auth_result: tuple[User | None, bool] = Depends(
require_ownership_permission(
Permission.LIBRARY_READ_ALL,
Permission.LIBRARY_READ_OWN,
)
),
):
"""Get filament requirements from a library file.
Parses the 3MF file to extract filament slot IDs, types, colors, and usage.
This enables AMS slot assignment when printing from the file manager.
Args:
file_id: The library file ID
plate_id: Optional plate index to get filaments for a specific plate
full_slots: Return one entry per *project* slot rather than only the
slots the plate consumes. See :func:`_expand_to_project_slots`.
Only the slice modal wants this; print-time AMS matching must keep
the used-only list.
"""
import defusedxml.ElementTree as ET
user, can_read_all = auth_result
# Get the library file
result = await db.execute(LibraryFile.active().where(LibraryFile.id == file_id))
lib_file = _ensure_library_file_visible(result.scalar_one_or_none(), user, can_read_all)
# Get the full file path
file_path = Path(app_settings.base_dir) / lib_file.file_path
if not file_path.exists():
raise HTTPException(status_code=404, detail="File not found on disk")
# Only 3MF files have parseable filament info
if not lib_file.filename.lower().endswith(".3mf"):
return {"file_id": file_id, "filename": lib_file.filename, "plate_id": plate_id, "filaments": []}
filaments = []
try:
with zipfile.ZipFile(file_path, "r") as zf:
# Parse slice_info.config for filament requirements
if "Metadata/slice_info.config" in zf.namelist():
content = zf.read("Metadata/slice_info.config").decode()
root = ET.fromstring(content)
if plate_id is not None:
# Find filaments for specific plate
for plate_elem in root.findall(".//plate"):
# Check if this is the requested plate
plate_index = None
for meta in plate_elem.findall("metadata"):
if meta.get("key") == "index":
try:
plate_index = int(meta.get("value", ""))
except ValueError:
pass # Skip plate with non-numeric index value
break
if plate_index == plate_id:
# Extract filaments from this plate
for filament_elem in plate_elem.findall("filament"):
filament_id = filament_elem.get("id")
filament_type = filament_elem.get("type", "")
filament_color = filament_elem.get("color", "")
used_g = filament_elem.get("used_g", "0")
used_m = filament_elem.get("used_m", "0")
tray_info_idx = filament_elem.get("tray_info_idx", "")
try:
used_grams = float(used_g)
except (ValueError, TypeError):
used_grams = 0
if used_grams > 0 and filament_id:
filaments.append(
{
"slot_id": int(filament_id),
"type": filament_type,
"color": filament_color,
"used_grams": round(used_grams, 1),
"used_meters": float(used_m) if used_m else 0,
"tray_info_idx": tray_info_idx,
# Sliced output already pre-filtered by used_g>0,
# so every entry that survives is in fact used by
# this plate. Print-dispatch consumers ignore the
# flag; SliceModal uses it to enable/disable rows.
"used_in_plate": True,
}
)
break
else:
# Extract all filaments with used_g > 0 (for single-plate or overview)
for filament_elem in root.findall(".//filament"):
filament_id = filament_elem.get("id")
filament_type = filament_elem.get("type", "")
filament_color = filament_elem.get("color", "")
used_g = filament_elem.get("used_g", "0")
used_m = filament_elem.get("used_m", "0")
tray_info_idx = filament_elem.get("tray_info_idx", "")
try:
used_grams = float(used_g)
except (ValueError, TypeError):
used_grams = 0
if used_grams > 0 and filament_id:
filaments.append(
{
"slot_id": int(filament_id),
"type": filament_type,
"color": filament_color,
"used_grams": round(used_grams, 1),
"used_meters": float(used_m) if used_m else 0,
"tray_info_idx": tray_info_idx,
"used_in_plate": True,
}
)
# Re-slicing a source that already carries slice_info (#2712).
# The block above answers "what does this plate consume", which is
# what print-time AMS matching needs. The slice modal needs "what
# slots exist", because its list is positional and the CLI binds
# entry N to slot N — so a source using only slot 4 handed the
# user's single pick to slot 1 and sliced slot 4 with the source's
# embedded default. Widen here rather than in the modal so the
# print path keeps the narrow list it depends on.
if full_slots and filaments:
filaments = expand_to_project_slots(zf, filaments)
# Unsliced project files: slice_info had no per-plate data.
# Return the FULL project_settings.config AMS slot list so
# the slicer CLI receives a profile for every project slot
# (otherwise it silently fills the gap from embedded
# defaults — surfaces as "I picked white but the print has
# grey" because the source's grey support filament leaks
# into the output). Use the preview slice to mark which
# slots the picked plate actually consumes; the SliceModal
# disables the unused rows so the user only interacts with
# the dropdowns that matter, while the backend still has
# the complete list to pass to the CLI.
if not filaments:
project_filaments = extract_project_filaments_from_3mf(zf)
used_slot_ids: set[int] = set()
if project_filaments and plate_id is not None:
preview = await _try_preview_slice_filaments(
db,
kind="library_file",
source_id=file_id,
plate_id=plate_id,
file_path=file_path,
request_id=request_id,
)
if preview is not None:
used_slot_ids = {f["slot_id"] for f in preview}
# Default to "every slot is used" when preview-slice
# didn't produce data: better to over-enable dropdowns
# than under-enable and have the user unable to pick a
# filament the plate actually uses.
fallback_all_used = not used_slot_ids
for f in project_filaments:
f["used_in_plate"] = fallback_all_used or f["slot_id"] in used_slot_ids
filaments = project_filaments
# Sort by slot ID
filaments.sort(key=lambda x: x["slot_id"])
# Enrich with nozzle mapping for dual-nozzle printers
nozzle_mapping = extract_nozzle_mapping_from_3mf(zf)
if nozzle_mapping:
for filament in filaments:
filament["nozzle_id"] = nozzle_mapping.get(filament["slot_id"])
# Nozzle-rack machines (#1784): the print dialog offers a rack
# position per filament group, which needs the group table as well
# as the carriage above.
annotate_rack_groups(filaments, file_path, plate_id)
except Exception as e:
logger.warning("Failed to parse filament requirements from library file %s: %s", file_id, e)
return {
"file_id": file_id,
"filename": lib_file.filename,
"plate_id": plate_id,
"filaments": filaments,
}
def _patch_process_bed_type(process_json: str, bed_type: str) -> str:
"""Overwrite ``curr_bed_type`` in a process-profile JSON before forwarding
to the slicer sidecar.
The slicer CLI reads the build-plate type from the process profile's
``curr_bed_type`` field. When the user picks a non-default plate in the
SliceModal (#1337), we patch the resolved JSON in place rather than
asking them to clone the preset just to switch a plate. Returns the
original string unchanged when the JSON can't be parsed or isn't a
dict — the slicer will then run with whatever the preset originally
specified, which is the safe fall-back path.
"""
try:
profile = json.loads(process_json)
except json.JSONDecodeError:
logger.warning("Bed-type override skipped: process profile is not valid JSON")
return process_json
if not isinstance(profile, dict):
return process_json
profile["curr_bed_type"] = bed_type
return json.dumps(profile)
def _source_plate_colours(model_bytes: bytes) -> list[str]:
"""Per-slot colours the source 3MF was designed with, or ``[]``.
Read from ``project_settings.config`` rather than ``slice_info.config``:
the latter records the colour the file was *last sliced* with, which for a
source that never carried one is the slicer's own #00AE42 default — the
exact value #2977 is about, so using it as a fallback would be circular.
STL and mesh-only 3MF sources have no project settings and yield ``[]``.
"""
from io import BytesIO
try:
with zipfile.ZipFile(BytesIO(model_bytes), "r") as zf:
return [str(f.get("color") or "") for f in extract_project_filaments_from_3mf(zf)]
except (zipfile.BadZipFile, OSError, ValueError):
return []
def _preset_default_colour(profile: dict) -> str:
"""A filament preset's own ``default_filament_colour``, or ``""``.
OrcaSlicer's third-party vendor profiles carry this; Bambu Studio's
bundled BBL filament profiles carry it nowhere (checked across the whole
shipped `resources/profiles/BBL/filament/` tree — zero occurrences), which
is why it can only ever be one link in the chain and never the whole fix.
It is read here and rewritten as ``filament_colour`` because the CLI does
not read it itself. Measured against a 02.08.02.61 sidecar: a profile
carrying only ``default_filament_colour: ["#FF00FF"]`` still slices to
``filament_colour: ["#00AE42"]``. Bambu Studio consumes the default in the
GUI when a project is created, not in ``--load-filaments``.
"""
raw = profile.get("default_filament_colour")
if isinstance(raw, list):
raw = raw[0] if raw else None
return raw.strip() if isinstance(raw, str) else ""
def _patch_filament_colours(
filament_jsons: list[str],
requested: list[str],
model_bytes: bytes,
) -> list[str]:
"""Write ``filament_colour`` onto each resolved filament profile (#2977).
Neither slicer stores a colour on a filament *preset* — it is a per-project
property their GUIs set from the plate — so a CLI slice with no colour
supplied records Bambu Studio's compiled-in default for every slot. That
default is `#00AE42`, which is why every internal-slicer output was green
regardless of the filament picked, and why the print dialog's AMS mapping
reported a colour mismatch against whatever was actually loaded.
Per slot, first non-empty of:
1. the caller's explicit colour (the SliceModal's per-slot swatch),
2. the preset's own ``default_filament_colour``,
3. the colour the source 3MF's plate was designed with.
All three empty means the slot is left untouched rather than being given a
guess: the slicer's default is then still wrong, but it is at least the
same wrong value the file would have had before this function existed.
Returns a new list; a profile that isn't parseable JSON is passed through
unchanged, on the same reasoning as ``_patch_process_bed_type`` — a colour
is not worth failing a slice that would otherwise succeed.
"""
source_colours = _source_plate_colours(model_bytes) if filament_jsons else []
patched: list[str] = []
for i, raw in enumerate(filament_jsons):
try:
profile = json.loads(raw)
except json.JSONDecodeError:
logger.warning("Filament colour skipped for slot %d: profile is not valid JSON", i + 1)
patched.append(raw)
continue
if not isinstance(profile, dict):
patched.append(raw)
continue
colour = (
(requested[i].strip() if i < len(requested) and requested[i] else "")
or _preset_default_colour(profile)
or (source_colours[i].strip() if i < len(source_colours) and source_colours[i] else "")
)
if not colour:
patched.append(raw)
continue
# One-element array: the same shape the CLI uses for every other
# per-filament field (`filament_type`, `filament_vendor`), and the
# shape a `--load-filaments` profile is parsed as. A bare string is
# accepted by the JSON parser but not by the config deserialiser.
profile["filament_colour"] = [colour]
patched.append(json.dumps(profile))
return patched
# Support-related keys we lift from the source 3MF's project_settings.config
# into the picked process preset before `--load-settings` sees it (#1881).
# BambuStudio's shipped process presets ("0.20mm Standard @BBL H2D" etc.)
# define `enable_support: 0` as their default — supports are a per-print
# decision, not a per-quality one. `--load-settings` is authoritative, so
# without preserving these fields the source's per-project support intent
# (supports on, PVA in the interface slot, tree vs normal) gets discarded
# and the slicer produces a single-material output with no supports at all.
_SOURCE_PROCESS_SUPPORT_KEYS_TO_PRESERVE = (
"enable_support",
"support_filament",
"support_interface_filament",
"support_type",
)
def _declined_source_keys(offered: list[DesignOverride], requested: list[str] | None) -> set[str]:
"""Settings the file offered and the caller left unticked (#2942).
The slice dialog lists what the designer changed and applies only the keys
that are switched on, so the answer to "which of these does this slice
want" is already in the request. This reads the other half of it — the
ones that were on offer and turned down — which the support carry-over
below must not put back.
``requested`` of ``None`` is a caller that predates the per-key choice and
so cannot have declined anything; an empty list is one that was shown the
file's settings and took none. Collapsing those two into "nothing
selected" is what made an empty panel indistinguishable from an old
client, and only one of them means the user said no.
"""
if requested is None:
return set()
return {override.key for override in offered} - set(requested)
def _patch_process_support_settings(
process_json: str,
source_3mf_bytes: bytes,
declined: set[str] | frozenset[str] = frozenset(),
) -> str:
"""Overlay the source 3MF's support configuration onto the process JSON.
The carry is deliberately one-way: a source can switch supports *on*,
never off (#2820). The original #1881 rule was "source wins in both
directions", which quietly stripped supports from every custom process
preset that enabled them — a MakerWorld download nearly always ships
`enable_support: 0`, so the reporter's own preset (supports on, normal
(auto)) came back out of the slicer disabled and set to tree(auto).
Nothing is lost by not carrying the off direction: a process preset
with supports *on* is by definition a deliberate user preset, since
Bambu's shipped ones all ship them off.
``declined`` names keys the caller offered the user as the file's own
(#2622) and that the user left unticked, which this carry must then not
reinstate behind their back (#2942). It is empty for a source that offers
nothing — an OrcaSlicer export carries no ``different_settings_to_system``,
so there is nothing to tick and #1881's blanket carry still applies — and
for a client that predates the per-key ticks.
Only fires on 3MF sources — STL / STEP don't carry `project_settings.
config`. Silently no-ops when the source doesn't have the config, has
a malformed one, or when the process JSON isn't parseable — the slice
then runs with the process preset's own defaults, which is the safe
fall-back for both this bug and the pre-fix behaviour.
"""
from io import BytesIO
try:
with zipfile.ZipFile(BytesIO(source_3mf_bytes), "r") as zf:
if "Metadata/project_settings.config" not in zf.namelist():
return process_json
src_cfg = json.loads(zf.read("Metadata/project_settings.config").decode("utf-8"))
except (zipfile.BadZipFile, json.JSONDecodeError, UnicodeDecodeError, OSError, KeyError):
return process_json
if not isinstance(src_cfg, dict):
return process_json
if not supports_enabled_in_config(src_cfg):
return process_json
try:
process_cfg = json.loads(process_json)
except json.JSONDecodeError:
return process_json
if not isinstance(process_cfg, dict):
return process_json
carried = {
key: src_cfg[key] for key in _SOURCE_PROCESS_SUPPORT_KEYS_TO_PRESERVE if key in src_cfg and key not in declined
}
if not carried:
return process_json
process_cfg.update(carried)
# Logged because this is the one layer of the process JSON the user
# can't see coming: the slice modal shows the picked preset's values,
# so a carried key silently disagrees with what was on screen.
logger.info(
"Carried support settings from the source 3MF onto the process preset: %s",
dict(sorted(carried.items())),
)
return json.dumps(process_cfg)
# The sidecar prefixes the slicer CLI's own error_string with this when the
# slicer ran and rejected the job (model off the bed, incompatible filament
# temps, range validation) — as opposed to the CLI crashing before it could
# evaluate the job at all.
_SLICER_REJECTION_MARKER = "Slicing failed with error from slicer:"
# The CLI writes its real diagnostic to stdout/stderr on the `[error]` level.
# Format is `[<timestamp>] [error] run <NNNN>: <message>` (or sometimes without
# the `run NNNN:` prefix). The bracketed timestamp is optional; the `[error]`
# tag is what we anchor on. Used to recover the actual rejection reason for
# the `error_string: "The input preset file is invalid and can not be parsed."`
# case (#1851) — the CLI emits that generic placeholder for every -5 exit
# including real preset-compat rejections, and the per-incident specifics
# only live in the stdout dump.
_CLI_ERROR_LINE_RE = re.compile(r"\[error\]\s*(?:run\s+\d+:\s*)?(.+?)\s*$", re.MULTILINE)
# The placeholder error_string Bambu Studio writes to result.json for any
# `--load-settings` parse / compat rejection (-5 exit). When the sidecar
# surfaces this, the real reason lives in the stdout `[error]` line that we
# mine via _CLI_ERROR_LINE_RE.
_INPUT_PRESET_INVALID_PLACEHOLDER = "The input preset file is invalid and can not be parsed."
def _slicer_rejection_message(error_text: str) -> str | None:
"""Extract the slicer's own rejection reason from a sidecar error string,
or ``None`` when the failure is not a slicer content rejection.
A content rejection means ``--load-settings`` *was* applied — the slicer
got far enough to evaluate the model against the chosen printer and say
no. Retrying with the 3MF's embedded settings would then only "succeed"
by silently reverting to the source file's original printer, masking the
real problem; such failures must reach the user instead.
When the sidecar's `error_string` is Bambu Studio's generic
"The input preset file is invalid and can not be parsed." placeholder
(#1851) — emitted for every -5 exit, including the actual preset-compat
rejections whose real reason is logged to stdout as
`[error] run NNNN: <diagnostic>` — prefer the stdout `[error]` line so
the user sees which preset clashed with which printer.
"""
if _SLICER_REJECTION_MARKER not in error_text:
return None
reason = error_text.split(_SLICER_REJECTION_MARKER, 1)[1]
# Mine the stdout/stderr dump for a more specific CLI diagnostic before
# we trim it off below. Done first so the lookup window covers the full
# response, not just the headline.
cli_diagnostic_match = _CLI_ERROR_LINE_RE.search(reason)
cli_diagnostic = cli_diagnostic_match.group(1).strip() if cli_diagnostic_match else None
# Trim the sidecar's trailing exit-code note and any stderr/stdout dump.
for cut in (": Slicer process failed", "\nstderr:", "\nstdout:"):
idx = reason.find(cut)
if idx != -1:
reason = reason[:idx]
reason = reason.strip() or None
# When the headline is Bambu Studio's catch-all placeholder, the real
# reason is in the stdout `[error]` line. Substitute it. The placeholder
# by itself tells the user nothing about why their slice was rejected.
if cli_diagnostic and (reason is None or reason == _INPUT_PRESET_INVALID_PLACEHOLDER):
return cli_diagnostic
return reason
async def _run_slicer_with_fallback(
db: AsyncSession,
*,
model_bytes: bytes,
model_filename: str,
request: SliceRequest,
current_user_id: int | None = None,
job_id: int | None = None,
):
"""Validate presets, dispatch to the right sidecar, run the slicer with
the auto-fallback for 3MF inputs whose `--load-settings` path crashes the
CLI. Returns ``(SliceResult, used_embedded_settings: bool)``. Raises
``HTTPException`` for any caller-facing error.
`current_user_id` is needed to resolve **cloud** presets — the cloud token
is per-user when auth is enabled. For the legacy / local-only path it can
be left ``None``.
`job_id`: when set, a request_id is generated and a parallel poller
pushes the sidecar's --pipe-fed progress events onto
``slice_dispatch.set_progress(job_id, ...)`` so the UI's persistent
toast can show "Generating G-code (75%)" instead of just elapsed
time. Pass None for synchronous routes that aren't tracked by the
dispatcher.
"""
from backend.app.api.routes.settings import get_setting
from backend.app.services.preset_resolver import resolve_preset_ref
from backend.app.services.slicer_api import (
SlicerApiServerError,
SlicerApiService,
SlicerApiUnavailableError,
SlicerInputError,
SlicerTimeoutError,
get_stall_timeout_seconds,
)
user: User | None = None
presets: dict[str, str] = {}
filament_jsons: list[str] = []
# Resolve each slot via the source-aware resolver. The schema
# validator has already normalised legacy `*_preset_id: int`
# fields into `PresetRef(source='local', id=str(int))`, so all
# three are guaranteed non-None here.
if current_user_id is not None:
user = await db.get(User, current_user_id)
refs = {
"printer": request.printer_preset,
"process": request.process_preset,
}
for slot, ref in refs.items():
assert ref is not None, "schema validator guarantees PresetRef is set"
presets[slot] = await resolve_preset_ref(db, user, ref, slot)
# Multi-color: resolve each filament slot in plate order. The schema
# validator backfilled `filament_presets` from the legacy `filament_preset`
# field for single-color callers, so this list is always non-empty.
for ref in request.filament_presets:
assert ref is not None, "schema validator guarantees filament list is non-None"
filament_jsons.append(await resolve_preset_ref(db, user, ref, "filament"))
# Give every slot a colour before anything else touches the list, so the
# unused-slot substitution below propagates a complete profile rather than
# one that still has to be patched afterwards (#2977).
filament_jsons = _patch_filament_colours(filament_jsons, request.filament_colours, model_bytes)
# Bed-type override (#1337): patch curr_bed_type onto the resolved
# process JSON so the slicer's StaticPrintConfig pass picks up the
# user's pick instead of whatever the process preset defaults to.
# Without this, slicing an STL of ABS onto a process preset whose
# default is "Cool Plate" fails with "Plate 1: Cool Plate does not
# support filament 1" — the reporter's exact scenario.
if request.bed_type:
presets["process"] = _patch_process_bed_type(presets["process"], request.bed_type)
# Slicer routing — pick the sidecar URL by preferred_slicer.
# The per-install URL setting (Settings UI → Slicer card) wins; an
# empty value falls back to the SLICER_API_URL / BAMBU_STUDIO_API_URL
# env defaults defined in core/config.py.
preferred = (await get_setting(db, "preferred_slicer")) or "bambu_studio"
if preferred == "orcaslicer":
configured = await get_setting(db, "orcaslicer_api_url")
api_url = (configured or app_settings.slicer_api_url).strip()
elif preferred == "bambu_studio":
configured = await get_setting(db, "bambu_studio_api_url")
api_url = (configured or app_settings.bambu_studio_api_url).strip()
else:
raise HTTPException(
status_code=400,
detail=f"Unknown preferred_slicer setting: '{preferred}'. Expected 'orcaslicer' or 'bambu_studio'.",
)
# Note: an earlier version of this code stripped Metadata/project_settings.
# config + model_settings.config + slice_info.config + cut_information.xml
# before forwarding the 3MF, the theory being that --load-settings would
# then take precedence cleanly. That theory was wrong: model_settings.
# config carries the plate definitions the CLI needs to map `--slice N`
# to a real plate, and slice_info / project_settings supply baseline
# config the CLI's StaticPrintConfig pass needs at all. Stripping ANY
# of them caused the CLI to silently exit immediately after
# "Initializing StaticPrintConfigs" — exit code 0, no result.json, no
# stderr — which Node's child_process treated as failure and Bambuddy
# then masked by falling back to slice_without_profiles using the
# un-stripped bytes (and the source's embedded printer). Net effect:
# every 3MF slice with profiles silently produced wrong-printer output.
# Forwarding the original bytes lets --load-settings override the
# specific fields the user changed (printer/process/filament) while
# the embedded plate / model definitions remain intact.
is_3mf = model_filename.lower().endswith(".3mf")
primary_bytes = model_bytes
if is_3mf:
# Strip inherit/unset sentinels from Metadata/project_settings.config
# so the CLI's StaticPrintConfig range validator accepts the file
# (#1201, #3030). Surgical — keeps the config present, just removes
# the offending keys; the supplied --load-settings (and the fallback's
# embedded values for keys we didn't touch) still drive the slice.
# The preview-slice path applies the same sanitiser in
# ``slice_preview.get_preview_filaments``.
primary_bytes = sanitize_project_settings_sentinels(primary_bytes)
# #2622: the process settings the file's designer moved off the stock
# preset. Read once — the support patch below needs to know which of
# them the user was shown, and the carry after it needs their values.
design_offered = extract_design_process_overrides(primary_bytes)
declined_from_file = _declined_source_keys(design_offered, request.design_overrides)
# #1881: preserve the source 3MF's support configuration on top of
# the picked process preset. Bambu's shipped process presets set
# `enable_support: 0` by default (supports are a per-print, not
# per-quality, decision); `--load-settings` is authoritative so
# without patching, the source's `enable_support: 1` + support-slot
# assignments get discarded and the slice comes out single-material
# with a PVA slot loaded but never used. Bounded by the ticks: this
# runs for a source that offers no per-key choice at all, and for the
# keys of one that does but whose ticks the user left on.
presets["process"] = _patch_process_support_settings(
presets["process"], primary_bytes, declined=declined_from_file
)
# Carry the designer's tweaks onto the picked preset. BambuStudio
# records exactly which keys deviate from the system preset in
# `different_settings_to_system`, so a MakerWorld author's 5 walls /
# 100% infill / 0.1mm first layer survive a re-slice for another printer
# instead of being flattened by --load-settings. Opt-in per key: only the
# keys the caller names are applied, and only if the source really lists
# them as changed. Runs after the #1881 support patch so an explicit
# design pick wins over the blanket support carry-over.
if request.design_overrides:
presets["process"] = apply_design_overrides(
presets["process"],
design_offered,
request.design_overrides,
)
# The user's own edits from the slice modal's settings panel. Applied last
# and for every model type (not just 3MF): unlike the two patches above this
# doesn't read anything out of the source file, it is what the user typed.
# Last write wins, so an explicit choice beats both the carried support
# config (#1881) and the designer's tweaks (#2622).
if request.process_overrides:
presets["process"] = apply_process_overrides(presets["process"], request.process_overrides)
used_embedded_settings = False
# "Slice as designed" (#2611): honour the file's embedded
# project_settings.config instead of the picked profile triplet. Only
# meaningful for a 3MF that actually carries embedded settings; the UI
# gates the toggle on the picked printer matching the design's target,
# so this path never re-targets across printer models.
embedded_mode = bool(request.use_embedded_settings and is_3mf)
# Bounds silence rather than total slicing time (#2730), so a heavy model
# that keeps reporting progress runs to completion however long it takes.
service = SlicerApiService(api_url, timeout_seconds=await get_stall_timeout_seconds(db))
# #1493: cross-nozzle-class re-slice (single <-> dual). Without
# intervention the slicer rejects with either "G-code in unprintable
# area of multi-extruder printers" (the source's X1C-coordinate layout
# lands in the H2D's per-nozzle dead zone) or — worse — segfaults
# inside ZFiller's polygon clipping when the geometry pipeline trips
# on the cross-class transition. Forwarding the sidecar's --arrange
# flag for these cases lets BambuStudio reposition objects for the
# target bed and reconcile the embedded project_settings.config
# against the new printer, the same way the GUI's "Switch Printer"
# operation does. --arrange WILL reposition objects, so we only
# enable it on a true class crossing — same-printer slices keep the
# user's deliberate layout. The bed-type and arrange flags are
# orthogonal so this decision doesn't interact with the #1337 build-
# plate override.
cross_class_arrange = False
if is_3mf:
from backend.app.services.slicer_3mf_convert import (
extract_source_printer_model,
)
from backend.app.utils.printer_models import is_dual_nozzle_model
source_model = extract_source_printer_model(primary_bytes)
target_model = await _resolve_target_printer_model(db, user, request)
if source_model and target_model and is_dual_nozzle_model(source_model) != is_dual_nozzle_model(target_model):
logger.info(
"Cross-nozzle-class re-slice (%s -> %s): enabling --arrange so BS reconciles "
"the embedded project layout against the target printer",
source_model,
target_model,
)
cross_class_arrange = True
# #2548: the user can also ask for either layout pass per-slice. Arrange
# is a union with the cross-class decision above — a user opt-out must
# not be able to switch off the flag that keeps a class-crossing slice
# from crashing — while orient is user-driven only.
arrange_flag = cross_class_arrange or request.auto_arrange
orient_flag = request.auto_orient
# When this slice is dispatcher-tracked, generate a request_id so
# the sidecar publishes progress under it, and wire a callback that
# forwards each frame onto SliceDispatchService.set_progress for the
# status-poll endpoint to surface to the UI.
progress_request_id: str | None = None
progress_callback = None
if job_id is not None:
from uuid import uuid4
from backend.app.services.slice_dispatch import slice_dispatch as _dispatch
progress_request_id = str(uuid4())
def _on_progress(snapshot: dict) -> None:
_dispatch.set_progress(job_id, snapshot)
progress_callback = _on_progress
# SliceModal lets the user pick a filament profile per slot, but each
# plate uses only a subset of the slots. The unused-slot dropdowns get
# whatever default the modal serves up — and a heterogeneous default
# (e.g. ABS in slot 2 next to a PLA in the used slot 1) makes
# BambuStudio reject the slice with "the temperature difference of
# the filaments used is too large" (exit 194) even though the G-code
# never touches the unused slot; a default scoped to another printer
# gets it rejected with "filament preset (slot N) is not compatible
# with printer …" (#2628). Replace unused-slot entries with the
# plate's lowest used slot before the real slice so the loaded set is
# materially homogeneous and printer-correct.
#
# ``plate`` is absent for single-plate and STL sources — the SliceModal
# skips the picker and omits the field — and absent means plate 1, the
# same reading as ``plate_num`` further down and as the schema's own
# description. Treating it as "unknown plate" instead is what left every
# single-plate 3MF unsubstituted (#2711): a MakerWorld project defining
# four filaments but painting only one reached the CLI with the other
# three still holding presets baked into the source for a different
# printer, and the slice died on the first of them.
#
# ``plate=0`` is the slice-all sentinel, not a plate: every slot is used
# by some plate, so there is nothing to substitute. It has to be excluded
# explicitly because the support-filament slots unioned in below are
# read from the project config and are not plate-scoped — they would
# survive the (empty) geometry lookup for plate 0 and become the anchor,
# collapsing every colour of a slice-all onto the support filament.
if is_3mf and request.plate != 0:
from backend.app.services.slicer_3mf_convert import substitute_unused_plate_filaments
filament_jsons = substitute_unused_plate_filaments(primary_bytes, request.plate or 1, filament_jsons)
# Arrange slice-all loop (#1493): when the user asks for ``plate=0``
# (all plates) AND arrange is on, ``--slice 0 --arrange 1``
# consolidates every plate's objects onto a single target bed (BS's
# ``--arrange`` is project-wide) — either packing them all together or
# rejecting with "Some objects are located over the boundary of the
# heated bed" when nothing fits. Slice each plate independently with
# ``--arrange 1`` and merge the per-plate outputs into one multi-plate
# 3MF instead. Slice-all without arrange goes through the regular path
# below — the sidecar's native ``--slice 0`` produces the right shape
# directly.
#
# Keyed on ``arrange_flag``, not just the cross-class decision: the
# project-wide collapse is a property of ``--arrange`` itself, so a
# user-requested arrange over all plates (#2548) hits it identically.
# Orient doesn't — it rotates objects where they stand and never moves
# one between plates — so it isn't part of this condition.
use_arrange_slice_all = arrange_flag and request.plate == 0 and request.export_3mf
try:
try:
if use_arrange_slice_all:
from backend.app.services.slicer_3mf_convert import (
count_plates_in_3mf,
merge_plate_3mfs,
)
plate_count = count_plates_in_3mf(primary_bytes)
if plate_count == 0:
raise HTTPException(
status_code=400,
detail=(
"Couldn't read plate count from the source 3MF for cross-class "
"slice-all. The source may be malformed or missing "
"Metadata/model_settings.config."
),
)
logger.info(
"Arrange slice-all: looping over %d plates with --arrange per plate, then merging "
"(embedded_settings=%s)",
plate_count,
embedded_mode,
)
from backend.app.services.slicer_api import SliceResult
per_plate_results: list[tuple[int, SliceResult]] = []
# Forward the same progress request_id + callback to each
# per-plate sub-call so the toast keeps showing the
# sidecar's stage messages ("Generating G-code 45%…").
# The sub-calls run sequentially, so the poller for plate
# N is cancelled before plate N+1's poller starts — no
# cross-talk between plate streams. Wrap the callback to
# surface "(plate N/M)" alongside the slicer's stage
# message so the user sees progress through the whole
# multi-plate loop, not just one plate at a time.
def _wrap_progress_for_plate(plate_num: int, total: int):
if progress_callback is None:
return None
def _cb(snapshot: dict) -> None:
snapshot = dict(snapshot)
snapshot["multi_plate_index"] = plate_num
snapshot["multi_plate_count"] = total
progress_callback(snapshot)
return _cb
for plate_num in range(1, plate_count + 1):
plate_cb = _wrap_progress_for_plate(plate_num, plate_count)
# "Slice as designed" has to take the loop too, not skip
# it: the project-wide collapse is caused by --arrange,
# and which config drives the slice has no bearing on
# that. Same call, minus --load-settings.
if embedded_mode:
per_plate = await service.slice_without_profiles(
model_bytes=primary_bytes,
model_filename=model_filename,
plate=plate_num,
export_3mf=True,
arrange=True,
orient=orient_flag,
request_id=progress_request_id,
on_progress=plate_cb,
)
else:
per_plate = await service.slice_with_profiles(
model_bytes=primary_bytes,
model_filename=model_filename,
printer_profile_json=presets["printer"],
process_profile_json=presets["process"],
filament_profile_jsons=filament_jsons,
plate=plate_num,
export_3mf=True,
arrange=True,
orient=orient_flag,
request_id=progress_request_id,
on_progress=plate_cb,
)
per_plate_results.append((plate_num, per_plate))
# Merge the N single-plate 3MFs into one multi-plate 3MF.
# ``primary_bytes`` is the source 3MF: it carries the
# original per-plate previews the slicer's --arrange
# pass doesn't regenerate, so the merger can fall back
# to those for each plate's cover image.
merged_bytes = merge_plate_3mfs(
[(n, r.content) for n, r in per_plate_results],
source_3mf_bytes=primary_bytes,
)
# Synthetic SliceResult: totals are the sum of each
# plate's so the archive card shows the project's print
# time and filament use, not just plate 1's.
result = SliceResult(
content=merged_bytes,
print_time_seconds=sum(r.print_time_seconds for _, r in per_plate_results),
filament_used_g=sum(r.filament_used_g for _, r in per_plate_results),
filament_used_mm=sum(r.filament_used_mm for _, r in per_plate_results),
)
# Report the path honestly: the loop can run either way, and
# the UI reads this flag to tell the user whose settings won.
used_embedded_settings = embedded_mode
elif embedded_mode:
# No --load-settings: feed the CLI the file's own
# project_settings.config untouched so the designer's tweaks
# (walls, infill, etc.) drive the slice. primary_bytes is
# already sentinel-sanitised above, the same bytes the
# crash-fallback uses. The resolved presets go unused here.
# Arrange / orient still apply: they are CLI actions on the
# geometry, not settings the embedded config could carry.
result = await service.slice_without_profiles(
model_bytes=primary_bytes,
model_filename=model_filename,
plate=request.plate,
export_3mf=request.export_3mf,
arrange=arrange_flag,
orient=orient_flag,
request_id=progress_request_id,
on_progress=progress_callback,
)
used_embedded_settings = True
else:
result = await service.slice_with_profiles(
model_bytes=primary_bytes,
model_filename=model_filename,
printer_profile_json=presets["printer"],
process_profile_json=presets["process"],
filament_profile_jsons=filament_jsons,
plate=request.plate,
export_3mf=request.export_3mf,
arrange=arrange_flag,
orient=orient_flag,
request_id=progress_request_id,
on_progress=progress_callback,
)
except SlicerApiServerError as exc:
rejection = _slicer_rejection_message(str(exc))
if rejection:
# The slicer ran and rejected the job for a content reason —
# the chosen printer/process/filament *were* applied. Falling
# back to embedded settings would silently re-slice for the
# source 3MF's original printer and hide the real problem
# (e.g. re-slicing an H2D model for an X1C: the object is off
# the smaller bed). Surface the slicer's reason instead.
raise HTTPException(status_code=400, detail=rejection) from exc
if not is_3mf or embedded_mode:
# embedded_mode already sliced with the file's own settings —
# there is nothing to fall back TO, so surface the server
# error (the outer handler turns it into a 502) instead of
# re-running the same embedded slice.
raise
if use_arrange_slice_all:
# The fallback is a single ``--slice 0`` call, and with
# arrange on that collapses every plate onto one bed — the
# exact outcome the per-plate loop above exists to avoid.
# Retrying would hand back a one-plate result for a job the
# user asked to slice as N, which reads as a Bambuddy bug
# rather than a slicer failure. Surface the error instead.
raise
logger.warning(
"Slicer CLI failed on the --load-settings path for %s (%s); retrying with embedded settings",
model_filename,
exc,
)
# Forward the same request_id + callback so the toast's live
# progress keeps updating across the fallback retry instead
# of going blank for the rest of the slice. Use the sanitised
# bytes — the embedded-settings path also reads the same
# project_settings.config and the same range validator runs
# there too, so without sanitisation the fallback would die
# on the same sentinel error (#1201). The SliceModal flags
# the difference to the user via used_embedded_settings.
# Carry the layout flags across too — the retry is meant to
# differ from the failed attempt only in where the print
# config came from, so dropping them here would silently
# produce an un-arranged result the user did ask for.
result = await service.slice_without_profiles(
model_bytes=primary_bytes,
model_filename=model_filename,
plate=request.plate,
export_3mf=request.export_3mf,
arrange=arrange_flag,
orient=orient_flag,
request_id=progress_request_id,
on_progress=progress_callback,
)
used_embedded_settings = True
except SlicerInputError as exc:
raise HTTPException(status_code=400, detail=str(exc)) from exc
except SlicerTimeoutError as exc:
# 504, not 502: the sidecar answered for the whole run, we stopped
# waiting. Reported separately so the user is told the slice ran out of
# time and where to change that, rather than that the sidecar is
# unreachable — which is what a read timeout used to look like (#2730).
raise HTTPException(status_code=504, detail=str(exc)) from exc
except SlicerApiServerError as exc:
raise HTTPException(status_code=502, detail=str(exc)) from exc
except SlicerApiUnavailableError as exc:
raise HTTPException(status_code=502, detail=str(exc)) from exc
finally:
await service.close()
# Backstop for #2838. Only the standard tier, and only when the presets we
# sent were actually used: there the sidecar resolved a bundled preset by
# name and the bundle guarantees the start G-code, so its absence is a
# sidecar defect we can name. A cloud, local or Orca-cloud preset carries
# its own start G-code, and the embedded-settings fallback prints the
# source file's — both are the user's to author, and refusing them here
# would be us second-guessing a profile we did not resolve.
if (
not used_embedded_settings
and request.printer_preset is not None
and request.printer_preset.source == "standard"
and start_gcode_is_missing(result.content, export_3mf=bool(request.export_3mf))
):
logger.error(
"Slice for printer preset %r came back without start G-code (%s); refusing it",
request.printer_preset.id,
"3mf" if request.export_3mf else "gcode",
)
raise HTTPException(status_code=502, detail=missing_start_gcode_message(request.printer_preset.id))
# Found while investigating #2977: a filament preset the sidecar's bundle
# cannot resolve is not an error there — the CLI inherits nothing and
# slices with its own defaults, so a PETG pick comes back as PLA at 200 C.
# Warned rather than refused: the file prints, and the user may well have
# meant to slice with a profile their sidecar image predates. Skipped on
# the embedded-settings path, which sends no filament profiles for the
# bundle to resolve in the first place.
if not used_embedded_settings:
unresolved = unresolved_filament_slots(result.content, export_3mf=bool(request.export_3mf))
if unresolved:
logger.warning(
"%s",
unresolved_filament_message(unresolved, [ref.id for ref in request.filament_presets]),
)
return result, used_embedded_settings
def _canonical_printer_model(raw: str | None) -> str | None:
"""Normalise a printer-preset name / ``printer_model`` field to a canonical
model code. Strips the BambuStudio ``"# "`` user-clone prefix and the
``" 0.4 nozzle"`` variant suffix that preset names carry but bare model
names don't — without this, ``"Bambu Lab H2D 0.4 nozzle"`` wouldn't
normalise to ``H2D``."""
import re
from backend.app.utils.printer_models import normalize_printer_model
if not raw:
return None
cleaned = str(raw).strip()
if cleaned.startswith("# "):
cleaned = cleaned[2:].strip()
cleaned = re.sub(r"\s+0\.\d+\s+nozzle$", "", cleaned, flags=re.IGNORECASE)
return normalize_printer_model(cleaned) if cleaned else None
async def _resolve_target_printer_model(db: AsyncSession, user: User | None, request: SliceRequest) -> str | None:
"""Best-effort: the printer model a slice request targets.
Returns ``None`` when it can't be determined (the nozzle-class guard
then simply doesn't fire — fail-open, never blocks a slice spuriously).
"""
from backend.app.services.preset_resolver import resolve_preset_ref
if request.printer_preset is None:
return None
try:
printer_json = await resolve_preset_ref(db, user, request.printer_preset, "printer")
data = json.loads(printer_json)
if not isinstance(data, dict):
return None
return _canonical_printer_model(
data.get("printer_model") or data.get("printer_settings_id") or data.get("name")
)
except Exception:
return None
async def guard_nozzle_class_reslice(
db: AsyncSession, user: User | None, request: SliceRequest, source_model: str | None
) -> None:
"""No-op guard, retained for call-site compatibility.
Cross-nozzle-class re-slicing is handled by ``_run_slicer_with_fallback``'s
two-pass conversion (#1493): a 1mm cube is sliced with the target triplet
via ``slice_with_profiles`` to produce a fresh target-shaped
``Metadata/project_settings.config``, which is then spliced into the
source 3MF before the real slice. So this guard never needs to block
anymore.
The function and its call sites in ``archives.py`` / the library re-slice
route are kept so external pinned-version forks and downstream patches
don't break, but it does nothing on a successful slice path. If the
two-pass conversion fails inside the slicer, the existing
``SlicerApiServerError`` / ``_slicer_rejection_message`` plumbing
surfaces the CLI's actual error to the user — which is more informative
than the old "isn't supported yet" 400 the guard used to raise.
"""
return None
async def slice_and_persist(
db: AsyncSession,
*,
model_bytes: bytes,
model_filename: str,
folder_id: int | None,
extra_metadata: dict | None,
request: SliceRequest,
current_user_id: int | None,
job_id: int | None = None,
) -> SliceResponse:
"""Slice a model and save the result as a new ``LibraryFile`` in
``folder_id`` (same folder as the source by convention).
Always exports as ``.gcode.3mf`` so the existing library thumbnail
pipeline works on the new file. Plain ``.gcode`` would have no
embedded thumbnail to extract.
"""
from backend.app.services.archive import ThreeMFParser
library_request = request.model_copy(update={"export_3mf": True})
result, used_embedded_settings = await _run_slicer_with_fallback(
db,
model_bytes=model_bytes,
model_filename=model_filename,
request=library_request,
current_user_id=current_user_id,
job_id=job_id,
)
# Same reduction as the archive sink: ``model_filename`` may be built from
# the source's embedded ``print_name``, which is free text (#2832). Managed
# storage names the file after a UUID and never sees this, but an external
# folder writes it verbatim, where a "/" would mean a directory nobody
# created -- and the library row shows it either way.
base_name = model_filename.rsplit(".", 1)[0]
safe_base = safe_path_component(base_name, fallback="sliced", max_bytes=MAX_FILENAME_BYTES - len(b".gcode.3mf"))
out_filename = f"{safe_base}.gcode.3mf"
# Write next to the source when the source lives on an external mount
# (#2810). The folder is loaded here rather than passed in because every
# caller already has only the id.
target_folder: LibraryFolder | None = None
if folder_id is not None:
folder_result = await db.execute(select(LibraryFolder).where(LibraryFolder.id == folder_id))
target_folder = folder_result.scalar_one_or_none()
out_path, out_is_external, external_fallback = _resolve_slice_destination(target_folder, out_filename)
if out_is_external:
# _unique_external_name may have suffixed it; the library row has to
# show the name the file actually has on the share, or the two drift.
out_filename = out_path.name
if external_fallback:
logger.warning(
"Slice output for %s stored in managed library instead of external folder %s: %s",
model_filename,
target_folder.external_path if target_folder else None,
external_fallback,
)
# BS/Orca CLIs skip plate_N.png in headless --export-3mf — render +
# inject server-side so the library card has a thumbnail. Best-effort:
# no-op when the slicer did embed thumbs (desktop Studio path), and
# falls through to the unmodified bytes on any render error. In a thread:
# a large plate renders for seconds, and on the event loop that stalled
# every request and printer connection for as long (#3135).
result = result._replace(content=await asyncio.to_thread(inject_plate_thumbnails_if_missing, result.content))
out_path.write_bytes(result.content)
# Extract thumbnail from the produced 3MF so the library card shows a
# preview. Failures here aren't fatal — the file is still useful
# without a thumbnail.
thumbnail_relative: str | None = None
parsed_metadata: dict = {}
try:
parser = ThreeMFParser(str(out_path))
parsed = parser.parse()
thumb_data = parsed.get("_thumbnail_data")
thumb_ext = parsed.get("_thumbnail_ext", ".png")
if thumb_data:
thumb_filename = f"{uuid.uuid4().hex}{thumb_ext}"
thumb_path = get_library_thumbnails_dir() / thumb_filename
thumb_path.write_bytes(thumb_data)
thumbnail_relative = to_relative_path(thumb_path)
cleaned = _clean_3mf_metadata(parsed)
if isinstance(cleaned, dict):
parsed_metadata = cleaned
except Exception as exc:
logger.warning("Failed to parse sliced 3MF metadata for %s: %s", out_filename, exc)
# Drop the embedded `print_name` (see _without_print_name) so the sliced
# row's display falls back to its ".gcode.3mf" filename instead of the
# source file's project title, which would make the two indistinguishable.
metadata: dict = dict(_without_print_name(parsed_metadata) or {})
# Some slicer-sidecar builds leave the X-Filament-Used-* response headers
# unset, so result.filament_used_g/_mm arrive as 0 even for a real
# multi-hour print. Fall back to the totals ThreeMFParser read from the
# produced 3MF's own G-code header.
filament_g = result.filament_used_g or parsed_metadata.get("filament_used_grams") or 0.0
filament_mm = result.filament_used_mm or parsed_metadata.get("filament_used_mm") or 0.0
metadata.update(
{
"print_time_seconds": result.print_time_seconds,
"filament_used_g": filament_g,
"filament_used_mm": filament_mm,
}
)
if used_embedded_settings:
metadata["used_embedded_settings"] = True
if external_fallback:
metadata["external_write_fallback"] = external_fallback
if extra_metadata:
metadata.update(extra_metadata)
new_file = LibraryFile(
folder_id=folder_id,
is_external=out_is_external,
filename=out_filename,
file_path=_stored_file_path(out_path, out_is_external),
# The on-disk payload is a ZIP container — the file_type must
# record that so the preview endpoint opens it as a 3MF instead
# of returning the ZIP bytes as text/plain (#1709 / yanglei1980).
# Earlier code mis-typed sliced rows as "gcode" to share the
# plain-G-code badge; that broke the embedded viewer. UI badges
# and gates for "gcode.3mf" are explicit at the call sites.
file_type="gcode.3mf",
file_size=len(result.content),
file_hash=hashlib.sha256(result.content).hexdigest(),
thumbnail_path=thumbnail_relative,
file_metadata=metadata,
source_type="sliced",
created_by_id=current_user_id,
)
db.add(new_file)
await db.commit()
# No refresh: expire_on_commit=False keeps id/filename accessible, and
# refreshing here flakes under pytest-xdist when teardown of a sibling
# test races the SELECT.
return SliceResponse(
library_file_id=new_file.id,
name=new_file.filename,
print_time_seconds=result.print_time_seconds,
filament_used_g=filament_g,
filament_used_mm=filament_mm,
used_embedded_settings=used_embedded_settings,
external_write_fallback=external_fallback,
)
async def slice_and_persist_as_archive(
db: AsyncSession,
*,
model_bytes: bytes,
model_filename: str,
request: SliceRequest,
source_archive, # PrintArchive — hint kept loose to avoid cyclic import
current_user_id: int | None,
job_id: int | None = None,
):
"""Slice a model and save the result as a new ``PrintArchive`` row,
inheriting printer / project / makerworld metadata from the source
archive. Always exports as a `.gcode.3mf` so the existing thumbnail
and plates infrastructure (which expects a zip-shaped 3MF) works on
the new archive. Returns ``SliceArchiveResponse``.
"""
from backend.app.models.archive import PrintArchive
from backend.app.schemas.slicer import SliceArchiveResponse
from backend.app.services.archive import ThreeMFParser
# Archive sinks always want a 3MF. The library route still respects the
# caller's `export_3mf` flag; here we override.
archive_request = request.model_copy(update={"export_3mf": True})
result, used_embedded_settings = await _run_slicer_with_fallback(
db,
model_bytes=model_bytes,
model_filename=model_filename,
request=archive_request,
job_id=job_id,
current_user_id=current_user_id,
)
timestamp = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S")
printer_folder = str(source_archive.printer_id) if source_archive.printer_id is not None else "unassigned"
# ``model_filename`` is built from the archive's display name, which comes
# from the 3MF's own metadata and is whatever the model's author typed. A
# "/" in it is a path separator, not a character: the joins below silently
# gain a level and the write lands on a parent that was never created
# (#2832). Reduce it to a single component first, leaving room for the
# prefix and the extension wrapped around it.
base_name = model_filename.rsplit(".", 1)[0]
reserve = max(len(f"{timestamp}__sliced".encode()), len(b".gcode.3mf"))
safe_base = safe_path_component(
base_name, fallback=f"archive_{source_archive.id}", max_bytes=MAX_FILENAME_BYTES - reserve
)
out_filename = f"{safe_base}.gcode.3mf"
archive_subdir = f"{timestamp}_{safe_base}_sliced"
archive_dir = (
app_settings.archive_dir / printer_folder / archive_subdir
) # SEC-PATH-OK: printer_folder = str(int|None); archive_subdir wraps safe_path_component output, asserted below
out_path = archive_dir / out_filename # SEC-PATH-OK: out_filename wraps safe_path_component output, asserted below
# The sanitiser is what makes the two joins single-component; this is the
# backstop that says so out loud, and would catch a future edit that reaches
# around it. Checked before mkdir so a rejected path creates nothing.
assert_under(app_settings.archive_dir, archive_dir, http=False)
assert_under(app_settings.archive_dir, out_path, http=False)
archive_dir.mkdir(parents=True, exist_ok=True)
# See library-slice path: BS/Orca sidecar CLIs don't embed plate_N.png
# in headless --export-3mf, so the produced 3MF often has no thumbnail
# at all. Server-side render fills the gap; no-op when the slicer did
# embed (desktop Studio path) and best-effort on any render error. Off the
# event loop, like the library-slice path (#3135).
result = result._replace(content=await asyncio.to_thread(inject_plate_thumbnails_if_missing, result.content))
out_path.write_bytes(result.content)
# Extract a thumbnail for the new archive card. Priority order:
# 1. Source archive's ``Metadata/plate_{N}.png`` — the GUI-rendered
# preview of the same plate the user is re-slicing. Closer to
# "what's actually printing" than any other available image
# (with --arrange the layout may differ slightly, but objects
# and colours match).
# 2. ``ThreeMFParser`` fallback chain on the sliced output: the
# slicer's own per-plate render if it wrote one, then the
# project-wide thumbnail under ``Auxiliaries/.thumbnails/``.
# BambuStudio CLI frequently doesn't emit a fresh per-plate render
# (slice writes the new gcode but leaves the preview slot empty),
# so without (1) the card falls all the way through to the
# MakerWorld-style cover art — visually unrelated to what the user
# picked, see #1493 follow-up. Failures don't fail the slice — the
# archive row is still useful without a thumbnail.
plate_num = request.plate or 1
thumbnail_path: str | None = None
parsed_metadata: dict = {}
src_3mf_path = app_settings.base_dir / source_archive.file_path
source_plate_bytes = _read_3mf_entry(src_3mf_path, f"Metadata/plate_{plate_num}.png")
if source_plate_bytes:
thumb_dest = archive_dir / "thumbnail.png"
thumb_dest.write_bytes(source_plate_bytes)
thumbnail_path = str(thumb_dest.relative_to(app_settings.base_dir))
try:
parser = ThreeMFParser(str(out_path), plate_number=plate_num)
parsed = parser.parse()
if thumbnail_path is None:
thumb_data = parsed.get("_thumbnail_data")
thumb_ext = parsed.get("_thumbnail_ext", ".png")
if thumb_data:
thumb_dest = archive_dir / f"thumbnail{thumb_ext}"
thumb_dest.write_bytes(thumb_data)
thumbnail_path = str(thumb_dest.relative_to(app_settings.base_dir))
parsed_metadata = {k: v for k, v in parsed.items() if not k.startswith("_")}
except Exception as exc:
logger.warning("Failed to parse sliced 3MF metadata for %s: %s", out_filename, exc)
metadata = dict(source_archive.extra_data) if source_archive.extra_data else {}
metadata.update(parsed_metadata)
# Fall back to the produced 3MF's G-code-header totals when the sidecar
# leaves the X-Filament-Used-* headers unset (result.filament_used_g == 0
# even for a real multi-hour print).
filament_g = result.filament_used_g or parsed_metadata.get("filament_used_grams") or 0.0
filament_mm = result.filament_used_mm or parsed_metadata.get("filament_used_mm") or 0.0
metadata.update(
{
"sliced_from_archive_id": source_archive.id,
"print_time_seconds": result.print_time_seconds,
"filament_used_g": filament_g,
"filament_used_mm": filament_mm,
}
)
if used_embedded_settings:
metadata["used_embedded_settings"] = True
# Prefer the actually-used filament list from the sliced output's
# slice_info.config (parsed_metadata.filament_* — only entries with
# used_g > 0). Falling back to the source_archive's list would
# surface every project-wide AMS slot, including ones the picked
# plate doesn't use (16+ swatches on the card for a 2-color print).
new_filament_type = parsed_metadata.get("filament_type") or source_archive.filament_type
new_filament_color = parsed_metadata.get("filament_color") or source_archive.filament_color
# When the user re-slices for a different printer model than the source,
# the source's printer_id (e.g. an H2D's "Workshop H2C") no longer
# represents where the new archive can be reprinted. The archive card
# and reprint modal both read printer_id first and only fall back to
# sliced_for_model when it's None, so leaving the inherited id makes
# the X1C-sliced card display the source H2D's printer name.
# Same pitfall as the sliced_for_model copy a few lines below.
new_target_model = parsed_metadata.get("sliced_for_model") or source_archive.sliced_for_model
is_cross_model_reslice = (
new_target_model is not None
and source_archive.sliced_for_model is not None
and new_target_model != source_archive.sliced_for_model
)
new_printer_id = None if is_cross_model_reslice else source_archive.printer_id
new_archive = PrintArchive(
printer_id=new_printer_id,
project_id=source_archive.project_id,
filename=out_filename,
file_path=str(out_path.relative_to(app_settings.base_dir)),
file_size=len(result.content),
content_hash=hashlib.sha256(result.content).hexdigest(),
thumbnail_path=thumbnail_path,
# Inherit identity from the source archive so the new entry shows
# up alongside its sibling in the archives list.
print_name=(source_archive.print_name or base_name) + " (re-sliced)",
print_time_seconds=result.print_time_seconds,
filament_used_grams=filament_g or None,
filament_type=new_filament_type,
filament_color=new_filament_color,
layer_height=source_archive.layer_height,
nozzle_diameter=source_archive.nozzle_diameter,
# The re-sliced output is for whatever printer the user just picked,
# not the source archive's printer — read the model the slicer baked
# into the new 3MF, falling back to the source only if it's absent.
# (Copying source_archive.sliced_for_model kept a cross-printer
# re-slice, e.g. X1C→H2D, showing the old "X1C sliced" model.)
sliced_for_model=parsed_metadata.get("sliced_for_model") or source_archive.sliced_for_model,
# Build plate type that the sliced output was produced for (#1493
# follow-up): the frontend's ArchiveCard reads ``archive.bed_type``
# off the top-level column, not extra_data, so without this lift the
# re-sliced card had no plate badge. ThreeMFParser pulls it from the
# sliced 3MF's ``slice_info.config`` ``curr_bed_type``; if that's
# absent (older sidecar / older slice profile) the source archive's
# bed_type is the right default.
bed_type=parsed_metadata.get("bed_type") or source_archive.bed_type,
makerworld_url=source_archive.makerworld_url,
designer=source_archive.designer,
# Sliced-but-not-printed: keep status default ("completed") so it
# surfaces in the normal archives list, but do not stamp
# started/completed_at — the user hasn't actually printed it yet.
extra_data=metadata,
created_by_id=current_user_id,
)
db.add(new_archive)
await db.commit()
await db.refresh(new_archive)
return SliceArchiveResponse(
archive_id=new_archive.id,
name=new_archive.print_name or out_filename,
print_time_seconds=result.print_time_seconds,
filament_used_g=filament_g,
filament_used_mm=filament_mm,
used_embedded_settings=used_embedded_settings,
)
@router.post("/files/{file_id}/slice", status_code=202)
async def slice_library_file(
file_id: int,
request: SliceRequest,
db: AsyncSession = Depends(get_db),
current_user: User | None = Depends(require_permission_if_auth_enabled(Permission.LIBRARY_UPLOAD)),
api_key_cloud_owner: User | None = Depends(resolve_api_key_cloud_owner),
):
"""Enqueue a slice job for a library file. Returns 202 + job_id; the
slice runs in the background, the caller polls `GET /slice-jobs/{id}`.
"""
from backend.app.core.database import async_session
from backend.app.services.slice_dispatch import (
http_exception_to_job_error,
slice_dispatch,
)
src_result = await db.execute(LibraryFile.active().where(LibraryFile.id == file_id))
lib_file = src_result.scalar_one_or_none()
# Per-row ownership gate. LIBRARY_UPLOAD alone let a READ_OWN caller (e.g. the
# built-in Operators group) slice another user's model by raw id even though
# GET on that id returned 404 — the sliced output was then attributed to and
# downloadable by the requester. Enforce the same visibility the read routes
# use before reading the source off disk. API-key / auth-disabled callers
# (current_user is None) keep can_read_all=True — no per-row identity.
can_read_all = current_user is None or current_user.has_permission(Permission.LIBRARY_READ_ALL.value)
lib_file = _ensure_library_file_visible(lib_file, current_user, can_read_all)
src_lower = (lib_file.filename or "").lower()
if src_lower.endswith(".step") or src_lower.endswith(".stp"):
# Neither slicer's CLI can load STEP: OrcaSlicer 2.4.2 and BambuStudio
# 02.07.01.62 both answer "Unknown file format. Input file must have
# .stl, .obj, .amf(.xml) extension." Accepting the job here meant
# reading the file, converting it and uploading it before the sidecar
# rejected it as unparseable -- which reads as a corrupt model rather
# than an unsupported format. Say so before any of that happens.
raise HTTPException(
status_code=400,
detail=(
"STEP files cannot be sliced. The OrcaSlicer and Bambu Studio command-line "
"slicers load only STL and 3MF -- open the STEP in your slicer and export it "
"as one of those first."
),
)
if not (src_lower.endswith(".stl") or src_lower.endswith(".3mf")):
raise HTTPException(status_code=400, detail="Source file must be STL or 3MF")
src_path = Path(app_settings.base_dir) / lib_file.file_path
if not src_path.exists():
raise HTTPException(status_code=404, detail="Source file missing on disk")
# Capture inputs the bg task needs — the request DB session is closed
# before the background task runs.
model_bytes = src_path.read_bytes()
folder_id = lib_file.folder_id
source_lib_file_id = lib_file.id
# API-keyed callers get None from the auth gate (auth.py keeps that
# behaviour to avoid a wider scope expansion). Fall back to the API
# key's owner so cloud-preset resolution can read the stored
# cloud_token (#1182 follow-up).
cloud_token_user = current_user or api_key_cloud_owner
user_id = cloud_token_user.id if cloud_token_user else None
# If the source has a `print_name` in its metadata (BambuStudio always
# sets this; OrcaSlicer often leaves it blank), derive the sliced
# output's filename from it instead of the raw filename. The source
# row's display already prefers print_name, so the sliced row's
# filename ("Piggo the piggy bank.gcode.3mf") will match the source's
# display name ("Piggo the piggy bank") with the gcode extension added.
src_print_name = None
if lib_file.file_metadata:
candidate = lib_file.file_metadata.get("print_name")
if isinstance(candidate, str) and candidate.strip():
src_print_name = candidate.strip()
src_ext = Path(lib_file.filename).suffix.lower() or ".3mf"
model_filename = f"{src_print_name}{src_ext}" if src_print_name else lib_file.filename
# Block a cross-nozzle-class re-slice (single-nozzle <-> H2D) up front.
# Fires only when the source is itself a sliced file (carries
# sliced_for_model); a plain un-sliced model has no source nozzle class.
await guard_nozzle_class_reslice(
db,
cloud_token_user,
request,
(lib_file.file_metadata or {}).get("sliced_for_model"),
)
async def _run(job_id: int):
async with async_session() as task_db:
try:
response = await slice_and_persist(
task_db,
model_bytes=model_bytes,
model_filename=model_filename,
folder_id=folder_id,
extra_metadata={"sliced_from_library_file_id": source_lib_file_id},
request=request,
current_user_id=user_id,
job_id=job_id,
)
except HTTPException as exc:
raise http_exception_to_job_error(exc) from exc
return response.model_dump()
job = await slice_dispatch.enqueue(
kind="library_file",
source_id=lib_file.id,
source_name=lib_file.filename,
owner_id=user_id,
run=_run,
)
return {
"job_id": job.id,
"status": job.status,
"status_url": f"/api/v1/slice-jobs/{job.id}",
}
@router.post("/files/{file_id}/print")
async def print_library_file(
file_id: int,
printer_id: int,
# SECURITY.md SEC-AUTH-1: every route either has an explicit auth dep or
# is in the route-auth-coverage allowlist. Gating the deprecation stub on
# QUEUE_CREATE matches the replacement route (POST /queue/) and means
# anonymous callers bounce at auth instead of seeing the deprecation
# message.
_: User | None = Depends(require_permission_if_auth_enabled(Permission.QUEUE_CREATE)),
):
"""Legacy direct library print endpoint. Use POST /queue/ instead."""
logger.warning(
"Gone API used: POST /library/files/%s/print?printer_id=%s; use POST /queue/ instead",
file_id,
printer_id,
)
raise HTTPException(
status_code=410,
detail="Direct library-file print has been removed. Create a print queue item with POST /queue/.",
)
# ============ File Detail Endpoints ============
@router.get("/files/{file_id}", response_model=FileResponseSchema)
async def get_file(
file_id: int,
db: AsyncSession = Depends(get_db),
auth_result: tuple[User | None, bool] = Depends(
require_ownership_permission(
Permission.LIBRARY_READ_ALL,
Permission.LIBRARY_READ_OWN,
)
),
):
"""Get a file by ID with full details."""
user, can_read_all = auth_result
result = await db.execute(
LibraryFile.active().options(selectinload(LibraryFile.created_by)).where(LibraryFile.id == file_id)
)
file = _ensure_library_file_visible(result.scalar_one_or_none(), user, can_read_all)
# Get folder name
folder_name = None
if file.folder_id:
folder_result = await db.execute(select(LibraryFolder.name).where(LibraryFolder.id == file.folder_id))
folder_name = folder_result.scalar()
# Get project name
project_name = None
if file.project_id:
project_result = await db.execute(select(Project.name).where(Project.id == file.project_id))
project_name = project_result.scalar()
# Get duplicates
duplicates = []
duplicate_count = 0
if file.file_hash:
dup_result = await db.execute(
select(LibraryFile, LibraryFolder.name)
.outerjoin(LibraryFolder, LibraryFile.folder_id == LibraryFolder.id)
.where(
LibraryFile.file_hash == file.file_hash,
LibraryFile.id != file.id,
LibraryFile.deleted_at.is_(None),
)
)
for dup_file, dup_folder_name in dup_result.all():
duplicates.append(
FileDuplicate(
id=dup_file.id,
filename=dup_file.filename,
folder_id=dup_file.folder_id,
folder_name=dup_folder_name,
created_at=dup_file.created_at,
)
)
duplicate_count = len(duplicates)
# Extract key metadata fields
print_name = None
print_time = None
filament_grams = None
sliced_for_model = None
if file.file_metadata:
print_name = file.file_metadata.get("print_name")
print_time = file.file_metadata.get("print_time_seconds")
filament_grams = file.file_metadata.get("filament_used_grams")
sliced_for_model = file.file_metadata.get("sliced_for_model")
return FileResponseSchema(
id=file.id,
folder_id=file.folder_id,
folder_name=folder_name,
project_id=file.project_id,
project_name=project_name,
filename=file.filename,
file_path=file.file_path,
file_type=file.file_type,
file_size=file.file_size,
file_hash=file.file_hash,
thumbnail_path=file.thumbnail_path,
metadata=file.file_metadata,
print_count=file.print_count,
last_printed_at=file.last_printed_at,
notes=file.notes,
external_url=file.external_url,
photos=list(file.photos or []),
source_url=file.source_url,
duplicates=duplicates if duplicates else None,
duplicate_count=duplicate_count,
created_by_id=file.created_by_id,
created_by_username=file.created_by.username if file.created_by else None,
created_at=file.created_at,
updated_at=file.updated_at,
print_name=print_name,
print_time_seconds=print_time,
filament_used_grams=filament_grams,
sliced_for_model=sliced_for_model,
)
@router.put("/files/{file_id}", response_model=FileResponseSchema)
async def update_file(
file_id: int,
data: FileUpdate,
db: AsyncSession = Depends(get_db),
auth_result: tuple[User | None, bool] = Depends(
require_ownership_permission(
Permission.LIBRARY_UPDATE_ALL,
Permission.LIBRARY_UPDATE_OWN,
)
),
):
"""Update a file's metadata."""
user, can_modify_all = auth_result
result = await db.execute(LibraryFile.active().where(LibraryFile.id == file_id))
file = result.scalar_one_or_none()
if not file:
raise HTTPException(status_code=404, detail="File not found")
# Ownership check
if not can_modify_all:
if file.created_by_id != user.id:
raise HTTPException(status_code=403, detail="You can only update your own files")
if data.filename is not None:
# Bambu printer SD cards are FAT32/exFAT; reject the same set Bambu
# Studio refuses on save so we fail here with a clear message
# instead of an obscure FTP 553 at print time (#1540).
try:
validate_print_filename(data.filename)
except InvalidFilenameError as e:
raise HTTPException(status_code=400, detail=str(e)) from e
file.filename = data.filename
# No print_name to keep in sync — library files display by filename,
# and _without_print_name strips the embedded 3MF Title on import (#1489).
if data.folder_id is not None:
if data.folder_id == 0:
file.folder_id = None
else:
# Verify folder exists
folder_result = await db.execute(select(LibraryFolder).where(LibraryFolder.id == data.folder_id))
if not folder_result.scalar_one_or_none():
raise HTTPException(status_code=404, detail="Folder not found")
file.folder_id = data.folder_id
if data.project_id is not None:
if data.project_id == 0:
file.project_id = None
else:
# Verify project exists
project_result = await db.execute(select(Project).where(Project.id == data.project_id))
if not project_result.scalar_one_or_none():
raise HTTPException(status_code=404, detail="Project not found")
file.project_id = data.project_id
if data.notes is not None:
file.notes = data.notes if data.notes else None
if data.external_url is not None:
file.external_url = data.external_url.strip() or None
await db.commit()
await db.refresh(file)
# Return full response. Bypass get_file's ownership gate — caller already
# passed update_file's ownership gate above, so we re-fetch + serialise
# directly instead of calling the route function (which would try to
# evaluate its own Depends() at call time and trip a TypeError).
return await get_file(file_id, db, auth_result=(None, True))
@router.delete("/files/{file_id}")
async def delete_file(
file_id: int,
db: AsyncSession = Depends(get_db),
auth_result: tuple[User | None, bool] = Depends(
require_ownership_permission(
Permission.LIBRARY_DELETE_ALL,
Permission.LIBRARY_DELETE_OWN,
)
),
):
"""Move a file to the trash (soft-delete).
The file's bytes and thumbnail stay on disk until the trash sweeper
hard-deletes the row after the retention window (see #1008). External
files skip the trash entirely — they can't be restored from disk and the
underlying file is outside Bambuddy's control, so we just drop the DB
record and thumbnail.
"""
user, can_modify_all = auth_result
result = await db.execute(LibraryFile.active().where(LibraryFile.id == file_id))
file = result.scalar_one_or_none()
if not file:
raise HTTPException(status_code=404, detail="File not found")
# Ownership check
if not can_modify_all:
if file.created_by_id != user.id:
raise HTTPException(status_code=403, detail="You can only delete your own files")
if file.is_external:
# External files bypass the trash — just drop the DB row + our thumbnail.
abs_thumb_path = to_absolute_path(file.thumbnail_path)
if abs_thumb_path and abs_thumb_path.exists():
try:
abs_thumb_path.unlink()
except OSError as e:
logger.warning("Failed to delete thumbnail from disk: %s", e)
from backend.app.services.library_trash import delete_dependent_variants, release_queue_references
await delete_dependent_variants(db, [file.id])
await release_queue_references(db, [file.id])
remove_library_photos_dir(file.id)
await db.delete(file)
await db.commit()
return {"status": "success", "message": "File deleted", "trashed": False}
# Managed file: soft-delete. Sweeper removes bytes + thumbnail after retention.
file.deleted_at = datetime.now(timezone.utc)
await db.commit()
return {"status": "success", "message": "File moved to trash", "trashed": True}
# ============ File Content Endpoints ============
@router.get("/files/{file_id}/download")
async def download_file(
file_id: int,
db: AsyncSession = Depends(get_db),
auth_result: tuple[User | None, bool] = Depends(
require_ownership_permission(
Permission.LIBRARY_READ_ALL,
Permission.LIBRARY_READ_OWN,
)
),
):
"""Download a file."""
user, can_read_all = auth_result
result = await db.execute(LibraryFile.active().where(LibraryFile.id == file_id))
file = _ensure_library_file_visible(result.scalar_one_or_none(), user, can_read_all)
abs_path = to_absolute_path(file.file_path)
if not abs_path or not abs_path.exists():
raise HTTPException(status_code=404, detail="File not found on disk")
return FastAPIFileResponse(
str(abs_path),
filename=file.filename,
media_type="application/octet-stream",
)
@router.post("/files/{file_id}/slicer-token")
async def create_library_slicer_token(
file_id: int,
db: AsyncSession = Depends(get_db),
auth_result: tuple[User | None, bool] = Depends(
require_ownership_permission(
Permission.LIBRARY_READ_ALL,
Permission.LIBRARY_READ_OWN,
)
),
):
"""Create a short-lived download token for opening files in slicer applications.
Slicer protocol handlers (bambustudioopen://, orcaslicer://) cannot send
auth headers, so they use this token in the URL path instead.
"""
from backend.app.core.auth import create_slicer_download_token
user, can_read_all = auth_result
result = await db.execute(LibraryFile.active().where(LibraryFile.id == file_id))
_ensure_library_file_visible(result.scalar_one_or_none(), user, can_read_all)
token = await create_slicer_download_token("library", file_id)
return {"token": token}
@router.get("/files/{file_id}/dl/{token}/{filename}")
async def download_library_file_for_slicer(
file_id: int,
token: str,
filename: str,
db: AsyncSession = Depends(get_db),
):
"""Download a library file using a slicer download token.
Token-authenticated (no auth headers needed). The token is short-lived and
file-bound, created by POST /files/{file_id}/slicer-token, and redeemable
for the rest of its TTL rather than exactly once -- the slicer is a separate
process that may fetch the URL more than once (#3029).
Filename is at the end of the URL so slicers can detect the file format.
"""
from backend.app.core.auth import verify_slicer_download_token
if not await verify_slicer_download_token(token, "library", file_id, single_use=False):
raise HTTPException(status_code=403, detail="Invalid or expired download token")
result = await db.execute(LibraryFile.active().where(LibraryFile.id == file_id))
file = result.scalar_one_or_none()
if not file:
raise HTTPException(status_code=404, detail="File not found")
abs_path = to_absolute_path(file.file_path)
if not abs_path or not abs_path.exists():
raise HTTPException(status_code=404, detail="File not found on disk")
return FastAPIFileResponse(
str(abs_path),
filename=file.filename,
media_type="application/octet-stream",
)
@router.get("/files/{file_id}/thumbnail")
async def get_thumbnail(
file_id: int,
db: AsyncSession = Depends(get_db),
auth_result: tuple[User | None, bool] = Depends(
require_media_token_ownership(
Permission.LIBRARY_READ_ALL,
Permission.LIBRARY_READ_OWN,
)
),
):
"""Get a file's thumbnail.
Accepts a media token in ``?token=`` because <img> cannot send headers.
Ownership is enforced here rather than assumed from the credential: until
#3025 this route took the anonymous camera-stream token, which carried no
principal, so any holder could read any user's thumbnail by walking IDs.
"""
user, can_read_all = auth_result
result = await db.execute(LibraryFile.active().where(LibraryFile.id == file_id))
file = _ensure_library_file_visible(result.scalar_one_or_none(), user, can_read_all)
abs_thumb_path = to_absolute_path(file.thumbnail_path)
if not abs_thumb_path or not abs_thumb_path.exists():
raise HTTPException(status_code=404, detail="Thumbnail not found")
# Detect media type from extension
thumb_ext = abs_thumb_path.suffix.lower()
media_types = {
".png": "image/png",
".jpg": "image/jpeg",
".jpeg": "image/jpeg",
".gif": "image/gif",
".webp": "image/webp",
}
media_type = media_types.get(thumb_ext, "image/png")
return FastAPIFileResponse(str(abs_thumb_path), media_type=media_type)
@router.post("/files/{file_id}/preview-thumbnail", response_model=ClientThumbnailResponse)
async def upload_preview_thumbnail(
file_id: int,
thumbnail: UploadFile = File(...),
db: AsyncSession = Depends(get_db),
auth_result: tuple[User | None, bool] = Depends(
require_ownership_permission(
Permission.LIBRARY_UPDATE_ALL,
Permission.LIBRARY_UPDATE_OWN,
)
),
):
"""Store a client-rendered preview thumbnail for a file (#2976).
STEP, PDF and spreadsheet previews are rendered in the browser; the FE
posts its first render here so the grid gets a thumbnail without the
server needing OpenCascade. A PDF normally has its PDFium thumbnail from
upload already, so for PDFs this only fills the gap for a file PDFium
could not read. Only file types in ``CLIENT_THUMBNAIL_TYPES`` are
accepted, and only while the file has no thumbnail yet — a stored
thumbnail is never replaced by this route.
"""
user, can_modify_all = auth_result
result = await db.execute(LibraryFile.active().where(LibraryFile.id == file_id))
file = result.scalar_one_or_none()
if not file:
raise HTTPException(status_code=404, detail="File not found")
# Ownership check (same shape as update_file)
if not can_modify_all:
if file.created_by_id != user.id:
raise HTTPException(status_code=403, detail="You can only update your own files")
if file.file_type not in CLIENT_THUMBNAIL_TYPES:
raise HTTPException(status_code=400, detail="File type does not accept client-rendered thumbnails")
if file.thumbnail_path:
return ClientThumbnailResponse(updated=False)
content = await thumbnail.read(MAX_CLIENT_THUMBNAIL_BYTES + 1)
if len(content) > MAX_CLIENT_THUMBNAIL_BYTES:
raise HTTPException(status_code=413, detail="Thumbnail too large")
# Decode and re-encode through PIL: validates the bytes are a real PNG
# and strips anything that isn't pixel data before it lands on disk.
import io
from PIL import Image, UnidentifiedImageError
try:
# Image.open() reads the header only. Both checks below happen before
# load(), so a declared-but-never-delivered canvas is refused rather
# than allocated. DecompressionBombError derives straight from
# Exception, so it has to be named explicitly — open() itself raises
# it once the declared size passes PIL's own limit.
with Image.open(io.BytesIO(content)) as source:
if source.format != "PNG":
raise HTTPException(status_code=400, detail="Thumbnail must be a PNG image")
if max(source.size) > MAX_CLIENT_THUMBNAIL_EDGE:
raise HTTPException(status_code=400, detail="Thumbnail image dimensions too large")
source.load()
img = source.convert("RGBA") if source.mode not in ("RGB", "RGBA") else source.copy()
except HTTPException:
raise
except (UnidentifiedImageError, OSError, ValueError, Image.DecompressionBombError) as e:
raise HTTPException(status_code=400, detail="Invalid thumbnail image") from e
if max(img.size) > STORED_CLIENT_THUMBNAIL_EDGE:
img.thumbnail((STORED_CLIENT_THUMBNAIL_EDGE, STORED_CLIENT_THUMBNAIL_EDGE), Image.Resampling.LANCZOS)
thumbnails_dir = get_library_thumbnails_dir()
thumb_filename = f"{uuid.uuid4().hex}.png"
thumb_path = thumbnails_dir / thumb_filename # SEC-PATH-OK: thumb_filename = uuid.uuid4().hex + ".png"
# Outside the decode guard on purpose: a full disk or an unwritable
# thumbnail directory is ours, not "Invalid thumbnail image".
try:
img.save(thumb_path, "PNG", optimize=True)
except OSError as e:
logger.error("Failed to store preview thumbnail for file %s: %s", file_id, e)
raise HTTPException(status_code=500, detail="Failed to store thumbnail") from e
# Two previews of the same file can reach this point together; the loser
# of the UPDATE takes its PNG back off disk instead of orphaning it.
result = await db.execute(
update(LibraryFile)
.where(LibraryFile.id == file_id, LibraryFile.thumbnail_path.is_(None))
.values(thumbnail_path=to_relative_path(thumb_path))
)
await db.commit()
if result.rowcount == 0:
thumb_path.unlink(missing_ok=True)
return ClientThumbnailResponse(updated=False)
return ClientThumbnailResponse(updated=True)
# ============ Photo Endpoints (#3077) ============
@router.post("/files/{file_id}/photos")
async def upload_file_photo(
file_id: int,
file: UploadFile = File(...),
db: AsyncSession = Depends(get_db),
auth_result: tuple[User | None, bool] = Depends(
require_ownership_permission(
Permission.LIBRARY_UPDATE_ALL,
Permission.LIBRARY_UPDATE_OWN,
)
),
):
"""Attach a photo of the printed result to a library file.
Photos are Bambuddy-side metadata, so external files take them too. Same
shape as the archive photo upload: extension allowlist, uuid-named on
disk, and the ``photos`` list re-assigned so SQLAlchemy sees the change.
"""
user, can_modify_all = auth_result
result = await db.execute(LibraryFile.active().where(LibraryFile.id == file_id))
library_file = result.scalar_one_or_none()
if not library_file:
raise HTTPException(status_code=404, detail="File not found")
# Ownership check (same shape as update_file)
if not can_modify_all:
if library_file.created_by_id != user.id:
raise HTTPException(status_code=403, detail="You can only update your own files")
if not file.filename or not file.filename.lower().endswith(PHOTO_EXTENSIONS):
raise HTTPException(status_code=400, detail="File must be an image (.jpg, .jpeg, .png, .webp)")
content = await file.read(MAX_PHOTO_BYTES + 1)
if len(content) > MAX_PHOTO_BYTES:
raise HTTPException(status_code=413, detail="Photo too large (max 10 MB)")
photos_dir = library_photos_dir(library_file.id)
photos_dir.mkdir(parents=True, exist_ok=True)
ext = Path(file.filename).suffix.lower()
photo_filename = f"{uuid.uuid4().hex[:8]}{ext}"
photo_path = photos_dir / photo_filename # SEC-PATH-OK: photo_filename = uuid.uuid4().hex[:8] + ext
photo_path.write_bytes(content)
photos = list(library_file.photos or [])
photos.append(photo_filename)
library_file.photos = photos
await db.commit()
await db.refresh(library_file)
return {"status": "uploaded", "filename": photo_filename, "photos": library_file.photos}
@router.get("/files/{file_id}/photos/{filename}")
async def get_file_photo(
file_id: int,
filename: str,
db: AsyncSession = Depends(get_db),
auth_result: tuple[User | None, bool] = Depends(
require_media_token_ownership(
Permission.LIBRARY_READ_ALL,
Permission.LIBRARY_READ_OWN,
)
),
):
"""Serve one photo. Media-token auth like the thumbnail route (#3025)."""
user, can_read_all = auth_result
result = await db.execute(LibraryFile.active().where(LibraryFile.id == file_id))
library_file = _ensure_library_file_visible(result.scalar_one_or_none(), user, can_read_all)
# Membership check first: names are uuid-generated on upload, so anything
# not in the stored list is not a photo, whatever is on disk.
if not library_file.photos or filename not in library_file.photos:
raise HTTPException(status_code=404, detail="Photo not found")
try:
photo_path = safe_join_under(library_photos_dir(library_file.id), filename, http=False)
except PathTraversalError:
raise HTTPException(status_code=404, detail="Photo not found") from None
if not photo_path.is_file():
raise HTTPException(status_code=404, detail="Photo not found")
media_type = PHOTO_MEDIA_TYPES.get(Path(filename).suffix.lower(), "image/jpeg")
return FastAPIFileResponse(str(photo_path), media_type=media_type)
@router.delete("/files/{file_id}/photos/{filename}")
async def delete_file_photo(
file_id: int,
filename: str,
db: AsyncSession = Depends(get_db),
auth_result: tuple[User | None, bool] = Depends(
require_ownership_permission(
Permission.LIBRARY_UPDATE_ALL,
Permission.LIBRARY_UPDATE_OWN,
)
),
):
"""Remove a photo from a library file."""
user, can_modify_all = auth_result
result = await db.execute(LibraryFile.active().where(LibraryFile.id == file_id))
library_file = result.scalar_one_or_none()
if not library_file:
raise HTTPException(status_code=404, detail="File not found")
if not can_modify_all:
if library_file.created_by_id != user.id:
raise HTTPException(status_code=403, detail="You can only update your own files")
if not library_file.photos or filename not in library_file.photos:
raise HTTPException(status_code=404, detail="Photo not found")
try:
photo_path = safe_join_under(library_photos_dir(library_file.id), filename, http=False)
except PathTraversalError:
raise HTTPException(status_code=404, detail="Photo not found") from None
if photo_path.is_file():
try:
photo_path.unlink()
except OSError as e:
logger.warning("Failed to delete photo from disk: %s", e)
photos = [p for p in library_file.photos if p != filename]
library_file.photos = photos if photos else None
await db.commit()
return {"status": "deleted", "photos": library_file.photos or []}
@router.get("/files/{file_id}/gcode")
async def get_gcode(
file_id: int,
plate: int | None = None,
db: AsyncSession = Depends(get_db),
auth_result: tuple[User | None, bool] = Depends(
require_ownership_permission(
Permission.LIBRARY_READ_ALL,
Permission.LIBRARY_READ_OWN,
)
),
):
"""Get gcode for a file (for preview).
Mirrors the archive route: ``?plate=2`` returns ``Metadata/plate_2.gcode``,
and omitting it returns the lowest-numbered plate. The viewer has been
sending ``plate`` since it gained a multi-plate URL, but this route took no
such parameter and FastAPI drops unknown query parameters silently — so
every multi-plate library file opened on whichever plate the slicer wrote
first into the zip, which is not plate 1.
"""
user, can_read_all = auth_result
result = await db.execute(LibraryFile.active().where(LibraryFile.id == file_id))
file = _ensure_library_file_visible(result.scalar_one_or_none(), user, can_read_all)
abs_path = to_absolute_path(file.file_path)
if not abs_path or not abs_path.exists():
raise HTTPException(status_code=404, detail="File not found on disk")
# Legacy sliced rows from before #1709 stored a `.gcode.3mf` ZIP body
# under file_type="gcode" — the on-disk filename is the truth in that
# case, so detect by suffix before checking the type column.
is_gcode_3mf = file.file_type in ("3mf", "gcode.3mf") or file.filename.lower().endswith(".gcode.3mf")
if plate is not None and plate < 1:
raise HTTPException(status_code=400, detail="Plate index must be >= 1")
if is_gcode_3mf:
try:
with zipfile.ZipFile(str(abs_path), "r") as zf:
gcode_files = [n for n in zf.namelist() if n.endswith(".gcode")]
if not gcode_files:
raise HTTPException(status_code=404, detail="No gcode found in 3MF file")
if plate is not None:
selected = select_plate_gcode_name(gcode_files, plate)
if selected is None:
raise HTTPException(status_code=404, detail=f"Plate {plate} not found in this file")
else:
selected = default_plate_gcode_name(gcode_files)
gcode_content = zf.read(selected)
from fastapi.responses import Response
return Response(content=gcode_content, media_type="text/plain")
except zipfile.BadZipFile:
raise HTTPException(status_code=400, detail="Invalid 3MF file")
elif file.file_type == "gcode":
return FastAPIFileResponse(str(abs_path), media_type="text/plain")
else:
raise HTTPException(status_code=400, detail="Unsupported file type")
# ============ Bulk Operations ============
@router.post("/files/move")
async def move_files(
data: FileMoveRequest,
db: AsyncSession = Depends(get_db),
auth_result: tuple[User | None, bool] = Depends(
require_ownership_permission(
Permission.LIBRARY_UPDATE_ALL,
Permission.LIBRARY_UPDATE_OWN,
)
),
):
"""Move multiple files to a folder.
Cross-boundary moves (managed ↔ external, or external ↔ external)
physically relocate the bytes — see ``_move_file_bytes``. Same-boundary
moves stay DB-only because the file's on-disk location doesn't depend
on which managed folder owns it.
Files not owned by the user are skipped (unless user has ``*_all``
permission). Each skip carries a structured reason so the UI can
surface "5 of 10 files were skipped: 3 had filename collisions on
the NAS, 2 are no longer on disk" rather than a blank "skipped: 5".
"""
user, can_modify_all = auth_result
# Verify folder exists if specified
target_folder: LibraryFolder | None = None
if data.folder_id is not None:
folder_result = await db.execute(select(LibraryFolder).where(LibraryFolder.id == data.folder_id))
target_folder = folder_result.scalar_one_or_none()
if not target_folder:
raise HTTPException(status_code=404, detail="Folder not found")
if target_folder.is_external and target_folder.external_readonly:
raise HTTPException(status_code=403, detail="Cannot move files to a read-only external folder")
target_is_external = target_folder is not None and target_folder.is_external
moved = 0
skipped = 0
skipped_reasons: list[dict] = []
for file_id in data.file_ids:
result = await db.execute(
LibraryFile.active().options(selectinload(LibraryFile.folder)).where(LibraryFile.id == file_id)
)
file = result.scalar_one_or_none()
if not file:
continue
# Ownership check
if not can_modify_all and file.created_by_id != user.id:
skipped += 1
skipped_reasons.append({"file_id": file_id, "code": "not_owner", "reason": "not the file owner"})
continue
# No bytes need to move when both ends are managed (same-boundary).
if not file.is_external and not target_is_external:
file.folder_id = data.folder_id
moved += 1
continue
# Block moves out of a read-only external mount. The user only has
# read access to the source, and a move is semantically a delete on
# the source — which a read-only mount can't fulfil. Without this
# guard we'd succeed at copying to the target, fail to unlink the
# source, and the same file would now exist in two places (with
# the DB pointing at only one).
if file.is_external and file.folder is not None and file.folder.external_readonly:
skipped += 1
skipped_reasons.append(
{"file_id": file_id, "code": "source_readonly", "reason": "source is on a read-only external folder"}
)
continue
# Otherwise relocate the bytes, then update the DB row to match.
try:
new_file_path = _move_file_bytes(file, target_folder)
except _MoveSkip as e:
skipped += 1
skipped_reasons.append({"file_id": file_id, "code": e.code, "reason": e.reason})
continue
file.is_external = target_is_external
file.folder_id = data.folder_id
file.file_path = new_file_path
# External rows historically carry `file_hash=None` (scan skips
# hashing). When pulling an external file into managed storage,
# compute the hash so dedup detection works for future uploads
# of the same content.
if not target_is_external and file.file_hash is None:
try:
abs_path = to_absolute_path(new_file_path)
if abs_path:
file.file_hash = calculate_file_hash(abs_path)
except OSError:
pass # leave hash null; dedup just won't match this row
moved += 1
await db.commit()
return {
"status": "success",
"moved": moved,
"skipped": skipped,
"skipped_reasons": skipped_reasons,
}
@router.post("/bulk-delete", response_model=BulkDeleteResponse)
async def bulk_delete(
data: BulkDeleteRequest,
db: AsyncSession = Depends(get_db),
auth_result: tuple[User | None, bool] = Depends(
require_ownership_permission(
Permission.LIBRARY_DELETE_ALL,
Permission.LIBRARY_DELETE_OWN,
)
),
):
"""Delete multiple files and/or folders.
Files not owned by the user are skipped (unless user has *_all permission).
"""
from backend.app.services.library_trash import delete_dependent_variants, release_queue_references
user, can_modify_all = auth_result
deleted_files = 0
deleted_folders = 0
skipped_files = 0
# External files bypass the trash and are removed for good, so the queue has
# to come off them. Collected here and dealt with once, below the loop.
hard_deleted: list[LibraryFile] = []
# Delete files first. Managed files go to trash (sweeper hard-deletes bytes
# later); external files bypass trash since their disk state is outside our
# control and can't be restored from trash anyway.
now = datetime.now(timezone.utc)
for file_id in data.file_ids:
result = await db.execute(LibraryFile.active().where(LibraryFile.id == file_id))
file = result.scalar_one_or_none()
if not file:
continue
if not can_modify_all and file.created_by_id != user.id:
skipped_files += 1
continue
if file.is_external:
abs_thumb_path = to_absolute_path(file.thumbnail_path)
if abs_thumb_path and abs_thumb_path.exists():
try:
abs_thumb_path.unlink()
except OSError as e:
logger.warning("Failed to delete thumbnail from disk: %s", e)
hard_deleted.append(file)
else:
file.deleted_at = now
deleted_files += 1
# After the loop and before any delete is issued (#2819). Order matters
# twice over: a query run while a delete is pending autoflushes it, taking
# the cascade with it, and releasing once for the whole set is a couple of
# statements rather than a couple per file.
if hard_deleted:
hard_deleted_ids = [f.id for f in hard_deleted]
await delete_dependent_variants(db, hard_deleted_ids)
await release_queue_references(db, hard_deleted_ids)
for file in hard_deleted:
remove_library_photos_dir(file.id)
await db.delete(file)
# Delete folders (cascade will handle contents). Folders have no ownership
# tracking, so users without *_all permission may only delete empty,
# non-external, non-linked folders (#1781) — same rule as DELETE /folders/{id}.
for folder_id in data.folder_ids:
result = await db.execute(select(LibraryFolder).where(LibraryFolder.id == folder_id))
folder = result.scalar_one_or_none()
if folder:
if not can_modify_all and await _restricted_folder_delete_blocker(db, folder):
continue
# Count files that will be deleted
file_count_result = await db.execute(
select(func.count(LibraryFile.id)).where(
LibraryFile.folder_id == folder_id,
LibraryFile.deleted_at.is_(None),
)
)
deleted_files += file_count_result.scalar() or 0
tree_file_ids = await _folder_tree_file_ids(db, folder_id)
await delete_dependent_variants(db, tree_file_ids)
await release_queue_references(db, tree_file_ids)
# The cascade hard-deletes every row in the subtree, so their
# photos go with them — same as DELETE /folders/{id} (#3077).
for doomed_id in tree_file_ids:
remove_library_photos_dir(doomed_id)
await db.delete(folder)
deleted_folders += 1
await db.commit()
return BulkDeleteResponse(deleted_files=deleted_files, deleted_folders=deleted_folders)
# ============ Stats Endpoint ============
@router.get("/stats")
async def get_library_stats(
db: AsyncSession = Depends(get_db),
auth_result: tuple[User | None, bool] = Depends(
require_ownership_permission(
Permission.LIBRARY_READ_ALL,
Permission.LIBRARY_READ_OWN,
)
),
):
"""Get library statistics."""
user, can_read_all = auth_result
# Stats exclude trashed files — users see counts/sizes for what's actually in the library.
# Without LIBRARY_READ_ALL the stats reflect only the caller's own files —
# match what the file list endpoint shows so the numbers stay consistent.
file_filters = [LibraryFile.deleted_at.is_(None)]
if user is not None and not can_read_all:
file_filters.append(LibraryFile.created_by_id == user.id)
# Total files
total_files_result = await db.execute(select(func.count(LibraryFile.id)).where(*file_filters))
total_files = total_files_result.scalar() or 0
# Total folders (folders are shared org structure, not per-user — count all)
total_folders_result = await db.execute(select(func.count(LibraryFolder.id)))
total_folders = total_folders_result.scalar() or 0
# Total size
total_size_result = await db.execute(select(func.sum(LibraryFile.file_size)).where(*file_filters))
total_size = total_size_result.scalar() or 0
# Files by type
type_result = await db.execute(
select(LibraryFile.file_type, func.count(LibraryFile.id)).where(*file_filters).group_by(LibraryFile.file_type)
)
files_by_type = dict(type_result.all())
# Total prints
total_prints_result = await db.execute(select(func.sum(LibraryFile.print_count)).where(*file_filters))
total_prints = total_prints_result.scalar() or 0
# Disk space info
library_dir = get_library_dir()
try:
disk_stat = shutil.disk_usage(library_dir)
disk_free_bytes = disk_stat.free
disk_total_bytes = disk_stat.total
disk_used_bytes = disk_stat.used
except OSError:
disk_free_bytes = 0
disk_total_bytes = 0
disk_used_bytes = 0
return {
"total_files": total_files,
"total_folders": total_folders,
"total_size_bytes": total_size,
"files_by_type": files_by_type,
"total_prints": total_prints,
"disk_free_bytes": disk_free_bytes,
"disk_total_bytes": disk_total_bytes,
"disk_used_bytes": disk_used_bytes,
}
Reference in New Issue View Git Blame Copy Permalink