Zones Services - Zone Assignment Business Logic

Zone (physical location within store)
 ↓
ZoneInventory (if zone has inventory configuration)
 ↓
ParLevelItem (assigns product variant to zone with ideal quantity)
 ↓
InventoryLevel (tracks actual inventory at store for that variant)

ParLevelItem:
- product: FK to Product
- variant_sku: String (matches ProductVariant.sku)
- quantity: Integer (par level - ideal quantity)
- zone_inventory: FK to ZoneInventory

InventoryLevel:
- product: FK to Product
- variant_sku: String (matches ProductVariant.sku)
- quantity_on_hand: Integer (actual quantity)
- store: FK to Store

class ZoneAssignmentService:
    """
    Service for managing zone-driven product assignments and inventory infrastructure.
    Implements the core logic for "what should be where" through par levels.
    """

    def __init__(self):
        self.Product = apps.get_model('products', 'Product')
        self.ProductVariant = apps.get_model('products', 'ProductVariant')
        self.InventoryLevel = apps.get_model('products', 'InventoryLevel')
        self.Store = apps.get_model('stores', 'Store')

def create_inventory_infrastructure_for_zone(self, zone: Zone) -> Dict[str, int]:
    """
    Create InventoryLevel records for all par level items in a zone.

    Returns:
        Dict with counts of created, existing, and failed records
    """

{
    'created': 5,        # New InventoryLevel records created
    'existing': 3,       # InventoryLevel records already existed
    'failed': 0,         # Failed to create
    'errors': []         # List of error messages
}

# 1. Check if zone has inventory configuration
if zone.zone_type != 'inventory':
    logger.info(f"Zone {zone.zone_name} is not an inventory zone, skipping")
    return stats

# 2. Get zone inventory and par levels
zone_inventory = zone.zone_inventory
par_levels = zone_inventory.par_levels.all()

# 3. For each par level item
for par_level in par_levels:
    try:
        # 4. Ensure InventoryLevel exists (calls ParLevelItem.ensure_inventory_level_exists())
        inventory_level = par_level.ensure_inventory_level_exists()

        # 5. Track statistics
        if inventory_level:
            if hasattr(inventory_level, '_created') or inventory_level.quantity_on_hand == 0:
                stats['created'] += 1
            else:
                stats['existing'] += 1
    except Exception as e:
        stats['failed'] += 1
        stats['errors'].append(str(e))

# New zone created in Sanity, synced to Django
zone = Zone.objects.get(sanity_id='zone_123')

service = ZoneAssignmentService()
stats = service.create_inventory_infrastructure_for_zone(zone)

logger.info(f"Created infrastructure: {stats}")
# Output: Created infrastructure: {'created': 10, 'existing': 0, 'failed': 0, 'errors': []}

# Admin added 5 new par level items to existing zone
zone = Zone.objects.get(pk=zone_id)

service = ZoneAssignmentService()
stats = service.create_inventory_infrastructure_for_zone(zone)

# Output might be: {'created': 5, 'existing': 15, 'failed': 0, 'errors': []}

# Rebuild infrastructure for all zones at a store
store = Store.objects.get(pk=store_id)
zones = Zone.objects.filter(store_location=store, zone_type='inventory')

service = ZoneAssignmentService()
for zone in zones:
    stats = service.create_inventory_infrastructure_for_zone(zone)
    logger.info(f"Zone {zone.zone_name}: {stats}")

def sync_par_levels_to_inventory(self, store: Optional['Store'] = None) -> Dict[str, int]:
    """
    Sync all par level assignments to inventory infrastructure.

    Args:
        store: Optional store to limit sync to specific store

    Returns:
        Dict with sync statistics
    """

{
    'zones_processed': 25,
    'par_levels_processed': 342,
    'inventory_levels_created': 98,
    'inventory_levels_existing': 244,
    'errors': []
}

# 1. Get zones to process
zones_queryset = Zone.objects.filter(zone_type='inventory')
if store:
    zones_queryset = zones_queryset.filter(store_location=store)

# 2. Optimize queries
zones = zones_queryset.select_related('zone_inventory').prefetch_related(
    'zone_inventory__par_levels__product'
)

# 3. Process each zone
for zone in zones:
    zone_stats = self.create_inventory_infrastructure_for_zone(zone)

    # 4. Aggregate statistics
    stats['zones_processed'] += 1
    stats['inventory_levels_created'] += zone_stats['created']
    stats['inventory_levels_existing'] += zone_stats['existing']
    stats['errors'].extend(zone_stats['errors'])

