bambuddy/backend/app/api/routes/spoolman_inventory.py

"""Spoolman inventory proxy endpoints.

Translates between Spoolman's data model and Bambuddy's internal
InventorySpool format so the frontend can use a single unified inventory UI
regardless of whether data comes from the local database or Spoolman.
"""

from __future__ import annotations

import asyncio
import json
import logging
import re
import time
from contextlib import asynccontextmanager, nullcontext
from datetime import datetime

from fastapi import APIRouter, Body, Depends, HTTPException, Path, Query, Response
from fastapi.responses import JSONResponse
from pydantic import BaseModel, Field, field_validator, model_validator
from sqlalchemy import delete, select, text, tuple_
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import selectinload

from backend.app.api.routes._spoolman_helpers import (
    BAMBU_LAST_DRIED_AT_KEY,
    BAMBU_LAST_DRIED_HOURS_KEY,
    BAMBU_LAST_DRIED_TEMP_KEY,
    NormalizedFilament,
    NormalizedVendorRef,
    _map_spoolman_spool,
    _safe_int,
    _safe_optional_float,
    assert_safe_spoolman_url,
    cost_per_kg_to_spoolman_price,
    spoolman_cost_per_kg,
    spoolman_net_weight,
    spoolman_price_weight,
    spoolman_tare,
)
from backend.app.core.auth import RequestPrinterScope, RequirePermissionIfAuthEnabled
from backend.app.core.database import get_db
from backend.app.core.permissions import Permission
from backend.app.core.printer_scope import PrinterScope
from backend.app.core.websocket import ws_manager
from backend.app.models.ams_label import AmsLabel
from backend.app.models.printer import Printer
from backend.app.models.settings import Settings
from backend.app.models.spool_filament_preset import SpoolmanFilamentPreset
from backend.app.models.spoolman_k_profile import SpoolmanKProfile
from backend.app.models.spoolman_slot_assignment import SpoolmanSlotAssignment
from backend.app.models.supplier import SpoolmanSpoolSupplier, Supplier
from backend.app.models.user import User
from backend.app.schemas.spool import SpoolFilamentPresetBase, SpoolKProfileBase, naive_utc
from backend.app.schemas.spoolman import SlotAmsId, SlotTrayId, SpoolmanFilamentPatch, SpoolmanSlotAssignmentEnriched
from backend.app.schemas.supplier import SpoolSupplierLinkInput
from backend.app.services import filament_profile_default as profile_defaults, slot_unlink_grace, spool_swaps
from backend.app.services.ams_assignment_lock import get_spool_move_lock
from backend.app.services.location_service import (
    enrich_spool_dicts_with_location_id,
    maybe_sync_spoolman_locations,
    resolve_spoolman_location_string,
)
from backend.app.services.printer_manager import PrinterManager, printer_manager
from backend.app.services.slicer_filament_resolver import resolve_slicer_filament
from backend.app.services.slot_nozzle import resolve_slot_nozzle
from backend.app.services.spool_filament_preset import forget_spoolman_spool_own_preset, resolve_spoolman_preset
from backend.app.services.spoolman import (
    SpoolmanClient,
    SpoolmanClientError,
    SpoolmanNotFoundError,
    SpoolmanUnavailableError,
    get_spoolman_client,
    init_spoolman_client,
)
from backend.app.services.spoolman_tracking import get_fallback_spool_tag_for_slot, is_slot_fallback_tag
from backend.app.services.tag_conflict import tag_already_linked
from backend.app.utils.color_utils import spoolman_color_hex
from backend.app.utils.filament_finish import finish_generic_id
from backend.app.utils.filament_ids import (
    GENERIC_FILAMENT_IDS,
    filament_id_to_setting_id,
    normalize_slicer_filament,
)
from backend.app.utils.filament_types import nozzle_temp_range, printer_filament_type

logger = logging.getLogger(__name__)

router = APIRouter(prefix="/spoolman/inventory", tags=["spoolman-inventory"])


# Cache the last successful health-check timestamp to avoid a round-trip on
# every request.  A failed check clears the cache immediately.
_health_check_cache: dict[str, float] = {}
_HEALTH_CHECK_TTL = 30.0  # seconds


def _tag_cleared(val: str | None) -> bool:
    """Return True when a PATCH field explicitly removes a tag (null)."""
    return val is None


async def _clear_stale_tag_links(
    client: SpoolmanClient,
    *,
    tag: str,
    keep_spool_id: int,
    log_context: str,
) -> int:
    """Clear extra.tag on OTHER spools still claiming the given tag (#1457).

    A given AMS slot tag — whether a real RFID (tray_uuid/tag_uid) or the
    deterministic fallback derived from (printer_serial, ams_id, tray_id) for
    non-RFID slots — uniquely identifies one physical slot. When a spool is
    (re)bound to that slot via Assign or Link, any other Spoolman spool whose
    extra.tag still holds the same value is stale and would resurface in the
    hover card / fill-level lookup.

    Best-effort: per-spool patch failures are logged and skipped, never raised.
    Returns the number of spools cleared.
    """
    if not tag:
        return 0
    tag_upper = tag.upper()

    try:
        spools = await client.get_spools()
    except (SpoolmanClientError, SpoolmanUnavailableError) as exc:
        logger.warning("Could not enumerate spools for stale-tag cleanup: %s", exc)
        return 0

    cleared = 0
    for spool in spools:
        spool_id = spool.get("id")
        if not spool_id or spool_id == keep_spool_id:
            continue
        extra = spool.get("extra") or {}
        raw_tag = extra.get("tag", "")
        if not raw_tag:
            continue
        clean_tag = raw_tag.strip('"').upper()
        if clean_tag != tag_upper:
            continue
        try:
            await client.merge_spool_extra(spool_id, {"tag": json.dumps("")})
            cleared += 1
            logger.info(
                "Cleared stale tag '%s' from Spoolman spool %s (%s; reassigned to spool %s)",
                tag_upper[:16],
                spool_id,
                log_context,
                keep_spool_id,
            )
        except (SpoolmanClientError, SpoolmanUnavailableError, SpoolmanNotFoundError) as exc:
            logger.warning(
                "Failed to clear stale tag on Spoolman spool %s: %s",
                spool_id,
                exc,
            )
    return cleared


async def _clear_stale_slot_fallback_tag_links(
    client: SpoolmanClient,
    *,
    printer_serial: str,
    ams_id: int,
    tray_id: int,
    keep_spool_id: int,
) -> int:
    """Convenience wrapper: compute the slot's fallback tag and clear it from
    other spools. Used by the assign route, which identifies the slot by
    (printer, ams, tray) rather than by an explicit tag value.
    """
    fallback_tag = get_fallback_spool_tag_for_slot(printer_serial, ams_id, tray_id)
    if not fallback_tag:
        return 0
    return await _clear_stale_tag_links(
        client,
        tag=fallback_tag,
        keep_spool_id=keep_spool_id,
        log_context=f"printer={printer_serial} ams={ams_id} tray={tray_id}",
    )


async def _clear_vacated_slot_fallback_tag(
    client: SpoolmanClient,
    db: AsyncSession,
    *,
    spool_id: int,
    spool: dict,
    vacated: list[tuple[int, int, int]],
) -> str | None:
    """Clear the spool's extra.tag when it is the fallback tag of a slot it leaves (#1574).

    A spool linked to a slot without RFID carries that slot's fallback tag, and
    the tag lookup keeps finding it for the old slot after it moved. A real RFID
    or any other tag is the spool's own and stays. Returns the tag it cleared,
    or None. Raises when Spoolman refuses, before anything local has changed,
    so the spool stays where it was.
    """
    tag = str((spool.get("extra") or {}).get("tag") or "").strip('"').upper()
    if not tag or not vacated:
        return None
    serials = dict(
        (await db.execute(select(Printer.id, Printer.serial_number).where(Printer.id.in_({p for p, _, _ in vacated}))))
        .tuples()
        .all()
    )
    if not any(
        tag == get_fallback_spool_tag_for_slot(serials.get(printer_id) or "", ams_id, tray_id)
        for printer_id, ams_id, tray_id in vacated
    ):
        return None
    try:
        await client.merge_spool_extra(spool_id, {"tag": json.dumps("")})
    except (SpoolmanClientError, SpoolmanUnavailableError, SpoolmanNotFoundError) as exc:
        logger.warning("Could not clear the old slot's fallback tag from Spoolman spool %s: %s", spool_id, exc)
        raise HTTPException(
            status_code=503 if isinstance(exc, SpoolmanUnavailableError) else 502,
            detail="Spoolman did not clear the spool's tag for its old slot, so the spool was not moved",
        ) from exc
    logger.info("Cleared the old slot's fallback tag from Spoolman spool %s before moving it", spool_id)
    return tag


async def _get_client(db: AsyncSession) -> SpoolmanClient:
    """Return a validated Spoolman client (URL checked, health-checked) or raise an HTTP error."""
    result = await db.execute(select(Settings))
    settings: dict[str, str] = {s.key: s.value for s in result.scalars().all()}

    enabled = settings.get("spoolman_enabled", "false").lower() == "true"
    url = settings.get("spoolman_url", "").strip()

    if not enabled:
        raise HTTPException(status_code=400, detail="Spoolman integration is not enabled")
    if not url:
        raise HTTPException(status_code=400, detail="Spoolman URL is not configured")

    # SSRF guard: reject dangerous schemes, cloud-metadata IPs (169.254.169.254, 100.100.100.200,
    # fd00:ec2::254), multicast and unspecified addresses — loopback and RFC-1918 ranges are
    # intentionally permitted (Spoolman commonly runs on the same host or home LAN).
    # Raises ValueError with a descriptive message on any violation.
    try:
        assert_safe_spoolman_url(url)
    except ValueError as exc:
        raise HTTPException(status_code=400, detail=str(exc)) from exc

    # Re-use the cached client when URL is unchanged; reinitialise on URL change (cache invalidation).
    client = await get_spoolman_client()
    if not client or client.base_url != url.rstrip("/"):
        try:
            client = await init_spoolman_client(url)
        except ValueError as exc:
            raise HTTPException(status_code=400, detail=str(exc)) from exc

    # Only call health_check() when the cached result has expired.
    # Evict stale entries when URL changes (only one Spoolman URL is active at a time).
    if url not in _health_check_cache and _health_check_cache:
        _health_check_cache.clear()
    now = time.monotonic()
    last_ok = _health_check_cache.get(url, 0.0)
    if now - last_ok > _HEALTH_CHECK_TTL:
        if not await client.health_check():
            _health_check_cache.pop(url, None)
            raise HTTPException(status_code=503, detail="Spoolman server is not reachable")
        _health_check_cache[url] = now

    return client


@asynccontextmanager
async def _translate_spoolman_errors():
    """Translate Spoolman typed exceptions to HTTP errors for all inventory endpoints."""
    try:
        yield
    except SpoolmanNotFoundError as exc:
        raise HTTPException(status_code=404, detail="Spool not found in Spoolman") from exc
    except SpoolmanClientError as exc:
        raise HTTPException(
            status_code=502,
            detail={
                "message": "Spoolman rejected the request",
                "upstream_status": exc.status_code,
                "upstream_body": getattr(exc, "response_text", ""),
            },
        ) from exc
    except SpoolmanUnavailableError as exc:
        raise HTTPException(status_code=503, detail="Spoolman server is not reachable") from exc


def _raise_if_partial_failure(spools: list[dict], results: list, operation: str) -> None:
    """Raise HTTP 502 if any gather result is an exception, logging each failure."""
    failures = [(s["id"], r) for s, r in zip(spools, results, strict=True) if isinstance(r, BaseException)]
    if failures:
        logger.error(
            "Partial %s failure: %d/%d spools failed: %s",
            operation,
            len(failures),
            len(spools),
            [(sid, type(exc).__name__) for sid, exc in failures],
        )
        raise HTTPException(
            status_code=502,
            detail=f"{operation} partially applied: {len(spools) - len(failures)}/{len(spools)} spools updated",
        )


async def _apply_price_if_set(client: SpoolmanClient, spool: dict, cost_per_kg: float | None) -> tuple[dict, list[str]]:
    """Patch the spool price; return (updated_spool, warnings).

    Spoolman's spool price is what the whole spool cost, so the per-kg rate is
    converted at the net weight of the spool Spoolman just created (#3194).

    Returns the original spool and a non-empty warnings list when the price
    update fails, so the caller can return HTTP 207 instead of silently
    discarding the price.
    """
    if cost_per_kg is None:
        return spool, []
    try:
        async with _translate_spoolman_errors():
            updated = await client.update_spool_full(
                spool["id"], price=cost_per_kg_to_spoolman_price(cost_per_kg, spoolman_price_weight(spool))
            )
        return updated, []
    except HTTPException as exc:
        if exc.status_code >= 500:
            raise  # Propagate network/server errors — don't swallow Spoolman outages
        logger.warning(
            "Price update failed for spool %d; spool created without price (cost_per_kg=%s, status=%d)",
            spool["id"],
            cost_per_kg,
            exc.status_code,
        )
        return spool, [f"price_not_set: Spoolman rejected the price update (HTTP {exc.status_code})"]


# ---------------------------------------------------------------------------
# Request / response schemas
# ---------------------------------------------------------------------------


_HEX_RE = re.compile(r"^[0-9A-Fa-f]{6}([0-9A-Fa-f]{2})?$")


def _validate_rgba(v: str | None) -> str | None:
    if v is None:
        return v
    clean = v.removeprefix("#")
    if not _HEX_RE.match(clean):
        raise ValueError("rgba must be a 6 or 8 character hex string (RRGGBB or RRGGBBAA)")
    return clean.upper()


def _validate_storage_location(v: str | None) -> str | None:
    if v is not None and any(c in v for c in ("\r", "\n", "\x00")):
        raise ValueError("storage_location must not contain control characters")
    return v


