514 lines
20 KiB
Python
514 lines
20 KiB
Python
"""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 two
|
|
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_lens_diameter_mm → ScopeInfo (descriptive only, no attachment point)
|
|
ocular_outer_diameter_mm → StaticConnection + AttachmentPoint "Ocular"
|
|
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"). If the referenced part has no attachment points at all, one
|
|
is created for it that mirrors the scope's attachment point's connection, so
|
|
the fit has something valid to attach to. If the scope itself has no Ocular/
|
|
Objective attachment point (no ocular_outer_diameter_mm/objective_thread, or
|
|
an unparseable one), a Manual attachment point — no connection details, never
|
|
picked up by the fit-matching algorithm — is created on the scope under the
|
|
appropriate label so the fit still has somewhere to attach. 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_diameter_mm",
|
|
"model_suffix",
|
|
"sub_part_type",
|
|
"product_number",
|
|
"url",
|
|
"ocular_outer_diameter_mm",
|
|
"ocular_fit",
|
|
"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 _mirrored_ap(db: AsyncSession, *, augmented_part_id: int, scope_ap):
|
|
"""Create an attachment point mirroring ``scope_ap``'s connection.
|
|
|
|
Used when the referenced part has no attachment point that already fits
|
|
the scope, so the explicit ocular_fit/objective_fit link has something
|
|
valid to attach to.
|
|
"""
|
|
return await attachment_point_crud.create(
|
|
db,
|
|
augmented_part_id=augmented_part_id,
|
|
attachment_point_name_id=scope_ap.attachment_point_name_id,
|
|
iso_thread_id=scope_ap.iso_thread_id,
|
|
unc_thread_id=scope_ap.unc_thread_id,
|
|
clamp_connection_id=scope_ap.clamp_connection_id,
|
|
static_connection_id=scope_ap.static_connection_id,
|
|
)
|
|
|
|
|
|
async def _ensure_manual_scope_ap(
|
|
db: AsyncSession,
|
|
*,
|
|
scope_ap,
|
|
augmented_part_id: int,
|
|
label: str,
|
|
messages: list[str],
|
|
):
|
|
"""Return ``scope_ap``, creating a Manual attachment point named ``label``
|
|
on the scope if it's None.
|
|
|
|
Used so an ocular_fit/objective_fit reference still has something to link
|
|
from even when the scope has no ocular_outer_diameter_mm/objective_thread
|
|
of its own.
|
|
"""
|
|
if scope_ap is not None:
|
|
return scope_ap
|
|
apn = await apn_crud.get_by_name(db, label)
|
|
if apn is None:
|
|
apn = await apn_crud.create(db, name=label)
|
|
messages.append(
|
|
f"scope has no {label.lower()} attachment point; created a manual one to record the fit"
|
|
)
|
|
return await attachment_point_crud.create(
|
|
db,
|
|
augmented_part_id=augmented_part_id,
|
|
attachment_point_name_id=apn.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.
|
|
|
|
``scope_ap`` must already exist — callers ensure this via
|
|
``_ensure_manual_scope_ap`` before calling. If the referenced part has no
|
|
attachment points at all, one is created mirroring the scope's attachment
|
|
point's connection.
|
|
"""
|
|
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 target_aps:
|
|
target_ap = target_aps[0]
|
|
else:
|
|
target_ap = await _mirrored_ap(
|
|
db, augmented_part_id=target.id, scope_ap=scope_ap
|
|
)
|
|
messages.append(
|
|
f"{scope_role}_fit '{target_sku}': referenced part had no attachment points, "
|
|
"created one matching the scope's attachment point"
|
|
)
|
|
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
|
|
model_suffix = _text(row.get("model_suffix"))
|
|
model = _text(row.get("model"))
|
|
magnification_min = _num(row.get("magnification_min"))
|
|
magnification_max = _num(row.get("magnification_max"))
|
|
objective_lens_diameter_mm = _num(
|
|
row.get("objective_lens_diameter_mm")
|
|
) or _num(row.get("objective_lens"))
|
|
|
|
# Construct the model number
|
|
default_product_number = model
|
|
if magnification_min:
|
|
default_product_number += f" {magnification_min}-{magnification_max}"
|
|
if objective_lens_diameter_mm:
|
|
default_product_number += f"x{objective_lens_diameter_mm}"
|
|
if model_suffix:
|
|
default_product_number += f" {model_suffix}"
|
|
product_number = _text(row.get("product_number")) or default_product_number
|
|
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_lens_diameter_mm=_num(row.get("objective_lens_diameter_mm")),
|
|
)
|
|
|
|
# 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,
|
|
)
|
|
|
|
# The objective attachment point comes from the thread alone — the
|
|
# lens diameter is a descriptive attribute, not a connection.
|
|
objective_thread_spec = _text(row.get("objective_thread"))
|
|
objective_ap = None
|
|
if objective_thread_spec:
|
|
objective_ap = await _thread_ap(
|
|
db,
|
|
augmented_part_id=part.id,
|
|
label=OBJECTIVE_LABEL,
|
|
spec=objective_thread_spec,
|
|
)
|
|
if objective_ap is None:
|
|
result.messages.append(
|
|
f"could not parse objective_thread '{objective_thread_spec}'"
|
|
)
|
|
|
|
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:
|
|
ocular_ap = await _ensure_manual_scope_ap(
|
|
db,
|
|
scope_ap=ocular_ap,
|
|
augmented_part_id=part_id,
|
|
label=OCULAR_LABEL,
|
|
messages=result.messages,
|
|
)
|
|
fits = ocular_fit
|
|
if "/" in fits:
|
|
fits = [x.strip() for x in fits.split("/")]
|
|
else:
|
|
fits = [fits]
|
|
for ocular_fit in fits:
|
|
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:
|
|
objective_ap = await _ensure_manual_scope_ap(
|
|
db,
|
|
scope_ap=objective_ap,
|
|
augmented_part_id=part_id,
|
|
label=OBJECTIVE_LABEL,
|
|
messages=result.messages,
|
|
)
|
|
fits = objective_fit
|
|
if "/" in fits:
|
|
fits = [x.strip() for x in fits.split("/")]
|
|
else:
|
|
fits = [fits]
|
|
for objective_fit in fits:
|
|
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)
|