"""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 "Occular"; 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 = "Occular" 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\d*\.?\d+)\s*[xX-]\s*P?(?P\d*\.?\d+)(?P.*)$" ) # 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\d*\.?\d+)\s*-\s*(?P\d*\.?\d+)" r"\s*(?PUNEF|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 = _parse_thread_attribute(thread_value) if thread_value else None 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)