class SpoolmanInventoryCreate(BaseModel):
    # When spoolman_filament_id is provided the caller has already chosen a filament from the
    # Spoolman catalog, so material (and other metadata) are optional — the backend skips
    # find_or_create_filament() and uses the supplied ID directly.
    spoolman_filament_id: int | None = Field(None, gt=0)
    material: str | None = Field(None, min_length=1, max_length=64)
    subtype: str | None = Field(None, max_length=64)
    brand: str | None = Field(None, max_length=128)
    color_name: str | None = Field(None, max_length=64)
    rgba: str | None = Field(None, max_length=8, description="6-digit hex (RRGGBB) or 8-digit (RRGGBBAA)")
    label_weight: int = Field(1000, ge=1, le=100_000)
    # Persisted to the Spoolman spool's own `spool_weight` (tare), which takes
    # priority over the filament-level value both in _map_spoolman_spool and in
    # the weigh endpoint. Only written when the request actually sets it: the
    # 250 default is the display fallback, and writing it on every create would
    # stamp an explicit tare on spools that should keep inheriting one (#2908).
    core_weight: int = Field(250, ge=0, le=10_000)
    weight_used: float = Field(0.0, ge=0.0, le=100_000.0)
    note: str | None = Field(None, max_length=1000)
    cost_per_kg: float | None = Field(None, ge=0.0, le=1_000_000.0)
    storage_location: str | None = Field(None, max_length=255)
    location_id: int | None = Field(None, gt=0)
    # BambuStudio slicer preset for this spool. Spoolman has no native field
    # for this, so we persist it under the bambu_slicer_filament[_name] keys
    # in the spool's extra dict and read it back in _map_spoolman_spool.
    slicer_filament: str | None = Field(None, max_length=128)
    slicer_filament_name: str | None = Field(None, max_length=255)

    @field_validator("rgba")
    @classmethod
    def validate_rgba(cls, v: str | None) -> str | None:
        return _validate_rgba(v)

    @field_validator("storage_location")
    @classmethod
    def validate_storage_location(cls, v: str | None) -> str | None:
        return _validate_storage_location(v)

    @model_validator(mode="after")
    def validate_weight_consistency(self) -> SpoolmanInventoryCreate:
        # material is required only when the caller has not pre-selected a Spoolman filament
        if self.spoolman_filament_id is None and not self.material:
            raise ValueError("material is required when spoolman_filament_id is not provided")
        if self.weight_used > self.label_weight:
            raise ValueError("weight_used must not exceed label_weight")
        return self


class SpoolmanInventoryUpdate(BaseModel):
    material: str | None = Field(None, min_length=1, max_length=64)
    subtype: str | None = Field(None, max_length=64)
    brand: str | None = Field(None, max_length=128)
    color_name: str | None = Field(None, max_length=64)
    rgba: str | None = Field(None, max_length=8, description="6-digit hex (RRGGBB) or 8-digit (RRGGBBAA)")
    label_weight: int | None = Field(None, ge=1, le=100_000)
    # Persisted to the spool's own `spool_weight` (see the Create schema).
    # Omitted / null leaves the current value alone, as with every other field
    # here. There is no per-spool "go back to inheriting" through this route;
    # the filament-level route already owns that decision (#2908).
    core_weight: int | None = Field(None, ge=0, le=10_000)
    weight_used: float | None = Field(None, ge=0.0, le=100_000.0)
    note: str | None = Field(None, max_length=1000)
    cost_per_kg: float | None = Field(None, ge=0.0, le=1_000_000.0)
    tag_uid: str | None = Field(None, min_length=8, max_length=30, pattern=r"^[0-9A-Fa-f]+$")
    tray_uuid: str | None = Field(None, min_length=32, max_length=32, pattern=r"^[0-9A-Fa-f]+$")
    storage_location: str | None = Field(None, max_length=255)
    location_id: int | None = Field(None, gt=0)
    # BambuStudio slicer preset — persisted to Spoolman extra dict (see Create
    # schema). Pass an empty string to clear; null/omitted leaves unchanged.
    slicer_filament: str | None = Field(None, max_length=128)
    slicer_filament_name: str | None = Field(None, max_length=255)
    # Set by hand for a drying done outside an AMS; null clears it. Persisted
    # to the spool's extra dict like the slicer preset (#2863).
    last_dried_at: datetime | None = None

    @field_validator("rgba")
    @classmethod
    def validate_rgba(cls, v: str | None) -> str | None:
        return _validate_rgba(v)

    @field_validator("storage_location")
    @classmethod
    def validate_storage_location(cls, v: str | None) -> str | None:
        return _validate_storage_location(v)

    @field_validator("last_dried_at")
    @classmethod
    def validate_last_dried_at(cls, v: datetime | None) -> datetime | None:
        return naive_utc(v)

    @model_validator(mode="after")
    def validate_tag_fields(self) -> SpoolmanInventoryUpdate:
        # null = remove tag; non-null values rejected (use /tag endpoint to write tags)
        if self.tag_uid is not None:
            raise ValueError("tag_uid cannot be set via this endpoint; use PATCH /spools/{id}/tag to write tags")
        if self.tray_uuid is not None:
            raise ValueError("tray_uuid cannot be set via this endpoint; use PATCH /spools/{id}/tag to write tags")
        return self

    @model_validator(mode="after")
    def validate_weight_consistency(self) -> SpoolmanInventoryUpdate:
        if self.weight_used is not None and self.label_weight is not None:
            if self.weight_used > self.label_weight:
                raise ValueError("weight_used must not exceed label_weight")
        return self


class SpoolmanInventoryBulkCreate(BaseModel):
    spool: SpoolmanInventoryCreate
    quantity: int = Field(1, ge=1, le=50)


class SpoolWeightUpdate(BaseModel):
    weight_grams: float = Field(..., ge=0.0, le=100_000.0)


class SpoolTagLinkRequest(BaseModel):
    # Minimum 8 hex chars = 4-byte NFC UID (Bambu Lab hardware tags use 4-byte UIDs).
    tag_uid: str | None = Field(None, min_length=8, max_length=30, pattern=r"^[0-9A-Fa-f]+$")
    tray_uuid: str | None = Field(None, min_length=32, max_length=32, pattern=r"^[0-9A-Fa-f]+$")

    @field_validator("tag_uid")
    @classmethod
    def tag_uid_not_all_zeros(cls, v: str | None) -> str | None:
        if v is not None and all(c in "0" for c in v):
            raise ValueError("tag_uid must not be all-zero bytes")
        return v

    @model_validator(mode="after")
    def at_least_one(self) -> SpoolTagLinkRequest:
        if not self.tag_uid and not self.tray_uuid:
            raise ValueError("tag_uid or tray_uuid is required")
        return self


class SpoolSlotAssignmentRequest(BaseModel):
    spoolman_spool_id: int = Field(..., gt=0)
    printer_id: int = Field(..., gt=0)
    ams_id: SlotAmsId
    tray_id: SlotTrayId
    # The spool also leaves any other slot it is assigned to, in the same
    # transaction (#1574): a scanned spool is the one in the user's hand.
    move: bool = False


# ---------------------------------------------------------------------------
# Endpoints
# ---------------------------------------------------------------------------


@router.get("/spools")
async def list_spools(
    include_archived: bool = Query(False),
    db: AsyncSession = Depends(get_db, scope="function"),
    _: User | None = RequirePermissionIfAuthEnabled(Permission.INVENTORY_READ),
) -> list[dict]:
    """Return all Spoolman spools in the InventorySpool format."""
    client = await _get_client(db)
    # Sync after we have the route-resolved client so tests that patch the
    # route module's get_spoolman_client/init_spoolman_client also catch the
    # sync's client lookup — otherwise the location_service path imports from
    # backend.app.services.spoolman directly and bypasses the patch.
    if await maybe_sync_spoolman_locations(db, client=client):
        await db.commit()

    async with _translate_spoolman_errors():
        spools = await client.get_all_spools(allow_archived=include_archived)

    mapped: list[dict] = []
    spool_ids: list[int] = []
    for s in spools:
        try:
            m = _map_spoolman_spool(s)
            mapped.append(m)
            spool_ids.append(m["id"])
        except ValueError as exc:
            logger.warning("Skipping malformed Spoolman spool (id=%r): %s", s.get("id"), exc)

    if spool_ids:
        kp_result = await db.execute(select(SpoolmanKProfile).where(SpoolmanKProfile.spoolman_spool_id.in_(spool_ids)))
        kp_by_spool: dict[int, list[dict]] = {}
        for kp in kp_result.scalars().all():
            kp_by_spool.setdefault(kp.spoolman_spool_id, []).append(_k_profile_to_dict(kp))
        for m in mapped:
            m["k_profiles"] = kp_by_spool.get(m["id"], [])

        # Supplier assignments (#2988) live Bambuddy-side even for Spoolman
        # spools, so the list carries them in both modes identically.
        link_result = await db.execute(
            select(SpoolmanSpoolSupplier)
            .options(selectinload(SpoolmanSpoolSupplier.supplier))
            .where(SpoolmanSpoolSupplier.spoolman_spool_id.in_(spool_ids))
        )
        links_by_spool: dict[int, list[dict]] = {}
        for link in link_result.scalars().all():
            links_by_spool.setdefault(link.spoolman_spool_id, []).append(_supplier_link_to_dict(link))
        for m in mapped:
            m["suppliers"] = links_by_spool.get(m["id"], [])

    await enrich_spool_dicts_with_location_id(db, mapped)
    return mapped


@router.get("/spools/{spool_id}")
async def get_spool(
    spool_id: int = Path(..., gt=0),
    db: AsyncSession = Depends(get_db, scope="function"),
    _: User | None = RequirePermissionIfAuthEnabled(Permission.INVENTORY_READ),
) -> dict:
    """Return a single Spoolman spool in the InventorySpool format."""
    client = await _get_client(db)
    async with _translate_spoolman_errors():
        spool = await client.get_spool(spool_id)
    try:
        mapped = _map_spoolman_spool(spool)
    except ValueError as exc:
        logger.warning("Malformed Spoolman spool (id=%r): %s", spool_id, exc)
        raise HTTPException(status_code=502, detail="Spoolman returned malformed spool data") from exc

    kp_result = await db.execute(select(SpoolmanKProfile).where(SpoolmanKProfile.spoolman_spool_id == spool_id))
    mapped["k_profiles"] = [_k_profile_to_dict(kp) for kp in kp_result.scalars().all()]
    link_result = await db.execute(
        select(SpoolmanSpoolSupplier)
        .options(selectinload(SpoolmanSpoolSupplier.supplier))
        .where(SpoolmanSpoolSupplier.spoolman_spool_id == spool_id)
    )
    mapped["suppliers"] = [_supplier_link_to_dict(link) for link in link_result.scalars().all()]
    await enrich_spool_dicts_with_location_id(db, [mapped])
    return mapped


async def _resolve_filament_id(data: SpoolmanInventoryCreate, client: SpoolmanClient) -> int:
    """Return the Spoolman filament ID for this spool creation request.

    If spoolman_filament_id is set the caller pre-selected a catalog entry,
    so find_or_create_filament() is skipped and the ID is used directly.
    """
    if data.spoolman_filament_id is not None:
        return data.spoolman_filament_id
    # Validator guarantees material is non-None when spoolman_filament_id is None
    assert data.material is not None  # noqa: S101
    # `or "808080"` on the result rather than on the input: spoolman_color_hex
    # returns None only for a missing value, so this is the same neutral grey the
    # old inline default produced, without handing an Optional to a str parameter.
    color_hex = spoolman_color_hex(data.rgba) or "808080"
    async with _translate_spoolman_errors():
        return await client.find_or_create_filament(
            material=data.material,
            subtype=data.subtype or "",
            brand=data.brand,
            color_hex=color_hex,
            label_weight=data.label_weight,
            color_name=data.color_name,
        )


async def _resolve_create_default(
    db: AsyncSession, data: SpoolmanInventoryCreate, spool: dict
) -> profile_defaults.ResolvedProfiles | None:
    """The profile default for a spool this request created (#1326).

    From the request's product fields; a request that only names a catalog
    filament (``spoolman_filament_id``) carries none, so then from the spool
    Spoolman made -- its colour name only when it is a real one, the mapper
    fills a missing one with the subtype.
    """
    if data.material:
        return await profile_defaults.resolve_profile_default(
            db, brand=data.brand, material=data.material, subtype=data.subtype, color_name=data.color_name
        )
    mapped = _map_spoolman_spool(spool)
    return await profile_defaults.resolve_profile_default(
        db,
        brand=mapped["brand"],
        material=mapped["material"],
        subtype=mapped["subtype"],
        color_name=data.color_name or (None if mapped["color_name_is_synthesized"] else mapped["color_name"]),
    )


async def _finish_created_spool(
    db: AsyncSession,
    client: SpoolmanClient,
    spool: dict,
    data: SpoolmanInventoryCreate,
    default: profile_defaults.ResolvedProfiles | None,
) -> dict:
    """What a create does after Spoolman made the spool: the fields Spoolman
    has no place for, and the product's profile default (#1326).

    The slicer preset and the colour name go under the spool's extra dict
    (mirror update_spool) -- Spoolman has no ``color_name`` on a filament
    (#1357). A preset the request did not bring comes from the default, and
    its preset overrides and K profiles land in the Bambuddy tables keyed by
    the new spool's id. The spool exists either way, so a failed write is
    logged and the create still succeeds.
    """
    slicer_filament = data.slicer_filament
    slicer_filament_name = data.slicer_filament_name
    if not slicer_filament and default is not None and default.slicer_filament:
        slicer_filament = default.slicer_filament
        slicer_filament_name = default.slicer_filament_name

    new_extra: dict = {}
    if slicer_filament is not None:
        new_extra["bambu_slicer_filament"] = json.dumps(slicer_filament)
    if slicer_filament_name is not None:
        new_extra["bambu_slicer_filament_name"] = json.dumps(slicer_filament_name)
    if data.color_name is not None:
        new_extra["bambu_color_name"] = json.dumps(data.color_name)
    if new_extra:
        try:
            async with _translate_spoolman_errors():
                # Registered before the write, or Spoolman refuses the keys.
                for name in new_extra:
                    await client.ensure_extra_field(name)
                spool = await client.merge_spool_extra(spool["id"], new_extra)
        except HTTPException:
            logger.warning("Failed to persist slicer_filament/color_name for spool %s", spool.get("id"))

    if default is not None:
        try:
            if await profile_defaults.fill_spoolman_rows(db, int(spool["id"]), default):
                await db.commit()
        except Exception as exc:  # noqa: BLE001 — the spool is created; its profiles can be set by hand
            await db.rollback()
            logger.warning("Profile default for Spoolman spool %s not applied: %s", spool.get("id"), exc)
    return spool


@router.post("/spools")
async def create_spool(
    data: SpoolmanInventoryCreate,
    db: AsyncSession = Depends(get_db, scope="function"),
    _: User | None = RequirePermissionIfAuthEnabled(Permission.INVENTORY_UPDATE),
) -> dict:
    """Create a new spool in Spoolman, auto-creating vendor and filament as needed."""
    client = await _get_client(db)
    filament_id = await _resolve_filament_id(data, client)

    storage_location = data.storage_location
    if "location_id" in data.model_fields_set or "storage_location" in data.model_fields_set:
        try:
            storage_location, _ = await resolve_spoolman_location_string(
                db,
                location_id=data.location_id,
                storage_location=data.storage_location,
                fields_set=set(data.model_fields_set),
            )
        except ValueError as exc:
            raise HTTPException(status_code=400, detail=str(exc)) from exc

    remaining = max(0.0, data.label_weight - data.weight_used)
    try:
        async with _translate_spoolman_errors():
            spool = await client.create_spool(
                filament_id=filament_id,
                remaining_weight=remaining,
                comment=data.note or None,
                location=storage_location or None,
                spool_weight=(data.core_weight if "core_weight" in data.model_fields_set else None),
                # Only a label weight the caller gave: left off, Spoolman uses
                # the filament's weight, which is the right size for a caller
                # that didn't say (#3194).
                initial_weight=(float(data.label_weight) if "label_weight" in data.model_fields_set else None),
            )
    except HTTPException as exc:
        if exc.status_code == 404 and data.spoolman_filament_id is not None:
            raise HTTPException(
                status_code=404,
                detail=f"Filament {data.spoolman_filament_id} not found in Spoolman",
            ) from exc
        raise

    spool, price_warnings = await _apply_price_if_set(client, spool, data.cost_per_kg)

    spool = await _finish_created_spool(db, client, spool, data, await _resolve_create_default(db, data, spool))

    result = _map_spoolman_spool(spool)
    await ws_manager.broadcast({"type": "inventory_changed"})
    if price_warnings:
        return JSONResponse(status_code=207, content={**result, "warnings": price_warnings})
    return result


