"""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 """ from sqlalchemy import func, select from sqlalchemy.ext.asyncio import AsyncSession 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.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.schemas.augmented_part import AugmentedPartSyncResult 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=None, 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, ) ) created += 1 await db.commit() return AugmentedPartSyncResult(created=created, deleted=deleted, skipped=skipped)