"""Synchronise AugmentedParts against the current ImportedParts table. Algorithm --------- Phase 1 – archive stale entries • Find association rows whose imported_part_id no longer exists. • Delete those rows. • Archive (rather than delete) any AugmentedPart that ends up with zero remaining associations (i.e. it was created purely from now-deleted ImportedParts), preserving its fit/attachment-point history. 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. UAC diameter range ← ImportedPart.UACDiameterRangemm, an 8-digit string with no separators (2 leading digits + 2 decimal digits for the minimum, then the same for the maximum) reformatted as AugmentedPart.uac_diameter_range_mm = "##.##-##.##". When present, also adds a second attachment point named "Objective" with a ClampConnection spanning that range. """ 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 clamp_connection as clamp_connection_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" UAC_ATTACHMENT_POINT_NAME = "Objective" 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, ) _CLAMP_RANGE_RE = re.compile( r"\((?P\d*\.?\d+)\s*-\s*(?P\d*\.?\d+)\s*mm\)", 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, } if m := _CLAMP_RANGE_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. return "clamp", { "min_diametere": float(m.group("min_diameter")), "max_diameter": float(m.group("max_diameter")), } return None def _parse_uac_diameter_range(value: str | None) -> tuple[str, float, float] | None: """Parse an 8-digit UACDiameterRangemm value into ("##.##-##.##", min, max). The source field is 8 digits with no separators: the first 4 are the minimum diameter (2 whole + 2 decimal), the last 4 the maximum, e.g. "03505500" -> ("03.50-55.00", 3.5, 55.0). """ if not value: return None digits = "".join(re.findall(r"\d", value)) if len(digits) != 8: return None min_str = f"{digits[0:2]}.{digits[2:4]}" max_str = f"{digits[4:6]}.{digits[6:8]}" return f"{min_str}-{max_str}", float(min_str), float(max_str) async def _add_uac_clamp_attachment_point( db: AsyncSession, ap: AugmentedPart, ip: ImportedPart ) -> None: """Add a second, "Objective"-named clamp attachment point derived from ImportedPart.UACDiameterRangemm, if that field is populated.""" parsed = _parse_uac_diameter_range(ip.UACDiameterRangemm) if parsed is None: return _, min_mm, max_mm = parsed min_in = min_mm / MM_PER_INCH max_in = max_mm / MM_PER_INCH apn = await _get_or_create_attachment_point_name(db, UAC_ATTACHMENT_POINT_NAME) clamp = await clamp_connection_crud.get_by_values(db, min_mm=min_mm, max_mm=max_mm) if clamp is None: clamp = await clamp_connection_crud.create( db, min_mm=min_mm, max_mm=max_mm, min_in=min_in, max_in=max_in ) await attachment_point_crud.create( db, augmented_part_id=ap.id, attachment_point_name_id=apn.id, clamp_connection_id=clamp.id, ) 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, ) return 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, ) return elif kind == "clamp": clamp = await clamp_connection_crud.get_by_values(db, **fields) if clamp is None: clamp = await clamp_connection_crud.create(db, **fields) await attachment_point_crud.create( db, augmented_part_id=ap.id, attachment_point_name_id=apn.id, clamp_connection_id=clamp.id, ) return else: # Unable to determine pass 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="") uac_range = _parse_uac_diameter_range(ip.UACDiameterRangemm) 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, uac_diameter_range_mm=uac_range[0] if uac_range else 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) await _add_uac_clamp_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 _archive_orphans(db: AsyncSession, candidate_ap_ids: set[int]) -> int: """Archive any of `candidate_ap_ids` left with zero linked ImportedParts. Idempotent: already-archived parts and parts that still have a link are left untouched. Does not commit — the caller commits as part of its own transaction. """ archived = 0 for ap_id in candidate_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 and not ap.archived: ap.archived = True archived += 1 return archived async def _detach_and_archive_orphans( db: AsyncSession, imported_part_ids: set[str] ) -> int: """Detach `imported_part_ids` from any AugmentedPart association and archive AugmentedParts left with zero remaining ImportedPart links as a result. Deletes the association rows directly rather than relying on the ImportedPart row's ON DELETE CASCADE, since SQLite (this app's dev/test database) doesn't enforce FK actions by default. Safe to call whether or not the ImportedPart row(s) still exist. Does not commit — the caller commits as part of its own transaction. """ if not imported_part_ids: return 0 result = await db.execute( select(augmented_part_imported_parts.c.augmented_part_id).where( augmented_part_imported_parts.c.imported_part_id.in_(imported_part_ids) ) ) candidate_ap_ids: set[int] = {row[0] for row in result.all()} await db.execute( augmented_part_imported_parts.delete().where( augmented_part_imported_parts.c.imported_part_id.in_(imported_part_ids) ) ) await db.flush() return await _archive_orphans(db, candidate_ap_ids) async def delete_imported_part_and_archive_orphans( db: AsyncSession, imported_part: ImportedPart ) -> int: """Delete `imported_part` and archive any AugmentedPart left with no remaining ImportedPart links as a result. Captures affected AugmentedPart ids before deleting so this is correct regardless of whether the DB enforces the association table's ON DELETE CASCADE. Returns the number of AugmentedParts newly archived. """ inventory_id = imported_part.InventoryID result = await db.execute( select(augmented_part_imported_parts.c.augmented_part_id).where( augmented_part_imported_parts.c.imported_part_id == inventory_id ) ) candidate_ap_ids: set[int] = {row[0] for row in result.all()} await db.delete(imported_part) await db.execute( augmented_part_imported_parts.delete().where( augmented_part_imported_parts.c.imported_part_id == inventory_id ) ) await db.flush() archived = await _archive_orphans(db, candidate_ap_ids) await db.commit() return archived 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 archived = 0 if orphaned_import_ids: archived = await _detach_and_archive_orphans(db, orphaned_import_ids) 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, archived=archived, 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) uac_range = _parse_uac_diameter_range(ip.UACDiameterRangemm) 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, uac_diameter_range_mm=uac_range[0] if uac_range else 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) await _add_uac_clamp_attachment_point(db, ap, ip) created += 1 await db.commit() return AugmentedPartSyncResult(created=created, archived=archived, skipped=skipped)