@router.post("/spools/bulk")
async def bulk_create_spools(
    payload: SpoolmanInventoryBulkCreate,
    db: AsyncSession = Depends(get_db, scope="function"),
    _: User | None = RequirePermissionIfAuthEnabled(Permission.INVENTORY_UPDATE),
) -> Response:
    """Create multiple identical spools in Spoolman."""
    client = await _get_client(db)
    data = payload.spool

    try:
        filament_id = await _resolve_filament_id(data, client)
    except HTTPException as exc:
        if exc.status_code == 404 and data.spoolman_filament_id is not None:
            raise HTTPException(
                status_code=404,
                detail=f"Filament {data.spoolman_filament_id} not found in Spoolman",
            ) from exc
        raise

    storage_location = data.storage_location
    if "location_id" in data.model_fields_set or "storage_location" in data.model_fields_set:
        try:
            storage_location, _ = await resolve_spoolman_location_string(
                db,
                location_id=data.location_id,
                storage_location=data.storage_location,
                fields_set=set(data.model_fields_set),
            )
        except ValueError as exc:
            raise HTTPException(status_code=400, detail=str(exc)) from exc

    remaining = max(0.0, data.label_weight - data.weight_used)
    created: list[dict] = []
    failures: list[str] = []
    # Every copy is the same product: one lookup for the batch, on the first.
    default: profile_defaults.ResolvedProfiles | None = None
    resolved_default = False
    for _ in range(payload.quantity):
        try:
            spool = await client.create_spool(
                filament_id=filament_id,
                remaining_weight=remaining,
                comment=data.note or None,
                location=storage_location or None,
                spool_weight=(data.core_weight if "core_weight" in data.model_fields_set else None),
                # Only a label weight the caller gave: left off, Spoolman uses
                # the filament's weight, which is the right size for a caller
                # that didn't say (#3194).
                initial_weight=(float(data.label_weight) if "label_weight" in data.model_fields_set else None),
            )
        except (SpoolmanUnavailableError, SpoolmanClientError, SpoolmanNotFoundError) as exc:
            logger.warning("Bulk spool creation: one spool failed: %s", exc)
            failures.append("spool creation failed")
            continue
        try:
            spool, price_warnings = await _apply_price_if_set(client, spool, data.cost_per_kg)
        except HTTPException as exc:
            logger.warning(
                "Bulk spool %d: price update failed (HTTP %d); spool not added to created list",
                spool.get("id", 0),
                exc.status_code,
            )
            failures.append("spool created but price update failed")
            continue
        if price_warnings:
            logger.warning("Bulk spool %s created without price: %s", spool.get("id"), price_warnings)
        # Each copy gets the preset and colour name the single create gives,
        # and the product's profile default (#1326).
        if not resolved_default:
            default = await _resolve_create_default(db, data, spool)
            resolved_default = True
        spool = await _finish_created_spool(db, client, spool, data, default)
        created.append(_map_spoolman_spool(spool))

    if not created:
        raise HTTPException(status_code=500, detail="Failed to create any spools in Spoolman")

    await ws_manager.broadcast({"type": "inventory_changed"})

    if len(created) < payload.quantity:
        # Some spool creations failed — return 207 Multi-Status so the caller
        # can distinguish a full success from a partial one and show a useful message.
        return JSONResponse(
            status_code=207,
            content={
                "created": created,
                "requested_count": payload.quantity,
                "failed_count": payload.quantity - len(created),
                "failures": failures,
            },
        )

    return JSONResponse(status_code=200, content=created)


@router.patch("/spools/{spool_id}")
async def update_spool(
    *,
    spool_id: int = Path(..., gt=0),
    data: SpoolmanInventoryUpdate,
    db: AsyncSession = Depends(get_db, scope="function"),
    _: User | None = RequirePermissionIfAuthEnabled(Permission.INVENTORY_UPDATE),
) -> dict:
    """Update an existing Spoolman spool, re-linking the filament if metadata changed."""
    client = await _get_client(db)

    async with _translate_spoolman_errors():
        current = await client.get_spool(spool_id)

    cur_filament: dict = current.get("filament") or {}
    cur_vendor: dict = cur_filament.get("vendor") or {}
    cur_mat: str = (cur_filament.get("material") or "").strip()
    cur_name: str = (cur_filament.get("name") or "").strip()
    if cur_mat and cur_name.upper().startswith(cur_mat.upper()):
        cur_subtype: str = cur_name[len(cur_mat) :].strip()
    else:
        cur_subtype = cur_name

    # Resolve final values: use request value if provided, else keep current
    material = data.material if data.material is not None else cur_mat
    subtype = data.subtype if data.subtype is not None else cur_subtype
    brand = data.brand if data.brand is not None else (cur_vendor.get("name") or None)
    # color_name uses model_fields_set so explicit null (clear) is distinguishable
    # from "field omitted" (don't touch). find_or_create_filament's convention:
    # None = don't touch, "" = explicit clear, "value" = set.
    if "color_name" in data.model_fields_set:
        color_name = data.color_name if data.color_name is not None else ""
    else:
        color_name = cur_filament.get("color_name") or None
    cur_color = (cur_filament.get("color_hex") or "808080").upper().removeprefix("#")
    # Handed over as stored. The opaque alpha this used to append was folded
    # straight back off by `spoolman_color_hex` below, so the two paths landed on
    # the same string and the append only obscured which shape was in hand (#2912).
    rgba = data.rgba if data.rgba is not None else cur_color
    # The label weight is the spool's own net weight (initial_weight), not the
    # filament's: one filament can have spools of different sizes (#3194).
    # The form shows it in whole grams, so it only counts as changed when it
    # differs from that; the exact stored value is otherwise left alone. A
    # spool with no size at either level takes the one it is given: Spoolman
    # refuses a remaining_weight until the spool has one.
    cur_net_weight = spoolman_net_weight(current)
    cur_has_size = cur_net_weight is not None and cur_net_weight > 0
    cur_label_weight = _safe_int(cur_net_weight, 1000) or 1000
    label_weight = data.label_weight if data.label_weight is not None else cur_label_weight
    label_weight_changed = data.label_weight is not None and (not cur_has_size or data.label_weight != cur_label_weight)
    # Default weight_used from the synthetic mapping (label - remaining) so an
    # edit that doesn't touch the weight field preserves Spoolman's real
    # remaining_weight after a "Reset usage to 0" — the previous code read
    # Spoolman's used_weight directly, which is 0 post-reset, so
    # `remaining = label - 0 = 1000` would overwrite the real remaining
    # the next time the user edited any other field (#1390).
    cur_remaining_raw = current.get("remaining_weight")
    if cur_remaining_raw is not None:
        synthetic_used = max(0.0, float(label_weight) - float(cur_remaining_raw))
    else:
        synthetic_used = float(current.get("used_weight") or 0)
    weight_used = data.weight_used if data.weight_used is not None else synthetic_used
    note = data.note if data.note is not None else current.get("comment")
    storage_location_changed = "storage_location" in data.model_fields_set or "location_id" in data.model_fields_set
    storage_location = data.storage_location if "storage_location" in data.model_fields_set else None
    if storage_location_changed:
        try:
            storage_location, _ = await resolve_spoolman_location_string(
                db,
                location_id=data.location_id,
                storage_location=storage_location,
                fields_set=set(data.model_fields_set),
            )
        except ValueError as exc:
            raise HTTPException(status_code=400, detail=str(exc)) from exc

    color_hex = spoolman_color_hex(rgba) or rgba

    # Resolve which filament this spool should be linked to AFTER the edit.
    #
    # The old behaviour was always `find_or_create_filament`, which proliferated
    # duplicate Spoolman filaments whenever the user changed any field that
    # made up the match key (material/subtype/brand/color) — every edit minted
    # a fresh row and orphaned the previous one (#1357 follow-up). To match
    # internal-mode behaviour ([[feedback_inventory_modes_parity]]: editing a
    # spool does not proliferate new entities), prefer PATCHing the current
    # filament in place when it's a singleton.
    cur_filament_id = cur_filament.get("id")
    desired_name = f"{material} {subtype}".strip() if subtype else material
    # Compare the stored shapes, not raw strings and not bare RGB prefixes. Raw
    # strings make an opaque spool's six characters differ from an incoming eight
    # and PATCH the filament on every no-op edit; bare prefixes make an
    # alpha-only edit invisible so the change never lands (#2912).
    cur_color_norm = spoolman_color_hex(cur_filament.get("color_hex")) or ""
    cur_vendor_name = (cur_vendor.get("name") or "").strip()
    # The label weight is not part of this: it is written to the spool's
    # initial_weight below. Comparing it against the filament's weight PATCHed
    # the filament for every spool of another size, or on a shared filament
    # relinked the spool to a new duplicate one (#3194).
    metadata_unchanged = (
        cur_filament_id
        and (cur_filament.get("name") or "").strip() == desired_name
        and (cur_filament.get("material") or "").upper() == material.upper()
        and cur_color_norm == (color_hex or "").upper()
        and cur_vendor_name.lower() == ((brand or "").strip().lower())
    )

    if metadata_unchanged:
        # No filament-side change at all — re-use the existing link, skip
        # find_or_create entirely so a no-op edit (e.g. just changing
        # weight_used or note) never even touches the filament catalogue.
        filament_id = cur_filament_id
    else:
        async with _translate_spoolman_errors():
            shared = await client.is_filament_shared(cur_filament_id, spool_id) if cur_filament_id else False
        if cur_filament_id and not shared:
            # Singleton filament — PATCH it in place so the user's edit lands
            # on the row their spool already points at instead of orphaning it.
            patch_body: dict = {
                "name": desired_name,
                "material": material,
                "color_hex": color_hex,
            }
            if brand:
                vendor_id = await client.find_or_create_vendor(brand)
                patch_body["vendor_id"] = vendor_id
            async with _translate_spoolman_errors():
                await client.patch_filament(cur_filament_id, patch_body)
            filament_id = cur_filament_id
        else:
            # Filament is shared with other spools — PATCHing it in place would
            # silently rewrite their metadata too. Fall back to find-or-create
            # so only this spool's link moves.
            async with _translate_spoolman_errors():
                filament_id = await client.find_or_create_filament(
                    material=material,
                    subtype=subtype or "",
                    brand=brand,
                    color_hex=color_hex,
                    label_weight=label_weight,
                    color_name=color_name,
                )
    if not filament_id:
        raise HTTPException(status_code=500, detail="Failed to find or create filament in Spoolman")

    remaining = max(0.0, label_weight - weight_used)

    # The spool's size lives on the spool (initial_weight), never the filament
    # (#3194). A relink also sends the current size: Spoolman stamps the new
    # filament's weight on a spool without an initial_weight of its own.
    new_initial_weight: float | None = None
    if label_weight_changed:
        new_initial_weight = float(label_weight)
    elif filament_id != cur_filament_id and cur_has_size:
        new_initial_weight = cur_net_weight

    # Spoolman's price is per spool, the form's per kg (#3194). An unchanged
    # rate on an unchanged size is not written, so a no-op edit cannot move the
    # price by rounding. On a resize the rate is kept, as in internal mode.
    cur_cost_per_kg = spoolman_cost_per_kg(current)
    cost_per_kg = data.cost_per_kg if data.cost_per_kg is not None else cur_cost_per_kg
    new_price: float | None = None
    if cost_per_kg is not None and (label_weight_changed or cost_per_kg != cur_cost_per_kg):
        price_weight = float(label_weight) if label_weight_changed else spoolman_price_weight(current)
        new_price = cost_per_kg_to_spoolman_price(cost_per_kg, price_weight)

    # Tag removal: clear only the "tag" key so other custom Spoolman extra fields
    # set outside Bambuddy are preserved.
    tag_nulled = (
        ("tag_uid" in data.model_fields_set or "tray_uuid" in data.model_fields_set)
        and _tag_cleared(data.tag_uid)
        and _tag_cleared(data.tray_uuid)
    )

    # Serialise tag-clear + PATCH under the per-spool extra lock to prevent a
    # concurrent merge_spool_extra call (e.g. NFC write-back) from overwriting
    # the tag key between our read and our write.
    #
    # Spoolman PATCHes extra dicts by MERGING — popping "tag" from a re-fetched
    # dict and sending the rest doesn't clear the key (Spoolman keeps the old
    # value because the key wasn't in the payload). Explicitly set the tag to
    # a JSON-encoded empty string; read-side filters strip the quotes.
    async with client.extra_lock(spool_id):
        if tag_nulled:
            # Re-fetch inside the lock so we work with fresh extra data.
            async with _translate_spoolman_errors():
                fresh = await client.get_spool(spool_id)
            cur_extra = dict(fresh.get("extra") or {})
            cur_extra["tag"] = json.dumps("")
            extra: dict | None = cur_extra
        else:
            extra = None

        async with _translate_spoolman_errors():
            updated = await client.update_spool_full(
                spool_id=spool_id,
                filament_id=filament_id,
                # Without a size Spoolman rejects the whole PATCH (HTTP 400),
                # so a spool that still has none gets no remaining_weight.
                remaining_weight=(remaining if new_initial_weight is not None or cur_has_size else None),
                comment=note or "",
                price=new_price,
                initial_weight=new_initial_weight,
                extra=extra,
                location=storage_location or None,
                clear_location=storage_location_changed and not storage_location,
                # No model_fields_set guard here, unlike create: this schema
                # already defaults core_weight to None, and None is what
                # update_spool_full reads as "leave the tare alone". A guard
                # would be a second spelling of the same condition.
                spool_weight=data.core_weight,
            )

        # Spoolman 0.27+: the native tags go as well, or the spool would still be
        # found by every one of them after "Clear RFID Tag". Only once the update
        # went through, so a failed one leaves the spool as it was; a failure here
        # is reported, and clearing again removes what is left.
        if tag_nulled:
            async with _translate_spoolman_errors():
                await client.unlink_all_native_tags(fresh)
            updated = {**updated, "tags": []}

    # Persist BambuStudio slicer preset AND color_name under spool.extra.
    # Spoolman has no native fields for these — color_name was confirmed
    # absent from the FilamentUpdateParameters schema in 0.23.1 (#1357), so
    # writing `filament.color_name` was a silent no-op that left every
    # edit looking "not saved". They all round-trip via extra and get
    # unpacked in _map_spoolman_spool. Only writes when the request
    # explicitly set the field — passing null/omitting leaves the existing
    # extra entry untouched (write empty string to clear).
    sf_set = "slicer_filament" in data.model_fields_set
    sfn_set = "slicer_filament_name" in data.model_fields_set
    cn_set = "color_name" in data.model_fields_set
    dried_set = "last_dried_at" in data.model_fields_set
    if sf_set or sfn_set or cn_set or dried_set:
        # Ensure extra fields are registered (Spoolman rejects PATCHes with
        # unknown keys with HTTP 400). Idempotent if startup already ran this.
        if sf_set:
            await client.ensure_extra_field("bambu_slicer_filament")
        if sfn_set:
            await client.ensure_extra_field("bambu_slicer_filament_name")
        if cn_set:
            await client.ensure_extra_field("bambu_color_name")
        new_extra: dict = {}
        if sf_set:
            new_extra["bambu_slicer_filament"] = json.dumps(data.slicer_filament or "")
        if sfn_set:
            new_extra["bambu_slicer_filament_name"] = json.dumps(data.slicer_filament_name or "")
        if cn_set:
            new_extra["bambu_color_name"] = json.dumps(data.color_name or "")
        if dried_set:
            # A hand-set date is not the AMS cycle the temperature and hours
            # describe, so they are cleared with it, as in internal mode.
            dried_at = data.last_dried_at.isoformat(timespec="seconds") if data.last_dried_at else ""
            new_extra[BAMBU_LAST_DRIED_AT_KEY] = json.dumps(dried_at)
            new_extra[BAMBU_LAST_DRIED_TEMP_KEY] = json.dumps("")
            new_extra[BAMBU_LAST_DRIED_HOURS_KEY] = json.dumps("")
        async with _translate_spoolman_errors():
            updated = await client.merge_spool_extra(spool_id, new_extra)
        if sf_set or sfn_set:
            forget_spoolman_spool_own_preset(spool_id)

    await ws_manager.broadcast({"type": "inventory_changed"})
    return _map_spoolman_spool(updated)


