bambuddy/backend/app/services/archive.py

import hashlib
import html
import json
import logging
import os
import re
import shutil
import zipfile
from datetime import date, datetime, time, timezone
from pathlib import Path

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

from backend.app.core.config import settings
from backend.app.core.tasks import spawn_background_task
from backend.app.models.archive import PrintArchive
from backend.app.models.filament import Filament
from backend.app.models.printer import Printer
from backend.app.utils.filename import clean_display_name
from backend.app.utils.safe_path import PathTraversalError, safe_join_under

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 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).
    """
    try:
        with zipfile.ZipFile(file_path, "r") as zf:
            if "Metadata/slice_info.config" not in zf.namelist():
                return None
            content = zf.read("Metadata/slice_info.config").decode()
            root = ET.fromstring(content)
            plates = root.findall(".//plate")
            if len(plates) != 1:
                return None
            return _read_plate_index(plates[0])
    except Exception:
        return 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}"


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 _parse_gcode_header(self, zf: zipfile.ZipFile):
        """Parse G-code file header for total layer count and printer model."""
        try:
            # Look for plate_1.gcode or similar
            gcode_files = [f for f in zf.namelist() if f.endswith(".gcode")]
            if not gcode_files:
                return

            # Read first 4KB of G-code (header contains metadata)
            gcode_path = gcode_files[0]
            with zf.open(gcode_path) as f:
                header = f.read(4096).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))

            # 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 - first layer or regular
            for key in ["bed_temperature_initial_layer", "bed_temperature"]:
                if key in data:
                    val = data[key]
                    if isinstance(val, list) and val:
                        self.metadata["bed_temperature"] = int(float(val[0]))
                    elif isinstance(val, (int, float, str)):
                        self.metadata["bed_temperature"] = int(float(val))
                    break

            # 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 _extract_settings_from_content(self, content: str):
        """Extract print settings from config content."""
        settings_map = {
            "layer_height": ("layer_height", float),
            "nozzle_diameter": ("nozzle_diameter", float),
            "bed_temperature": ("bed_temperature", int),
            "nozzle_temperature": ("nozzle_temperature", int),
        }

        for key, (search_key, converter) in settings_map.items():
            if key not in self.metadata:
                try:
                    # Try JSON format
                    if f'"{search_key}"' in content:
                        start = content.find(f'"{search_key}"')
                        value_start = content.find(":", start) + 1
                        value_end = content.find(",", value_start)
                        if value_end == -1:
                            value_end = content.find("}", value_start)
                        value = content[value_start:value_end].strip().strip('"')
                        self.metadata[key] = converter(value)
                except (ValueError, TypeError):
                    pass  # Skip settings with unconvertible values

    def _parse_3dmodel(self, zf: zipfile.ZipFile):
        """Parse 3D/3dmodel.model for MakerWorld metadata."""
        try:
            model_path = "3D/3dmodel.model"
            if model_path not in zf.namelist():
                return

            content = zf.read(model_path).decode("utf-8", errors="ignore")

            # Parse XML metadata elements
            # MakerWorld adds metadata like: <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_3mf(
    data: bytes, plate_number: int | None = None, include_positions: bool = False
) -> dict[int, str] | dict[int, dict] | tuple[dict[int, dict], list | None]:
    """Extract printable objects from 3MF file bytes.

    This is a lightweight function used during print start to get the list
    of objects that can be skipped.

    Args:
        data: Raw bytes of the 3MF file
        plate_number: Which plate was printed (1-based), or None for first plate
        include_positions: If True, return tuple of (objects dict, bbox_all)

    Returns:
        If include_positions=False: Dictionary mapping identify_id (int) to object name (str)
        If include_positions=True: Tuple of (dict mapping identify_id to {name, x, y}, bbox_all list or None)
    """
    from io import BytesIO

    printable_objects: dict = {}
    bbox_all: list | None = None

    try:
        with zipfile.ZipFile(BytesIO(data), "r") as zf:
            if "Metadata/slice_info.config" not in zf.namelist():
                return printable_objects

            content = zf.read("Metadata/slice_info.config").decode()
            root = ET.fromstring(content)

            plates = root.findall(".//plate")
            if not plates:
                return printable_objects

            # Pick the plate that is actually printing. An all-plates export
            # lists every plate, so without this we offered the objects (and
            # the marker positions) of plate 1 whatever the printer was
            # running (#2522). Falling back to the first plate keeps the
            # single-plate export — the common case — working when the caller
            # has no plate to give us.
            plate = None
            if plate_number is not None:
                plate = next((p for p in plates if _read_plate_index(p) == plate_number), None)
            if plate is None:
                plate = plates[0]

            # Derive plate_idx from the plate we settled on, never from the
            # requested one: on a fallback they differ, and plate_idx also
            # selects the plate_N.json the positions come from.
            plate_idx = _read_plate_index(plate) or 1

            # Load position data from plate_N.json if we need positions
            # Build a lookup by name - use list to handle duplicate names
            bbox_by_name: dict[str, list[list]] = {}
            if include_positions:
                plate_json_path = f"Metadata/plate_{plate_idx}.json"
                if plate_json_path in zf.namelist():
                    try:
                        plate_json = json.loads(zf.read(plate_json_path).decode())
                        # Get bbox_all - the bounding box of all objects (used for image bounds)
                        bbox_all = plate_json.get("bbox_all")
                        for bbox_obj in plate_json.get("bbox_objects", []):
                            obj_name = bbox_obj.get("name")
                            bbox = bbox_obj.get("bbox", [])
                            if obj_name and len(bbox) >= 4:
                                if obj_name not in bbox_by_name:
                                    bbox_by_name[obj_name] = []
                                bbox_by_name[obj_name].append(bbox)
                    except (json.JSONDecodeError, KeyError):
                        pass  # Position data is optional; objects will lack x/y coordinates

            # Extract objects from slice_info.config
            for obj in plate.findall("object"):
                identify_id = obj.get("identify_id")
                name = obj.get("name")
                skipped = obj.get("skipped", "false")

                if identify_id and name and skipped.lower() != "true":
                    try:
                        obj_id = int(identify_id)
                        if include_positions:
                            x, y = None, None
                            # Match by name - pop first bbox to handle duplicates
                            bboxes = bbox_by_name.get(name)
                            if bboxes:
                                bbox = bboxes.pop(0)
                                # Calculate center from bbox [x_min, y_min, x_max, y_max]
                                x = (bbox[0] + bbox[2]) / 2
                                y = (bbox[1] + bbox[3]) / 2
                            printable_objects[obj_id] = {"name": name, "x": x, "y": y}
                        else:
                            printable_objects[obj_id] = name
                    except ValueError:
                        pass  # Skip objects with non-numeric identify_id

    except Exception:
        pass  # Return empty dict if 3MF is corrupt or unreadable

    if include_positions:
        return printable_objects, bbox_all
    return printable_objects


class ProjectPageParser:
    """Parser for extracting project page data from Bambu Lab 3MF files."""

    def __init__(self, file_path: Path):
        self.file_path = file_path

    def parse(self, archive_id: int) -> dict:
        """Extract project page metadata and images from 3MF file."""
        import html

        result = {
            "title": None,
            "description": None,
            "designer": None,
            "designer_user_id": None,
            "license": None,
            "copyright": None,
            "creation_date": None,
            "modification_date": None,
            "origin": None,
            "profile_title": None,
            "profile_description": None,
            "profile_cover": None,
            "profile_user_id": None,
            "profile_user_name": None,
            "design_model_id": None,
            "design_profile_id": None,
            "design_region": None,
            "model_pictures": [],
            "profile_pictures": [],
            "thumbnails": [],
        }

        try:
            with zipfile.ZipFile(self.file_path, "r") as zf:
                # Parse 3D/3dmodel.model for metadata
                model_path = "3D/3dmodel.model"
                if model_path in zf.namelist():
                    content = zf.read(model_path).decode("utf-8", errors="ignore")

                    # Extract metadata elements using regex
                    # Format: <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) -> tuple[set[str], set[tuple[str, str]]]:
        """Get all content hashes and (print name, hash) pairs that appear more than once.

        For hashes: returns all hashes with > 1 archive (true duplicates).
        For name/hash pairs: returns only pairs that have > 1 archive
                     (i.e., same file archived multiple times, not different files with same name).

        Returns a tuple of (duplicate_hashes, duplicate_name_hash_pairs).
        """
        from sqlalchemy import func

        # Soft-deleted archives don't appear in the listing (#1343), so they
        # mustn't influence the duplicate-group counts either — otherwise a
        # group with 1 live + 4 soft-deleted would still be flagged as a
        # duplicate even though the user only sees one row.
        result = await self.db.execute(
            select(PrintArchive.content_hash)
            .where(PrintArchive.content_hash.isnot(None), PrintArchive.deleted_at.is_(None))
            .group_by(PrintArchive.content_hash)
            .having(func.count(PrintArchive.id) > 1)
        )
        duplicate_hashes = {row[0] for row in result.all()}

        # Find print names that have multiple archives with the SAME hash
        # This avoids marking different files with the same name as duplicates
        result = await self.db.execute(
            select(func.lower(PrintArchive.print_name), PrintArchive.content_hash)
            .where(
                PrintArchive.print_name.isnot(None),
                PrintArchive.content_hash.isnot(None),
                PrintArchive.deleted_at.is_(None),
            )
            .group_by(func.lower(PrintArchive.print_name), PrintArchive.content_hash)
            .having(func.count(PrintArchive.id) > 1)
        )
        duplicate_name_hash_pairs = {(row[0], row[1]) for row in result.all()}

        return duplicate_hashes, duplicate_name_hash_pairs

    async def find_duplicates(
        self,
        archive_id: int,
        content_hash: str | None = None,
        print_name: str | None = None,
        makerworld_model_id: str | None = None,
    ) -> list[dict]:
        """Find duplicate archives based on hash or name matching.

        Returns list of dicts with id, print_name, created_at, match_type.
        """
        duplicates = []

        # First, find exact matches by content hash
        if content_hash:
            result = await self.db.execute(
                select(PrintArchive)
                .where(
                    and_(
                        PrintArchive.content_hash == content_hash,
                        PrintArchive.id != archive_id,
                        PrintArchive.deleted_at.is_(None),
                    )
                )
                .order_by(PrintArchive.created_at.desc())
                .limit(10)
            )
            for archive in result.scalars().all():
                duplicates.append(
                    {
                        "id": archive.id,
                        "print_name": archive.print_name,
                        "created_at": archive.created_at,
                        "match_type": "exact",
                    }
                )

        # Then, find similar matches by print name or MakerWorld ID
        # Prefer strict name+hash matching when hash exists; fallback to name-only for legacy/manual
        # archives that may not have a content_hash.
        if print_name or makerworld_model_id:
            conditions = [PrintArchive.id != archive_id, PrintArchive.deleted_at.is_(None)]

            name_conditions = []
            if print_name:
                if content_hash:
                    # Match if print names are similar AND have the same hash (same file)
                    name_conditions.append(
                        and_(PrintArchive.print_name.ilike(print_name), PrintArchive.content_hash == content_hash)
                    )
                else:
                    # Fallback for archives without hash data: match by print name only.
                    name_conditions.append(PrintArchive.print_name.ilike(print_name))
            if makerworld_model_id:
                # Match by MakerWorld model ID stored in extra_data
                from backend.app.core.db_dialect import is_sqlite

                if is_sqlite():
                    from sqlalchemy import func

                    name_conditions.append(
                        func.json_extract(PrintArchive.extra_data, "$.makerworld_model_id") == str(makerworld_model_id)
                    )
                else:
                    name_conditions.append(
                        text("(extra_data::jsonb->>'makerworld_model_id') = :mw_id").bindparams(
                            mw_id=str(makerworld_model_id)
                        )
                    )

            if name_conditions:
                conditions.append(or_(*name_conditions))

                result = await self.db.execute(
                    select(PrintArchive).where(and_(*conditions)).order_by(PrintArchive.created_at.desc()).limit(10)
                )
                for archive in result.scalars().all():
                    # Don't add if already in duplicates (exact match)
                    if not any(d["id"] == archive.id for d in duplicates):
                        duplicates.append(
                            {
                                "id": archive.id,
                                "print_name": archive.print_name,
                                "created_at": archive.created_at,
                                "match_type": "similar",
                            }
                        )

        return duplicates

    async def archive_print(
        self,
        printer_id: int | None,
        source_file: Path,
        print_data: dict | None = None,
        created_by_id: int | None = None,
        original_filename: str | None = None,
        project_id: int | None = None,
        cost_center_id: int | None = None,
        subtask_id: str | None = None,
        prefer_filename_for_name: bool = False,
        plate_id: int | None = None,
        library_file_id: int | None = None,
        slicer_ams_mapping: list[int] | None = None,
        slicer_ams_mapping_printer_id: int | None = None,
    ) -> 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.
        """
        # Verify printer exists if specified
        if printer_id is not None:
            result = await self.db.execute(select(Printer).where(Printer.id == printer_id))
            printer = result.scalar_one_or_none()
            if not printer:
                return None

        # Create archive directory structure
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        display_stem = resolve_display_stem(original_filename if original_filename else source_file.name)
        archive_name = f"{timestamp}_{display_stem}"
        # Use "unassigned" folder for archives without a printer
        printer_folder = str(printer_id) if printer_id is not None else "unassigned"
        archive_dir = (
            settings.archive_dir / printer_folder / archive_name
        )  # SEC-PATH-OK: printer_folder = str(int|None) → digits or "unassigned"; archive_name = f"{timestamp}_{display_stem}" where resolve_display_stem strips path components via Path(filename).name
        archive_dir.mkdir(parents=True, exist_ok=True)

        # Copy 3MF file with an explicit fsync'd loop (avoids a sendfile
        # short-read quirk that silently truncated 3MF archives on some
        # platforms — see _copy_and_fsync and #1032).
        dest_file = archive_dir / source_file.name
        _copy_and_fsync(source_file, dest_file)

        # If we just archived a 3MF, verify the dest is a valid ZIP before
        # going any further. Staying quiet here is how #1032 escaped review —
        # the archive row was written but every later zipfile.ZipFile() call
        # on the dest failed with "File is not a zip file".
        if (
            source_file.suffix.lower() == ".3mf"
            and zipfile.is_zipfile(source_file)
            and not zipfile.is_zipfile(dest_file)
        ):
            try:
                src_size = source_file.stat().st_size
                dst_size = dest_file.stat().st_size
            except OSError:
                src_size = dst_size = -1
            logger.error(
                "Archive copy corrupted 3MF: src=%s (%s bytes, valid ZIP) -> dst=%s (%s bytes, NOT a ZIP). Refusing to create archive row.",
                source_file,
                src_size,
                dest_file,
                dst_size,
            )
            # Narrow cleanup: remove only the truncated file and the archive
            # directory if it's now empty. archive_dir was created with
            # exist_ok=True so it could in theory pre-date this call (e.g.
            # same-second same-filename collision); rmtree would be too broad.
            try:
                dest_file.unlink()
            except OSError:
                pass
            try:
                archive_dir.rmdir()
            except OSError:
                pass  # directory not empty — leave untouched
            return None

        # Compute content hash for duplicate detection
        content_hash = self.compute_file_hash(dest_file)

        # Extract plate number from filename (e.g., "plate_5" from "/data/Metadata/plate_5.gcode")
        plate_number = None
        if print_data:
            filename = print_data.get("filename", "")
            match = re.search(r"plate_(\d+)", filename)
            if match:
                plate_number = int(match.group(1))

        # Parse 3MF metadata
        parser = ThreeMFParser(dest_file, plate_number=plate_number)
        metadata = parser.parse()

        # Save thumbnail if present
        thumbnail_path = None
        if "_thumbnail_data" in metadata:
            thumb_file = archive_dir / f"thumbnail{metadata['_thumbnail_ext']}"
            thumb_file.write_bytes(metadata["_thumbnail_data"])
            thumbnail_path = str(thumb_file.relative_to(settings.base_dir))
            del metadata["_thumbnail_data"]
            del metadata["_thumbnail_ext"]

        # Merge with print data from MQTT
        if print_data:
            metadata["_print_data"] = print_data

        # Promote the slicer's own live-resolved AMS-slot pick, when the caller
        # explicitly opted in (see the `slicer_ams_mapping` param docstring for
        # why this is NOT read off `print_data["ams_mapping"]`), to a stable
        # top-level extra_data key. Lets a later reprint reuse the exact tray
        # the user picked/BambuStudio auto-matched at slice time instead of the
        # scheduler re-deriving one from just the file's static type/color,
        # which can land on the wrong physical spool when that match isn't
        # unique. Top-level (not nested under the `_print_data` diagnostic bag)
        # so API consumers have a single stable path:
        # `archive.extra_data.slicer_ams_mapping`. Stored together with the
        # printer it was resolved against — see `slicer_ams_mapping_printer_id`
        # param docstring — so a later reprint can tell whether it's even
        # applicable before trying to reuse it.
        if slicer_ams_mapping and slicer_ams_mapping_printer_id is not None:
            metadata["slicer_ams_mapping"] = {
                "mapping": slicer_ams_mapping,
                "printer_id": slicer_ams_mapping_printer_id,
            }

        # Determine status and timestamps
        status = print_data.get("status", "completed") if print_data else "archived"
        started_at = datetime.now(timezone.utc) if status == "printing" else None
        completed_at = datetime.now(timezone.utc) if status in ("completed", "failed", "archived") else None

        # Calculate cost based on filament usage and type
        cost = None
        filament_grams = metadata.get("filament_used_grams")
        filament_type = metadata.get("filament_type")
        if filament_grams and filament_type:
            # For multi-material prints, use the first filament type for cost calculation
            primary_type = filament_type.split(",")[0].strip()
            # Look up filament cost_per_kg from database
            filament_result = await self.db.execute(select(Filament).where(Filament.type == primary_type).limit(1))
            filament = filament_result.scalar_one_or_none()
            if filament:
                cost = round((filament_grams / 1000) * filament.cost_per_kg, 2)
            else:
                # Use default filament cost from settings
                from backend.app.api.routes.settings import get_setting

                default_cost_setting = await get_setting(self.db, "default_filament_cost")
                default_cost_per_kg = float(default_cost_setting) if default_cost_setting else 25.0
                cost = round((filament_grams / 1000) * default_cost_per_kg, 2)

        # Calculate quantity from printable objects count
        # printable_objects is a dict of {identify_id: name} for non-skipped objects
        quantity = 1  # Default to 1
        printable_objects = metadata.get("printable_objects")
        if printable_objects and isinstance(printable_objects, dict):
            quantity = len(printable_objects)
            logger.debug("Auto-detected %s parts from 3MF printable objects", quantity)

        # Create archive record
        archive = PrintArchive(
            printer_id=printer_id,
            filename=original_filename or source_file.name,
            file_path=str(dest_file.relative_to(settings.base_dir)),
            file_size=dest_file.stat().st_size,
            content_hash=content_hash,
            thumbnail_path=thumbnail_path,
            # clean_display_name because the 3MF's own metadata reaches this
            # verbatim, and a control character in it renders nowhere and
            # truncates somewhere (#2832). The schema does the same for names
            # arriving over the API. Cleaned before the fallback rather than
            # after it, so an embedded name that is only whitespace still falls
            # through to the filename instead of leaving the archive nameless.
            print_name=(
                clean_display_name(display_stem)
                if prefer_filename_for_name
                else (clean_display_name(metadata.get("print_name")) or clean_display_name(display_stem))
            ),
            print_time_seconds=metadata.get("print_time_seconds"),
            filament_used_grams=metadata.get("filament_used_grams"),
            filament_type=metadata.get("filament_type"),
            filament_color=metadata.get("filament_color"),
            layer_height=metadata.get("layer_height"),
            total_layers=metadata.get("total_layers"),
            nozzle_diameter=metadata.get("nozzle_diameter"),
            bed_temperature=metadata.get("bed_temperature"),
            bed_type=metadata.get("bed_type"),
            nozzle_temperature=metadata.get("nozzle_temperature"),
            sliced_for_model=metadata.get("sliced_for_model"),
            makerworld_url=metadata.get("makerworld_url"),
            designer=metadata.get("designer"),
            status=status,
            started_at=started_at,
            completed_at=completed_at,
            cost=cost,
            quantity=quantity,
            extra_data=metadata,
            created_by_id=created_by_id,
            project_id=project_id,
            library_file_id=library_file_id,
            cost_center_id=cost_center_id,
            subtask_id=subtask_id,
            plate_id=plate_id,
        )

        self.db.add(archive)
        await self.db.commit()
        await self.db.refresh(archive)

        return archive

    async def get_archive(self, archive_id: int) -> PrintArchive | None:
        """Get an archive by ID with relationships loaded."""
        from sqlalchemy.orm import selectinload

        result = await self.db.execute(
            select(PrintArchive)
            .options(selectinload(PrintArchive.created_by), selectinload(PrintArchive.project))
            .where(PrintArchive.id == archive_id)
        )
        return result.scalar_one_or_none()

    async def update_archive_status(
        self,
        archive_id: int,
        status: str,
        completed_at: datetime | None = None,
        failure_reason: str | None = None,
    ) -> bool:
        """Update the status of an archive."""
        archive = await self.get_archive(archive_id)
        if not archive:
            return False

        archive.status = status
        if completed_at:
            archive.completed_at = completed_at
        if failure_reason:
            archive.failure_reason = failure_reason

        await self.db.commit()
        return True

    async def list_archives(
        self,
        printer_id: int | None = None,
        project_id: int | None = None,
        date_from: date | None = None,
        date_to: date | None = None,
        limit: int = 50,
        offset: int = 0,
        visible_to_user_id: int | None = None,
    ) -> list[PrintArchive]:
        """List archives with optional filtering.

        ``visible_to_user_id`` scopes results to archives that user owns. Used
        when the caller has ARCHIVES_READ_OWN but not ARCHIVES_READ_ALL — pass
        ``None`` to skip the filter (caller has read-all or auth is disabled).
        """
        from sqlalchemy.orm import selectinload

        query = (
            select(PrintArchive)
            .options(selectinload(PrintArchive.project), selectinload(PrintArchive.created_by))
            # Hide soft-deleted rows from the listings (#1343). The stats
            # endpoint deliberately does NOT add this filter so deleted
            # archives keep contributing to Quick Stats.
            .where(PrintArchive.deleted_at.is_(None))
            .order_by(PrintArchive.created_at.desc())
        )

        if printer_id:
            query = query.where(PrintArchive.printer_id == printer_id)

        if project_id:
            query = query.where(PrintArchive.project_id == project_id)

        if date_from:
            dt_from = datetime.combine(date_from, time.min, tzinfo=timezone.utc)
            query = query.where(PrintArchive.created_at >= dt_from)

        if date_to:
            dt_to = datetime.combine(date_to, time.max, tzinfo=timezone.utc)
            query = query.where(PrintArchive.created_at <= dt_to)

        if visible_to_user_id is not None:
            query = query.where(PrintArchive.created_by_id == visible_to_user_id)

        query = query.limit(limit).offset(offset)
        result = await self.db.execute(query)
        return list(result.scalars().all())

    async def soft_delete_archive(self, archive_id: int) -> bool:
        """Soft-delete an archive (#1343).

        Removes the archive's files from disk (it disappears from the listings
        and frees the storage) but flips the row's ``deleted_at`` so the stats
        endpoint keeps counting its filament / energy / time / cost. The user
        can opt into a hard delete via the "Also remove from statistics"
        checkbox in the delete dialog — that path calls ``delete_archive``
        instead and removes the row entirely.
        """
        archive = await self.get_archive(archive_id)
        if not archive:
            return False
        if archive.deleted_at is not None:
            # Already soft-deleted; nothing to do. The files were purged on
            # the first soft-delete pass so there is nothing left on disk.
            return True

        dir_to_delete = self._resolve_archive_dir_for_delete(archive)

        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()

        if dir_to_delete:
            shutil.rmtree(dir_to_delete, ignore_errors=True)
        return True

    def _resolve_archive_dir_for_delete(self, archive: PrintArchive) -> Path | None:
        """Return the on-disk directory that backs *archive*, after the same
        two safety checks ``delete_archive`` enforces.

        Extracted so soft-delete and hard-delete share the path-resolution
        rules. Returns ``None`` when nothing should be removed from disk
        (no file_path, path outside archive_dir, or path not deep enough).
        """
        if not archive.file_path or not archive.file_path.strip():
            logger.error(
                f"SECURITY: Refusing to delete files for archive {archive.id} - "
                f"file_path is empty or invalid: '{archive.file_path}'"
            )
            return None

        file_path = settings.base_dir / archive.file_path
        if not file_path.exists():
            return None

        archive_dir = file_path.parent
        try:
            relative_path = archive_dir.resolve().relative_to(settings.archive_dir.resolve())
        except ValueError:
            logger.error(
                f"SECURITY: Refusing to delete archive {archive.id} - "
                f"path {archive_dir} is outside archive directory {settings.archive_dir}"
            )
            return None
        if len(relative_path.parts) < 1:
            logger.error(
                f"SECURITY: Refusing to delete archive {archive.id} - "
                f"path {archive_dir} is not deep enough inside archive directory"
            )
            return None
        return archive_dir

    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

        # Resolve the directory to delete BEFORE committing the DB change
        dir_to_delete: Path | None = None

        if archive.file_path and archive.file_path.strip():
            file_path = settings.base_dir / archive.file_path
            if file_path.exists():
                archive_dir = file_path.parent

                # Safety check 1: archive_dir must be inside archive_dir
                try:
                    archive_dir.resolve().relative_to(settings.archive_dir.resolve())
                except ValueError:
                    logger.error(
                        f"SECURITY: Refusing to delete archive {archive_id} - "
                        f"path {archive_dir} is outside archive directory {settings.archive_dir}"
                    )
                    await self.db.delete(archive)
                    await self.db.commit()
                    return True

                # Safety check 2: archive_dir must be at least 1 level deep inside archive_dir
                try:
                    relative_path = archive_dir.resolve().relative_to(settings.archive_dir.resolve())
                    if len(relative_path.parts) < 1:
                        logger.error(
                            f"SECURITY: Refusing to delete archive {archive_id} - "
                            f"path {archive_dir} is not deep enough inside archive directory"
                        )
                        await self.db.delete(archive)
                        await self.db.commit()
                        return True
                except ValueError:
                    pass  # Already handled above

                dir_to_delete = archive_dir
        else:
            logger.error(
                f"SECURITY: Refusing to delete files for archive {archive_id} - "
                f"file_path is empty or invalid: '{archive.file_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
        if dir_to_delete:
            shutil.rmtree(dir_to_delete, ignore_errors=True)

        return True

    async def attach_timelapse(
        self,
        archive_id: int,
        timelapse_data: bytes,
        filename: str = "timelapse.mp4",
    ) -> bool:
        """Attach a timelapse video to an archive.

        Non-MP4 videos (e.g. AVI from P1S) are saved as-is and a background
        task converts them to MP4 for browser compatibility.
        """
        import asyncio

        archive = await self.get_archive(archive_id)
        if not archive:
            return False

        # Get archive directory
        file_path = (
            settings.base_dir / archive.file_path
        )  # SEC-PATH-OK: archive.file_path is DB-stored, set by archive_print() under settings.archive_dir
        archive_dir = file_path.parent

        # 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
        await asyncio.to_thread(timelapse_file.write_bytes, timelapse_data)

        # Update archive record
        archive.timelapse_path = str(timelapse_file.relative_to(settings.base_dir))
        await self.db.commit()

        # For non-MP4 videos (e.g. AVI from P1S), kick off background conversion
        if not filename.lower().endswith(".mp4"):
            spawn_background_task(
                _convert_timelapse_to_mp4(archive_id, timelapse_file),
                name=f"timelapse-convert-{archive_id}",
            )

        return True


async def _convert_timelapse_to_mp4(archive_id: int, source_path: Path) -> None:
    """Background task: convert non-MP4 timelapse (e.g. AVI from P1S) to MP4.

    Runs with low CPU priority (-threads 1, nice) so it doesn't starve
    other processes on resource-constrained devices like Raspberry Pi.
    """
    import asyncio

    from backend.app.core.database import async_session
    from backend.app.services.camera import get_ffmpeg_path

    logger = logging.getLogger(__name__)

    ffmpeg = get_ffmpeg_path()
    if not ffmpeg:
        logger.info(
            "FFmpeg not available, skipping timelapse conversion for archive %s (file saved as %s)",
            archive_id,
            source_path.suffix,
        )
        return

    mp4_path = source_path.with_suffix(".mp4")

    try:
        cmd = [
            ffmpeg,
            "-y",
            "-i",
            str(source_path),
            "-c:v",
            "libx264",
            "-preset",
            "fast",
            "-crf",
            "23",
            "-threads",
            "1",
            "-movflags",
            "+faststart",
            str(mp4_path),
        ]

        # Try with nice for lower CPU priority (standard on Linux/macOS)
        try:
            process = await asyncio.create_subprocess_exec(
                "nice",
                "-n",
                "19",
                *cmd,
                stdout=asyncio.subprocess.PIPE,
                stderr=asyncio.subprocess.PIPE,
            )
        except FileNotFoundError:
            # nice not available (e.g. Windows), run without
            process = await asyncio.create_subprocess_exec(
                *cmd,
                stdout=asyncio.subprocess.PIPE,
                stderr=asyncio.subprocess.PIPE,
            )

        _, stderr = await process.communicate()

        if process.returncode != 0:
            logger.warning(
                "Timelapse conversion failed for archive %s: %s",
                archive_id,
                stderr.decode()[-500:],
            )
            if mp4_path.exists():
                mp4_path.unlink()
            return

        # Update DB path to the new MP4 file
        async with async_session() as db:
            from backend.app.models.archive import PrintArchive

            result = await db.execute(select(PrintArchive).where(PrintArchive.id == archive_id))
            archive = result.scalar_one_or_none()
            if archive:
                archive.timelapse_path = str(mp4_path.relative_to(settings.base_dir))
                await db.commit()

        # Remove original non-MP4 file
        if source_path.exists():
            source_path.unlink()

        logger.info(
            "Converted timelapse to MP4 for archive %s (%s → %s)",
            archive_id,
            source_path.name,
            mp4_path.name,
        )

    except Exception as e:
        logger.warning("Timelapse conversion error for archive %s: %s", archive_id, e)
        if mp4_path.exists():
            mp4_path.unlink()