import hashlib
import html
import json
import logging
import os
import re
import shutil
import zipfile
from datetime import date, datetime, time, timezone
from pathlib import Path
from defusedxml import ElementTree as ET
from sqlalchemy import and_, or_, select, text
from sqlalchemy.ext.asyncio import AsyncSession
from backend.app.core.config import settings
from backend.app.core.tasks import spawn_background_task
from backend.app.models.archive import PrintArchive
from backend.app.models.filament import Filament
from backend.app.models.printer import Printer
from backend.app.utils.archive_paths import archive_dir as resolve_archive_dir
from backend.app.utils.ffmpeg_output import NO_FFMPEG_OUTPUT, summarize_ffmpeg_stderr
from backend.app.utils.filename import clean_display_name
from backend.app.utils.safe_path import PathTraversalError, assert_under, safe_join_under
from backend.app.utils.threemf_tools import bed_temperature_from_config
logger = logging.getLogger(__name__)
def _copy_and_fsync(src: Path, dst: Path, chunk_size: int = 1024 * 1024) -> None:
"""Copy src to dst with an explicit chunked read/write and fsync the dst.
Replacement for shutil.copy2 in the archive pipeline. shutil.copy2 uses
Linux sendfile(), which on some kernels/filesystems has returned a short
count on the first call and truncated the destination for larger 3MF
uploads (#1032, observed on Raspberry Pi OS bookworm / armv7l). An
explicit loop with fsync avoids that path and guarantees the dest bytes
are on disk before the caller inspects them as a ZIP.
"""
with src.open("rb") as rf, dst.open("wb") as wf:
while True:
buf = rf.read(chunk_size)
if not buf:
break
wf.write(buf)
wf.flush()
os.fsync(wf.fileno())
shutil.copystat(src, dst)
def resolve_display_stem(filename: str) -> str:
"""Return a clean human-readable stem from a 3MF/gcode filename.
Bambu Studio's "Send to printer" dialog typically writes files like
``Plate_1.gcode.3mf`` (a sliced gcode payload wrapped in a 3MF container).
The naive ``Path(filename).stem`` only drops the last suffix, leaving
``Plate_1.gcode`` — which then surfaces in the archive UI as a confusing
``Plate_1.gcode`` rather than ``Plate_1`` (#1152 follow-up).
Strip the recognised print-format suffixes in order:
- ``.gcode.3mf`` → bare stem (Bambu Studio FTP send)
- ``.3mf`` → bare stem
- ``.gcode`` → bare stem (rare standalone gcode upload)
Anything else passes through unchanged.
"""
name = Path(filename).name # drop any path components
lower = name.lower()
for suffix in (".gcode.3mf", ".3mf", ".gcode"):
if lower.endswith(suffix):
return name[: -len(suffix)]
return Path(name).stem
def _read_plate_index(plate) -> int | None:
"""Return the 1-based index of a ``slice_info.config`` ```` element, or None.
Bambu Studio and OrcaSlicer record it as a ```` child — there is no ``plate_idx`` attribute on ````
itself, so an XPath predicate on one never matches (#2522).
"""
for meta in plate.findall("metadata"):
if meta.get("key") == "index":
value = meta.get("value")
if not value:
return None
try:
return int(value)
except ValueError:
return None
return None
def plate_indexes_in_3mf(file_path: Path) -> list[int | None]:
"""Return one entry per ```` a Bambu 3MF declares, in file order.
Reads only ``Metadata/slice_info.config``. An entry is None when that plate
carries no readable index, and the list is empty for a file that could not
be read at all — callers must not confuse either with "this file has plates
and yours is not among them". An unreadable 3MF is a parse this code does
not understand, not evidence about which plate it holds (#2957).
"""
try:
with zipfile.ZipFile(file_path, "r") as zf:
if "Metadata/slice_info.config" not in zf.namelist():
return []
root = ET.fromstring(zf.read("Metadata/slice_info.config").decode())
return [_read_plate_index(plate) for plate in root.findall(".//plate")]
except Exception:
return []
def peek_plate_index_in_3mf(file_path: Path) -> int | None:
"""Return the plate index a single-plate Bambu 3MF represents, or None.
Reads only ``Metadata/slice_info.config`` to keep this cheap — used by
the print-start callback to verify that the 3MF we just downloaded over
FTP actually matches the plate the printer is running (#1204). The full
ThreeMFParser does much more work and runs later inside ArchiveService.
An all-plates export carries every plate, so "which plate is this file"
has no answer; returning None there keeps the #1204 guard from reading
plate 1 out of such a file, declaring a mismatch against the plate that
is really running, and discarding a perfectly good 3MF (#2522).
"""
plates = plate_indexes_in_3mf(file_path)
return plates[0] if len(plates) == 1 else None
_PLATE_SUFFIX_RE = re.compile(r"^(.*?)(\s*-\s*Plate\s+|_plate_)(\d+)$", re.IGNORECASE)
def swap_plate_suffix(name: str | None, target_plate: int) -> str | None:
"""Return ``name`` with its trailing plate number replaced, or None.
Bambu Studio names multi-plate uploads ``" - Plate "`` (and
a lowercase ``"_plate_"`` variant exists too — see
test_print_start_expected_promotion). When MQTT subtask_name lags
across consecutive plates of the same model (#1204) the suffix points
at the previous plate; swapping it gives us the correct upload to
re-fetch from FTP. Returns None if no recognised suffix is present.
"""
if not name:
return None
m = _PLATE_SUFFIX_RE.match(name)
if not m:
return None
base, separator, _ = m.groups()
return f"{base}{separator}{target_plate}"
# How much of a plate's G-code to scan for header/config values. The header
# block ends in the first kilobyte; the CONFIG_BLOCK that follows it carries
# layer_height 14-25KB in (measured across the sliced 3MFs on hand, Bambu
# Studio and OrcaSlicer alike), so 4KB — what this used to read — could only
# ever see the header.
_GCODE_SCAN_BYTES = 64 * 1024
class ThreeMFParser:
"""Parser for Bambu Lab 3MF files."""
def __init__(self, file_path: Path, plate_number: int | None = None):
self.file_path = file_path
self.plate_number = plate_number # Which plate was printed (1, 2, 3, etc.)
self.metadata: dict = {}
def parse(self) -> dict:
"""Extract metadata from 3MF file."""
try:
with zipfile.ZipFile(self.file_path, "r") as zf:
self._parse_slice_info(zf) # Now sets self.plate_number from slice_info
self._parse_project_settings(zf)
self._parse_gcode_header(zf)
self._parse_3dmodel(zf)
self._extract_thumbnail(zf) # Uses correct plate_number for thumbnail
# Enhance print_name with plate info if this is a multi-plate export
plate_index = self.metadata.get("_plate_index")
if plate_index and plate_index > 1:
# Append plate number to distinguish from other plates
existing_name = self.metadata.get("print_name", "")
if existing_name and f"Plate {plate_index}" not in existing_name:
self.metadata["print_name"] = f"{existing_name} - Plate {plate_index}"
# ALWAYS prefer slice_info values - they contain ONLY filaments actually used in print
# project_settings contains ALL configured filaments (AMS slots), not just used ones
if self.metadata.get("_slice_filament_type"):
self.metadata["filament_type"] = self.metadata["_slice_filament_type"]
if self.metadata.get("_slice_filament_color"):
self.metadata["filament_color"] = self.metadata["_slice_filament_color"]
# Clean up internal keys
self.metadata.pop("_slice_filament_type", None)
self.metadata.pop("_slice_filament_color", None)
self.metadata.pop("_plate_index", None)
except Exception as e:
# Return whatever metadata was extracted before the error, but
# surface the failure so corrupted / truncated 3MF archives are
# visible in support bundles (#1032).
logger.warning(
"ThreeMFParser: failed to parse %s: %s(%s) — returning partial metadata",
self.file_path,
type(e).__name__,
e,
)
return self.metadata
def _parse_slice_info(self, zf: zipfile.ZipFile):
"""Parse slice_info.config for print settings and printable objects."""
try:
if "Metadata/slice_info.config" in zf.namelist():
content = zf.read("Metadata/slice_info.config").decode()
root = ET.fromstring(content)
# Extract printer_model_id from plate metadata
# Format:
for meta in root.findall(".//metadata"):
key = meta.get("key")
value = meta.get("value")
if key == "printer_model_id" and value:
from backend.app.utils.printer_models import normalize_printer_model_id
normalized = normalize_printer_model_id(value)
if normalized:
self.metadata["sliced_for_model"] = normalized
break
# Loop every so multi-plate exports get summed file-level
# totals. Pre-fix, this used `root.find(".//plate")` which
# returned only the first plate — file-level `print_time_seconds`
# / `filament_used_grams` reflected plate 1 alone, and the
# archive card / project rollup under-reported by the number
# of plates (#1593). Per-plate breakdown is still served by
# the dedicated `/plates` endpoint.
plates = root.findall(".//plate")
summed_time = 0
summed_grams = 0.0
any_time_seen = False
any_grams_seen = False
for plate in plates:
# Plate-level fields that only make sense at the file
# level when there's exactly one plate. ``plate_number``
# / ``_plate_index`` describe which plate the export
# represents — meaningless for an all-plates 3MF, so we
# only record them in the single-plate case. ``bed_type``
# is also single-valued; we take the first plate's value
# as a best-effort default for the archive metadata.
plate_index_value: int | None = None
for meta in plate.findall("metadata"):
key = meta.get("key")
value = meta.get("value")
if key == "index" and value:
try:
plate_index_value = int(value)
except ValueError:
pass # Skip non-numeric plate index
elif key == "prediction" and value:
try:
summed_time += int(value)
any_time_seen = True
except ValueError:
pass
elif key == "weight" and value:
try:
summed_grams += float(value)
any_grams_seen = True
except ValueError:
pass
elif key == "curr_bed_type" and value and "bed_type" not in self.metadata:
self.metadata["bed_type"] = value
# Per-plate object lists are only kept at the file level
# when there's one plate — the skip-object affordance
# operates on the plate being printed, which is the
# `/plates` endpoint's job for multi-plate exports.
if len(plates) == 1:
if plate_index_value is not None:
if not self.plate_number:
self.plate_number = plate_index_value
self.metadata["_plate_index"] = plate_index_value
printable_objects: dict[int, str] = {}
for obj in plate.findall("object"):
identify_id = obj.get("identify_id")
name = obj.get("name")
skipped = obj.get("skipped", "false")
if identify_id and name and skipped.lower() != "true":
try:
printable_objects[int(identify_id)] = name
except ValueError:
pass # Skip objects with non-numeric identify_id
if printable_objects:
self.metadata["printable_objects"] = printable_objects
if any_time_seen:
self.metadata["print_time_seconds"] = summed_time
if any_grams_seen:
self.metadata["filament_used_grams"] = round(summed_grams, 2)
# Get filament info from filaments ACTUALLY USED in the print
# slice_info has
# Only include filaments where used_g > 0
filaments = root.findall(".//filament")
if filaments:
# Collect unique filament types and colors for filaments that are actually used
types = []
colors = []
for f in filaments:
# Check if this filament is actually used in the print
used_g = f.get("used_g", "0")
try:
used_amount = float(used_g)
except (ValueError, TypeError):
used_amount = 0
# Only include if used_g > 0 (filament is actually consumed)
if used_amount > 0:
ftype = f.get("type")
fcolor = f.get("color")
if ftype and ftype not in types:
types.append(ftype)
if fcolor and fcolor not in colors:
colors.append(fcolor)
if types:
self.metadata["_slice_filament_type"] = ", ".join(types)
if colors:
self.metadata["_slice_filament_color"] = ",".join(colors)
# Collect per-slot filament usage for tracking & notifications
filament_slots = []
for f in filaments:
slot_id = f.get("id")
used_g_str = f.get("used_g", "0")
try:
used_g = float(used_g_str)
except (ValueError, TypeError):
used_g = 0
if used_g > 0 and slot_id:
filament_slots.append(
{
"slot_id": int(slot_id),
"used_g": round(used_g, 2),
"type": f.get("type", ""),
"color": f.get("color", ""),
}
)
if filament_slots:
self.metadata["filament_slots"] = filament_slots
except Exception:
pass # Skip unparseable slice_info metadata
def _parse_project_settings(self, zf: zipfile.ZipFile):
"""Parse project settings for print configuration."""
try:
if "Metadata/project_settings.config" in zf.namelist():
content = zf.read("Metadata/project_settings.config").decode()
try:
data = json.loads(content)
self._extract_filament_info(data)
self._extract_print_settings(data)
except json.JSONDecodeError:
pass # Skip malformed project_settings JSON
except Exception:
pass # Skip unreadable project settings file
def _printed_plate_gcode(self, gcode_files: list[str]) -> str:
"""Return the G-code entry for the plate this archive is about.
``plate_number`` is known for a plate-specific export (slice_info sets
it) and picking blindly is wrong there: a project sliced with plate 2
at 0.08 and plate 1 at 0.2 would otherwise report plate 1's numbers.
Falls back to the lowest plate index, then to zip order, so a file
whose entries are named some other way still parses as it did before.
"""
if self.plate_number:
wanted = f"Metadata/plate_{self.plate_number}.gcode"
if wanted in gcode_files:
return wanted
def plate_index(name: str) -> int:
match = re.search(r"plate_(\d+)\.gcode$", name)
return int(match.group(1)) if match else 10**6
return min(gcode_files, key=lambda name: (plate_index(name), gcode_files.index(name)))
def _parse_gcode_header(self, zf: zipfile.ZipFile):
"""Parse the printed plate's G-code for what only it can settle.
The plate's own G-code is the file the printer executes, so where it
disagrees with ``project_settings.config`` — the *project's* record,
which a multi-plate or per-plate-modified export can leave describing
a different plate entirely — the G-code wins.
"""
try:
gcode_files = [f for f in zf.namelist() if f.endswith(".gcode")]
if not gcode_files:
return
gcode_path = self._printed_plate_gcode(gcode_files)
# 64KB, not 4KB: the header block ends within the first kilobyte,
# but the CONFIG_BLOCK that carries layer_height starts right after
# it and the keys are alphabetical, so layer_height lands 14-25KB
# in on real files. The read is decompress-on-demand, so the cost
# of the wider window is a few tens of KB per archived file.
with zf.open(gcode_path) as f:
header = f.read(_GCODE_SCAN_BYTES).decode("utf-8", errors="ignore")
# Look for "; total layer number: XX" pattern
match = re.search(r";\s*total\s+layer\s+number[:\s]+(\d+)", header, re.IGNORECASE)
if match:
self.metadata["total_layers"] = int(match.group(1))
# Layer height, overriding project_settings.config when both are
# present. The project config records the project's settings and can
# describe a plate other than this one; the plate's G-code is what
# the printer executes, so it decides. Anchored to the line start so keys ending in
# "layer_height" (independent_support_layer_height) can't match.
match = re.search(r"^;\s*layer_height\s*=\s*([\d.]+)\s*$", header, re.IGNORECASE | re.MULTILINE)
if match:
try:
self.metadata["layer_height"] = float(match.group(1))
except ValueError:
pass # Malformed value: keep whatever project_settings gave us
# Total filament usage. The slicer writes the print's totals into
# the G-code header ("; total filament weight [g] : 126.26"). Only
# a fallback — slice_info.config is more authoritative when present
# — but it covers sliced outputs whose slice_info lacks per-filament
# used_g, and it's the slicer's own figure regardless.
if "filament_used_grams" not in self.metadata:
match = re.search(r";\s*total\s+filament\s+weight\s*\[g\]\s*:\s*([\d.]+)", header, re.IGNORECASE)
if match:
self.metadata["filament_used_grams"] = float(match.group(1))
if "filament_used_mm" not in self.metadata:
match = re.search(r";\s*total\s+filament\s+length\s*\[mm\]\s*:\s*([\d.]+)", header, re.IGNORECASE)
if match:
self.metadata["filament_used_mm"] = float(match.group(1))
# Look for printer_model in gcode header (fallback if not found in slice_info)
# Format: "; printer_model = Bambu Lab X1 Carbon" or "; printer_model = X1C"
if "sliced_for_model" not in self.metadata:
match = re.search(r";\s*printer_model\s*=\s*(.+)", header, re.IGNORECASE)
if match:
from backend.app.utils.printer_models import normalize_printer_model
raw_model = match.group(1).strip()
self.metadata["sliced_for_model"] = normalize_printer_model(raw_model)
except Exception:
pass # G-code header parsing is best-effort; metadata may come from other sources
def _extract_filament_info(self, data: dict):
"""Extract filament info from project settings — includes support
materials so a PLA-model / PVA-support project shows both on the
archive card badge (#1881).
Earlier code filtered by ``filament_is_support``; that hid PVA
(and any other soluble/breakaway support material) from the card
even when the user had explicitly configured it, and made source
3MFs look single-material until the print completed. slice_info
(parsed separately) is still preferred when present — it lists
only filaments the print actually consumes, this fallback only
runs on unsliced source 3MFs.
"""
try:
filament_types = data.get("filament_type", [])
filament_colors = data.get("filament_colour", [])
if not filament_types:
return
unique_types: list[str] = []
for ftype in filament_types:
if ftype and ftype not in unique_types:
unique_types.append(ftype)
unique_colors: list[str] = []
for color in filament_colors:
if color and color not in unique_colors:
unique_colors.append(color)
if unique_types:
self.metadata["filament_type"] = ", ".join(unique_types)
if unique_colors:
self.metadata["filament_color"] = ",".join(unique_colors)
except Exception:
pass # Filament info is optional; fall back to slice_info values
def _extract_print_settings(self, data: dict):
"""Extract print settings from JSON config."""
try:
# Layer height - usually an array, get first value
if "layer_height" in data:
val = data["layer_height"]
if isinstance(val, list) and val:
self.metadata["layer_height"] = float(val[0])
elif isinstance(val, (int, float, str)):
self.metadata["layer_height"] = float(val)
# Nozzle diameter
if "nozzle_diameter" in data:
val = data["nozzle_diameter"]
if isinstance(val, list) and val:
self.metadata["nozzle_diameter"] = float(val[0])
elif isinstance(val, (int, float, str)):
self.metadata["nozzle_diameter"] = float(val)
# Bed temperature, for the plate this project is sliced for. This
# used to look for `bed_temperature` alone, a key BambuStudio does
# not write -- so every archive from a Bambu slice stored NULL, and
# preheat fell back to a configured bed temperature on every job
# (#2989). Orca-exported 3MFs keep working through the generic keys.
bed_temperature = bed_temperature_from_config(data)
if bed_temperature is not None:
self.metadata["bed_temperature"] = bed_temperature
# Nozzle temperature
for key in ["nozzle_temperature_initial_layer", "nozzle_temperature"]:
if key in data:
val = data[key]
if isinstance(val, list) and val:
self.metadata["nozzle_temperature"] = int(float(val[0]))
elif isinstance(val, (int, float, str)):
self.metadata["nozzle_temperature"] = int(float(val))
break
# Printer model (extract and normalize)
if "printer_model" in data:
from backend.app.utils.printer_models import normalize_printer_model
self.metadata["sliced_for_model"] = normalize_printer_model(data["printer_model"])
# Build plate type — only set from project_settings if slice_info didn't already
# provide it (slice_info is more authoritative as it reflects the exported plate).
if "bed_type" not in self.metadata and "curr_bed_type" in data:
val = data["curr_bed_type"]
if isinstance(val, str) and val.strip():
self.metadata["bed_type"] = val.strip()
except Exception:
pass # Print settings are optional; missing values are left unset
def _parse_3dmodel(self, zf: zipfile.ZipFile):
"""Parse 3D/3dmodel.model for MakerWorld metadata."""
try:
model_path = "3D/3dmodel.model"
if model_path not in zf.namelist():
return
content = zf.read(model_path).decode("utf-8", errors="ignore")
# Parse XML metadata elements
# MakerWorld adds metadata like: username
metadata_pattern = r']*>([^<]*)'
matches = re.findall(metadata_pattern, content)
# 3MF metadata values are XML-encoded — `&` becomes `&`, etc.
# ProjectPageParser learned this the hard way: BambuStudio sometimes
# writes triple-encoded payloads (`&amp;`), so we unescape
# in a loop until the string stabilises. Without this, a Title like
# "Foo & Bar" lands in the DB as raw "Foo & Bar" and React then
# double-escapes it on render to "Foo & Bar" (#1658).
makerworld_fields = {}
for name, value in matches:
decoded = value.strip()
prev = None
while prev != decoded:
prev = decoded
decoded = html.unescape(decoded)
makerworld_fields[name] = decoded
# Check for direct MakerWorld URL in content
url_pattern = r'https?://makerworld\.com/[^\s<>"\']+/models/(\d+)'
url_match = re.search(url_pattern, content)
if url_match:
self.metadata["makerworld_url"] = url_match.group(0)
self.metadata["makerworld_model_id"] = url_match.group(1)
# Extract model ID from DSM reference in image URLs
# Format: https://makerworld.bblmw.com/makerworld/model/DSM00000001275614/...
# The numeric part (1275614) is the MakerWorld model ID
if "makerworld_url" not in self.metadata:
dsm_pattern = r"DSM0+(\d+)"
dsm_match = re.search(dsm_pattern, content)
if dsm_match:
model_id = dsm_match.group(1)
self.metadata["makerworld_url"] = f"https://makerworld.com/en/models/{model_id}"
self.metadata["makerworld_model_id"] = model_id
# Store designer info
if "Designer" in makerworld_fields:
self.metadata["designer"] = makerworld_fields["Designer"]
if "Title" in makerworld_fields:
self.metadata["print_name"] = makerworld_fields["Title"]
except Exception:
pass # MakerWorld/3dmodel metadata is optional
def _extract_thumbnail(self, zf: zipfile.ZipFile):
"""Extract thumbnail image from 3MF.
If a plate_number was specified, try to use that plate's thumbnail first.
"""
thumbnail_paths = []
# If a specific plate was printed, try that thumbnail first
if self.plate_number:
thumbnail_paths.append(f"Metadata/plate_{self.plate_number}.png")
# Fallback to default paths
thumbnail_paths.extend(
[
"Metadata/plate_1.png",
"Metadata/thumbnail.png",
"Metadata/model_thumbnail.png",
# Project-wide thumbnail BambuStudio embeds at upload time. We
# only reach this when BS hasn't written a per-plate
# ``Metadata/plate_N.png`` — most notably the #1493 cross-class
# re-slice path where ``--arrange`` rearranges objects but the
# CLI then doesn't emit a fresh per-plate preview. The
# ``_middle`` size is the editor-quality variant (~500 KB);
# ``_small`` and ``_3mf`` are smaller alternates if it's not
# present. Without this fallback the re-sliced archive cards
# render without a cover image.
"Auxiliaries/.thumbnails/thumbnail_middle.png",
"Auxiliaries/.thumbnails/thumbnail_small.png",
"Auxiliaries/.thumbnails/thumbnail_3mf.png",
]
)
for thumb_path in thumbnail_paths:
if thumb_path in zf.namelist():
self.metadata["_thumbnail_data"] = zf.read(thumb_path)
self.metadata["_thumbnail_ext"] = ".png"
break
def extract_printable_objects_from_archive(
file_path: Path, plate_number: int | None = None
) -> tuple[dict[int, dict], list | None]:
"""Objects and plate bbox for an archived print, read off local disk.
The archive of a running print usually holds the very 3MF the printer is
executing, so the object list can be rebuilt without asking the printer for
a file we already have -- 15 MB over FTPS from a machine that is mid-print,
in the case this was written for. Returns empty when the archive has
no readable 3MF, which is the caller's signal to fall back to the printer.
"""
if not file_path.is_file() or not str(file_path).endswith(".3mf"):
return {}, None
try:
data = file_path.read_bytes()
except OSError:
return {}, None
return extract_printable_objects_from_3mf(data, plate_number=plate_number, include_positions=True)
def extract_printable_objects_from_3mf(
data: bytes, plate_number: int | None = None, include_positions: bool = False
) -> dict[int, str] | dict[int, dict] | tuple[dict[int, dict], list | None]:
"""Extract printable objects from 3MF file bytes.
This is a lightweight function used during print start to get the list
of objects that can be skipped.
Args:
data: Raw bytes of the 3MF file
plate_number: Which plate was printed (1-based), or None for first plate
include_positions: If True, return tuple of (objects dict, bbox_all)
Returns:
If include_positions=False: Dictionary mapping identify_id (int) to object name (str)
If include_positions=True: Tuple of (dict mapping identify_id to {name, x, y}, bbox_all list or None)
"""
from io import BytesIO
printable_objects: dict = {}
bbox_all: list | None = None
try:
with zipfile.ZipFile(BytesIO(data), "r") as zf:
if "Metadata/slice_info.config" not in zf.namelist():
return printable_objects
content = zf.read("Metadata/slice_info.config").decode()
root = ET.fromstring(content)
plates = root.findall(".//plate")
if not plates:
return printable_objects
# Pick the plate that is actually printing. An all-plates export
# lists every plate, so without this we offered the objects (and
# the marker positions) of plate 1 whatever the printer was
# running (#2522). Falling back to the first plate keeps the
# single-plate export — the common case — working when the caller
# has no plate to give us.
plate = None
if plate_number is not None:
plate = next((p for p in plates if _read_plate_index(p) == plate_number), None)
if plate is None:
plate = plates[0]
# Derive plate_idx from the plate we settled on, never from the
# requested one: on a fallback they differ, and plate_idx also
# selects the plate_N.json the positions come from.
plate_idx = _read_plate_index(plate) or 1
# Load position data from plate_N.json if we need positions
# Build a lookup by name - use list to handle duplicate names
bbox_by_name: dict[str, list[list]] = {}
if include_positions:
plate_json_path = f"Metadata/plate_{plate_idx}.json"
if plate_json_path in zf.namelist():
try:
plate_json = json.loads(zf.read(plate_json_path).decode())
# Get bbox_all - the bounding box of all objects (used for image bounds)
bbox_all = plate_json.get("bbox_all")
for bbox_obj in plate_json.get("bbox_objects", []):
obj_name = bbox_obj.get("name")
bbox = bbox_obj.get("bbox", [])
if obj_name and len(bbox) >= 4:
if obj_name not in bbox_by_name:
bbox_by_name[obj_name] = []
bbox_by_name[obj_name].append(bbox)
except (json.JSONDecodeError, KeyError):
pass # Position data is optional; objects will lack x/y coordinates
# Extract objects from slice_info.config
for obj in plate.findall("object"):
identify_id = obj.get("identify_id")
name = obj.get("name")
skipped = obj.get("skipped", "false")
if identify_id and name and skipped.lower() != "true":
try:
obj_id = int(identify_id)
if include_positions:
x, y = None, None
# Match by name - pop first bbox to handle duplicates
bboxes = bbox_by_name.get(name)
if bboxes:
bbox = bboxes.pop(0)
# Calculate center from bbox [x_min, y_min, x_max, y_max]
x = (bbox[0] + bbox[2]) / 2
y = (bbox[1] + bbox[3]) / 2
printable_objects[obj_id] = {"name": name, "x": x, "y": y}
else:
printable_objects[obj_id] = name
except ValueError:
pass # Skip objects with non-numeric identify_id
except Exception:
pass # Return empty dict if 3MF is corrupt or unreadable
if include_positions:
return printable_objects, bbox_all
return printable_objects
class ProjectPageParser:
"""Parser for extracting project page data from Bambu Lab 3MF files."""
def __init__(self, file_path: Path):
self.file_path = file_path
def parse(self, archive_id: int) -> dict:
"""Extract project page metadata and images from 3MF file."""
import html
result = {
"title": None,
"description": None,
"designer": None,
"designer_user_id": None,
"license": None,
"copyright": None,
"creation_date": None,
"modification_date": None,
"origin": None,
"profile_title": None,
"profile_description": None,
"profile_cover": None,
"profile_user_id": None,
"profile_user_name": None,
"design_model_id": None,
"design_profile_id": None,
"design_region": None,
"model_pictures": [],
"profile_pictures": [],
"thumbnails": [],
}
try:
with zipfile.ZipFile(self.file_path, "r") as zf:
# Parse 3D/3dmodel.model for metadata
model_path = "3D/3dmodel.model"
if model_path in zf.namelist():
content = zf.read(model_path).decode("utf-8", errors="ignore")
# Extract metadata elements using regex
# Format: Value or
metadata_pattern = r']*>([^<]*)'
matches = re.findall(metadata_pattern, content)
field_mapping = {
"Title": "title",
"Description": "description",
"Designer": "designer",
"DesignerUserId": "designer_user_id",
"License": "license",
"Copyright": "copyright",
"CreationDate": "creation_date",
"ModificationDate": "modification_date",
"Origin": "origin",
"ProfileTitle": "profile_title",
"ProfileDescription": "profile_description",
"ProfileCover": "profile_cover",
"ProfileUserId": "profile_user_id",
"ProfileUserName": "profile_user_name",
"DesignModelId": "design_model_id",
"DesignProfileId": "design_profile_id",
"DesignRegion": "design_region",
}
for name, value in matches:
if name in field_mapping:
# Decode HTML entities multiple times (content is often triple-encoded)
decoded = value.strip()
prev = None
while prev != decoded:
prev = decoded
decoded = html.unescape(decoded)
# Normalize non-breaking spaces to regular spaces
decoded = decoded.replace("\xa0", " ")
result[field_mapping[name]] = decoded if decoded else None
# List images in Auxiliaries folder
from urllib.parse import quote
for name in zf.namelist():
if name.startswith("Auxiliaries/Model Pictures/"):
filename = name.split("/")[-1]
if filename:
result["model_pictures"].append(
{
"name": filename,
"path": name,
"url": f"/api/v1/archives/{archive_id}/project-image/{quote(name, safe='')}",
}
)
elif name.startswith("Auxiliaries/Profile Pictures/"):
filename = name.split("/")[-1]
if filename:
result["profile_pictures"].append(
{
"name": filename,
"path": name,
"url": f"/api/v1/archives/{archive_id}/project-image/{quote(name, safe='')}",
}
)
elif name.startswith("Auxiliaries/.thumbnails/"):
filename = name.split("/")[-1]
if filename:
result["thumbnails"].append(
{
"name": filename,
"path": name,
"url": f"/api/v1/archives/{archive_id}/project-image/{quote(name, safe='')}",
}
)
except Exception as e:
result["_error"] = str(e)
return result
def get_image(self, image_path: str) -> tuple[bytes, str] | None:
"""Extract an image from the 3MF file.
Returns tuple of (image_data, content_type) or None if not found.
"""
try:
with zipfile.ZipFile(self.file_path, "r") as zf:
if image_path in zf.namelist():
data = zf.read(image_path)
# Determine content type from extension
ext = image_path.lower().split(".")[-1]
content_types = {
"png": "image/png",
"jpg": "image/jpeg",
"jpeg": "image/jpeg",
"webp": "image/webp",
"gif": "image/gif",
}
content_type = content_types.get(ext, "application/octet-stream")
return (data, content_type)
except Exception:
pass # Return None if image cannot be extracted from 3MF
return None
def update_metadata(self, updates: dict) -> bool:
"""Update project page metadata in the 3MF file.
Args:
updates: Dict with fields to update (title, description, designer, etc.)
Returns:
True if successful, False otherwise.
"""
import html
import tempfile
try:
# Read the 3MF file
with zipfile.ZipFile(self.file_path, "r") as zf_read:
# Find and read the 3dmodel.model file
model_path = "3D/3dmodel.model"
if model_path not in zf_read.namelist():
return False
content = zf_read.read(model_path).decode("utf-8")
# Update metadata fields
field_mapping = {
"title": "Title",
"description": "Description",
"designer": "Designer",
"license": "License",
"copyright": "Copyright",
"profile_title": "ProfileTitle",
"profile_description": "ProfileDescription",
}
for field, xml_name in field_mapping.items():
if field in updates and updates[field] is not None:
new_value = html.escape(updates[field])
# Replace existing metadata or we'd need to add it
pattern = rf'(]*>)[^<]*()'
replacement = rf"\g<1>{new_value}\g<2>"
content = re.sub(pattern, replacement, content)
# Write to a temporary file first
with tempfile.NamedTemporaryFile(delete=False, suffix=".3mf") as tmp:
tmp_path = Path(tmp.name)
# Create new zip with updated content
with zipfile.ZipFile(tmp_path, "w", zipfile.ZIP_DEFLATED) as zf_write:
for item in zf_read.namelist():
if item == model_path:
zf_write.writestr(item, content.encode("utf-8"))
else:
zf_write.writestr(item, zf_read.read(item))
# Replace original file with updated one
shutil.move(tmp_path, self.file_path)
return True
except Exception:
# Clean up temp file if it exists
if "tmp_path" in locals() and tmp_path.exists():
tmp_path.unlink()
return False
async def _null_print_log_thumbnail_paths(db: AsyncSession, archive_id: int) -> None:
"""NULL thumbnail_path on PrintLogEntry rows linked to *archive_id*.
Called from both soft- and hard-delete paths before the archive's files
leave disk. The FK on PrintLogEntry.archive_id is ON DELETE SET NULL so
log rows survive the archive — without this clear, their cached
thumbnail_path would still point at a deleted file and the print-log
view would 404-storm on every render (#1348 follow-up). Lazy-NULL on
the GET route self-heals stragglers (e.g. failed prints that never had
a thumbnail written), but eager clear here avoids the one-time storm.
"""
from sqlalchemy import update as sa_update
from backend.app.models.print_log import PrintLogEntry
await db.execute(sa_update(PrintLogEntry).where(PrintLogEntry.archive_id == archive_id).values(thumbnail_path=None))
async def _delete_related_queue_items(db: AsyncSession, archive_id: int) -> int:
"""Delete every queue item pointing at *archive_id* (#1734).
Called from ``soft_delete_archive``. Hard-delete is covered by the
``ON DELETE CASCADE`` on ``print_queue.archive_id`` — same end state
via the FK. Pre-#1734 this helper merely flipped pending rows to
``status='cancelled'`` while leaving every other status alone and
leaving the rows in the DB, which surprised users who expected the
queue lines to disappear when their backing archive went away. Worse,
a Send-All archive backed N queue items (one per plate, #1733) — soft-
deleting that archive left N "cancelled" rows behind, none of which
could ever dispatch.
Now we delete unconditionally regardless of status. ``printing`` rows
are blocked one layer up at the route (``delete_archive`` returns 409
when a related row is mid-print) so we never delete an actively-
running queue row out from under the dispatcher. Completed / failed
/ cancelled rows go too — they're queue history, not print history.
PrintLogEntry rows are the authoritative print history and are
untouched (FK ``ON DELETE SET NULL``).
Returns the number of rows removed so the caller can report it.
"""
from sqlalchemy import delete as sa_delete
from backend.app.models.print_queue import PrintQueueItem
result = await db.execute(sa_delete(PrintQueueItem).where(PrintQueueItem.archive_id == archive_id))
return result.rowcount or 0
async def _count_related_queue_items(db: AsyncSession, archive_id: int) -> tuple[int, int]:
"""Return ``(total, printing)`` queue items linked to *archive_id*.
Used by the archive GET response so the frontend delete-confirm modal
can surface how much the deletion will wipe out, and by the delete
route so it can 409 when a related row is currently printing (#1734).
"""
from sqlalchemy import func as sa_func, select as sa_select
from backend.app.models.print_queue import PrintQueueItem
total = (
await db.execute(
sa_select(sa_func.count()).select_from(PrintQueueItem).where(PrintQueueItem.archive_id == archive_id)
)
).scalar_one()
printing = (
await db.execute(
sa_select(sa_func.count())
.select_from(PrintQueueItem)
.where(
PrintQueueItem.archive_id == archive_id,
PrintQueueItem.status == "printing",
)
)
).scalar_one()
return int(total or 0), int(printing or 0)
class ArchiveService:
"""Service for archiving print jobs."""
def __init__(self, db: AsyncSession):
self.db = db
@staticmethod
def compute_file_hash(file_path: Path) -> str:
"""Compute SHA256 hash of a file for duplicate detection."""
sha256 = hashlib.sha256()
with open(file_path, "rb") as f:
# Read in chunks to handle large files
for chunk in iter(lambda: f.read(8192), b""):
sha256.update(chunk)
return sha256.hexdigest()
async def get_duplicate_hashes_and_names(self) -> tuple[set[str], set[tuple[str, str]]]:
"""Get all content hashes and (print name, hash) pairs that appear more than once.
For hashes: returns all hashes with > 1 archive (true duplicates).
For name/hash pairs: returns only pairs that have > 1 archive
(i.e., same file archived multiple times, not different files with same name).
Returns a tuple of (duplicate_hashes, duplicate_name_hash_pairs).
"""
from sqlalchemy import func
# Soft-deleted archives don't appear in the listing (#1343), so they
# mustn't influence the duplicate-group counts either — otherwise a
# group with 1 live + 4 soft-deleted would still be flagged as a
# duplicate even though the user only sees one row.
result = await self.db.execute(
select(PrintArchive.content_hash)
.where(PrintArchive.content_hash.isnot(None), PrintArchive.deleted_at.is_(None))
.group_by(PrintArchive.content_hash)
.having(func.count(PrintArchive.id) > 1)
)
duplicate_hashes = {row[0] for row in result.all()}
# Find print names that have multiple archives with the SAME hash
# This avoids marking different files with the same name as duplicates
result = await self.db.execute(
select(func.lower(PrintArchive.print_name), PrintArchive.content_hash)
.where(
PrintArchive.print_name.isnot(None),
PrintArchive.content_hash.isnot(None),
PrintArchive.deleted_at.is_(None),
)
.group_by(func.lower(PrintArchive.print_name), PrintArchive.content_hash)
.having(func.count(PrintArchive.id) > 1)
)
duplicate_name_hash_pairs = {(row[0], row[1]) for row in result.all()}
return duplicate_hashes, duplicate_name_hash_pairs
async def find_duplicates(
self,
archive_id: int,
content_hash: str | None = None,
print_name: str | None = None,
makerworld_model_id: str | None = None,
) -> list[dict]:
"""Find duplicate archives based on hash or name matching.
Returns list of dicts with id, print_name, created_at, match_type.
"""
duplicates = []
# First, find exact matches by content hash
if content_hash:
result = await self.db.execute(
select(PrintArchive)
.where(
and_(
PrintArchive.content_hash == content_hash,
PrintArchive.id != archive_id,
PrintArchive.deleted_at.is_(None),
)
)
.order_by(PrintArchive.created_at.desc())
.limit(10)
)
for archive in result.scalars().all():
duplicates.append(
{
"id": archive.id,
"print_name": archive.print_name,
"created_at": archive.created_at,
"match_type": "exact",
}
)
# Then, find similar matches by print name or MakerWorld ID
# Prefer strict name+hash matching when hash exists; fallback to name-only for legacy/manual
# archives that may not have a content_hash.
if print_name or makerworld_model_id:
conditions = [PrintArchive.id != archive_id, PrintArchive.deleted_at.is_(None)]
name_conditions = []
if print_name:
if content_hash:
# Match if print names are similar AND have the same hash (same file)
name_conditions.append(
and_(PrintArchive.print_name.ilike(print_name), PrintArchive.content_hash == content_hash)
)
else:
# Fallback for archives without hash data: match by print name only.
name_conditions.append(PrintArchive.print_name.ilike(print_name))
if makerworld_model_id:
# Match by MakerWorld model ID stored in extra_data
from backend.app.core.db_dialect import is_sqlite
if is_sqlite():
from sqlalchemy import func
name_conditions.append(
func.json_extract(PrintArchive.extra_data, "$.makerworld_model_id") == str(makerworld_model_id)
)
else:
name_conditions.append(
text("(extra_data::jsonb->>'makerworld_model_id') = :mw_id").bindparams(
mw_id=str(makerworld_model_id)
)
)
if name_conditions:
conditions.append(or_(*name_conditions))
result = await self.db.execute(
select(PrintArchive).where(and_(*conditions)).order_by(PrintArchive.created_at.desc()).limit(10)
)
for archive in result.scalars().all():
# Don't add if already in duplicates (exact match)
if not any(d["id"] == archive.id for d in duplicates):
duplicates.append(
{
"id": archive.id,
"print_name": archive.print_name,
"created_at": archive.created_at,
"match_type": "similar",
}
)
return duplicates
async def archive_print(
self,
printer_id: int | None,
source_file: Path,
print_data: dict | None = None,
created_by_id: int | None = None,
original_filename: str | None = None,
project_id: int | None = None,
cost_center_id: int | None = None,
subtask_id: str | None = None,
prefer_filename_for_name: bool = False,
plate_id: int | None = None,
library_file_id: int | None = None,
slicer_ams_mapping: list[int] | None = None,
slicer_ams_mapping_printer_id: int | None = None,
update_archive_id: int | None = None,
) -> PrintArchive | None:
"""Archive a 3MF file with metadata.
Args:
printer_id: ID of the printer (optional)
source_file: Path to the 3MF file
print_data: Print data from MQTT (optional)
created_by_id: User ID who created this archive (optional, for user tracking)
original_filename: Original human-readable filename (optional, for library files
stored with UUID names)
project_id: Project to associate this archive with (optional, set when triggered
from the project view)
library_file_id: Library file this run was dispatched from (optional,
set by the queue scheduler — powers per-file project progress, #1897)
subtask_id: MQTT-provided task identifier (optional). Used to match an
existing archive across a backend restart mid-print so the
original row can be resumed instead of cancelled (#972).
prefer_filename_for_name: When True, use the uploaded filename stem as the
archive's display name even if the 3MF embeds a `print_name` in its
metadata. Used by virtual-printer flows so users who rename a job in
BambuStudio's "send to printer" dialog see that name instead of the
creator-baked title (#1152).
slicer_ams_mapping: The slicer's own live-resolved AMS-slot pick, to persist
onto `extra_data.slicer_ams_mapping` for a later reprint to reuse. Deliberately
a distinct parameter, not read off `print_data["ams_mapping"]` — that key is
populated on every MQTT print-start callback regardless of source (bambu_mqtt's
request-topic interception captures it for slicer-direct LAN prints too), so
promoting it unconditionally would stamp every archive on installs with no
virtual printer at all. Callers that gate this behind an opt-in (the VP-queue
"Save AMS mapping" toggle) pass it explicitly; everyone else leaves it unset.
slicer_ams_mapping_printer_id: The printer `slicer_ams_mapping`'s tray IDs were
resolved against. Required alongside `slicer_ams_mapping` — a global tray ID
only means something relative to one printer's specific AMS layout, so a
mapping saved without knowing which printer it came from can't be safely
reused later on any printer, including the same one (there'd be no way to
tell). A model-based VP with no fixed target printer has no valid value to
pass here and must leave both params unset.
update_archive_id: Fill in an existing archive row instead of adding one.
Used to upgrade a no-3MF fallback archive once the file finally arrives
(#2957). Everything above the row itself — the copy, the parse, the
thumbnail, the cost — is exactly what a fresh archive does; only the
destination differs. The row must keep its id: the energy-start reading,
the timelapse session, ``_active_prints``, the start notification and any
queue link were all written against it while the print was running, and a
second row would orphan every one of them. Fields the fallback path
already established from MQTT (``started_at``, ``subtask_id``,
``created_by_id``, ``project_id``) are left alone; the 3MF has nothing
better to say about them.
"""
# Verify printer exists if specified
if printer_id is not None:
result = await self.db.execute(select(Printer).where(Printer.id == printer_id))
printer = result.scalar_one_or_none()
if not printer:
return None
# Create archive directory structure
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
display_stem = resolve_display_stem(original_filename if original_filename else source_file.name)
archive_name = f"{timestamp}_{display_stem}"
# Use "unassigned" folder for archives without a printer
printer_folder = str(printer_id) if printer_id is not None else "unassigned"
archive_dir = (
settings.archive_dir / printer_folder / archive_name
) # SEC-PATH-OK: printer_folder = str(int|None) → digits or "unassigned"; archive_name = f"{timestamp}_{display_stem}" where resolve_display_stem strips path components via Path(filename).name
archive_dir.mkdir(parents=True, exist_ok=True)
# Copy 3MF file with an explicit fsync'd loop (avoids a sendfile
# short-read quirk that silently truncated 3MF archives on some
# platforms — see _copy_and_fsync and #1032).
dest_file = archive_dir / source_file.name
_copy_and_fsync(source_file, dest_file)
# If we just archived a 3MF, verify the dest is a valid ZIP before
# going any further. Staying quiet here is how #1032 escaped review —
# the archive row was written but every later zipfile.ZipFile() call
# on the dest failed with "File is not a zip file".
if (
source_file.suffix.lower() == ".3mf"
and zipfile.is_zipfile(source_file)
and not zipfile.is_zipfile(dest_file)
):
try:
src_size = source_file.stat().st_size
dst_size = dest_file.stat().st_size
except OSError:
src_size = dst_size = -1
logger.error(
"Archive copy corrupted 3MF: src=%s (%s bytes, valid ZIP) -> dst=%s (%s bytes, NOT a ZIP). Refusing to create archive row.",
source_file,
src_size,
dest_file,
dst_size,
)
# Narrow cleanup: remove only the truncated file and the archive
# directory if it's now empty. archive_dir was created with
# exist_ok=True so it could in theory pre-date this call (e.g.
# same-second same-filename collision); rmtree would be too broad.
try:
dest_file.unlink()
except OSError:
pass
try:
archive_dir.rmdir()
except OSError:
pass # directory not empty — leave untouched
return None
# Compute content hash for duplicate detection
content_hash = self.compute_file_hash(dest_file)
# Extract plate number from filename (e.g., "plate_5" from "/data/Metadata/plate_5.gcode")
plate_number = None
if print_data:
filename = print_data.get("filename", "")
match = re.search(r"plate_(\d+)", filename)
if match:
plate_number = int(match.group(1))
# Parse 3MF metadata
parser = ThreeMFParser(dest_file, plate_number=plate_number)
metadata = parser.parse()
# Save thumbnail if present
thumbnail_path = None
if "_thumbnail_data" in metadata:
thumb_file = archive_dir / f"thumbnail{metadata['_thumbnail_ext']}"
thumb_file.write_bytes(metadata["_thumbnail_data"])
thumbnail_path = str(thumb_file.relative_to(settings.base_dir))
del metadata["_thumbnail_data"]
del metadata["_thumbnail_ext"]
# Merge with print data from MQTT
if print_data:
metadata["_print_data"] = print_data
# Promote the slicer's own live-resolved AMS-slot pick, when the caller
# explicitly opted in (see the `slicer_ams_mapping` param docstring for
# why this is NOT read off `print_data["ams_mapping"]`), to a stable
# top-level extra_data key. Lets a later reprint reuse the exact tray
# the user picked/BambuStudio auto-matched at slice time instead of the
# scheduler re-deriving one from just the file's static type/color,
# which can land on the wrong physical spool when that match isn't
# unique. Top-level (not nested under the `_print_data` diagnostic bag)
# so API consumers have a single stable path:
# `archive.extra_data.slicer_ams_mapping`. Stored together with the
# printer it was resolved against — see `slicer_ams_mapping_printer_id`
# param docstring — so a later reprint can tell whether it's even
# applicable before trying to reuse it.
if slicer_ams_mapping and slicer_ams_mapping_printer_id is not None:
metadata["slicer_ams_mapping"] = {
"mapping": slicer_ams_mapping,
"printer_id": slicer_ams_mapping_printer_id,
}
# Determine status and timestamps
status = print_data.get("status", "completed") if print_data else "archived"
started_at = datetime.now(timezone.utc) if status == "printing" else None
completed_at = datetime.now(timezone.utc) if status in ("completed", "failed", "archived") else None
# Calculate cost based on filament usage and type
cost = None
filament_grams = metadata.get("filament_used_grams")
filament_type = metadata.get("filament_type")
if filament_grams and filament_type:
# For multi-material prints, use the first filament type for cost calculation
primary_type = filament_type.split(",")[0].strip()
# Look up filament cost_per_kg from database
filament_result = await self.db.execute(select(Filament).where(Filament.type == primary_type).limit(1))
filament = filament_result.scalar_one_or_none()
if filament:
cost = round((filament_grams / 1000) * filament.cost_per_kg, 2)
else:
# Use default filament cost from settings
from backend.app.api.routes.settings import get_setting
default_cost_setting = await get_setting(self.db, "default_filament_cost")
default_cost_per_kg = float(default_cost_setting) if default_cost_setting else 25.0
cost = round((filament_grams / 1000) * default_cost_per_kg, 2)
# Calculate quantity from printable objects count
# printable_objects is a dict of {identify_id: name} for non-skipped objects
quantity = 1 # Default to 1
printable_objects = metadata.get("printable_objects")
if printable_objects and isinstance(printable_objects, dict):
quantity = len(printable_objects)
logger.debug("Auto-detected %s parts from 3MF printable objects", quantity)
# Recovery of an existing fallback row: assign the freshly-parsed values
# onto it rather than adding a second archive for the same print (#2957).
if update_archive_id is not None:
existing = await self.db.get(PrintArchive, update_archive_id)
if existing is None:
logger.warning("archive_print: archive %s to update no longer exists", update_archive_id)
return None
# `metadata` is freshly parsed from the 3MF, so assigning it drops
# the row's `no_3mf_available` / `no_3mf_reason` markers as a side
# effect — which is correct, the archive is no longer a fallback,
# and it is what stops the Archives banner counting it.
# `_print_data` is diagnostic history rather than something the 3MF
# knows about: keep the row's copy for a caller that passed no
# print_data of its own.
merged = dict(metadata)
preserved = (existing.extra_data or {}).get("_print_data")
if preserved is not None and "_print_data" not in merged:
merged["_print_data"] = preserved
# A record that this row started life without a 3MF, which the
# dropped markers no longer say.
merged["recovered_no_3mf"] = True
existing.filename = original_filename or source_file.name
existing.file_path = str(dest_file.relative_to(settings.base_dir))
existing.file_size = dest_file.stat().st_size
existing.content_hash = content_hash
existing.thumbnail_path = thumbnail_path
existing.print_name = (
clean_display_name(display_stem)
if prefer_filename_for_name
else (clean_display_name(metadata.get("print_name")) or clean_display_name(display_stem))
)
# Only overwrite what the 3MF actually knows. A fallback archive
# recovered mid-print has a real print_time_seconds from MQTT and a
# filament type/colour from the AMS; a 3MF that omits a field must
# not blank them back out.
for field in (
"print_time_seconds",
"filament_used_grams",
"filament_type",
"filament_color",
"layer_height",
"total_layers",
"nozzle_diameter",
"bed_temperature",
"bed_type",
"nozzle_temperature",
"sliced_for_model",
"makerworld_url",
"designer",
):
value = metadata.get(field)
if value is not None:
setattr(existing, field, value)
if cost is not None:
existing.cost = cost
existing.quantity = quantity
existing.extra_data = merged
if plate_id is not None:
existing.plate_id = plate_id
if library_file_id is not None:
existing.library_file_id = library_file_id
await self.db.commit()
await self.db.refresh(existing)
return existing
# Create archive record
archive = PrintArchive(
printer_id=printer_id,
filename=original_filename or source_file.name,
file_path=str(dest_file.relative_to(settings.base_dir)),
file_size=dest_file.stat().st_size,
content_hash=content_hash,
thumbnail_path=thumbnail_path,
# clean_display_name because the 3MF's own metadata reaches this
# verbatim, and a control character in it renders nowhere and
# truncates somewhere (#2832). The schema does the same for names
# arriving over the API. Cleaned before the fallback rather than
# after it, so an embedded name that is only whitespace still falls
# through to the filename instead of leaving the archive nameless.
print_name=(
clean_display_name(display_stem)
if prefer_filename_for_name
else (clean_display_name(metadata.get("print_name")) or clean_display_name(display_stem))
),
print_time_seconds=metadata.get("print_time_seconds"),
filament_used_grams=metadata.get("filament_used_grams"),
filament_type=metadata.get("filament_type"),
filament_color=metadata.get("filament_color"),
layer_height=metadata.get("layer_height"),
total_layers=metadata.get("total_layers"),
nozzle_diameter=metadata.get("nozzle_diameter"),
bed_temperature=metadata.get("bed_temperature"),
bed_type=metadata.get("bed_type"),
nozzle_temperature=metadata.get("nozzle_temperature"),
sliced_for_model=metadata.get("sliced_for_model"),
makerworld_url=metadata.get("makerworld_url"),
designer=metadata.get("designer"),
status=status,
started_at=started_at,
completed_at=completed_at,
cost=cost,
quantity=quantity,
extra_data=metadata,
created_by_id=created_by_id,
project_id=project_id,
library_file_id=library_file_id,
cost_center_id=cost_center_id,
subtask_id=subtask_id,
plate_id=plate_id,
)
self.db.add(archive)
await self.db.commit()
await self.db.refresh(archive)
return archive
async def get_archive(self, archive_id: int) -> PrintArchive | None:
"""Get an archive by ID with relationships loaded."""
from sqlalchemy.orm import selectinload
result = await self.db.execute(
select(PrintArchive)
.options(selectinload(PrintArchive.created_by), selectinload(PrintArchive.project))
.where(PrintArchive.id == archive_id)
)
return result.scalar_one_or_none()
async def update_archive_status(
self,
archive_id: int,
status: str,
completed_at: datetime | None = None,
failure_reason: str | None = None,
) -> bool:
"""Update the status of an archive."""
archive = await self.get_archive(archive_id)
if not archive:
return False
archive.status = status
if completed_at:
archive.completed_at = completed_at
if failure_reason:
archive.failure_reason = failure_reason
await self.db.commit()
return True
async def list_archives(
self,
printer_id: int | None = None,
project_id: int | None = None,
date_from: date | None = None,
date_to: date | None = None,
limit: int = 50,
offset: int = 0,
visible_to_user_id: int | None = None,
) -> list[PrintArchive]:
"""List archives with optional filtering.
``visible_to_user_id`` scopes results to archives that user owns. Used
when the caller has ARCHIVES_READ_OWN but not ARCHIVES_READ_ALL — pass
``None`` to skip the filter (caller has read-all or auth is disabled).
"""
from sqlalchemy.orm import selectinload
query = (
select(PrintArchive)
.options(selectinload(PrintArchive.project), selectinload(PrintArchive.created_by))
# Hide soft-deleted rows from the listings (#1343). The stats
# endpoint deliberately does NOT add this filter so deleted
# archives keep contributing to Quick Stats.
.where(PrintArchive.deleted_at.is_(None))
.order_by(PrintArchive.created_at.desc())
)
if printer_id:
query = query.where(PrintArchive.printer_id == printer_id)
if project_id:
query = query.where(PrintArchive.project_id == project_id)
if date_from:
dt_from = datetime.combine(date_from, time.min, tzinfo=timezone.utc)
query = query.where(PrintArchive.created_at >= dt_from)
if date_to:
dt_to = datetime.combine(date_to, time.max, tzinfo=timezone.utc)
query = query.where(PrintArchive.created_at <= dt_to)
if visible_to_user_id is not None:
query = query.where(PrintArchive.created_by_id == visible_to_user_id)
query = query.limit(limit).offset(offset)
result = await self.db.execute(query)
return list(result.scalars().all())
async def soft_delete_archive(self, archive_id: int) -> bool:
"""Soft-delete an archive (#1343).
Removes the archive's files from disk (it disappears from the listings
and frees the storage) but flips the row's ``deleted_at`` so the stats
endpoint keeps counting its filament / energy / time / cost. The user
can opt into a hard delete via the "Also remove from statistics"
checkbox in the delete dialog — that path calls ``delete_archive``
instead and removes the row entirely.
"""
archive = await self.get_archive(archive_id)
if not archive:
return False
if archive.deleted_at is not None:
# Already soft-deleted; nothing to do. The files were purged on
# the first soft-delete pass so there is nothing left on disk.
return True
dirs_to_delete = self._resolve_archive_dirs_for_delete(archive)
recorded_paths = (archive.timelapse_path, archive.thumbnail_path)
await _null_print_log_thumbnail_paths(self.db, archive_id)
await _delete_related_queue_items(self.db, archive_id)
archive.deleted_at = datetime.now(timezone.utc)
await self.db.commit()
for directory in dirs_to_delete:
shutil.rmtree(directory, ignore_errors=True)
self._purge_id_named_dir(archive_id, recorded_paths)
return True
def _resolve_archive_dirs_for_delete(self, archive: PrintArchive) -> list[Path]:
"""Directories belonging to *archive* alone, safe to remove whole.
Shared by soft-delete and hard-delete so the two cannot drift apart
again — the previous helper said it was extracted for that reason, and
``delete_archive`` was still doing its own copy of the same rules.
An archive with a 3MF owns the directory its ``file_path`` sits in,
``//_/``. Any archive may also
own ``archive/no_source//``, where a source 3MF uploaded onto a
no-3MF archive lands (#1531); that one was never removed, so deleting
such an archive freed the row and left the upload behind.
Two directories are deliberately absent. ``/photos`` is the
legacy location *every* no-3MF archive wrote into at once, so removing
it on one delete would take the others' photos with it. And
``/`` — the directory :func:`resolve_archive_dir` gives
an archive with no ``file_path`` — is handled by
:meth:`_purge_id_named_dir` instead, for the reason given there.
"""
candidates: list[Path] = []
# Only when there is a path to derive it from. Without one,
# ``resolve_archive_dir`` returns the id-named directory, which must not
# be removed wholesale -- see _purge_id_named_dir.
if archive.file_path and archive.file_path.strip():
candidates.append(resolve_archive_dir(archive))
candidates.append(settings.archive_dir / "no_source" / str(archive.id))
resolved: list[Path] = []
for candidate in candidates:
if candidate in resolved or not candidate.is_dir():
continue
try:
relative_path = candidate.resolve().relative_to(settings.archive_dir.resolve())
except ValueError:
# A genuine guard trip, unlike the empty ``file_path`` this used
# to shout about: the row points somewhere outside the archive
# tree, which only a corrupted import or hand-edited SQL can do.
logger.error(
f"SECURITY: Refusing to delete archive {archive.id} - "
f"path {candidate} is outside archive directory {settings.archive_dir}"
)
continue
# Two deep, not one. An archive directory has been
# ``//_/`` since the first
# commit, so nothing legitimate sits one level down -- but the
# per-printer folder does, and it holds every print that printer
# ever made. Under the old ``< 1`` a row whose file_path had lost a
# path component took the whole folder with it.
if len(relative_path.parts) < 2:
logger.error(
f"SECURITY: Refusing to delete archive {archive.id} - "
f"path {candidate} is not deep enough inside archive directory"
)
continue
resolved.append(candidate)
return resolved
def _purge_id_named_dir(self, archive_id: int, recorded_paths: tuple[str | None, ...]) -> None:
"""Remove one archive's own files from ``/``, carefully.
Takes the recorded paths rather than the row because ``delete_archive``
removes the row before it touches the disk, deliberately: a failed
commit must leave the files alone. Reading ``archive.timelapse_path``
off a deleted instance afterwards would raise or silently refresh.
That directory is where an archive with no 3MF keeps its timelapse and
its finish photos (:func:`resolve_archive_dir`). It is emphatically NOT
an ``rmtree`` target, because it shares a namespace with the per-printer
folders: a normal archive lives at
``//_/``, so ``archive/1`` is
printer 1's folder *and* the directory the helper hands archive id 1.
Archive ids and printer ids are both small integers from unrelated
sequences, so on any install the first few archives collide with the
printers. Removing the directory would take every print that printer
ever made — measured on a scratch tree before this guard existed.
So nothing is removed that has not been identified as this archive's.
``photos`` is a fixed subdirectory name and an archive directory is
always ``_``, so the two cannot be confused; the video
is removed by the path the row itself records. The directory then goes
only if that left it empty, which a printer folder holding prints never
will. Anything unrecognised keeps it alive and is leaked rather than
guessed at — the safe direction for a recursive delete.
"""
directory = settings.archive_dir / str(archive_id)
if not directory.is_dir():
return
try:
relative_path = directory.resolve().relative_to(settings.archive_dir.resolve())
except ValueError:
return
if len(relative_path.parts) != 1:
return
shutil.rmtree(directory / "photos", ignore_errors=True) # SEC-PATH-OK: constant subdirectory
for recorded in recorded_paths:
if not recorded:
continue
try:
# Two checks, not one. safe_join_under rejects the absolute and
# ``..`` shapes and proves the result is inside the data
# directory; assert_under then narrows it to *this* archive's
# own directory, because a row whose timelapse_path names
# another archive's file must not take it with this delete.
# The column is written by Bambuddy from a filename the printer
# supplied over FTP, so it is not a trusted constant.
candidate = safe_join_under(settings.base_dir, recorded, http=False)
assert_under(directory, candidate, http=False)
except PathTraversalError:
continue
if candidate.is_file():
candidate.unlink(missing_ok=True)
try:
directory.rmdir()
except OSError:
# Not empty (a printer folder, or a file this archive did not
# record) or already gone. Both are fine: the point of rmdir over
# rmtree is that it cannot take anything with it.
pass
async def delete_archive(self, archive_id: int) -> bool:
"""Delete an archive and its files."""
archive = await self.get_archive(archive_id)
if not archive:
return False
# Resolved BEFORE committing the DB change, since the row is what says
# where the files are. Shared with soft-delete rather than repeated
# here: this was a second copy of the same checks and it had already
# diverged from the one it was extracted from.
dirs_to_delete = self._resolve_archive_dirs_for_delete(archive)
recorded_paths = (archive.timelapse_path, archive.thumbnail_path)
# NULL stale thumbnail_path on linked PrintLogEntries before the FK
# SET-NULL cascade fires. The on-disk file is about to be removed by
# the rmtree below, so the path on any surviving log entry (archive_id
# gets SET NULL by the FK) would otherwise point at a missing file
# and produce 404 storms in the print-log view (#1348-followup).
await _null_print_log_thumbnail_paths(self.db, archive_id)
# Delete database record FIRST — if the commit fails (e.g. database locked
# during concurrent bulk deletes), the files stay on disk and nothing is lost.
await self.db.delete(archive)
await self.db.commit()
# Only delete files AFTER the DB commit succeeds to avoid orphaned records
for directory in dirs_to_delete:
shutil.rmtree(directory, ignore_errors=True)
self._purge_id_named_dir(archive_id, recorded_paths)
return True
async def attach_timelapse(
self,
archive_id: int,
timelapse_data: bytes,
filename: str = "timelapse.mp4",
) -> bool:
"""Attach a timelapse video to an archive.
Non-MP4 videos (e.g. AVI from P1S) are saved as-is and a background
task converts them to MP4 for browser compatibility.
"""
import asyncio
archive = await self.get_archive(archive_id)
if not archive:
return False
# Where this archive's files live. Deliberately the shared helper: an
# archive created without a 3MF has ``file_path == ""``, and deriving the
# directory here as ``(base_dir / "").parent`` resolved to the parent of
# base_dir — outside the data directory entirely. In Docker that is /app,
# so the write failed with EACCES and the timelapse was retried and
# discarded 25 times; where the parent happens to be writable it
# succeeded, dropped a stray video next to the install, and then failed
# anyway on the relative_to() below. Every H2-series and P2S print sent
# from the slicer takes that path, because the file goes to internal
# storage and no 3MF can be fetched.
archive_dir = resolve_archive_dir(archive)
# Save timelapse - use thread pool to avoid blocking event loop
# (timelapse files can be 100MB+, sync write blocks for seconds).
# `filename` ultimately comes from a printer's FTP listing (compromised-
# printer threat model) or a query param on /archives/{id}/timelapse/select;
# the safe-join helper rejects ``..`` segments and absolute paths so a
# crafted name can't escape the archive directory. Use http=False so a
# service-layer reject surfaces as a return False (matching the existing
# not-found contract) rather than a 400 raised from inside a background
# task.
try:
timelapse_file = safe_join_under(archive_dir, filename, http=False)
except PathTraversalError:
logger.warning(
"Refusing to attach timelapse with unsafe filename %r to archive %s",
filename,
archive_id,
)
return False
# Created only once the name has been vetted, so a rejected filename
# leaves nothing behind. A no-3MF archive has never had a directory of
# its own, and the timelapse can be the first thing to want one.
await asyncio.to_thread(lambda: timelapse_file.parent.mkdir(parents=True, exist_ok=True))
await asyncio.to_thread(timelapse_file.write_bytes, timelapse_data)
# Update archive record
archive.timelapse_path = str(timelapse_file.relative_to(settings.base_dir))
await self.db.commit()
# For non-MP4 videos (e.g. AVI from P1S), kick off background conversion
if not filename.lower().endswith(".mp4"):
spawn_background_task(
_convert_timelapse_to_mp4(archive_id, timelapse_file),
name=f"timelapse-convert-{archive_id}",
)
return True
async def _convert_timelapse_to_mp4(archive_id: int, source_path: Path) -> None:
"""Background task: convert non-MP4 timelapse (e.g. AVI from P1S) to MP4.
Runs with low CPU priority (-threads 1, nice) so it doesn't starve
other processes on resource-constrained devices like Raspberry Pi.
"""
import asyncio
from backend.app.core.database import async_session
from backend.app.services.camera import get_ffmpeg_path
logger = logging.getLogger(__name__)
ffmpeg = get_ffmpeg_path()
if not ffmpeg:
logger.info(
"FFmpeg not available, skipping timelapse conversion for archive %s (file saved as %s)",
archive_id,
source_path.suffix,
)
return
mp4_path = source_path.with_suffix(".mp4")
try:
cmd = [
ffmpeg,
"-y",
"-i",
str(source_path),
"-c:v",
"libx264",
"-preset",
"fast",
"-crf",
"23",
"-threads",
"1",
"-movflags",
"+faststart",
str(mp4_path),
]
# Try with nice for lower CPU priority (standard on Linux/macOS)
try:
process = await asyncio.create_subprocess_exec(
"nice",
"-n",
"19",
*cmd,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
)
except FileNotFoundError:
# nice not available (e.g. Windows), run without
process = await asyncio.create_subprocess_exec(
*cmd,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
)
_, stderr = await process.communicate()
if process.returncode != 0:
logger.warning(
"Timelapse conversion failed for archive %s: %s",
archive_id,
summarize_ffmpeg_stderr(stderr) or NO_FFMPEG_OUTPUT,
)
if mp4_path.exists():
mp4_path.unlink()
return
# Update DB path to the new MP4 file
async with async_session() as db:
from backend.app.models.archive import PrintArchive
result = await db.execute(select(PrintArchive).where(PrintArchive.id == archive_id))
archive = result.scalar_one_or_none()
if archive:
archive.timelapse_path = str(mp4_path.relative_to(settings.base_dir))
await db.commit()
# Remove original non-MP4 file
if source_path.exists():
source_path.unlink()
logger.info(
"Converted timelapse to MP4 for archive %s (%s → %s)",
archive_id,
source_path.name,
mp4_path.name,
)
except Exception as e:
logger.warning("Timelapse conversion error for archive %s: %s", archive_id, e)
if mp4_path.exists():
mp4_path.unlink()