Adding scope import

app/api/router.py

@@ -14,6 +14,7 @@ from app.api import (
     part_colours,
     part_types,
     roles,
+    scope_import,
     scope_models,
     static_connections,
     sub_part_types,
@@ -66,3 +67,6 @@ api_router.include_router(
 )
 api_router.include_router(audit_log.router, prefix="/audit-log", tags=["audit-log"])
 api_router.include_router(fits.router, prefix="/fits", tags=["fits"])
+api_router.include_router(
+    scope_import.router, prefix="/scope-import", tags=["scope-import"]
+)

app/api/scope_import.py

@@ -0,0 +1,53 @@
+from fastapi import APIRouter, Depends, File, HTTPException, UploadFile, status
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.core.database import get_db
+from app.crud import audit as audit_crud
+from app.deps import require_permission
+from app.models.user import User
+from app.schemas.scope_import import ScopeImportResult
+from app.services import scope_import as scope_import_service
+
+router = APIRouter()
+
+_XLSX_TYPES = {
+    "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
+    "application/octet-stream",
+    "",
+}
+
+
+@router.post("", response_model=ScopeImportResult)
+async def import_scopes(
+    file: UploadFile = File(...),
+    db: AsyncSession = Depends(get_db),
+    actor: User = Depends(require_permission("augmented_parts:write")),
+):
+    filename = file.filename or ""
+    if not filename.lower().endswith(".xlsx"):
+        raise HTTPException(
+            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+            detail="Expected an .xlsx spreadsheet.",
+        )
+    data = await file.read()
+    try:
+        result = await scope_import_service.import_from_bytes(db, data)
+    except Exception as exc:  # openpyxl / malformed workbook
+        raise HTTPException(
+            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+            detail=f"Could not read spreadsheet: {exc}",
+        ) from exc
+
+    await audit_crud.log(
+        db,
+        actor=actor,
+        action="update",
+        resource_type="AugmentedPart",
+        resource_id=None,
+        description=(
+            f"Scope import ({filename}): {result.created} created, "
+            f"{result.updated} updated, {result.skipped} skipped, "
+            f"{result.errors} errors"
+        ),
+    )
+    return result

app/crud/augmented_part.py

@@ -12,6 +12,13 @@ async def get(db: AsyncSession, augmented_part_id: int) -> AugmentedPart | None:
     return await db.get(AugmentedPart, augmented_part_id)
 
 
+async def get_by_sku(db: AsyncSession, sku: str) -> AugmentedPart | None:
+    result = await db.execute(
+        select(AugmentedPart).where(AugmentedPart.sku == sku).order_by(AugmentedPart.id)
+    )
+    return result.scalars().first()
+
+
 async def list_all(
     db: AsyncSession, skip: int = 0, limit: int = 100
 ) -> list[AugmentedPart]:

app/crud/scope_info.py

@@ -0,0 +1,56 @@
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.models.scope_info import ScopeInfo
+
+
+async def get(db: AsyncSession, scope_info_id: int) -> ScopeInfo | None:
+    return await db.get(ScopeInfo, scope_info_id)
+
+
+async def get_by_augmented_part(
+    db: AsyncSession, augmented_part_id: int
+) -> ScopeInfo | None:
+    result = await db.execute(
+        select(ScopeInfo).where(ScopeInfo.augmented_part_id == augmented_part_id)
+    )
+    return result.scalar_one_or_none()
+
+
+async def create(
+    db: AsyncSession,
+    *,
+    augmented_part_id: int,
+    magnification_min: float | None = None,
+    magnification_max: float | None = None,
+    model_suffix: str | None = None,
+    objective_thread_pitch: str | None = None,
+) -> ScopeInfo:
+    info = ScopeInfo(
+        augmented_part_id=augmented_part_id,
+        magnification_min=magnification_min,
+        magnification_max=magnification_max,
+        model_suffix=model_suffix,
+        objective_thread_pitch=objective_thread_pitch,
+    )
+    db.add(info)
+    await db.commit()
+    await db.refresh(info)
+    return info
+
+
+async def upsert(db: AsyncSession, *, augmented_part_id: int, **fields) -> ScopeInfo:
+    """Create the ScopeInfo for an AugmentedPart, or update it if it exists."""
+    info = await get_by_augmented_part(db, augmented_part_id)
+    if info is None:
+        return await create(db, augmented_part_id=augmented_part_id, **fields)
+    for field, value in fields.items():
+        setattr(info, field, value)
+    await db.commit()
+    await db.refresh(info)
+    return info
+
+
+async def delete(db: AsyncSession, info: ScopeInfo) -> None:
+    await db.delete(info)
+    await db.commit()

app/crud/scope_model.py

@@ -8,6 +8,18 @@ async def get(db: AsyncSession, scope_model_id: int) -> ScopeModel | None:
     return await db.get(ScopeModel, scope_model_id)
 
 
+async def get_by_name_and_manufacturer(
+    db: AsyncSession, name: str, manufacturer_id: int
+) -> ScopeModel | None:
+    result = await db.execute(
+        select(ScopeModel).where(
+            ScopeModel.name == name,
+            ScopeModel.manufacturer_id == manufacturer_id,
+        )
+    )
+    return result.scalar_one_or_none()
+
+
 async def list_all(
     db: AsyncSession, skip: int = 0, limit: int = 100
 ) -> list[ScopeModel]:

app/models/__init__.py

@@ -18,6 +18,7 @@ from app.models.manufacturer import Manufacturer
 from app.models.part_colour import PartColour
 from app.models.part_type import PartType
 from app.models.permission import Permission
+from app.models.scope_info import ScopeInfo
 from app.models.scope_model import ScopeModel
 from app.models.static_connection import StaticConnection
 from app.models.sub_part_type import SubPartType
@@ -42,6 +43,7 @@ __all__ = [
     "PartColour",
     "PartType",
     "Permission",
+    "ScopeInfo",
     "ScopeModel",
     "StaticConnection",
     "SubPartType",

app/models/scope_info.py

@@ -0,0 +1,34 @@
+from sqlalchemy import Float, ForeignKey, String, UniqueConstraint
+from sqlalchemy.orm import Mapped, mapped_column
+
+from app.core.database import Base
+
+
+class ScopeInfo(Base):
+    """Scope-specific attributes for an AugmentedPart of PartType "Scope".
+
+    One-to-one with AugmentedPart (unique augmented_part_id). Holds fields that
+    have no home on AugmentedPart itself; populated by the scope xlsx importer.
+    """
+
+    __tablename__ = "scope_infos"
+    __table_args__ = (UniqueConstraint("augmented_part_id"),)
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    augmented_part_id: Mapped[int] = mapped_column(
+        ForeignKey("augmented_parts.id", ondelete="CASCADE"), nullable=False
+    )
+    magnification_min: Mapped[float | None] = mapped_column(
+        Float, nullable=True, default=None
+    )
+    magnification_max: Mapped[float | None] = mapped_column(
+        Float, nullable=True, default=None
+    )
+    model_suffix: Mapped[str | None] = mapped_column(
+        String, nullable=True, default=None
+    )
+    # Kept as the raw string from the sheet (e.g. "0.75", "1/16") so no precision
+    # or notation is lost; the parsed thread lives on its own AttachmentPoint.
+    objective_thread_pitch: Mapped[str | None] = mapped_column(
+        String, nullable=True, default=None
+    )

app/schemas/scope_import.py

@@ -0,0 +1,17 @@
+from pydantic import BaseModel
+
+
+class ScopeImportRowResult(BaseModel):
+    row: int
+    product_number: str | None = None
+    status: str  # "created" | "updated" | "skipped" | "error"
+    messages: list[str] = []
+
+
+class ScopeImportResult(BaseModel):
+    created: int
+    updated: int
+    skipped: int
+    errors: int
+    total: int
+    rows: list[ScopeImportRowResult] = []

app/schemas/scope_info.py

@@ -0,0 +1,15 @@
+from pydantic import BaseModel, ConfigDict
+
+
+class ScopeInfoBase(BaseModel):
+    magnification_min: float | None = None
+    magnification_max: float | None = None
+    model_suffix: str | None = None
+    objective_thread_pitch: str | None = None
+
+
+class ScopeInfoRead(ScopeInfoBase):
+    model_config = ConfigDict(from_attributes=True)
+
+    id: int
+    augmented_part_id: int

app/services/scope_import.py

@@ -0,0 +1,427 @@
+"""Import third-party scopes from an xlsx spreadsheet.
+
+The spreadsheet describes parts of PartType "Scope". Each row maps onto an
+AugmentedPart (keyed on ``product_number`` → ``sku``, so re-importing upserts),
+a ScopeModel, a ScopeInfo record for the scope-specific fields, up to three
+attachment points, and up to two Fits against other parts.
+
+Column → destination
+--------------------
+manufacturer                        → Manufacturer (get-or-create) → AugmentedPart.manufacturer
+model                               → ScopeModel.name (get-or-create) → AugmentedPart.scope_model
+url                                 → ScopeModel.url
+model_suffix                        → ScopeInfo.model_suffix (+ description)
+sub_part_type                       → SubPartType (get-or-create under "Scope")
+product_number                      → AugmentedPart.sku  (also the upsert / fit-lookup key)
+magnification_min / _max            → ScopeInfo
+objective_thread_pitch              → ScopeInfo
+ocular_outer_diameter_mm            → StaticConnection + AttachmentPoint "Ocular"
+objective_lens_outside_diameter_mm  → StaticConnection + AttachmentPoint "Objective"
+objective_thread                    → ISO/UNC thread + AttachmentPoint "Objective"
+ocular_fit                          → Fit: scope's Ocular AP ↔ referenced part's first AP
+objective_fit                       → Fit: scope's Objective AP ↔ referenced part's first AP
+
+AugmentedPart.description is synthesised from "{manufacturer} {model} {suffix}"
+and colour defaults to an "Unknown" PartColour, since the sheet carries neither.
+
+Fits are created with label POSSIBLE. The referenced part is looked up by
+``product_number`` and linked to its first attachment point (chosen answer:
+"sole/first AP"). Both directions are resolved in a second pass so a fit between
+two scopes in the same spreadsheet resolves regardless of row order.
+
+Bad rows (missing key, unknown fit target, unparseable thread) are skipped and
+reported per-row; good rows are committed as they go.
+"""
+
+import io
+
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.crud import attachment_point as attachment_point_crud
+from app.crud import attachment_point_name as apn_crud
+from app.crud import augmented_part as augmented_part_crud
+from app.crud import fit as fit_crud
+from app.crud import iso_thread as iso_thread_crud
+from app.crud import manufacturer as manufacturer_crud
+from app.crud import part_colour as part_colour_crud
+from app.crud import part_type as part_type_crud
+from app.crud import scope_info as scope_info_crud
+from app.crud import scope_model as scope_model_crud
+from app.crud import static_connection as static_connection_crud
+from app.crud import sub_part_type as sub_part_type_crud
+from app.crud import unc_thread as unc_thread_crud
+from app.models.fit import FitLabel
+from app.schemas.scope_import import ScopeImportResult, ScopeImportRowResult
+from app.services.augmented_part_sync import (
+    MM_PER_INCH,
+    _parse_thread_attribute,
+)
+
+PART_TYPE_NAME = "Scope"
+DEFAULT_COLOUR = "Unknown"
+OCULAR_LABEL = "Ocular"
+OBJECTIVE_LABEL = "Objective"
+
+# Columns we read from the sheet (normalised: stripped + lower-cased headers).
+EXPECTED_COLUMNS = [
+    "manufacturer",
+    "model",
+    "magnification_min",
+    "magnification_max",
+    "objective_lens_outside_diameter_mm",
+    "model_suffix",
+    "sub_part_type",
+    "product_number",
+    "url",
+    "ocular_outer_diameter_mm",
+    "ocular_fit",
+    "objective_thread_pitch",
+    "objective_thread",
+    "objective_fit",
+]
+
+
+def _text(value) -> str | None:
+    """Coerce a spreadsheet cell to a trimmed string (None if empty)."""
+    if value is None:
+        return None
+    # openpyxl surfaces integer-valued cells as floats (12345.0); render cleanly
+    # so product numbers/suffixes don't gain a stray ".0".
+    if isinstance(value, float) and value.is_integer():
+        return str(int(value))
+    s = str(value).strip()
+    return s or None
+
+
+def _num(value) -> float | None:
+    """Coerce a spreadsheet cell to a float (None if empty/non-numeric)."""
+    if value is None:
+        return None
+    if isinstance(value, (int, float)):
+        return float(value)
+    s = str(value).strip()
+    if not s:
+        return None
+    try:
+        return float(s)
+    except ValueError:
+        return None
+
+
+def parse_workbook(data: bytes) -> list[dict]:
+    """Read the first worksheet into a list of {column_name: cell} dicts.
+
+    The header row is matched case-insensitively; unknown columns are ignored
+    and fully-blank rows are dropped.
+    """
+    from openpyxl import load_workbook
+
+    wb = load_workbook(io.BytesIO(data), read_only=True, data_only=True)
+    try:
+        ws = wb.active
+        rows = ws.iter_rows(values_only=True)
+        try:
+            header = next(rows)
+        except StopIteration:
+            return []
+        headers = [(_text(h) or "").lower() for h in header]
+
+        records: list[dict] = []
+        for raw in rows:
+            if all(c is None or (isinstance(c, str) and not c.strip()) for c in raw):
+                continue
+            record = {}
+            for idx, name in enumerate(headers):
+                if name:
+                    record[name] = raw[idx] if idx < len(raw) else None
+            records.append(record)
+        return records
+    finally:
+        wb.close()
+
+
+async def _static_ap(
+    db: AsyncSession, *, augmented_part_id: int, label: str, od_mm: float
+):
+    """Create an attachment point carrying a StaticConnection for ``od_mm``."""
+    sc = await static_connection_crud.get_by_od_mm(db, od_mm)
+    if sc is None:
+        sc = await static_connection_crud.create(
+            db, od_mm=od_mm, od_in=od_mm / MM_PER_INCH
+        )
+    apn = await apn_crud.get_by_name(db, label)
+    if apn is None:
+        apn = await apn_crud.create(db, name=label)
+    return await attachment_point_crud.create(
+        db,
+        augmented_part_id=augmented_part_id,
+        attachment_point_name_id=apn.id,
+        static_connection_id=sc.id,
+    )
+
+
+async def _thread_ap(
+    db: AsyncSession, *, augmented_part_id: int, label: str, spec: str
+):
+    """Create an attachment point for a parsed ISO/UNC thread, or None."""
+    parsed = _parse_thread_attribute(spec)
+    if parsed is None:
+        return None
+    kind, fields = parsed
+    apn = await apn_crud.get_by_name(db, label)
+    if apn is None:
+        apn = await apn_crud.create(db, name=label)
+
+    if kind == "iso":
+        thread = await iso_thread_crud.get_by_values(db, depth_mm=None, **fields)
+        if thread is None:
+            thread = await iso_thread_crud.create(db, depth_mm=None, **fields)
+        return await attachment_point_crud.create(
+            db,
+            augmented_part_id=augmented_part_id,
+            attachment_point_name_id=apn.id,
+            iso_thread_id=thread.id,
+        )
+    # unc
+    thread = await unc_thread_crud.get_by_values(db, depth_in=None, **fields)
+    if thread is None:
+        thread = await unc_thread_crud.create(db, depth_in=None, **fields)
+    return await attachment_point_crud.create(
+        db,
+        augmented_part_id=augmented_part_id,
+        attachment_point_name_id=apn.id,
+        unc_thread_id=thread.id,
+    )
+
+
+async def _link_fit(
+    db: AsyncSession,
+    *,
+    scope_ap,
+    scope_role: str,
+    target_sku: str,
+    scope_part_id: int,
+    messages: list[str],
+) -> None:
+    """Create a POSSIBLE fit from the scope's AP to the referenced part's first AP."""
+    if scope_ap is None:
+        messages.append(
+            f"{scope_role}_fit '{target_sku}' skipped: scope has no "
+            f"{scope_role} attachment point"
+        )
+        return
+    target = await augmented_part_crud.get_by_sku(db, target_sku)
+    if target is None:
+        messages.append(f"{scope_role}_fit references unknown product '{target_sku}'")
+        return
+    if target.id == scope_part_id:
+        messages.append(f"{scope_role}_fit '{target_sku}' skipped: self-reference")
+        return
+    target_aps = await attachment_point_crud.list_for_augmented_part(db, target.id)
+    if not target_aps:
+        messages.append(
+            f"{scope_role}_fit '{target_sku}' skipped: referenced part has no "
+            "attachment points"
+        )
+        return
+    target_ap = target_aps[0]
+    existing = await fit_crud.get_by_attachment_points(db, scope_ap, target_ap)
+    if existing is None:
+        await fit_crud.create(
+            db, ap_a=scope_ap, ap_b=target_ap, label=FitLabel.POSSIBLE
+        )
+
+
+async def import_scopes(db: AsyncSession, rows: list[dict]) -> ScopeImportResult:
+    # Shared lookups created once up front.
+    part_type = await part_type_crud.get_by_name(db, PART_TYPE_NAME)
+    if part_type is None:
+        part_type = await part_type_crud.create(db, name=PART_TYPE_NAME)
+    colour = await part_colour_crud.get_by_name(db, DEFAULT_COLOUR)
+    if colour is None:
+        colour = await part_colour_crud.create(db, name=DEFAULT_COLOUR)
+
+    created = updated = skipped = errors = 0
+    row_results: list[ScopeImportRowResult] = []
+    # (row_dict, scope_ap_ocular, scope_ap_objective, part_id, row_result)
+    staged: list[tuple] = []
+
+    # ---- Phase 1: parts, scope info, attachment points ----------------------
+    for offset, row in enumerate(rows):
+        # +2: 1-based, and the header occupies row 1.
+        row_num = offset + 2
+        product_number = _text(row.get("product_number")) or _text(row.get("model"))
+        result = ScopeImportRowResult(
+            row=row_num, product_number=product_number, status="skipped", messages=[]
+        )
+
+        try:
+            if not product_number:
+                result.messages.append("missing product_number")
+                skipped += 1
+                row_results.append(result)
+                continue
+
+            manufacturer_name = _text(row.get("manufacturer"))
+            if not manufacturer_name:
+                result.messages.append("missing manufacturer")
+                skipped += 1
+                row_results.append(result)
+                continue
+
+            manufacturer = await manufacturer_crud.get_by_name(db, manufacturer_name)
+            if manufacturer is None:
+                manufacturer = await manufacturer_crud.create(
+                    db, name=manufacturer_name, url=""
+                )
+
+            model_name = _text(row.get("model"))
+            url = _text(row.get("url"))
+            scope_model = None
+            if model_name:
+                scope_model = await scope_model_crud.get_by_name_and_manufacturer(
+                    db, model_name, manufacturer.id
+                )
+                if scope_model is None:
+                    scope_model = await scope_model_crud.create(
+                        db, name=model_name, manufacturer_id=manufacturer.id, url=url
+                    )
+
+            sub_part_type_name = _text(row.get("sub_part_type"))
+            sub_part_type = None
+            if sub_part_type_name:
+                sub_part_type = await sub_part_type_crud.get_by_name_and_part_type(
+                    db, sub_part_type_name, part_type.id
+                )
+                if sub_part_type is None:
+                    sub_part_type = await sub_part_type_crud.create(
+                        db, name=sub_part_type_name, part_type_id=part_type.id
+                    )
+
+            model_suffix = _text(row.get("model_suffix"))
+            description = (
+                " ".join(p for p in (manufacturer_name, model_name, model_suffix) if p)
+                or product_number
+            )
+
+            fields = dict(
+                sku=product_number,
+                description=description,
+                colour_id=colour.id,
+                manufacturer_id=manufacturer.id,
+                scope_model_id=scope_model.id if scope_model else None,
+                part_type_id=part_type.id,
+                sub_part_type_id=sub_part_type.id if sub_part_type else None,
+            )
+
+            existing = await augmented_part_crud.get_by_sku(db, product_number)
+            if existing is not None:
+                part = await augmented_part_crud.update(db, existing, fields)
+                result.status = "updated"
+                updated += 1
+            else:
+                part = await augmented_part_crud.create(db, **fields)
+                result.status = "created"
+                created += 1
+
+            await scope_info_crud.upsert(
+                db,
+                augmented_part_id=part.id,
+                magnification_min=_num(row.get("magnification_min")),
+                magnification_max=_num(row.get("magnification_max")),
+                model_suffix=model_suffix,
+                objective_thread_pitch=_text(row.get("objective_thread_pitch")),
+            )
+
+            # Rebuild attachment points from scratch so re-imports stay idempotent
+            # (cascades delete any fits that referenced the old points).
+            for old_ap in await attachment_point_crud.list_for_augmented_part(
+                db, part.id
+            ):
+                await attachment_point_crud.delete(db, old_ap)
+
+            ocular_ap = None
+            ocular_od = _num(row.get("ocular_outer_diameter_mm"))
+            if ocular_od is not None:
+                ocular_ap = await _static_ap(
+                    db,
+                    augmented_part_id=part.id,
+                    label=OCULAR_LABEL,
+                    od_mm=ocular_od,
+                )
+
+            objective_static_ap = None
+            objective_od = _num(row.get("objective_lens_outside_diameter_mm"))
+            if objective_od is not None:
+                objective_static_ap = await _static_ap(
+                    db,
+                    augmented_part_id=part.id,
+                    label=OBJECTIVE_LABEL,
+                    od_mm=objective_od,
+                )
+
+            objective_thread_spec = _text(row.get("objective_thread"))
+            objective_thread_ap = None
+            if objective_thread_spec:
+                objective_thread_ap = await _thread_ap(
+                    db,
+                    augmented_part_id=part.id,
+                    label=OBJECTIVE_LABEL,
+                    spec=objective_thread_spec,
+                )
+                if objective_thread_ap is None:
+                    result.messages.append(
+                        f"could not parse objective_thread "
+                        f"'{objective_thread_spec}'"
+                    )
+
+            # For objective_fit prefer the lens-OD point, falling back to the thread.
+            objective_ap = objective_static_ap or objective_thread_ap
+
+            staged.append((row, ocular_ap, objective_ap, part.id, result))
+            row_results.append(result)
+        except Exception as exc:  # noqa: BLE001 - report, don't abort the batch
+            result.status = "error"
+            result.messages.append(f"unexpected error: {exc}")
+            errors += 1
+            row_results.append(result)
+
+    # ---- Phase 2: fits (all parts from this import now exist) ----------------
+    for row, ocular_ap, objective_ap, part_id, result in staged:
+        try:
+            ocular_fit = _text(row.get("ocular_fit"))
+            if ocular_fit:
+                await _link_fit(
+                    db,
+                    scope_ap=ocular_ap,
+                    scope_role=OCULAR_LABEL.lower(),
+                    target_sku=ocular_fit,
+                    scope_part_id=part_id,
+                    messages=result.messages,
+                )
+            objective_fit = _text(row.get("objective_fit"))
+            if objective_fit:
+                await _link_fit(
+                    db,
+                    scope_ap=objective_ap,
+                    scope_role=OBJECTIVE_LABEL.lower(),
+                    target_sku=objective_fit,
+                    scope_part_id=part_id,
+                    messages=result.messages,
+                )
+        except Exception as exc:  # noqa: BLE001
+            result.messages.append(f"fit resolution error: {exc}")
+
+    return ScopeImportResult(
+        created=created,
+        updated=updated,
+        skipped=skipped,
+        errors=errors,
+        total=len(rows),
+        rows=row_results,
+    )
+
+
+async def import_from_bytes(db: AsyncSession, data: bytes) -> ScopeImportResult:
+    rows = parse_workbook(data)
+    return await import_scopes(db, rows)

app/templates/_nav.html

@@ -112,6 +112,11 @@
     <line x1="12" x2="12" y1="8" y2="22"/>
     <path d="M5 12H2a10 10 0 0 0 20 0h-3"/>
   </symbol>
+  <symbol id="icon-upload" viewBox="0 0 24 24">
+    <path d="M21 15v4a2 2 0 0 1-2 2H5a2 2 0 0 1-2-2v-4"/>
+    <polyline points="17 8 12 3 7 8"/>
+    <line x1="12" x2="12" y1="3" y2="15"/>
+  </symbol>
 </svg>
 
 {# ── Nav macro for DRY icon use ────────────────────────────────────────────── #}
@@ -206,6 +211,9 @@
             {% if su or "augmented_parts:write" in p %}
               <a href="/admin/augmented-parts/skipped">{{ icon("alert-circle") }} Skipped Parts</a>
             {% endif %}
+            {% if su or "augmented_parts:write" in p %}
+              <a href="/admin/scope-import">{{ icon("upload") }} Scope Import</a>
+            {% endif %}
           </div>
         </div>
         {% endif %}

app/templates/admin/_scope_import_result.html

@@ -0,0 +1,41 @@
+{% if error %}
+<div class="card">
+  <p style="color:var(--color-danger,#c00)">{{ error }}</p>
+</div>
+{% elif result %}
+<div class="card">
+  <h2>Import result</h2>
+  <p>
+    <strong>{{ result.created }}</strong> created,
+    <strong>{{ result.updated }}</strong> updated,
+    <strong>{{ result.skipped }}</strong> skipped,
+    <strong>{{ result.errors }}</strong> errors
+    ({{ result.total }} rows).
+  </p>
+
+  {% set flagged = result.rows | selectattr("messages") | list %}
+  {% if flagged %}
+  <table class="table">
+    <thead>
+      <tr><th>Row</th><th>Product #</th><th>Status</th><th>Messages</th></tr>
+    </thead>
+    <tbody>
+      {% for r in flagged %}
+      <tr>
+        <td>{{ r.row }}</td>
+        <td>{{ r.product_number or "—" }}</td>
+        <td>{{ r.status }}</td>
+        <td>
+          <ul style="margin:0">
+            {% for m in r.messages %}<li>{{ m }}</li>{% endfor %}
+          </ul>
+        </td>
+      </tr>
+      {% endfor %}
+    </tbody>
+  </table>
+  {% else %}
+  <p class="muted">All rows imported cleanly.</p>
+  {% endif %}
+</div>
+{% endif %}

app/templates/admin/scope_import.html

@@ -0,0 +1,37 @@
+{% extends "base.html" %}
+{% block title %}Scope Import · {{ app_name }}{% endblock %}
+{% block body %}
+{% include "_nav.html" %}
+<main class="container">
+  <h1>Scope Import</h1>
+
+  <div class="card">
+    <h2>Upload spreadsheet</h2>
+    <p class="muted">
+      Upload an <code>.xlsx</code> file describing scopes (PartType = Scope). Rows are
+      matched on <code>product_number</code>; re-uploading updates existing scopes.
+    </p>
+    <form hx-post="/admin/scope-import"
+          hx-target="#scope-import-result"
+          hx-swap="innerHTML"
+          hx-encoding="multipart/form-data"
+          hx-indicator="#import-indicator">
+      <label>Spreadsheet
+        <input type="file" name="file" accept=".xlsx" required>
+      </label>
+      <button type="submit" class="btn btn-primary">Import</button>
+      <span id="import-indicator" class="htmx-indicator muted"> Importing…</span>
+    </form>
+    <details style="margin-top:1rem">
+      <summary class="muted">Expected columns</summary>
+      <ul class="muted">
+        {% for col in expected_columns %}
+        <li><code>{{ col }}</code></li>
+        {% endfor %}
+      </ul>
+    </details>
+  </div>
+
+  <div id="scope-import-result"></div>
+</main>
+{% endblock %}

app/web/admin.py

@@ -3,7 +3,16 @@
 import math
 import urllib.parse
 
-from fastapi import APIRouter, Depends, Form, HTTPException, Request, status
+from fastapi import (
+    APIRouter,
+    Depends,
+    File,
+    Form,
+    HTTPException,
+    Request,
+    UploadFile,
+    status,
+)
 from fastapi.responses import HTMLResponse, RedirectResponse, Response
 from sqlalchemy.ext.asyncio import AsyncSession
 
@@ -34,6 +43,7 @@ from app.models.user import User
 from app.services import augmented_part_sync as augmented_part_sync_service
 from app.services import fit_matching as fit_matching_service
 from app.services import odata as odata_service
+from app.services import scope_import as scope_import_service
 from app.web.templates import templates
 
 router = APIRouter(prefix="/admin", include_in_schema=False)
@@ -1344,6 +1354,64 @@ async def filter_imported_parts(
     )
 
 
+# --- Scope Import ------------------------------------------------------------
+
+
+@router.get("/scope-import", response_class=HTMLResponse)
+async def scope_import_page(
+    request: Request,
+    db: AsyncSession = Depends(get_db),
+    user: User = Depends(web_require("augmented_parts:write")),
+):
+    return templates.TemplateResponse(
+        request,
+        "admin/scope_import.html",
+        {"user": user, "expected_columns": scope_import_service.EXPECTED_COLUMNS},
+    )
+
+
+@router.post("/scope-import", response_class=HTMLResponse)
+async def upload_scope_import(
+    request: Request,
+    file: UploadFile = File(...),
+    db: AsyncSession = Depends(get_db),
+    actor: User = Depends(web_require("augmented_parts:write")),
+):
+    filename = file.filename or ""
+    if not filename.lower().endswith(".xlsx"):
+        return templates.TemplateResponse(
+            request,
+            "admin/_scope_import_result.html",
+            {"error": "Expected an .xlsx spreadsheet."},
+        )
+    data = await file.read()
+    try:
+        result = await scope_import_service.import_from_bytes(db, data)
+    except Exception as exc:  # noqa: BLE001
+        return templates.TemplateResponse(
+            request,
+            "admin/_scope_import_result.html",
+            {"error": f"Could not read spreadsheet: {exc}"},
+        )
+    await audit_crud.log(
+        db,
+        actor=actor,
+        action="update",
+        resource_type="AugmentedPart",
+        resource_id=None,
+        description=(
+            f"Scope import ({filename}): {result.created} created, "
+            f"{result.updated} updated, {result.skipped} skipped, "
+            f"{result.errors} errors"
+        ),
+    )
+    return templates.TemplateResponse(
+        request,
+        "admin/_scope_import_result.html",
+        {"result": result},
+    )
+
+
 # --- Augmented Parts ---------------------------------------------------------
 
 
@@ -2026,7 +2094,12 @@ async def _fit_view(db: AsyncSession, fit, part_id: int) -> dict:
             fit.augmented_part_a_id,
         )
     other_part = await augmented_part_crud.get(db, other_part_id)
-    return {"fit": fit, "own_ap": own_ap, "other_ap": other_ap, "other_part": other_part}
+    return {
+        "fit": fit,
+        "own_ap": own_ap,
+        "other_ap": other_ap,
+        "other_part": other_part,
+    }
 
 
 async def _augmented_part_fits_ctx(
@@ -2224,7 +2297,9 @@ async def delete_augmented_part_fit(
 ):
     fit = await fit_crud.get(db, fit_id)
     if fit is not None:
-        label = f"attachment points {fit.attachment_point_a_id}↔{fit.attachment_point_b_id}"
+        label = (
+            f"attachment points {fit.attachment_point_a_id}↔{fit.attachment_point_b_id}"
+        )
         await fit_crud.delete(db, fit)
         await audit_crud.log(
             db,