async def _release_spool_slots(db: AsyncSession, spool_id: int, *, keep_tracked: bool) -> list[tuple[int, int, int]]:
    """Delete a spool's slot rows; returns the freed (printer, ams, tray) slots.

    ``keep_tracked`` keeps the rows of a printer with a print being tracked:
    its usage is charged at completion by the slot's spool, which only the row
    knows for a spool without a tag. ``release_archived_slot_rows`` frees them
    once that print is reported. The caller commits, then announces.
    """
    from backend.app.models.active_print_spoolman import ActivePrintSpoolman

    query = select(SpoolmanSlotAssignment).where(SpoolmanSlotAssignment.spoolman_spool_id == spool_id)
    if keep_tracked:
        query = query.where(SpoolmanSlotAssignment.printer_id.not_in(select(ActivePrintSpoolman.printer_id)))
    rows = (await db.execute(query)).scalars().all()
    freed = [(r.printer_id, r.ams_id, r.tray_id) for r in rows]
    for row in rows:
        await db.delete(row)
    return freed


async def _announce_freed_slots(freed: list[tuple[int, int, int]]) -> None:
    for printer_id, ams_id, tray_id in freed:
        await ws_manager.broadcast(
            {"type": "spool_assignment_changed", "printer_id": printer_id, "ams_id": ams_id, "tray_id": tray_id}
        )


async def _purge_local_rows_for_spool(db: AsyncSession, spool_id: int) -> None:
    """Drop the Bambuddy-side rows a deleted Spoolman spool leaves behind.

    Spoolman owns the spool; Bambuddy owns the K profiles, the filament preset
    overrides and the supplier assignments, each keyed by the remote id with no
    foreign key that could cascade. The K-profile and preset leaks are inert,
    but a leaked ``spoolman_spool_suppliers`` row keeps the supplier's
    reference count non-zero, so deleting that supplier answers 409 forever
    with no way for the user to find the phantom reference (#2988).

    The caller commits.
    """
    for model in (SpoolmanKProfile, SpoolmanFilamentPreset, SpoolmanSpoolSupplier):
        await db.execute(delete(model).where(model.spoolman_spool_id == spool_id))


@router.delete("/spools/{spool_id}")
async def delete_spool(
    spool_id: int = Path(..., gt=0),
    db: AsyncSession = Depends(get_db, scope="function"),
    _: User | None = RequirePermissionIfAuthEnabled(Permission.INVENTORY_UPDATE),
) -> dict:
    """Permanently delete a spool from Spoolman."""
    client = await _get_client(db)
    async with _translate_spoolman_errors():
        await client.delete_spool(spool_id)
    await _purge_local_rows_for_spool(db, spool_id)
    # Its slots go too: a row left behind points tracking and the scheduler
    # at an id Spoolman no longer has.
    freed = await _release_spool_slots(db, spool_id, keep_tracked=False)
    await db.commit()
    await ws_manager.broadcast({"type": "inventory_changed"})
    await _announce_freed_slots(freed)
    return {"status": "deleted"}


@router.post("/spools/{spool_id}/archive")
async def archive_spool(
    spool_id: int = Path(..., gt=0),
    db: AsyncSession = Depends(get_db, scope="function"),
    _: User | None = RequirePermissionIfAuthEnabled(Permission.INVENTORY_UPDATE),
) -> dict:
    """Archive a spool in Spoolman (soft-delete)."""
    client = await _get_client(db)
    async with _translate_spoolman_errors():
        spool = await client.set_spool_archived(spool_id, archived=True)
    # An archived spool gives its slots up, as in the built-in inventory.
    freed = await _release_spool_slots(db, spool_id, keep_tracked=True)
    await db.commit()
    await ws_manager.broadcast({"type": "inventory_changed"})
    await _announce_freed_slots(freed)
    try:
        mapped = _map_spoolman_spool(spool)
    except ValueError as exc:
        logger.warning("Malformed Spoolman spool (id=%r): %s", spool_id, exc)
        raise HTTPException(status_code=502, detail="Spoolman returned malformed spool data") from exc
    return mapped


@router.post("/spools/{spool_id}/restore")
async def restore_spool(
    spool_id: int = Path(..., gt=0),
    db: AsyncSession = Depends(get_db, scope="function"),
    _: User | None = RequirePermissionIfAuthEnabled(Permission.INVENTORY_UPDATE),
) -> dict:
    """Restore an archived spool in Spoolman."""
    client = await _get_client(db)
    async with _translate_spoolman_errors():
        spool = await client.set_spool_archived(spool_id, archived=False)
    try:
        mapped = _map_spoolman_spool(spool)
    except ValueError as exc:
        logger.warning("Malformed Spoolman spool (id=%r): %s", spool_id, exc)
        raise HTTPException(status_code=502, detail="Spoolman returned malformed spool data") from exc
    await ws_manager.broadcast({"type": "inventory_changed"})
    return mapped


@router.post("/spools/{spool_id}/reset-consumed-counter")
async def reset_spool_consumed_counter(
    spool_id: int = Path(..., gt=0),
    db: AsyncSession = Depends(get_db, scope="function"),
    _: User | None = RequirePermissionIfAuthEnabled(Permission.INVENTORY_UPDATE),
) -> dict:
    """Zero the displayed "Total Consumed" counter for a Spoolman spool.

    Spoolman has no native baseline field, so the baseline lives in
    ``spool.extra`` — the same mechanism internal mode uses for its
    ``weight_used_baseline`` column. ``_map_spoolman_spool`` folds it back in,
    so the Inventory page's ``weight_used - baseline`` reads 0 while every
    native Spoolman field is left exactly as it was.

    It used to PATCH ``used_weight = 0`` upstream, which is not the same thing:
    Spoolman recomputes ``remaining_weight`` from initial minus used, so the
    spool jumped back to full and the measured remaining filament was gone
    (#2906). Parity with the internal-mode endpoint (#1644, #1390, see also
    ``backend/app/api/routes/inventory.py::reset_spool_consumed_counter``).
    """
    client = await _get_client(db)
    async with _translate_spoolman_errors():
        spool = await client.reset_spool_consumed_counter(spool_id)
    try:
        mapped = _map_spoolman_spool(spool)
    except ValueError as exc:
        logger.warning("Malformed Spoolman spool (id=%r): %s", spool_id, exc)
        raise HTTPException(status_code=502, detail="Spoolman returned malformed spool data") from exc
    await ws_manager.broadcast({"type": "inventory_changed"})
    return mapped


class SpoolmanBulkUpdateRequest(BaseModel):
    ids: list[int] = Field(..., min_length=1, max_length=500)
    update: SpoolmanInventoryUpdate


class SpoolmanBulkIdsRequest(BaseModel):
    ids: list[int] = Field(..., min_length=1, max_length=500)


@router.post("/spools/bulk-update")
async def bulk_update_spools(
    payload: SpoolmanBulkUpdateRequest,
    db: AsyncSession = Depends(get_db, scope="function"),
    _: User | None = RequirePermissionIfAuthEnabled(Permission.INVENTORY_UPDATE),
) -> dict:
    """Apply the same partial update to every listed Spoolman spool.

    Loops the per-spool ``update_spool`` route so the filament re-linking +
    extra-dict + location-resolution rules stay in sync with the single-spool
    PATCH path. Per-spool errors are collected; one bad ID doesn't abort the
    batch.
    """
    update_fields = payload.update.model_dump(exclude_unset=True)
    if not update_fields:
        raise HTTPException(status_code=400, detail="update must include at least one field")

    updated = 0
    errors: list[dict] = []
    for sid in payload.ids:
        try:
            await update_spool(spool_id=sid, data=payload.update, db=db, _=None)
            updated += 1
        except HTTPException as exc:
            errors.append({"id": sid, "status": exc.status_code, "detail": exc.detail})
        except Exception as exc:  # noqa: BLE001 — surface unexpected failures per-row
            logger.exception("Spoolman bulk-update failed for spool %s", sid)
            errors.append({"id": sid, "status": 500, "detail": str(exc)})
    if updated:
        await ws_manager.broadcast({"type": "inventory_changed"})
    return {"updated": updated, "errors": errors}


@router.post("/spools/bulk-delete")
async def bulk_delete_spools(
    payload: SpoolmanBulkIdsRequest,
    db: AsyncSession = Depends(get_db, scope="function"),
    _: User | None = RequirePermissionIfAuthEnabled(Permission.INVENTORY_UPDATE),
) -> dict:
    """Hard-delete every listed Spoolman spool. Per-spool failures are collected."""
    client = await _get_client(db)
    deleted = 0
    freed: list[tuple[int, int, int]] = []
    errors: list[dict] = []
    for sid in payload.ids:
        try:
            async with _translate_spoolman_errors():
                await client.delete_spool(sid)
            await _purge_local_rows_for_spool(db, sid)
            freed += await _release_spool_slots(db, sid, keep_tracked=False)
            deleted += 1
        except HTTPException as exc:
            errors.append({"id": sid, "status": exc.status_code, "detail": exc.detail})
        except Exception as exc:  # noqa: BLE001 — surface unexpected failures per-row
            logger.exception("Spoolman bulk-delete failed for spool %s", sid)
            errors.append({"id": sid, "status": 500, "detail": str(exc)})
    if deleted:
        await db.commit()
        await ws_manager.broadcast({"type": "inventory_changed"})
        await _announce_freed_slots(freed)
    return {"deleted": deleted, "errors": errors}


@router.post("/spools/bulk-archive")
async def bulk_archive_spools(
    payload: SpoolmanBulkIdsRequest,
    db: AsyncSession = Depends(get_db, scope="function"),
    _: User | None = RequirePermissionIfAuthEnabled(Permission.INVENTORY_UPDATE),
) -> dict:
    """Archive every listed Spoolman spool. Per-spool failures are collected."""
    client = await _get_client(db)
    archived = 0
    freed: list[tuple[int, int, int]] = []
    errors: list[dict] = []
    for sid in payload.ids:
        try:
            async with _translate_spoolman_errors():
                await client.set_spool_archived(sid, archived=True)
            freed += await _release_spool_slots(db, sid, keep_tracked=True)
            archived += 1
        except HTTPException as exc:
            errors.append({"id": sid, "status": exc.status_code, "detail": exc.detail})
        except Exception as exc:  # noqa: BLE001 — surface unexpected failures per-row
            logger.exception("Spoolman bulk-archive failed for spool %s", sid)
            errors.append({"id": sid, "status": 500, "detail": str(exc)})
    if archived:
        await db.commit()
        await ws_manager.broadcast({"type": "inventory_changed"})
        await _announce_freed_slots(freed)
    return {"archived": archived, "errors": errors}


@router.post("/spools/bulk-restore")
async def bulk_restore_spools(
    payload: SpoolmanBulkIdsRequest,
    db: AsyncSession = Depends(get_db, scope="function"),
    _: User | None = RequirePermissionIfAuthEnabled(Permission.INVENTORY_UPDATE),
) -> dict:
    """Restore every listed archived Spoolman spool. Per-spool failures are collected."""
    client = await _get_client(db)
    restored = 0
    errors: list[dict] = []
    for sid in payload.ids:
        try:
            async with _translate_spoolman_errors():
                await client.set_spool_archived(sid, archived=False)
            restored += 1
        except HTTPException as exc:
            errors.append({"id": sid, "status": exc.status_code, "detail": exc.detail})
        except Exception as exc:  # noqa: BLE001 — surface unexpected failures per-row
            logger.exception("Spoolman bulk-restore failed for spool %s", sid)
            errors.append({"id": sid, "status": 500, "detail": str(exc)})
    if restored:
        await ws_manager.broadcast({"type": "inventory_changed"})
    return {"restored": restored, "errors": errors}


@router.post("/spools/reset-consumed-counter-bulk")
async def bulk_reset_spool_consumed_counter(
    payload: dict = Body(...),
    db: AsyncSession = Depends(get_db, scope="function"),
    _: User | None = RequirePermissionIfAuthEnabled(Permission.INVENTORY_UPDATE),
) -> dict:
    """Bulk reset the "Total Consumed" counter across the given Spoolman spool IDs.

    Caller passes an explicit list of IDs — no "reset all" shortcut, since
    a typo on a wildcard would wipe the entire inventory's tracking.
    Returns the count of spools reset and, like the other bulk actions, the
    ones Spoolman refused in ``errors``: an individual failure does not abort
    the batch, and a batch where every reset failed is not reported as a
    success (FO-C8).
    """
    spool_ids = payload.get("spool_ids")
    if not isinstance(spool_ids, list) or not spool_ids:
        raise HTTPException(status_code=400, detail="spool_ids must be a non-empty list")
    if not all(isinstance(sid, int) for sid in spool_ids):
        raise HTTPException(status_code=400, detail="spool_ids must contain integers")

    client = await _get_client(db)
    reset_count = 0
    errors: list[dict] = []
    for spool_id in spool_ids:
        try:
            async with _translate_spoolman_errors():
                await client.reset_spool_consumed_counter(spool_id)
            reset_count += 1
        except HTTPException as exc:
            logger.warning("Spoolman reset-consumed-counter failed for spool %s: %s", spool_id, exc.detail)
            errors.append({"id": spool_id, "status": exc.status_code, "detail": exc.detail})
    if reset_count:
        await ws_manager.broadcast({"type": "inventory_changed"})
    return {"reset": reset_count, "errors": errors}