# Set up inventory infrastructure for entire system
service = ZoneAssignmentService()
stats = service.sync_par_levels_to_inventory()

logger.info(f"System sync complete: {stats}")
# Output: System sync complete: {'zones_processed': 50, 'par_levels_processed': 1250, ...}

# Sync just one store after bulk par level import
store = Store.objects.get(name='Downtown Location')

service = ZoneAssignmentService()
stats = service.sync_par_levels_to_inventory(store=store)

logger.info(f"Store sync complete: {stats}")

# management/commands/sync_zone_inventory.py
from nextango.apps.zones.services import ZoneAssignmentService

class Command(BaseCommand):
    def handle(self, *args, **options):
        service = ZoneAssignmentService()
        stats = service.sync_par_levels_to_inventory()

        self.stdout.write(f"Zones processed: {stats['zones_processed']}")
        self.stdout.write(f"Inventory levels created: {stats['inventory_levels_created']}")

def get_zone_assignments_for_product(
    self,
    product: 'Product',
    variant_sku: str
) -> List[Dict[str, Any]]:
    """
    Get all zone assignments for a specific product variant.

    Returns:
        List of zone assignment dictionaries
    """

[
    {
        'zone_id': 'uuid-123',
        'zone_name': 'Coffee Bar',
        'store_id': 'uuid-456',
        'store_name': 'Downtown Location',
        'par_quantity': 10,
        'zone_type': 'inventory'
    },
    {
        'zone_id': 'uuid-789',
        'zone_name': 'Coffee Display',
        'store_id': 'uuid-012',
        'store_name': 'Uptown Location',
        'par_quantity': 15,
        'zone_type': 'inventory'
    }
]

# 1. Find all par level assignments for product variant
par_levels = ParLevelItem.objects.filter(
    product=product,
    variant_sku=variant_sku
).select_related(
    'zone_inventory__zone__store_location'
)

# 2. Build assignment list
for par_level in par_levels:
    zone = par_level.zone_inventory.zone
    assignments.append({
        'zone_id': zone.pk,
        'zone_name': zone.zone_name,
        'store_id': zone.store_location.pk,
        'store_name': zone.store_location.name,
        'par_quantity': par_level.quantity,
        'zone_type': zone.zone_type
    })

# Show where a product should be located
product = Product.objects.get(name='Dark Roast Coffee')
variant_sku = 'DARK-12OZ'

service = ZoneAssignmentService()
assignments = service.get_zone_assignments_for_product(product, variant_sku)

for assignment in assignments:
    print(f"{assignment['store_name']} - {assignment['zone_name']}: {assignment['par_quantity']} units")

# Output:
# Downtown Location - Coffee Bar: 10 units
# Uptown Location - Coffee Display: 15 units

# Compare par levels across all stores for a variant
assignments = service.get_zone_assignments_for_product(product, variant_sku)

for assignment in assignments:
    # Get actual inventory level
    inventory_level = InventoryLevel.objects.get(
        product=product,
        variant_sku=variant_sku,
        store_id=assignment['store_id']
    )

    discrepancy = inventory_level.quantity_on_hand - assignment['par_quantity']
    print(f"{assignment['store_name']}: {discrepancy} units {'over' if discrepancy > 0 else 'under'} par")

# /api/products/{product_id}/assignments/
@action(detail=True, methods=['get'])
def assignments(self, request, pk=None):
    product = self.get_object()
    variant_sku = request.query_params.get('variant_sku')

    service = ZoneAssignmentService()
    assignments = service.get_zone_assignments_for_product(product, variant_sku)

    return Response(assignments)

def get_products_for_zone(self, zone: Zone) -> List[Dict[str, Any]]:
    """
    Get all products assigned to a specific zone through par levels.

    Returns:
        List of product assignment dictionaries
    """

[
    {
        'product_id': 'uuid-123',
        'product_name': 'Dark Roast Coffee',
        'variant_sku': 'DARK-12OZ',
        'variant_name': '12 oz Bag',
        'par_quantity': 10,
        'par_level_id': 'uuid-456'
    },
    {
        'product_id': 'uuid-789',
        'product_name': 'Light Roast Coffee',
        'variant_sku': 'LIGHT-12OZ',
        'variant_name': '12 oz Bag',
        'par_quantity': 8,
        'par_level_id': 'uuid-012'
    }
]

# 1. Check if zone has inventory configuration
if zone.zone_type != 'inventory' or not hasattr(zone, 'zone_inventory'):
    return []

