bambuddy/backend/app/services/archive.py

import hashlib
import html
import json
import logging
import os
import re
import shutil
import zipfile
from collections.abc import Sequence
from datetime import date, datetime, time, timezone
from pathlib import Path
from typing import Any

from defusedxml import ElementTree as ET
from sqlalchemy import and_, or_, select, text
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncSession

from backend.app.core.config import settings
from backend.app.core.printer_scope import PrinterScope
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`` ``<plate>`` element, or None.

    Bambu Studio and OrcaSlicer record it as a ``<metadata key="index"
    value="N"/>`` child — there is no ``plate_idx`` attribute on ``<plate>``
    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 ``<plate>`` 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 ``"<Project> - Plate <N>"`` (and
    a lowercase ``"_plate_<N>"`` 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}"


_MESH_OPEN = b"<mesh"
_MESH_CLOSE = b"</mesh>"
_MODEL_READ_CHUNK = 1024 * 1024


def read_3dmodel_without_meshes(zf: zipfile.ZipFile, model_path: str = "3D/3dmodel.model") -> str:
    """Return the model XML with every ``<mesh>...</mesh>`` body left out.

    The metadata scans only want ``<metadata>`` elements and the URLs inside
    them, but a plain 3MF keeps its geometry inline in this same entry: a
    combined plate's is hundreds of MB of vertex and triangle rows (#3162).
    Streaming past the meshes keeps memory to the non-geometry part. The
    entry can't simply be cut at ``<resources>``: BambuStudio writes its
    MakerWorld metadata after ``</build>``. Mesh bodies hold only numbers, and
    a ``<`` inside a metadata value is escaped, so nothing a scan looks for is
    dropped.
    """
    out = bytearray()
    pending = b""
    in_mesh = False
    with zf.open(model_path) as fh:
        while chunk := fh.read(_MODEL_READ_CHUNK):
            data = pending + chunk
            pos = 0
            while True:
                if in_mesh:
                    end = data.find(_MESH_CLOSE, pos)
                    if end == -1:
                        # Keep enough of the tail to match a tag split across chunks.
                        pending = data[max(pos, len(data) - len(_MESH_CLOSE) + 1) :]
                        break
                    pos = end + len(_MESH_CLOSE)
                    in_mesh = False
                else:
                    start = data.find(_MESH_OPEN, pos)
                    if start == -1:
                        keep = max(pos, len(data) - len(_MESH_OPEN) + 1)
                        out += data[pos:keep]
                        pending = data[keep:]
                        break
                    out += data[pos:start]
                    pos = start + len(_MESH_OPEN)
                    in_mesh = True
    if not in_mesh:
        out += pending
    return out.decode("utf-8", errors="ignore")


# 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