@router.patch("/spools/{spool_id}/weight")
async def sync_spool_weight(
    *,
    spool_id: int = Path(..., gt=0),
    data: SpoolWeightUpdate,
    db: AsyncSession = Depends(get_db, scope="function"),
    _: User | None = RequirePermissionIfAuthEnabled(Permission.INVENTORY_UPDATE),
) -> dict:
    """Update a spool's remaining weight from a measured gross weight.

    Computes remaining = gross_weight - tare, with the tare resolved the way
    Spoolman does it (spool, filament, vendor, then 250 g; see spoolman_tare).
    """
    client = await _get_client(db)

    async with _translate_spoolman_errors():
        current = await client.get_spool(spool_id)

    core_weight, _source = spoolman_tare(current)
    remaining = max(0.0, data.weight_grams - core_weight)

    async with _translate_spoolman_errors():
        updated = await client.update_spool_full(spool_id=spool_id, remaining_weight=remaining)

    label_weight = _safe_int(spoolman_net_weight(updated), 1000)
    weight_used = max(0.0, label_weight - remaining)
    await ws_manager.broadcast({"type": "inventory_changed"})
    return {"status": "ok", "weight_used": weight_used}


def _extra_tag(spool: dict) -> str:
    """The tag stored in a Spoolman spool's ``extra``, normalised for comparison.

    Anything that is not a string reads as no tag. ``extra`` is free-form and
    edited outside Bambuddy, and ``.get("tag", "")`` does not default a key
    that is present and null -- that returns None, which has no ``.strip``.
    """
    extra = spool.get("extra")
    raw = extra.get("tag") if isinstance(extra, dict) else None
    return raw.strip('"').upper() if isinstance(raw, str) else ""


@router.patch("/spools/{spool_id}/tag")
async def link_tag_to_spoolman_spool(
    *,
    spool_id: int = Path(..., gt=0),
    data: SpoolTagLinkRequest,
    db: AsyncSession = Depends(get_db, scope="function"),
    _: User | None = RequirePermissionIfAuthEnabled(Permission.INVENTORY_UPDATE),
) -> dict:
    """Write an NFC tag UID or Bambu tray UUID into Spoolman's extra.tag for a spool.

    tray_uuid takes precedence over tag_uid when both are supplied.
    Uses extra_lock to serialise against concurrent extra-field writes.

    A tag another active spool already carries is refused with the shared
    ``tag_already_linked`` 409, identical to the built-in route's (#3110).
    """
    client = await _get_client(db)
    tag = (data.tray_uuid or data.tag_uid).upper()
    tag_json = json.dumps(tag)

    async with client.extra_lock(spool_id):
        # Duplicate check: scan all spools for the same tag on a different spool.
        # Sorted, because Spoolman has no unique constraint on extra.tag either,
        # and a caller offered whichever row the scan happened to reach first
        # could not tell two holders apart. The built-in route names the lowest
        # id for the same reason (#3110).
        #
        # Sorting means every row is read, where the old loop stopped at its
        # first match, so one malformed row after the holder must not be able
        # to take the whole request down: _extra_tag refuses a non-string, and
        # a row without an integer id cannot be named and so is not treated as
        # a holder. Bambuddy only ever writes a JSON string here; a third party
        # editing extra.tag in Spoolman is what puts anything else in reach.
        async with _translate_spoolman_errors():
            all_spools = await client.get_all_spools()
        holders = sorted(
            (s for s in all_spools if _extra_tag(s) == tag and isinstance(s.get("id"), int) and s["id"] != spool_id),
            key=lambda s: s["id"],
        )
        if holders:
            raise tag_already_linked("tray_uuid" if data.tray_uuid else "tag_uid", holders[0]["id"])

        # Re-fetch inside the lock so cur_extra reflects any concurrent update.
        async with _translate_spoolman_errors():
            current = await client.get_spool(spool_id)

        # Spoolman 0.27+: both identifiers become native tags of the spool, added to
        # what it already carries rather than replacing it. Spoolman refuses a UID
        # another spool holds with a 409 naming that spool, the same answer as above.
        # Whatever this request added is taken back if anything after it fails, a
        # refused second identifier or the extra.tag write, so the spool is left as
        # it was found.
        added: list[str] = []
        try:
            if await client.has_tag_api():
                had = {t.get("uid") for t in current.get("tags") or []}
                for field, uid in (("tray_uuid", data.tray_uuid), ("tag_uid", data.tag_uid)):
                    if not uid:
                        continue
                    async with _translate_spoolman_errors():
                        holder = await client.claim_native_tag(
                            spool_id, uid.upper(), "bambu" if data.tray_uuid else None
                        )
                    if holder is not None and holder != spool_id:
                        if holder < 0:
                            # A filament or a location holds it: no spool to name or move it from.
                            raise HTTPException(
                                status_code=409,
                                detail={
                                    "code": "tag_linked_elsewhere",
                                    "message": f"{field} is linked to a filament or location in Spoolman",
                                    "field": field,
                                },
                            )
                        raise tag_already_linked(field, holder)
                    if uid.upper() not in had:
                        added.append(uid.upper())

            cur_extra = dict(current.get("extra") or {})
            cur_extra["tag"] = tag_json
            async with _translate_spoolman_errors():
                updated = await client.update_spool_full(spool_id=spool_id, extra=cur_extra)
        except HTTPException:
            for uid_added in added:
                try:
                    await client.unlink_native_tag(spool_id, uid_added)
                except (SpoolmanClientError, SpoolmanUnavailableError) as exc:
                    logger.warning("Could not take back native tag %s from spool %s: %s", uid_added, spool_id, exc)
            raise

    logger.info("Linked tag %s to Spoolman spool %s", tag, spool_id)
    await ws_manager.broadcast({"type": "inventory_changed"})
    return _map_spoolman_spool(updated)


_HEX_TAG_RE = re.compile(r"^[0-9A-F]{8,64}$")
_AMS_CHIP_UID_PADDING = "00000100"


@router.post("/tags/migrate")
async def migrate_tags_to_native(
    dry_run: bool = Query(True, description="Only count what would move, change nothing"),
    db: AsyncSession = Depends(get_db, scope="function"),
    _: User | None = RequirePermissionIfAuthEnabled(Permission.INVENTORY_UPDATE),
) -> dict:
    """Move tags from Spoolman's extra.tag into its native tags (Spoolman 0.27+).

    extra.tag itself is left as it is, so older readers keep working and a
    downgrade loses nothing. A slot's fallback ID is not a physical tag and
    stays behind. Archived spools are skipped, and a tag an archived spool still
    holds moves to the active one. A tag Spoolman holds on another active spool
    is reported as a conflict for the user to settle, never moved by guessing.
    """
    client = await _get_client(db)
    if not await client.has_tag_api():
        raise HTTPException(
            status_code=409,
            detail={"code": "spoolman_without_tags", "message": "Spoolman 0.27 or later is required"},
        )
    serials = (await db.execute(select(Printer.serial_number))).scalars().all()
    async with _translate_spoolman_errors():
        spools = await client.get_all_spools(allow_archived=True)

    # Who holds which native tag, from the listing: the dry run sees the same
    # conflicts the real one would, and two spools sharing one extra.tag are
    # caught on the second, with no request spent on either. Archived spools are
    # listed only for this: nothing looks an archived spool up, so they get no
    # tags, and one that still holds a tag gives it up to the active spool.
    archived = {spool["id"] for spool in spools if spool.get("archived")}
    native_holder = {t.get("uid"): spool["id"] for spool in spools for t in spool.get("tags") or []}
    report: dict = {"dry_run": dry_run, "moved": [], "already": 0, "slot_ids": 0, "conflicts": []}
    for spool in spools:
        if spool["id"] in archived:
            continue
        tag = _extra_tag(spool)
        if not _HEX_TAG_RE.match(tag) or set(tag) == {"0"}:
            continue
        if is_slot_fallback_tag(tag, serials):
            report["slot_ids"] += 1
            continue
        # The AMS pads a Bambu chip's 4-byte UID to 8 bytes ("D3E68F32" arrives as
        # "D3E68F3200000100"). The native tag is the chip's own UID, as the AMS
        # sync stores it, so it matches what a reader sees.
        bambu = len(tag) == 32
        if len(tag) == 16 and tag.endswith(_AMS_CHIP_UID_PADDING):
            tag, bambu = tag[:8], True
        holder = native_holder.get(tag)
        if holder == spool["id"]:
            report["already"] += 1
            continue
        if holder in archived:
            holder = None
        if holder is None and not dry_run:
            async with _translate_spoolman_errors():
                holder = await client.claim_native_tag(spool["id"], tag, "bambu" if bambu else None)
        if holder is not None:
            report["conflicts"].append({"spool_id": spool["id"], "tag": tag, "holder": holder})
            continue
        native_holder[tag] = spool["id"]
        report["moved"].append(spool["id"])

    logger.info(
        "Native tag migration (dry_run=%s): %d moved, %d already, %d slot ids, %d conflicts",
        dry_run,
        len(report["moved"]),
        report["already"],
        report["slot_ids"],
        len(report["conflicts"]),
    )
    if not dry_run and report["moved"]:
        await ws_manager.broadcast({"type": "inventory_changed"})
    return report


@router.get("/slot-assignments/all", response_model=list[SpoolmanSlotAssignmentEnriched])
async def get_all_spoolman_slot_assignments(
    printer_id: int | None = Query(None, gt=0),
    db: AsyncSession = Depends(get_db, scope="function"),
    _: User | None = RequirePermissionIfAuthEnabled(Permission.INVENTORY_READ),
    printer_scope: PrinterScope = RequestPrinterScope,
) -> list[SpoolmanSlotAssignmentEnriched]:
    """Return all Spoolman slot assignments enriched with printer_name and ams_label.

    ``printer_name`` is null only when the printer relation is missing
    (cascade-deleted edge case). ``ams_label`` is null when no AmsLabel row
    matches the slot's MQTT serial (or the synthetic ``f"p{pid}a{ams_id}"``
    fallback key).
    """
    query = select(SpoolmanSlotAssignment).options(selectinload(SpoolmanSlotAssignment.printer))
    if printer_id is not None:
        query = query.where(SpoolmanSlotAssignment.printer_id == printer_id)
    if (clause := printer_scope.where_strict(SpoolmanSlotAssignment.printer_id)) is not None:
        query = query.where(clause)
    result = await db.execute(query)
    slots = list(result.scalars().all())

    # Build (printer_id, ams_id) -> ams_serial map from live printer states.
    # Same pattern as inventory.py:765-806 for the local /assignments endpoint.
    printer_ids = {s.printer_id for s in slots}
    serial_map: dict[tuple[int, int], str] = {}
    all_statuses = printer_manager.get_all_statuses()
    for pid in printer_ids:
        state = all_statuses.get(pid)
        if not (state and state.raw_data):
            continue
        # Some printer firmware variants wrap the AMS list in an outer dict
        # (`{"ams": [...]}`). Mirror the defense used in sync_spoolman_ams_weights
        # (line 842-844) so a wrapped payload still resolves to a list.
        ams_raw = state.raw_data.get("ams", [])
        if isinstance(ams_raw, dict):
            ams_raw = ams_raw.get("ams", [])
        if not isinstance(ams_raw, list):
            continue
        for ams_unit in ams_raw:
            if not isinstance(ams_unit, dict):
                continue
            sn = str(ams_unit.get("sn") or ams_unit.get("serial_number") or "")
            if not sn:
                continue
            try:
                serial_map[(pid, int(ams_unit.get("id", 0)))] = sn
            except (ValueError, TypeError):
                continue

    # Add the synthetic position key (f"p{pid}a{ams_id}"): a label saved before
    # the unit reported its serial (the AMS Lite before #3348) is still under
    # one, and is shown until it is saved again, as on the printer card.
    all_serials: set[str] = set(serial_map.values())
    for s in slots:
        all_serials.add(f"p{s.printer_id}a{s.ams_id}")

    label_by_serial: dict[str, str] = {}
    if all_serials:
        lbl_result = await db.execute(select(AmsLabel).where(AmsLabel.ams_serial_number.in_(all_serials)))
        for lbl in lbl_result.scalars().all():
            label_by_serial[lbl.ams_serial_number] = lbl.label

    def _ams_label_for(pid: int, ams_id: int) -> str | None:
        sn = serial_map.get((pid, ams_id))
        if sn and sn in label_by_serial:
            return label_by_serial[sn]
        return label_by_serial.get(f"p{pid}a{ams_id}")

    enriched: list[SpoolmanSlotAssignmentEnriched] = []
    for s in slots:
        if s.printer is None:
            # FK is ondelete=CASCADE so this should be unreachable in normal
            # operation; surface it loudly if a stale row ever appears.
            logger.warning(
                "Orphaned Spoolman slot assignment: printer_id=%d (ams=%d, tray=%d, spoolman_spool_id=%d) has no Printer row",
                s.printer_id,
                s.ams_id,
                s.tray_id,
                s.spoolman_spool_id,
            )
        enriched.append(
            SpoolmanSlotAssignmentEnriched(
                printer_id=s.printer_id,
                printer_name=s.printer.name if s.printer else None,
                ams_id=s.ams_id,
                tray_id=s.tray_id,
                spoolman_spool_id=s.spoolman_spool_id,
                ams_label=_ams_label_for(s.printer_id, s.ams_id),
            )
        )
    return enriched