# 2. Get par levels with related products
par_levels = zone.zone_inventory.par_levels.select_related('product')

# 3. Build product list
for par_level in par_levels:
    variant = par_level.get_product_variant()  # Calls ProductVariant lookup
    products.append({
        'product_id': par_level.product.pk,
        'product_name': par_level.product.name,
        'variant_sku': par_level.variant_sku,
        'variant_name': variant.name if variant else 'Unknown Variant',
        'par_quantity': par_level.quantity,
        'par_level_id': par_level.pk
    })

# Generate checklist for zone stockingzone = Zone.objects.get(zone_name='Coffee Bar')

service = ZoneAssignmentService()
products = service.get_products_for_zone(zone)

print(f"Stocking checklist for {zone.zone_name}:")
for product in products:
    print(f"- {product['product_name']} ({product['variant_name']}): {product['par_quantity']} units")

# Output:
# Stocking checklist for Coffee Bar:
# - Dark Roast Coffee (12 oz Bag): 10 units
# - Light Roast Coffee (12 oz Bag): 8 units

# Check if zone has correct products stocked
zone = Zone.objects.get(pk=zone_id)
products = service.get_products_for_zone(zone)

for product in products:
    inventory_level = InventoryLevel.objects.get(
        product_id=product['product_id'],
        variant_sku=product['variant_sku'],
        store=zone.store_location
    )

    actual = inventory_level.quantity_on_hand
    par = product['par_quantity']

    if actual < par:
        print(f"{product['product_name']} below par: {actual}/{par}")
    else:
        print(f"{product['product_name']} at par: {actual}/{par}")

# /api/zones/{zone_id}/products/
@action(detail=True, methods=['get'])
def products(self, request, pk=None):
    zone = self.get_object()

    service = ZoneAssignmentService()
    products = service.get_products_for_zone(zone)

    return Response(products)

def validate_sku_assignments(self, zone_inventory: ZoneInventory) -> Dict[str, Any]:
    """
    Validate that all par level SKUs match actual ProductVariant records.

    Returns:
        Dict with validation results
    """

{
    'valid_count': 18,
    'invalid_count': 2,
    'total_count': 20,
    'invalid_items': [
        {
            'par_level_id': 'uuid-123',
            'product_name': 'Dark Roast Coffee',
            'variant_sku': 'DARK-16OZ-TYPO',
            'error': "No ProductVariant with SKU 'DARK-16OZ-TYPO' found for product 'Dark Roast Coffee'"
        },
        {
            'par_level_id': 'uuid-456',
            'product_name': 'Light Roast Coffee',
            'variant_sku': 'DELETED-SKU',
            'error': "No ProductVariant with SKU 'DELETED-SKU' found for product 'Light Roast Coffee'"
        }
    ]
}

# 1. Get all par levels for zone inventory
par_levels = zone_inventory.par_levels.select_related('product')

# 2. Validate each par level
for par_level in par_levels:
    variant = par_level.get_product_variant()  # Calls ProductVariant lookup

    if variant:
        results['valid_count'] += 1
    else:
        results['invalid_count'] += 1
        results['invalid_items'].append({
            'par_level_id': par_level.pk,
            'product_name': par_level.product.name,
            'variant_sku': par_level.variant_sku,
            'error': f"No ProductVariant with SKU '{par_level.variant_sku}' found"
        })

# After bulk import from Sanity, validate all SKUs
zones = Zone.objects.filter(zone_type='inventory')

service = ZoneAssignmentService()
total_invalid = 0

for zone in zones:
    if hasattr(zone, 'zone_inventory'):
        results = service.validate_sku_assignments(zone.zone_inventory)

        if results['invalid_count'] > 0:
            print(f"{zone.zone_name}: {results['invalid_count']} invalid SKUs")
            for item in results['invalid_items']:
                print(f"  - {item['product_name']}: {item['variant_sku']}")
            total_invalid += results['invalid_count']

if total_invalid == 0:
    print("All SKU assignments are valid")

# management/commands/validate_zone_skus.py
class Command(BaseCommand):
    def handle(self, *args, **options):
        service = ZoneAssignmentService()
        zone_inventories = ZoneInventory.objects.all()

        for zone_inv in zone_inventories:
            results = service.validate_sku_assignments(zone_inv)

            if results['invalid_count'] > 0:
                self.stdout.write(self.style.WARNING(
                    f"{zone_inv.zone.zone_name}: {results['invalid_count']} invalid"
                ))

                for item in results['invalid_items']:
                    self.stdout.write(f"  {item['error']}")

