204 lines
7.0 KiB
Python
204 lines
7.0 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
|
||
"""
|
||
|
||
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
|
||
|
||
|
||
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=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=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 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=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)
|