@router.post("/sync-ams-weights")
async def sync_spoolman_ams_weights(
    db: AsyncSession = Depends(get_db, scope="function"),
    _: User | None = RequirePermissionIfAuthEnabled(Permission.INVENTORY_UPDATE),
    printer_scope: PrinterScope = RequestPrinterScope,
):
    """Sync remaining weight back to Spoolman for all slot-assigned spools.

    Reads live AMS remain% from connected printers, computes
    remaining = label_weight * remain% / 100, and PATCHes Spoolman.
    """
    client = await _get_client(db)

    # Fetch all non-archived Spoolman spools once for label_weight lookup
    async with _translate_spoolman_errors():
        raw_spools = await client.get_all_spools(allow_archived=False)
    spool_lookup: dict[int, dict] = {s["id"]: s for s in raw_spools if s.get("id") is not None}

    result = await db.execute(select(SpoolmanSlotAssignment))
    # Only slots on printers the caller may see (#1727)
    assignments = [a for a in result.scalars().all() if printer_scope.allows(a.printer_id)]

    from backend.app.services.usage_tracker import _active_sessions

    synced = 0
    skipped = 0
    # A printing printer is left alone: its print's usage is charged when it
    # completes, on top of whatever remaining weight the spool has by then.
    printing = 0

    def _find_tray(ams_data: list, ams_id: int, tray_id: int) -> dict | None:
        if not ams_data:
            return None
        for ams_unit in ams_data:
            if _safe_int(ams_unit.get("id"), -1) != ams_id:
                continue
            for tray in ams_unit.get("tray", []):
                if _safe_int(tray.get("id"), -1) == tray_id:
                    return tray
        return None

    for assignment in assignments:
        spool_dict = spool_lookup.get(assignment.spoolman_spool_id)
        if not spool_dict:
            logger.debug("Spoolman AMS sync: spool %d not found in Spoolman, skipping", assignment.spoolman_spool_id)
            skipped += 1
            continue

        label_weight = _safe_int(spoolman_net_weight(spool_dict), 1000)
        if label_weight <= 0:
            logger.debug("Spoolman AMS sync: spool %d has no label_weight, skipping", assignment.spoolman_spool_id)
            skipped += 1
            continue

        state = printer_manager.get_status(assignment.printer_id)
        # A disconnected printer keeps its last report, which is not a reading.
        if not state or not state.raw_data or getattr(state, "connected", True) is False:
            logger.info(
                "Spoolman AMS sync: printer %d not connected, skipping spool %d",
                assignment.printer_id,
                assignment.spoolman_spool_id,
            )
            skipped += 1
            continue

        # Until completion has charged the print its session stays open, also
        # after the printer has reported the print finished.
        if (
            str(getattr(state, "state", "") or "").upper() in PrinterManager.ACTIVE_PRINT_STATES
            or assignment.printer_id in _active_sessions
        ):
            logger.info(
                "Spoolman AMS sync: printer %d is printing, skipping spool %d",
                assignment.printer_id,
                assignment.spoolman_spool_id,
            )
            skipped += 1
            printing += 1
            continue

        ams_raw = state.raw_data.get("ams", [])
        if isinstance(ams_raw, dict):
            ams_raw = ams_raw.get("ams", [])
        tray = _find_tray(ams_raw, assignment.ams_id, assignment.tray_id)
        if not tray:
            logger.info(
                "Spoolman AMS sync: no tray data for spool %d (printer %d AMS%d-T%d)",
                assignment.spoolman_spool_id,
                assignment.printer_id,
                assignment.ams_id,
                assignment.tray_id,
            )
            skipped += 1
            continue

        remain_raw = tray.get("remain")
        if remain_raw is None:
            logger.debug(
                "Spoolman AMS sync: no remain value for spool %d (tray %d/%d), skipping",
                assignment.spoolman_spool_id,
                assignment.ams_id,
                assignment.tray_id,
            )
            skipped += 1
            continue

        try:
            remain_val = int(remain_raw)
        except (TypeError, ValueError):
            logger.debug(
                "Spoolman AMS sync: non-numeric remain=%r for spool %d, skipping",
                remain_raw,
                assignment.spoolman_spool_id,
            )
            skipped += 1
            continue

        if remain_val < 0 or remain_val > 100:
            logger.debug("Spoolman AMS sync: invalid remain=%s for spool %d", remain_raw, assignment.spoolman_spool_id)
            skipped += 1
            continue

        remaining = round(label_weight * remain_val / 100.0, 1)
        try:
            async with _translate_spoolman_errors():
                await client.update_spool_full(assignment.spoolman_spool_id, remaining_weight=remaining)
            logger.info(
                "Spoolman AMS sync: spool %d remaining set to %s g (remain=%d%%)",
                assignment.spoolman_spool_id,
                remaining,
                remain_val,
            )
            synced += 1
        except HTTPException as exc:
            if exc.status_code == 404:
                logger.warning(
                    "Spoolman AMS sync: spool %d not found in Spoolman (404), skipping",
                    assignment.spoolman_spool_id,
                )
            else:
                logger.warning(
                    "Spoolman AMS sync: failed to update spool %d (HTTP %d)",
                    assignment.spoolman_spool_id,
                    exc.status_code,
                )
            skipped += 1

    return {"synced": synced, "skipped": skipped, "printing": printing}


@router.post("/slot-assignments")
async def assign_spoolman_slot(
    body: SpoolSlotAssignmentRequest,
    db: AsyncSession = Depends(get_db, scope="function"),
    current_user: User | None = RequirePermissionIfAuthEnabled(Permission.INVENTORY_UPDATE),
    printer_scope: PrinterScope = RequestPrinterScope,
) -> dict:
    """Assign a Spoolman spool to a printer AMS slot (stored in local DB only).

    Raises 404 if the printer does not exist or the spool is not found in Spoolman.
    Spoolman's own ``spool.location`` field is NOT touched — it is user-managed.
    """

    client = await _get_client(db)
    printer_scope.ensure(body.printer_id)
    result = await db.execute(select(Printer).where(Printer.id == body.printer_id))
    printer = result.scalar_one_or_none()
    if not printer:
        raise HTTPException(status_code=404, detail="Printer not found")

    # Verify the Spoolman spool exists before committing to local DB.
    # This prevents ghost rows pointing at non-existent spool IDs.
    async with _translate_spoolman_errors():
        spool = await client.get_spool(body.spoolman_spool_id)
    # As the built-in inventory refuses it: an archived spool on a slot would
    # be charged the next print's usage (SSA-8).
    if spool.get("archived"):
        raise HTTPException(status_code=400, detail="Cannot assign an archived spool")

    # A move (#1574) takes the spool out of every other slot it holds, in the
    # same commit as the upsert below, so a failed move leaves it where it was
    # instead of in no slot. Its fallback tag for a slot it leaves is cleared
    # first. Moves of one spool take turns (get_spool_move_lock): on PostgreSQL
    # a move's DELETE doesn't see the slot another move inserts and commits
    # after it began, and both left the spool assigned. A slot on a printer
    # outside the caller's scope stays assigned, as unassigning leaves it
    # (#1727).
    target = (body.printer_id, body.ams_id, body.tray_id)
    slot_columns = (SpoolmanSlotAssignment.printer_id, SpoolmanSlotAssignment.ams_id, SpoolmanSlotAssignment.tray_id)
    others = [SpoolmanSlotAssignment.spoolman_spool_id == body.spoolman_spool_id, tuple_(*slot_columns) != target]
    if (in_scope := printer_scope.where_strict(SpoolmanSlotAssignment.printer_id)) is not None:
        others.append(in_scope)
    vacated: list[tuple[int, int, int]] = []
    cleared_tag: str | None = None
    async with get_spool_move_lock("spoolman", body.spoolman_spool_id) if body.move else nullcontext():
        if body.move:
            vacated = list((await db.execute(select(*slot_columns).where(*others))).tuples())
            cleared_tag = await _clear_vacated_slot_fallback_tag(
                client, db, spool_id=body.spoolman_spool_id, spool=spool, vacated=vacated
            )

        # Spool confirmed in Spoolman — upsert into local slot-assignment table
        # assigned_at is intentionally not refreshed on re-assign (original timestamp preserved)
        try:
            if body.move:
                # Every other slot of the spool, not only those read above: a
                # link or the AMS sync may have given it one more meanwhile.
                result = await db.execute(delete(SpoolmanSlotAssignment).where(*others).returning(*slot_columns))
                vacated = list(result.tuples())
            await db.execute(
                text(
                    "INSERT INTO spoolman_slot_assignments"
                    " (printer_id, ams_id, tray_id, spoolman_spool_id)"
                    " VALUES (:printer_id, :ams_id, :tray_id, :spool_id)"
                    " ON CONFLICT(printer_id, ams_id, tray_id)"
                    " DO UPDATE SET spoolman_spool_id = excluded.spoolman_spool_id"
                ),
                {
                    "printer_id": body.printer_id,
                    "ams_id": body.ams_id,
                    "tray_id": body.tray_id,
                    "spool_id": body.spoolman_spool_id,
                },
            )
            await spool_swaps.record_assignment(
                db, body.printer_id, body.ams_id, body.tray_id, body.spoolman_spool_id, spool_swaps.SPOOLMAN
            )
            await db.commit()
            slot_unlink_grace.forget_slot(body.printer_id, body.ams_id, body.tray_id)
        except Exception as exc:
            await db.rollback()
            logger.error("Failed to persist slot assignment: %s", exc)
            if cleared_tag:
                # The spool stays in its old slot, so it gets that slot's tag back.
                try:
                    await client.merge_spool_extra(body.spoolman_spool_id, {"tag": json.dumps(cleared_tag)})
                except (SpoolmanClientError, SpoolmanUnavailableError, SpoolmanNotFoundError) as restore_exc:
                    logger.warning(
                        "Could not put back the fallback tag of Spoolman spool %s: %s",
                        body.spoolman_spool_id,
                        restore_exc,
                    )
            raise HTTPException(status_code=500, detail="Failed to save slot assignment") from exc
    if vacated:
        logger.info(
            "Moved Spoolman spool %d to printer %d AMS%d-T%d from %s",
            body.spoolman_spool_id,
            body.printer_id,
            body.ams_id,
            body.tray_id,
            ", ".join(f"printer {p} AMS{a}-T{t}" for p, a, t in vacated),
        )
    # Other tabs and the SpoolBuddy kiosk follow slot changes over the
    # websocket, as the built-in assign announces them: the slots the spool
    # left, and the one it went to.
    await _announce_freed_slots([*vacated, target])

    # #1457: clear stale fallback-tag links on OTHER spools still bound to this
    # slot. Without this, a non-RFID slot's deterministic fallback tag stays
    # attached to the previous spool in Spoolman's extra.tag and re-surfaces in
    # the hover card whenever the local slot assignment is removed.
    if printer.serial_number:
        await _clear_stale_slot_fallback_tag_links(
            client,
            printer_serial=printer.serial_number,
            ams_id=body.ams_id,
            tray_id=body.tray_id,
            keep_spool_id=body.spoolman_spool_id,
        )

    mapped = _map_spoolman_spool(spool)

    # Fetch K-profiles before the MQTT try block so we can use async DB access.
    kp_rows_result = await db.execute(
        select(SpoolmanKProfile).where(
            SpoolmanKProfile.spoolman_spool_id == body.spoolman_spool_id,
            SpoolmanKProfile.printer_id == body.printer_id,
        )
    )
    kp_rows = kp_rows_result.scalars().all()

    # Auto-configure AMS slot via MQTT (best-effort; slot assignment is already persisted)
    try:
        mqtt_client = printer_manager.get_client(body.printer_id)
        if mqtt_client:
            # Spoolman's material is free text, so it arrives as whatever the
            # user typed there -- "PLA+", "PolyTerra PLA". The sub-brand keeps
            # that wording; the slot's type has to be one the printer and the
            # slicer know (issue #2902).
            material = mapped.get("material") or ""
            tray_type = printer_filament_type(material)
            brand = mapped.get("brand") or ""
            subtype = mapped.get("subtype") or ""
            if brand:
                tray_sub_brands = f"{brand} {material} {subtype}".strip()
            elif subtype:
                tray_sub_brands = f"{material} {subtype}".strip()
            else:
                tray_sub_brands = material

            tray_color = (mapped.get("rgba") or "808080FF").upper()
            if len(tray_color) == 6:
                tray_color = tray_color + "FF"

            # Printer state, read here rather than further down because the
            # per-model preset override below needs the slot's nozzle
            # diameter and the K-profile cascade further down needs the same
            # value -- one read, so they cannot disagree. (The previous
            # `mqtt_client.printer_state` access via hasattr always returned
            # None -- the attribute is `state`, not `printer_state` -- so the
            # K-profile cascade silently skipped state.kprofiles, defaulted
            # nozzle_diameter to 0.4, and left slot_extruder unset.)
            state = printer_manager.get_status(body.printer_id)
            slot_nozzle = resolve_slot_nozzle(
                state, body.ams_id, body.tray_id, printer_manager.get_model(body.printer_id)
            )
            nozzle_diameter = slot_nozzle.diameter

            # Per-printer-model preset override, same cascade as internal
            # mode: a cloud/Orca preset is bound to a model, so one stored
            # preset per spool is wrong across two models. Returns Spoolman's
            # own value when no override is set.
            slot_slicer_filament, slot_slicer_filament_name = await resolve_spoolman_preset(
                db,
                spoolman_spool_id=body.spoolman_spool_id,
                printer_model=printer_manager.get_model(body.printer_id),
                nozzle_diameter=nozzle_diameter,
                fallback_filament=mapped.get("slicer_filament"),
                fallback_name=mapped.get("slicer_filament_name"),
            )

            # #1713: resolve the spool's stored slicer_filament reference
            # (cloud preset, local preset, GF-prefix builtin, or numeric
            # LocalPreset id) to the printer-side tray_info_idx + setting_id.
            # Previously the Spoolman path dropped slicer_filament on the
            # floor and only the generic-material fallback fired; the user-
            # configured profile never reached the printer. Shared with the
            # internal-mode route via the same helper so the two flows can't
            # drift again.
            tray_info_idx, setting_id, sub_brand_override, type_override = await resolve_slicer_filament(
                db=db,
                current_user=current_user,
                slicer_filament=slot_slicer_filament,
                slicer_filament_name=slot_slicer_filament_name,
                material=material,
            )
            if sub_brand_override:
                tray_sub_brands = sub_brand_override
            # A preset carries its own type; the reduction above only infers
            # one from Spoolman's free-text material. The preset wins when the
            # spool has one (issue #2902, @doncaruana).
            if type_override:
                tray_type = printer_filament_type(type_override)

            material_upper = material.upper().strip()
            # Fall back to generic-material id when slicer_filament is empty
            # or the resolver discarded an unresolvable value. Matches the
            # internal-mode tail in inventory.py:_apply_spool_to_slot_inner,
            # including the order: the spool's own wording first and the
            # reduced type only after it, so "PETG HF" keeps its own generic
            # preset (GFG96) rather than trading it for "PETG"'s GFG99.
            if not tray_info_idx:
                tray_info_idx = (
                    # The finish first, as in internal mode: a Silk spool
                    # gets Generic PLA Silk.
                    finish_generic_id(tray_type, f"{material} {subtype}")
                    or GENERIC_FILAMENT_IDS.get(material_upper)
                    or GENERIC_FILAMENT_IDS.get(material_upper.split("-")[0].split(" ")[0])
                    or GENERIC_FILAMENT_IDS.get(tray_type.upper())
                    or ""
                )

            # Ensure setting_id is always derivable from tray_info_idx. The
            # local-preset path can leave it empty when the LP's setting JSON
            # has no filament_id and falls through to the generic material id;
            # without this fallback the slicer gets a half-configured slot
            # (filament id without setting id) and the slot detail modal
            # renders empty fields. Same pattern as the internal-mode tail.
            if tray_info_idx and not setting_id:
                setting_id = filament_id_to_setting_id(tray_info_idx)

            temp_defaults = nozzle_temp_range(material, tray_type)
            temp_min = mapped.get("nozzle_temp_min") or temp_defaults[0]
            temp_max = temp_defaults[1]

            # Pull printer state from printer_manager. The previous
            # `mqtt_client.printer_state` access via hasattr always returned
            # None (the attribute is `state`, not `printer_state`), so the
            # K-profile cascade silently skipped state.kprofiles, defaulted
            # nozzle_diameter to 0.4, and left slot_extruder unset.
            slot_extruder = slot_nozzle.extruder

            # Prefer exact extruder match, fall back to extruder-agnostic kp
            # for the same nozzle. Hard-skipping on mismatch silently dropped
            # valid stored profiles when the AMS-extruder mapping had shifted.
            exact_kp = None
            fallback_kp = None
            for kp in kp_rows:
                if kp.nozzle_diameter != nozzle_diameter or kp.cali_idx is None:
                    continue
                if not slot_nozzle.flow_matches(kp.nozzle_type):
                    continue
                if slot_extruder is not None and kp.extruder is not None and kp.extruder == slot_extruder:
                    exact_kp = kp
                    break
                if fallback_kp is None:
                    fallback_kp = kp
            matching_kp = exact_kp or fallback_kp

            # Resolve the printer-side calibration entry by cali_idx so we
            # know the authoritative filament_id (the printer indexes its
            # calibration table by filament_id, not setting_id).
            printer_kp = None
            if matching_kp and state and state.kprofiles:
                for pkp in state.kprofiles:
                    if pkp.slot_id == matching_kp.cali_idx and pkp.nozzle_diameter == nozzle_diameter:
                        printer_kp = pkp
                        break
                if printer_kp is None:
                    logger.warning(
                        "Spoolman assign: cali_idx=%d not present in printer's "
                        "calibration table — stored kp may be stale.",
                        matching_kp.cali_idx,
                    )

            # Realign the slot's filament context (tray_info_idx + setting_id)
            # to the kp's calibration context. Without this, ams_filament_setting
            # declares the slot under generic PLA while extrusion_cali_sel points
            # the cali_idx at a different preset — the printer can't link them
            # and falls back to the default profile. P-prefix local presets are
            # valid for tray_info_idx; PFUS-prefix cloud-user presets are not
            # (the slicer rejects them).
            effective_tray_info_idx = tray_info_idx
            effective_setting_id = setting_id
            if printer_kp and printer_kp.filament_id:
                if not printer_kp.filament_id.startswith("PFUS"):
                    effective_tray_info_idx = printer_kp.filament_id
                if printer_kp.setting_id:
                    effective_setting_id = printer_kp.setting_id
            elif matching_kp and matching_kp.setting_id:
                derived = normalize_slicer_filament(matching_kp.setting_id)[0]
                if derived and not derived.startswith("PFUS"):
                    effective_tray_info_idx = derived
                effective_setting_id = matching_kp.setting_id
            if effective_tray_info_idx != tray_info_idx or effective_setting_id != setting_id:
                logger.info(
                    "Spoolman assign: realigning tray_info_idx %r → %r, setting_id %r → %r (kp_id=%s, source=%s)",
                    tray_info_idx,
                    effective_tray_info_idx,
                    setting_id,
                    effective_setting_id,
                    matching_kp.id if matching_kp else None,
                    "printer" if printer_kp else "stored",
                )

            mqtt_client.ams_set_filament_setting(
                ams_id=body.ams_id,
                tray_id=body.tray_id,
                tray_info_idx=effective_tray_info_idx,
                tray_type=tray_type,
                tray_sub_brands=tray_sub_brands,
                tray_color=tray_color,
                nozzle_temp_min=temp_min,
                nozzle_temp_max=temp_max,
                setting_id=effective_setting_id,
            )

            if matching_kp and matching_kp.cali_idx is not None:
                # Use printer-reported filament_id when available, otherwise
                # fall back to the realigned tray_info_idx so both commands
                # reference the same filament context.
                cali_filament_id = (
                    printer_kp.filament_id if printer_kp and printer_kp.filament_id else None
                ) or effective_tray_info_idx
                mqtt_client.extrusion_cali_sel(
                    ams_id=body.ams_id,
                    tray_id=body.tray_id,
                    cali_idx=matching_kp.cali_idx,
                    filament_id=cali_filament_id,
                    nozzle_diameter=nozzle_diameter,
                )
                logger.info(
                    "Spoolman assign: applied K-profile cali_idx=%d "
                    "(kp_id=%d, filament_id=%s) for spool %d on printer %d AMS%d-T%d",
                    matching_kp.cali_idx,
                    matching_kp.id,
                    cali_filament_id,
                    body.spoolman_spool_id,
                    body.printer_id,
                    body.ams_id,
                    body.tray_id,
                )
            else:
                # No stored K-profile for this spool — always reset the slot to
                # Default K (cali_idx=-1). The live cali_idx belongs to whatever
                # filament was there before, so preserving it would apply the
                # wrong filament's calibration to the new spool.
                mqtt_client.extrusion_cali_sel(
                    ams_id=body.ams_id,
                    tray_id=body.tray_id,
                    cali_idx=-1,
                    filament_id=effective_tray_info_idx,
                    nozzle_diameter=nozzle_diameter,
                )
                logger.info(
                    "No stored K-profile for Spoolman spool %d — reset slot to Default K (cali_idx=-1)",
                    body.spoolman_spool_id,
                )

            # Read-back verification, as the built-in assign registers it (#2582):
            # the AMS pushes that follow confirm the tray took what was sent.
            # Not for a slot the firmware calls empty, which echoes nothing
            # back until a spool goes in (the built-in check, inventory.py).
            from backend.app.api.routes.inventory import _find_tray_in_ams_data  # noqa: PLC0415
            from backend.app.services.ams_slot_presence import spool_present  # noqa: PLC0415

            live_tray = None
            if state and state.raw_data:
                ams_raw = state.raw_data.get("ams", [])
                if isinstance(ams_raw, dict):
                    ams_raw = ams_raw.get("ams", [])
                live_tray = _find_tray_in_ams_data(ams_raw, body.ams_id, body.tray_id)
            tray_state = (live_tray or {}).get("state")
            if not (spool_present(live_tray) is not True and tray_state in (9, 10)):
                mqtt_client.register_assignment_verification(
                    ams_id=body.ams_id,
                    tray_id=body.tray_id,
                    tray_info_idx=effective_tray_info_idx,
                    tray_color=tray_color,
                    cali_idx=matching_kp.cali_idx if matching_kp and matching_kp.cali_idx is not None else -1,
                )
                # Fresh telemetry for the check to compare against.
                mqtt_client.request_status_update()

            logger.info(
                "Auto-configured AMS slot ams=%d tray=%d for Spoolman spool %d on printer %d",
                body.ams_id,
                body.tray_id,
                body.spoolman_spool_id,
                body.printer_id,
            )
    except Exception:
        logger.exception(
            "Failed to auto-configure AMS slot for Spoolman spool %d (printer=%d, ams=%d, tray=%d)",
            body.spoolman_spool_id,
            body.printer_id,
            body.ams_id,
            body.tray_id,
        )

    return mapped