# Before syncing par levels to inventory, validate SKUs
def safe_sync_par_levels_to_inventory(store):
    service = ZoneAssignmentService()

    # Step 1: Validate all SKUs
    zones = Zone.objects.filter(store_location=store, zone_type='inventory')
    has_errors = False

    for zone in zones:
        if hasattr(zone, 'zone_inventory'):
            results = service.validate_sku_assignments(zone.zone_inventory)
            if results['invalid_count'] > 0:
                logger.error(f"Zone {zone.zone_name} has invalid SKUs")
                has_errors = True

    if has_errors:
        raise ValueError("Cannot sync - fix invalid SKUs first")

    # Step 2: Sync infrastructure
    stats = service.sync_par_levels_to_inventory(store=store)
    return stats

def cleanup_orphaned_inventory_levels(self, store: Optional['Store'] = None) -> Dict[str, int]:
    """
    Remove InventoryLevel records that no longer have corresponding par level assignments.

    Args:
        store: Optional store to limit cleanup to specific store

    Returns:
        Dict with cleanup statistics
    """

{
    'checked': 150,
    'removed': 8,
    'errors': []
}

# 1. Get inventory levels to check
inventory_levels_queryset = self.InventoryLevel.objects.all()
if store:
    inventory_levels_queryset = inventory_levels_queryset.filter(store=store)

# 2. Check each inventory level
for inventory_level in inventory_levels:
    # 3. Check if there's a corresponding par level assignment
    par_level_exists = ParLevelItem.objects.filter(
        product=inventory_level.product,
        variant_sku=inventory_level.variant_sku,
        zone_inventory__zone__store_location=inventory_level.store
    ).exists()

    # 4. If no par level exists, remove inventory level
    if not par_level_exists:
        logger.info(f"Removing orphaned InventoryLevel: {inventory_level}")
        inventory_level.delete()
        stats['removed'] += 1

# Admin removed products from zone in Sanity, cleanup orphaned inventory
store = Store.objects.get(name='Downtown Location')

service = ZoneAssignmentService()
stats = service.cleanup_orphaned_inventory_levels(store=store)

logger.info(f"Cleanup complete: {stats['removed']} orphaned inventory levels removed")

# Scheduled job to cleanup orphaned inventory levels
from nextango.apps.zones.services import ZoneAssignmentService

def scheduled_inventory_cleanup():
    service = ZoneAssignmentService()

    # Cleanup all stores
    stats = service.cleanup_orphaned_inventory_levels()

    logger.info(f"System-wide cleanup: {stats}")
    return stats

# management/commands/cleanup_orphaned_inventory.py
class Command(BaseCommand):
    def add_arguments(self, parser):
        parser.add_argument('--store-id', type=str, help='Store ID to cleanup')

    def handle(self, *args, **options):
        service = ZoneAssignmentService()

        if options['store_id']:
            store = Store.objects.get(pk=options['store_id'])
            stats = service.cleanup_orphaned_inventory_levels(store=store)
            self.stdout.write(f"Store cleanup: {stats}")
        else:
            stats = service.cleanup_orphaned_inventory_levels()
            self.stdout.write(f"System cleanup: {stats}")

# Before major zone reconfiguration, cleanup orphaned records
def prepare_for_zone_migration(store):
    service = ZoneAssignmentService()

    # Step 1: Cleanup orphaned inventory levels
    cleanup_stats = service.cleanup_orphaned_inventory_levels(store=store)
    logger.info(f"Pre-migration cleanup: {cleanup_stats}")

    # Step 2: Validate remaining SKUs
    zones = Zone.objects.filter(store_location=store, zone_type='inventory')
    for zone in zones:
        if hasattr(zone, 'zone_inventory'):
            validation = service.validate_sku_assignments(zone.zone_inventory)
            if validation['invalid_count'] > 0:
                raise ValueError(f"Invalid SKUs found in {zone.zone_name}")

    # Step 3: Sync new infrastructure
    sync_stats = service.sync_par_levels_to_inventory(store=store)
    logger.info(f"Post-migration sync: {sync_stats}")

try:
    # Service logic
    inventory_level = par_level.ensure_inventory_level_exists()
    stats['created'] += 1
except Exception as e:
    # Log error
    error_msg = f"Failed to create inventory level for par level {par_level.pk}: {e}"
    stats['errors'].append(error_msg)
    stats['failed'] += 1
    logger.error(error_msg)