async def estimate_filament_cost(db: AsyncSession, grams: float | None, filament_type: str | None) -> float | None:
    """An archive's filament cost from its grams and type: the first type's
    cost per kg on record, else the default filament cost (25 when unset).
    ``None`` without grams or a type."""
    if not grams or not filament_type:
        return None
    # For multi-material prints, the first filament type prices the whole print
    primary_type = filament_type.split(",")[0].strip()
    filament = (await db.execute(select(Filament).where(Filament.type == primary_type).limit(1))).scalar_one_or_none()
    if filament:
        return round((grams / 1000) * filament.cost_per_kg, 2)
    from backend.app.api.routes.settings import get_setting

    default_cost_setting = await get_setting(db, "default_filament_cost")
    default_cost_per_kg = float(default_cost_setting) if default_cost_setting else 25.0
    return round((grams / 1000) * default_cost_per_kg, 2)


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: <plate><metadata key="printer_model_id" value="C11" /></plate>
                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 <plate> 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 <filament id="1" type="PLA" color="#FFFFFF" used_g="100" />
                # 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 = read_3dmodel_without_meshes(zf, model_path)

            # Parse XML metadata elements
            # MakerWorld adds metadata like: <metadata name="Designer">username</metadata>
            metadata_pattern = r'<metadata\s+name="([^"]+)"[^>]*>([^<]*)</metadata>'
            matches = re.findall(metadata_pattern, content)

            # 3MF metadata values are XML-encoded — `&` becomes `&amp;`, etc.
            # ProjectPageParser learned this the hard way: BambuStudio sometimes
            # writes triple-encoded payloads (`&amp;amp;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 &amp; Bar" and React then
            # double-escapes it on render to "Foo &amp;amp; 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 = read_3dmodel_without_meshes(zf, model_path)

                    # Extract metadata elements using regex
                    # Format: <metadata name="Key">Value</metadata> or <metadata name="Key" />
                    metadata_pattern = r'<metadata\s+name="([^"]+)"[^>]*>([^<]*)</metadata>'
                    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'(<metadata\s+name="{xml_name}"[^>]*>)[^<]*(</metadata>)'
                        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, visible: Sequence[Any] = ()
    ) -> tuple[set[str], set[tuple[str, str]]]:
        """Get all content hashes and (print name, hash) pairs that appear more than once.

        ``visible`` limits the count to archives the caller may see.

        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), *visible)
            .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),
                *visible,
            )
            .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,
        visible: Sequence[Any] = (),
    ) -> list[dict]:
        """Find duplicate archives based on hash or name matching.

        ``visible`` holds WHERE clauses limiting matches to archives the caller may see.

        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),
                        *visible,
                    )
                )
                .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), *visible]

            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

    @staticmethod
    def _has_sliced_plate(file_path: Path, plate_number: int) -> bool:
        try:
            with zipfile.ZipFile(file_path) as zf:
                return f"Metadata/plate_{plate_number}.gcode" in zf.namelist()
        except (OSError, zipfile.BadZipFile):
            return False

    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
        # Removed again if the row is not written (below); only a folder this
        # call created, never another archive's of the same name.
        created_archive_dir = not archive_dir.exists()
        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))
        if plate_id is None:
            # A print Bambuddy did not dispatch only names its plate in the
            # gcode path. Kept as the archive's plate when the file really has
            # it: some firmware reports the .3mf name there instead, and a user
            # name like "x_plate_2" must not label a single-plate file.
            if plate_number is not None and self._has_sliced_plate(dest_file, plate_number):
                plate_id = plate_number
        else:
            # The dispatcher knows the plate; the path may name none, or not
            # the plate sent (#1166). Without it the parser describes another.
            plate_number = plate_id

        # 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

        cost = await estimate_filament_cost(self.db, metadata.get("filament_used_grams"), metadata.get("filament_type"))

        # 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)
        try:
            await self.db.commit()
        except IntegrityError:
            # Refused, so the row was not written (a printer deleted meanwhile,
            # by its foreign key): the folder its files went into would be
            # orphaned. Not on other errors: a connection lost during COMMIT may
            # still have written the row, whose files these are.
            if created_archive_dir:
                shutil.rmtree(archive_dir, ignore_errors=True)
            raise
        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,
        printer_scope: PrinterScope | 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)

        # Only archives from printers the caller may see (#1727)
        if printer_scope is not None and (clause := printer_scope.where(PrintArchive.printer_id)) is not None:
            query = query.where(clause)

        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,
        ``<archive_dir>/<printer_id>/<timestamp>_<name>/``. Any archive may also
        own ``archive/no_source/<id>/``, 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. ``<base_dir>/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
        ``<archive_dir>/<id>`` — 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
            # ``<archive_dir>/<printer_id>/<timestamp>_<name>/`` 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 ``<archive_dir>/<id>``, 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
        ``<archive_dir>/<printer_id>/<timestamp>_<name>/``, 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 ``<timestamp>_<name>``, 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",
        *,
        plate_id: int | None = None,
    ) -> 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.

        ``plate_id`` is the plate whose run produced the video, when the caller
        knows it; print start uses it to tell a reprint from the next plate of
        a shared archive (#3275).
        """
        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))

        # A file of this name that the archive no longer points at is an earlier
        # run's video, kept because another plate of a shared archive started
        # (#3275). Writing over it would destroy the copy that was kept: the
        # external-camera stitch always names its output layer_timelapse.mp4.
        # Re-attaching the file the archive does point at still replaces it.
        timelapse_file = await asyncio.to_thread(
            _free_timelapse_name, timelapse_file, archive.timelapse_path, settings.base_dir
        )
        await asyncio.to_thread(timelapse_file.write_bytes, timelapse_data)

        # Update archive record
        archive.timelapse_path = str(timelapse_file.relative_to(settings.base_dir))
        archive.timelapse_plate_id = plate_id
        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


def _free_timelapse_name(target: Path, current_relpath: str | None, base_dir: Path) -> Path:
    """``target``, or ``<stem>_<n><suffix>`` beside it when ``target`` would
    replace a video the archive does not point at (see ``attach_timelapse``).

    A non-MP4 is converted to ``<stem>.mp4`` afterwards, so that name counts as
    taken too.
    """

    def outputs(path: Path) -> set[Path]:
        return {path.resolve(), path.with_suffix(".mp4").resolve()}

    # Only compared against ``target``, which attach_timelapse has already
    # safe-joined; nothing is read or written through this path.
    current = (base_dir / current_relpath).resolve() if current_relpath else None  # SEC-PATH-OK: comparison only
    taken = {p for p in outputs(target) if p.exists()}
    if not taken or current in taken:
        return target
    n = 2
    while True:
        candidate = target.with_name(f"{target.stem}_{n}{target.suffix}")
        if not any(p.exists() for p in outputs(candidate)):
            return candidate
        n += 1


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()
            source_rel = str(source_path.relative_to(settings.base_dir))
            # Only if the archive still shows this video: one deleted while it
            # converted, or replaced by a reprint's, must not come back.
            if archive is None or archive.timelapse_path != source_rel:
                logger.info(
                    "Archive %s no longer shows %s; discarding its MP4 conversion", archive_id, source_path.name
                )
                mp4_path.unlink(missing_ok=True)
                return
            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()