@router.delete("/slot-assignments/{spoolman_spool_id}")
async def unassign_spoolman_slot(
    spoolman_spool_id: int = Path(..., gt=0),
    db: AsyncSession = Depends(get_db, scope="function"),
    _: User | None = RequirePermissionIfAuthEnabled(Permission.INVENTORY_UPDATE),
    printer_scope: PrinterScope = RequestPrinterScope,
) -> dict:
    """Remove the local slot assignment for a Spoolman spool.

    Spoolman's own ``spool.location`` field is NOT touched — it is user-managed.
    """
    client = await _get_client(db)

    try:
        # A slot on a printer the caller can't see stays assigned (#1727)
        unassign = delete(SpoolmanSlotAssignment).where(SpoolmanSlotAssignment.spoolman_spool_id == spoolman_spool_id)
        if (clause := printer_scope.where_strict(SpoolmanSlotAssignment.printer_id)) is not None:
            unassign = unassign.where(clause)
        await db.execute(unassign)
        await db.commit()
    except Exception as exc:
        await db.rollback()
        logger.error("Failed to delete slot assignment: %s", exc)
        raise HTTPException(status_code=500, detail="Failed to remove slot assignment") from exc

    # Fetch the spool from Spoolman to return in InventorySpool format.
    # If the spool no longer exists in Spoolman, the local unassignment still succeeded.
    try:
        async with _translate_spoolman_errors():
            spool = await client.get_spool(spoolman_spool_id)
        return _map_spoolman_spool(spool)
    except HTTPException as exc:
        if exc.status_code != 404:
            raise
        # Spool no longer exists in Spoolman; unassignment still succeeded.
        return {"id": spoolman_spool_id}


@router.get("/slot-assignments")
async def get_spoolman_slot_assignment(
    printer_id: int = Query(..., gt=0),
    ams_id: int = Query(..., ge=0, le=7),
    tray_id: int = Query(..., ge=0, le=3),
    db: AsyncSession = Depends(get_db, scope="function"),
    _: User | None = RequirePermissionIfAuthEnabled(Permission.INVENTORY_READ),
    printer_scope: PrinterScope = RequestPrinterScope,
) -> dict | None:
    """Return the Spoolman spool assigned to a specific printer slot, or null if unassigned."""
    client = await _get_client(db)
    printer_scope.ensure(printer_id)
    result = await db.execute(select(Printer).where(Printer.id == printer_id))
    printer = result.scalar_one_or_none()
    if not printer:
        raise HTTPException(status_code=404, detail="Printer not found")

    slot_result = await db.execute(
        select(SpoolmanSlotAssignment).where(
            SpoolmanSlotAssignment.printer_id == printer_id,
            SpoolmanSlotAssignment.ams_id == ams_id,
            SpoolmanSlotAssignment.tray_id == tray_id,
        )
    )
    slot = slot_result.scalar_one_or_none()
    if not slot:
        return None

    try:
        async with _translate_spoolman_errors():
            spool = await client.get_spool(slot.spoolman_spool_id)
        return _map_spoolman_spool(spool)
    except HTTPException as exc:
        if exc.status_code != 404:
            raise
        # Spool deleted in Spoolman — clean up stale assignment.
        # Include spoolman_spool_id in WHERE to avoid a TOCTOU race where a
        # concurrent re-assign changed the slot to a different spool between
        # the GET and this DELETE.
        try:
            await db.execute(
                delete(SpoolmanSlotAssignment).where(
                    SpoolmanSlotAssignment.id == slot.id,
                    SpoolmanSlotAssignment.spoolman_spool_id == slot.spoolman_spool_id,
                )
            )
            await db.commit()
        except Exception as cleanup_exc:
            await db.rollback()
            logger.warning(
                "Failed to remove stale slot assignment for spool %s: %s",
                slot.spoolman_spool_id,
                cleanup_exc,
            )
        return None


def _k_profile_to_dict(p: SpoolmanKProfile) -> dict:
    """Manually map SpoolmanKProfile → SpoolKProfileResponse-compatible dict."""
    return {
        "id": p.id,
        "spool_id": p.spoolman_spool_id,
        "printer_id": p.printer_id,
        "extruder": p.extruder,
        "nozzle_diameter": p.nozzle_diameter,
        "nozzle_type": p.nozzle_type,
        "k_value": p.k_value,
        "name": p.name,
        "cali_idx": p.cali_idx,
        "setting_id": p.setting_id,
        "created_at": p.created_at,
    }


def _filament_preset_to_dict(p: SpoolmanFilamentPreset) -> dict:
    """Manually map SpoolmanFilamentPreset → SpoolFilamentPresetResponse-compatible dict."""
    return {
        "id": p.id,
        "spool_id": p.spoolman_spool_id,
        "printer_model": p.printer_model,
        "nozzle_diameter": p.nozzle_diameter,
        "slicer_filament": p.slicer_filament,
        "slicer_filament_name": p.slicer_filament_name,
        "created_at": p.created_at,
    }


@router.get("/spools/{spool_id}/filament-presets")
async def get_spoolman_filament_presets(
    spool_id: int = Path(..., gt=0),
    db: AsyncSession = Depends(get_db, scope="function"),
    _: User | None = RequirePermissionIfAuthEnabled(Permission.INVENTORY_READ),
) -> list[dict]:
    """Return all per-printer-model preset overrides for a Spoolman spool."""
    await _get_client(db)
    result = await db.execute(
        select(SpoolmanFilamentPreset).where(SpoolmanFilamentPreset.spoolman_spool_id == spool_id)
    )
    return [_filament_preset_to_dict(p) for p in result.scalars().all()]


@router.put("/spools/{spool_id}/filament-presets")
async def save_spoolman_filament_presets(
    spool_id: int = Path(..., gt=0),
    presets: list[SpoolFilamentPresetBase] = Body(...),
    db: AsyncSession = Depends(get_db, scope="function"),
    _: User | None = RequirePermissionIfAuthEnabled(Permission.INVENTORY_UPDATE),
) -> list[dict]:
    """Replace all per-printer-model preset overrides for a Spoolman spool."""
    client = await _get_client(db)
    async with _translate_spoolman_errors():
        await client.get_spool(spool_id)

    # Same as the internal route: reject a duplicated (model, diameter) before
    # touching the stored rows, so a bad payload cannot clear what it fails to
    # replace.
    seen: set[tuple[str, str]] = set()
    for preset in presets:
        key = (preset.printer_model, preset.nozzle_diameter)
        if key in seen:
            raise HTTPException(
                422,
                f"Duplicate override for model {preset.printer_model!r} nozzle {preset.nozzle_diameter or 'any'!r}",
            )
        seen.add(key)

    saved: list[SpoolmanFilamentPreset] = []
    try:
        await db.execute(delete(SpoolmanFilamentPreset).where(SpoolmanFilamentPreset.spoolman_spool_id == spool_id))
        await db.flush()
        for preset in presets:
            obj = SpoolmanFilamentPreset(
                spoolman_spool_id=spool_id,
                printer_model=preset.printer_model,
                nozzle_diameter=preset.nozzle_diameter,
                slicer_filament=preset.slicer_filament,
                slicer_filament_name=preset.slicer_filament_name,
            )
            db.add(obj)
            saved.append(obj)
        await db.commit()
    except IntegrityError as exc:
        await db.rollback()
        raise HTTPException(422, "Duplicate or invalid preset override (check model and nozzle uniqueness)") from exc
    except HTTPException:
        raise
    except Exception as exc:
        await db.rollback()
        logger.error("Filament preset save for spool %d failed: %s", spool_id, exc)
        raise HTTPException(500, "Failed to save filament presets") from exc

    for obj in saved:
        await db.refresh(obj)

    return [_filament_preset_to_dict(p) for p in saved]


