347 lines
12 KiB
Python
347 lines
12 KiB
Python
"""Synchronise AugmentedParts against the current ImportedParts table.
|
||
|
||
Algorithm
|
||
---------
|
||
Phase 1 – prune stale entries
|
||
• Find association rows whose imported_part_id no longer exists.
|
||
• Delete those rows.
|
||
• Delete any AugmentedPart that ends up with zero remaining associations
|
||
(i.e. it was created purely from now-deleted ImportedParts).
|
||
|
||
Phase 2 – create entries for new ImportedParts
|
||
• Any ImportedPart not yet linked to an AugmentedPart gets one created.
|
||
• Description ← ImportedPart.Description (fallback: InventoryID)
|
||
• Colour ← ImportedPart.TBXColour (fallback: "Unknown"; created if needed)
|
||
• Manufacturer ← "Armament" (created if needed)
|
||
• Model ← null
|
||
• Type ← first PartType whose name is a substring of the description
|
||
(longest name wins; skipped if no match)
|
||
• SubType ← null
|
||
• Attachment point ← always named "Ocular"; one of:
|
||
- ISO/UNC thread parsed from TBXThreadAttribute (using the
|
||
matching ProductMaxThreadDepth mm/in if populated), or
|
||
- StaticConnection from ProductODmm/ProductODin (whichever
|
||
unit is missing is derived via mm/in conversion) if the
|
||
thread attribute is absent or unparseable
|
||
Skipped entirely if none of those fields are usable.
|
||
"""
|
||
|
||
import re
|
||
|
||
from sqlalchemy import func, select
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from app.crud import attachment_point as attachment_point_crud
|
||
from app.crud import attachment_point_name as attachment_point_name_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 static_connection as static_connection_crud
|
||
from app.crud import unc_thread as unc_thread_crud
|
||
from app.models.associations import augmented_part_imported_parts
|
||
from app.models.augmented_part import AugmentedPart
|
||
from app.models.imported_part import ImportedPart
|
||
from app.models.part_type import PartType
|
||
from app.models.unc_thread import UncSeries
|
||
from app.schemas.augmented_part import AugmentedPartSyncResult
|
||
|
||
ATTACHMENT_POINT_NAME = "Ocular"
|
||
MM_PER_INCH = 25.4
|
||
|
||
# e.g.
|
||
# M8-1.25 - 8mm OD, 1.25 mm pitch
|
||
# M8 x 1.25
|
||
# M50x.75 - No leading digit on floats
|
||
# Float pattern: \d*\.?\d+
|
||
_ISO_THREAD_RE = re.compile(
|
||
r"M\s*(?P<diameter>\d*\.?\d+)\s*[xX-]\s*P?(?P<pitch>\d*\.?\d+)(?P<extra>.*)"
|
||
)
|
||
|
||
# e.g.
|
||
# 1.750-32 UNS
|
||
# 1.750-32
|
||
# 1.5625-40
|
||
#
|
||
# Leading digit / period, one or more digits, a dash, then
|
||
_UNC_THREAD_RE = re.compile(
|
||
r"(?P<diameter>\d*\.?\d+)\s*-\s*(?P<tpi>\d*\.?\d+)"
|
||
r"\s*(?P<series>UNEF|UNC|UNF|UNS|UN)?",
|
||
re.IGNORECASE,
|
||
)
|
||
|
||
|
||
def _parse_float(value: str | None) -> float | None:
|
||
if value is None:
|
||
return None
|
||
value = value.strip()
|
||
if not value:
|
||
return None
|
||
try:
|
||
return float(value)
|
||
except ValueError:
|
||
return None
|
||
|
||
|
||
def _parse_thread_attribute(value: str) -> tuple[str, dict] | None:
|
||
"""Parse a TBXThreadAttribute string as an ISO metric or UNC/UNF thread spec."""
|
||
if m := _ISO_THREAD_RE.match(value):
|
||
return "iso", {
|
||
"diameter_mm": float(m.group("diameter")),
|
||
"pitch_mm": float(m.group("pitch")),
|
||
}
|
||
|
||
if m := _UNC_THREAD_RE.match(value):
|
||
# Source data often omits the series letters; UNS ("unified special") is the
|
||
# correct designation when the diameter/TPI pairing isn't a standard pitch.
|
||
series = UncSeries((m.group("series") or "UNS").upper())
|
||
return "unc", {
|
||
"diameter_fraction": m.group("diameter"),
|
||
"tpi": int(m.group("tpi")),
|
||
"series": series,
|
||
}
|
||
|
||
return None
|
||
|
||
|
||
async def _get_or_create_attachment_point_name(db: AsyncSession, name: str):
|
||
apn = await attachment_point_name_crud.get_by_name(db, name)
|
||
if apn is None:
|
||
apn = await attachment_point_name_crud.create(db, name=name)
|
||
return apn
|
||
|
||
|
||
async def _add_attachment_point(db: AsyncSession, ap: AugmentedPart, ip: ImportedPart) -> None:
|
||
"""Infer and create an attachment point for a newly-synced AugmentedPart.
|
||
|
||
Preference order: TBXThreadAttribute (ISO/UNC thread) then ProductODmm/in
|
||
(StaticConnection). Falls through to the OD-based connection if the thread
|
||
attribute is present but unparseable.
|
||
"""
|
||
thread_value = (ip.TBXThreadAttribute or "").strip()
|
||
parsed = None
|
||
if thread_value:
|
||
parsed = _parse_thread_attribute(thread_value)
|
||
else:
|
||
parsed = _parse_thread_attribute(ip.Description or "")
|
||
|
||
if parsed is not None:
|
||
kind, fields = parsed
|
||
apn = await _get_or_create_attachment_point_name(db, ATTACHMENT_POINT_NAME)
|
||
|
||
if kind == "iso":
|
||
depth_mm = _parse_float(ip.ProductMaxThreadDepthmm)
|
||
thread = await iso_thread_crud.get_by_values(db, depth_mm=depth_mm, **fields)
|
||
if thread is None:
|
||
thread = await iso_thread_crud.create(db, depth_mm=depth_mm, **fields)
|
||
await attachment_point_crud.create(
|
||
db,
|
||
augmented_part_id=ap.id,
|
||
attachment_point_name_id=apn.id,
|
||
iso_thread_id=thread.id,
|
||
)
|
||
elif kind == "unc":
|
||
depth_in = _parse_float(ip.ProductMaxThreadDepthin)
|
||
thread = await unc_thread_crud.get_by_values(db, depth_in=depth_in, **fields)
|
||
if thread is None:
|
||
thread = await unc_thread_crud.create(db, depth_in=depth_in, **fields)
|
||
await attachment_point_crud.create(
|
||
db,
|
||
augmented_part_id=ap.id,
|
||
attachment_point_name_id=apn.id,
|
||
unc_thread_id=thread.id,
|
||
)
|
||
else:
|
||
pass
|
||
return
|
||
|
||
od_mm = _parse_float(ip.ProductODmm)
|
||
od_in = _parse_float(ip.ProductODin)
|
||
if od_mm is None and od_in is None:
|
||
return
|
||
if od_mm is None:
|
||
od_mm = od_in * MM_PER_INCH
|
||
elif od_in is None:
|
||
od_in = od_mm / MM_PER_INCH
|
||
|
||
apn = await _get_or_create_attachment_point_name(db, ATTACHMENT_POINT_NAME)
|
||
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_in)
|
||
|
||
await attachment_point_crud.create(
|
||
db,
|
||
augmented_part_id=ap.id,
|
||
attachment_point_name_id=apn.id,
|
||
static_connection_id=sc.id,
|
||
)
|
||
|
||
|
||
async def sync_one(
|
||
db: AsyncSession,
|
||
*,
|
||
inventory_id: str,
|
||
part_type_id: int,
|
||
sub_part_type_id: int | None,
|
||
) -> AugmentedPart:
|
||
"""Create an AugmentedPart for a single ImportedPart with manually supplied type."""
|
||
ip = await db.get(ImportedPart, inventory_id)
|
||
if ip is None:
|
||
raise ValueError(f"ImportedPart {inventory_id!r} not found")
|
||
|
||
description = (ip.Description or "").strip() or ip.InventoryID
|
||
|
||
colour_name = (ip.TBXColour or "").strip() or "Unknown"
|
||
colour = await part_colour_crud.get_by_name(db, colour_name)
|
||
if colour is None:
|
||
colour = await part_colour_crud.create(db, name=colour_name)
|
||
|
||
manufacturer = await manufacturer_crud.get_by_name(db, "Armament")
|
||
if manufacturer is None:
|
||
manufacturer = await manufacturer_crud.create(db, name="Armament", url="")
|
||
|
||
ap = AugmentedPart(
|
||
sku=ip.InventoryID,
|
||
description=description,
|
||
colour_id=colour.id,
|
||
manufacturer_id=manufacturer.id,
|
||
scope_model_id=None,
|
||
part_type_id=part_type_id,
|
||
sub_part_type_id=sub_part_type_id,
|
||
)
|
||
db.add(ap)
|
||
await db.flush()
|
||
|
||
await db.execute(
|
||
augmented_part_imported_parts.insert().values(
|
||
augmented_part_id=ap.id,
|
||
imported_part_id=ip.InventoryID,
|
||
)
|
||
)
|
||
await _add_attachment_point(db, ap, ip)
|
||
await db.commit()
|
||
return ap
|
||
|
||
|
||
def _infer_part_type(description: str, part_types: list[PartType]) -> PartType | None:
|
||
"""Return the best-matching PartType or None.
|
||
|
||
Sorted by name length descending so more specific types (e.g. "Flip Cover")
|
||
are preferred over shorter ones (e.g. "Scope").
|
||
"""
|
||
desc_lower = description.lower()
|
||
for pt in sorted(part_types, key=lambda p: len(p.name), reverse=True):
|
||
if pt.name.lower() in desc_lower:
|
||
return pt
|
||
return None
|
||
|
||
|
||
async def sync(db: AsyncSession) -> AugmentedPartSyncResult:
|
||
# ------------------------------------------------------------------ phase 1
|
||
|
||
# IDs referenced in existing associations
|
||
result = await db.execute(select(augmented_part_imported_parts.c.imported_part_id).distinct())
|
||
all_linked_ids: set[str] = {row[0] for row in result.all()}
|
||
|
||
# IDs that still exist in imported_parts
|
||
result = await db.execute(select(ImportedPart.InventoryID))
|
||
current_ids: set[str] = {row[0] for row in result.all()}
|
||
|
||
orphaned_import_ids = all_linked_ids - current_ids
|
||
deleted = 0
|
||
|
||
if orphaned_import_ids:
|
||
# Which AugmentedParts are affected?
|
||
result = await db.execute(
|
||
select(augmented_part_imported_parts.c.augmented_part_id).where(
|
||
augmented_part_imported_parts.c.imported_part_id.in_(orphaned_import_ids)
|
||
)
|
||
)
|
||
orphaned_ap_ids: set[int] = {row[0] for row in result.all()}
|
||
|
||
# Remove the stale association rows
|
||
await db.execute(
|
||
augmented_part_imported_parts.delete().where(
|
||
augmented_part_imported_parts.c.imported_part_id.in_(orphaned_import_ids)
|
||
)
|
||
)
|
||
await db.flush()
|
||
|
||
# Delete AugmentedParts that now have no remaining associations
|
||
for ap_id in orphaned_ap_ids:
|
||
result = await db.execute(
|
||
select(func.count())
|
||
.select_from(augmented_part_imported_parts)
|
||
.where(augmented_part_imported_parts.c.augmented_part_id == ap_id)
|
||
)
|
||
if result.scalar() == 0:
|
||
ap = await db.get(AugmentedPart, ap_id)
|
||
if ap is not None:
|
||
await db.delete(ap)
|
||
deleted += 1
|
||
|
||
await db.commit()
|
||
|
||
# ------------------------------------------------------------------ phase 2
|
||
|
||
# Re-query linked IDs after phase 1 cleanup
|
||
result = await db.execute(select(augmented_part_imported_parts.c.imported_part_id).distinct())
|
||
now_linked_ids: set[str] = {row[0] for row in result.all()}
|
||
unlinked_ids = current_ids - now_linked_ids
|
||
|
||
created = 0
|
||
skipped = 0
|
||
|
||
if not unlinked_ids:
|
||
return AugmentedPartSyncResult(created=created, deleted=deleted, skipped=skipped)
|
||
|
||
result = await db.execute(
|
||
select(ImportedPart).where(ImportedPart.InventoryID.in_(unlinked_ids))
|
||
)
|
||
new_parts = list(result.scalars().all())
|
||
|
||
# Ensure "Armament" manufacturer exists
|
||
manufacturer = await manufacturer_crud.get_by_name(db, "Armament")
|
||
if manufacturer is None:
|
||
manufacturer = await manufacturer_crud.create(db, name="Armament", url="")
|
||
|
||
part_types = await part_type_crud.list_all(db)
|
||
|
||
for ip in new_parts:
|
||
description = (ip.Description or "").strip() or ip.InventoryID
|
||
|
||
part_type = _infer_part_type(description, part_types)
|
||
if part_type is None:
|
||
skipped += 1
|
||
continue
|
||
|
||
colour_name = (ip.TBXColour or "").strip() or "Unknown"
|
||
colour = await part_colour_crud.get_by_name(db, colour_name)
|
||
if colour is None:
|
||
colour = await part_colour_crud.create(db, name=colour_name)
|
||
|
||
ap = AugmentedPart(
|
||
sku=ip.InventoryID,
|
||
description=description,
|
||
colour_id=colour.id,
|
||
manufacturer_id=manufacturer.id,
|
||
scope_model_id=None,
|
||
part_type_id=part_type.id,
|
||
sub_part_type_id=None,
|
||
)
|
||
db.add(ap)
|
||
await db.flush()
|
||
|
||
await db.execute(
|
||
augmented_part_imported_parts.insert().values(
|
||
augmented_part_id=ap.id,
|
||
imported_part_id=ip.InventoryID,
|
||
)
|
||
)
|
||
await _add_attachment_point(db, ap, ip)
|
||
created += 1
|
||
|
||
await db.commit()
|
||
|
||
return AugmentedPartSyncResult(created=created, deleted=deleted, skipped=skipped)
|