def _normalize_filament(raw: dict) -> NormalizedFilament | None:
    """Normalise a raw Spoolman filament dict for the frontend catalog picker.

    Returns None for entries with missing/zero IDs — those are malformed and
    must be filtered out before returning to the client.
    weight=0 is collapsed to None — 0g is not a valid filament weight.
    """
    filament_id = _safe_int(raw.get("id"), 0)
    if filament_id <= 0:
        logger.warning("Skipping Spoolman filament with missing or invalid id: %r", raw.get("name"))
        return None
    vendor = raw.get("vendor") or {}
    vendor_ref: NormalizedVendorRef | None = None
    if vendor:
        vendor_id = _safe_int(vendor.get("id"), 0)
        if vendor_id <= 0:
            logger.warning("Spoolman filament %d has vendor without valid id — vendor omitted", filament_id)
        else:
            vendor_ref = {"id": vendor_id, "name": str(vendor.get("name") or "").strip() or "Unknown"}
    return NormalizedFilament(
        id=filament_id,
        name=str(raw.get("name") or ""),
        material=raw.get("material") or None,
        color_hex=raw.get("color_hex") or None,
        color_name=raw.get("color_name") or None,
        weight=_safe_int(raw.get("weight"), 0) or None,  # 0g is not a valid weight
        spool_weight=_safe_optional_float(raw.get("spool_weight")),
        vendor=vendor_ref,
    )


@router.get("/filaments")
async def list_spoolman_filaments(
    db: AsyncSession = Depends(get_db, scope="function"),
    _: User | None = RequirePermissionIfAuthEnabled(Permission.INVENTORY_READ),
) -> list[NormalizedFilament]:
    """Return all filaments from Spoolman, normalised for the frontend catalog picker."""
    client = await _get_client(db)
    async with _translate_spoolman_errors():
        raw_filaments = await client.get_filaments()
    if not isinstance(raw_filaments, list):
        logger.warning("Spoolman get_filaments() returned non-list type: %s", type(raw_filaments).__name__)
        return []
    return [f for raw in raw_filaments if (f := _normalize_filament(raw)) is not None]


@router.patch("/filaments/{filament_id}")
async def patch_spoolman_filament(
    *,
    filament_id: int = Path(..., gt=0),
    body: SpoolmanFilamentPatch = Body(...),
    db: AsyncSession = Depends(get_db, scope="function"),
    _: User | None = RequirePermissionIfAuthEnabled(Permission.INVENTORY_UPDATE),
) -> NormalizedFilament:
    """Update a Spoolman filament's name and/or spool_weight.

    When spool_weight changes, Option A (keep_existing_spools=True) stamps the old
    weight (the filament's, or its vendor's empty_spool_weight when the filament had
    none) onto spools currently inheriting it (spool.spool_weight is None) so their
    tare calculations are unaffected by the filament change.
    Option B (keep_existing_spools=False, the default): when spool_weight is a
    concrete value, stamps it onto every affected spool explicitly; when spool_weight
    is null, clears per-spool overrides so spools fall back to the filament value.
    """
    client = await _get_client(db)

    async with _translate_spoolman_errors():
        current = await client.get_filament(filament_id)

    patch_data = {k: v for k, v in body.model_dump(exclude_unset=True).items() if k != "keep_existing_spools"}
    if not patch_data:
        normalized = _normalize_filament(current)
        if normalized is None:
            raise HTTPException(status_code=404, detail="Filament not found")
        return normalized

    async with _translate_spoolman_errors():
        updated = await client.patch_filament(filament_id, patch_data)

    if "spool_weight" in body.model_fields_set:
        async with _translate_spoolman_errors():
            all_spools = await client.get_all_spools()
        affected_spools = [s for s in all_spools if (s.get("filament") or {}).get("id") == filament_id]

        if affected_spools:
            if body.keep_existing_spools:
                # What the inheriting spools weigh against today: the filament's
                # own value, or when it has none, its vendor's (#3195). With
                # neither they sit on the 250 g fallback, which is Bambuddy's
                # alone, so there is nothing real to keep.
                old_weight = _safe_optional_float(current.get("spool_weight"))
                if old_weight is None:
                    old_weight = _safe_optional_float((current.get("vendor") or {}).get("empty_spool_weight"))
                if old_weight is not None:
                    spools_to_fix = [s for s in affected_spools if s.get("spool_weight") is None]
                    if spools_to_fix:
                        async with _translate_spoolman_errors():
                            results = await asyncio.gather(
                                *(
                                    client.update_spool_full(spool_id=s["id"], spool_weight=old_weight)
                                    for s in spools_to_fix
                                ),
                                return_exceptions=True,
                            )
                        _raise_if_partial_failure(spools_to_fix, results, "spool_weight stamp (option A)")
            else:
                new_weight = body.spool_weight
                if new_weight is not None:
                    # Stamp the new weight onto every spool of this filament type so
                    # each spool carries the value explicitly rather than inheriting.
                    async with _translate_spoolman_errors():
                        results = await asyncio.gather(
                            *(
                                client.update_spool_full(spool_id=s["id"], spool_weight=new_weight)
                                for s in affected_spools
                            ),
                            return_exceptions=True,
                        )
                    _raise_if_partial_failure(affected_spools, results, "spool_weight stamp (option B)")
                else:
                    # Filament weight is being cleared — remove any per-spool override
                    # so spools fall back to the vendor's empty_spool_weight, or 250 g.
                    spools_to_clear = [s for s in affected_spools if s.get("spool_weight") is not None]
                    if spools_to_clear:
                        async with _translate_spoolman_errors():
                            results = await asyncio.gather(
                                *(
                                    client.update_spool_full(spool_id=s["id"], clear_spool_weight=True)
                                    for s in spools_to_clear
                                ),
                                return_exceptions=True,
                            )
                        _raise_if_partial_failure(spools_to_clear, results, "spool_weight clear (option B null)")

    normalized = _normalize_filament(updated)
    if normalized is None:
        raise HTTPException(status_code=502, detail="Spoolman returned malformed filament data")
    return normalized


@router.post("/profile-defaults/{default_id}/apply")
async def apply_profile_default_to_spoolman_spools(
    default_id: int = Path(..., gt=0),
    db: AsyncSession = Depends(get_db, scope="function"),
    _: User | None = RequirePermissionIfAuthEnabled(Permission.INVENTORY_UPDATE),
) -> dict:
    """Spoolman twin of ``POST /inventory/profile-defaults/{id}/apply`` (#1326).

    Fills in what active Spoolman spools of the default's product do not have
    yet: the spool's own preset (in its extra dict), and its preset overrides
    and K profiles (Bambuddy tables keyed by the spool id). Never overwrites.
    """
    client = await _get_client(db)
    row = await profile_defaults.load_default(db, default_id)
    if row is None:
        raise HTTPException(404, "Profile default not found")
    async with _translate_spoolman_errors():
        spools = await client.get_all_spools()

    updated = 0
    cache: dict[str, profile_defaults.ResolvedProfiles | None] = {}
    slicer_field_ready = False
    for raw in spools:
        try:
            mapped = _map_spoolman_spool(raw)
        except ValueError:
            continue
        # The mapper fills a missing colour name with the subtype; that is no colour.
        colour = None if mapped["color_name_is_synthesized"] else mapped["color_name"]
        if mapped["archived_at"] or not profile_defaults.default_matches(
            row, brand=mapped["brand"], material=mapped["material"], subtype=mapped["subtype"], color_name=colour
        ):
            continue
        colour_key = (colour or "").strip().lower()
        if colour_key not in cache:
            cache[colour_key] = await profile_defaults.resolve_profile_default(
                db, brand=row.brand, material=row.material, subtype=row.subtype, color_name=colour
            )
        resolved = cache[colour_key]
        if resolved is None:
            continue
        changed = await profile_defaults.fill_spoolman_rows(db, mapped["id"], resolved)
        if resolved.slicer_filament and not mapped["slicer_filament"]:
            new_extra = {"bambu_slicer_filament": json.dumps(resolved.slicer_filament)}
            if resolved.slicer_filament_name:
                new_extra["bambu_slicer_filament_name"] = json.dumps(resolved.slicer_filament_name)
            try:
                async with _translate_spoolman_errors():
                    if not slicer_field_ready:
                        await client.ensure_extra_field("bambu_slicer_filament")
                        await client.ensure_extra_field("bambu_slicer_filament_name")
                        slicer_field_ready = True
                    await client.merge_spool_extra(mapped["id"], new_extra)
                forget_spoolman_spool_own_preset(mapped["id"])
                changed = True
            except HTTPException:
                logger.warning("Profile default: preset not written to Spoolman spool %s", mapped["id"])
        if changed:
            updated += 1
    await db.commit()
    if updated:
        await ws_manager.broadcast({"type": "inventory_changed"})
    return {"updated": updated}


@router.get("/spools/{spool_id}/k-profiles")
async def get_spoolman_k_profiles(
    spool_id: int = Path(..., gt=0),
    db: AsyncSession = Depends(get_db, scope="function"),
    _: User | None = RequirePermissionIfAuthEnabled(Permission.INVENTORY_READ),
) -> list[dict]:
    """Return all local K-value calibration profiles for a Spoolman spool."""
    await _get_client(db)
    result = await db.execute(select(SpoolmanKProfile).where(SpoolmanKProfile.spoolman_spool_id == spool_id))
    profiles = result.scalars().all()
    return [_k_profile_to_dict(p) for p in profiles]


@router.put("/spools/{spool_id}/k-profiles")
async def save_spoolman_k_profiles(
    spool_id: int = Path(..., gt=0),
    profiles: list[SpoolKProfileBase] = Body(...),
    db: AsyncSession = Depends(get_db, scope="function"),
    _: User | None = RequirePermissionIfAuthEnabled(Permission.INVENTORY_UPDATE),
    printer_scope: PrinterScope = RequestPrinterScope,
) -> list[dict]:
    """Replace all K-value calibration profiles for a Spoolman spool.

    As the internal route: a caller limited to some printers (#1727) replaces
    only theirs, and the other printers' profiles stay.
    """
    client = await _get_client(db)
    async with _translate_spoolman_errors():
        await client.get_spool(spool_id)

    saved: list[SpoolmanKProfile] = []
    try:
        stmt = delete(SpoolmanKProfile).where(SpoolmanKProfile.spoolman_spool_id == spool_id)
        if (in_scope := printer_scope.where_strict(SpoolmanKProfile.printer_id)) is not None:
            stmt = stmt.where(in_scope)
        await db.execute(stmt)
        for profile in profiles:
            if not printer_scope.allows(profile.printer_id):
                continue
            obj = SpoolmanKProfile(
                spoolman_spool_id=spool_id,
                printer_id=profile.printer_id,
                extruder=profile.extruder,
                nozzle_diameter=profile.nozzle_diameter,
                nozzle_type=profile.nozzle_type,
                k_value=profile.k_value,
                name=profile.name,
                cali_idx=profile.cali_idx,
                setting_id=profile.setting_id,
            )
            db.add(obj)
            saved.append(obj)
        await db.commit()
    except IntegrityError as exc:
        await db.rollback()
        raise HTTPException(422, "Duplicate or invalid K-profile (check printer_id and nozzle uniqueness)") from exc
    except Exception as exc:
        await db.rollback()
        logger.error("K-profile save for spool %d failed: %s", spool_id, exc)
        raise HTTPException(500, "Failed to save K-profiles") from exc

    for obj in saved:
        await db.refresh(obj)

    return [_k_profile_to_dict(p) for p in saved]


def _supplier_link_to_dict(link: SpoolmanSpoolSupplier) -> dict:
    """Same shape as ``SpoolSupplierResponse`` so the frontend renders both
    inventories with one component."""
    return {
        "id": link.id,
        "supplier_id": link.supplier_id,
        "supplier_name": link.supplier_name,
        "supplier_article_number": link.supplier_article_number,
        "quoted_price_per_kg": link.quoted_price_per_kg,
        "is_purchase_source": link.is_purchase_source,
    }


@router.get("/spools/{spool_id}/suppliers")
async def get_spoolman_spool_suppliers(
    spool_id: int = Path(..., gt=0),
    db: AsyncSession = Depends(get_db, scope="function"),
    _: User | None = RequirePermissionIfAuthEnabled(Permission.INVENTORY_READ),
) -> list[dict]:
    """Supplier assignments for a Spoolman spool (#2988, Bambuddy-side rows)."""
    await _get_client(db)
    result = await db.execute(
        select(SpoolmanSpoolSupplier)
        .options(selectinload(SpoolmanSpoolSupplier.supplier))
        .where(SpoolmanSpoolSupplier.spoolman_spool_id == spool_id)
    )
    return [_supplier_link_to_dict(link) for link in result.scalars().all()]


@router.put("/spools/{spool_id}/suppliers")
async def save_spoolman_spool_suppliers(
    spool_id: int = Path(..., gt=0),
    links: list[SpoolSupplierLinkInput] = Body(...),
    db: AsyncSession = Depends(get_db, scope="function"),
    _: User | None = RequirePermissionIfAuthEnabled(Permission.INVENTORY_UPDATE),
) -> list[dict]:
    """Replace a Spoolman spool's supplier assignments (#2988).

    Mirror of the built-in inventory's replace-all endpoint — Spoolman owns
    the spool, Bambuddy owns the assignment (``SpoolmanKProfile`` precedent),
    so the rows are local and the spool is only verified to exist remotely.
    """
    client = await _get_client(db)
    async with _translate_spoolman_errors():
        await client.get_spool(spool_id)

    supplier_ids = [link.supplier_id for link in links]
    if len(set(supplier_ids)) != len(supplier_ids):
        raise HTTPException(400, "Duplicate supplier in assignment list")
    if sum(1 for link in links if link.is_purchase_source) > 1:
        raise HTTPException(400, "Only one assignment can be the purchase source")
    if supplier_ids:
        found = await db.execute(select(Supplier.id).where(Supplier.id.in_(supplier_ids)))
        missing = set(supplier_ids) - {row[0] for row in found.all()}
        if missing:
            raise HTTPException(404, f"Supplier(s) not found: {sorted(missing)}")

    await db.execute(delete(SpoolmanSpoolSupplier).where(SpoolmanSpoolSupplier.spoolman_spool_id == spool_id))
    saved: list[SpoolmanSpoolSupplier] = []
    for link in links:
        row = SpoolmanSpoolSupplier(spoolman_spool_id=spool_id, **link.model_dump())
        db.add(row)
        saved.append(row)
    await db.commit()
    refreshed = await db.execute(
        select(SpoolmanSpoolSupplier)
        .options(selectinload(SpoolmanSpoolSupplier.supplier))
        .where(SpoolmanSpoolSupplier.id.in_([row.id for row in saved]))
        .order_by(SpoolmanSpoolSupplier.id)
    )
    await ws_manager.broadcast({"type": "inventory_changed"})
    return [_supplier_link_to_dict(link) for link in refreshed.scalars().all()]