""" Import spare parts from the legacy Excel workbook (e.g. SAMS AUTO SPARE.xlsx). Dry-run by default: prints a report of what would happen and writes nothing. Pass --commit to actually create records. Usage: python manage.py import_legacy_excel --file "/path/to/SAMS AUTO SPARE.xlsx" --user manager1 python manage.py import_legacy_excel --file "/path/to/SAMS AUTO SPARE.xlsx" --user manager1 --commit """ import os import re from decimal import Decimal, InvalidOperation from django.contrib.auth import get_user_model from django.core.management.base import BaseCommand, CommandError from django.db import transaction from django.utils import timezone from inventory.models import BatchImport, SparePart, StockMovement, Unit from inventory.services import HEADER_ALIASES, _normalize_header, _parse_quantity, _to_decimal, derive_prices try: import openpyxl except ImportError: openpyxl = None # Column meanings (QNTY, KUNUNUA, KUUZA, BEI YA MWISHO; sold/remain/TOTAL ignored) are defined once in # inventory.services so the web upload and this command can never disagree. REQUIRED_COLUMNS = ('item_code', 'description', 'cost_price', 'quantity') class Command(BaseCommand): help = 'Import spare parts from the legacy Excel workbook (dry-run unless --commit is passed).' def add_arguments(self, parser): parser.add_argument('--file', required=True, help='Path to the .xlsx workbook.') parser.add_argument('--sheet', default='KUUZIA', help="Sheet name to import (default: 'KUUZIA').") parser.add_argument('--user', required=True, help='Username to attribute this import to.') parser.add_argument( '--commit', action='store_true', help='Actually write to the database. Without this flag, only a report is printed.', ) def handle(self, *args, **options): if openpyxl is None: raise CommandError('openpyxl is not installed. Run: pip install -r requirements.txt') User = get_user_model() try: user = User.objects.get(username=options['user']) except User.DoesNotExist: raise CommandError(f"No user named '{options['user']}'.") file_path = options['file'] sheet_name = options['sheet'] commit = options['commit'] try: workbook = openpyxl.load_workbook(file_path, data_only=True) except FileNotFoundError: raise CommandError(f'File not found: {file_path}') if sheet_name not in workbook.sheetnames: raise CommandError( f"Sheet '{sheet_name}' not found. Available sheets: {', '.join(workbook.sheetnames)}" ) worksheet = workbook[sheet_name] rows = list(worksheet.iter_rows(values_only=True)) if not rows: raise CommandError('Sheet is empty.') columns = self._map_columns(rows[0]) missing = [key for key in REQUIRED_COLUMNS if key not in columns] if missing: raise CommandError( f"Sheet '{sheet_name}' is missing recognizable column(s): {', '.join(missing)}" ) to_create, skipped_blank, skipped_duplicate, errors = self._parse_rows(rows[1:], columns) total_rows = len(rows) - 1 self._print_report(sheet_name, total_rows, to_create, skipped_blank, skipped_duplicate, errors) if not commit: self.stdout.write(self.style.WARNING('Dry run only — nothing was written. Pass --commit to import.')) return batch_id, created_count = self._commit_import( to_create, skipped_blank, skipped_duplicate, errors, file_path, user, ) self.stdout.write(self.style.SUCCESS(f'Batch {batch_id}: created {created_count} spare parts.')) def _map_columns(self, header_row): columns = {} for index, raw_header in enumerate(header_row): key = HEADER_ALIASES.get(_normalize_header(raw_header)) if key: columns[key] = index return columns def _parse_rows(self, data_rows, columns): def cell(row, key): index = columns.get(key) return row[index] if index is not None and index < len(row) else None existing_codes = set(SparePart.objects.values_list('part_number', flat=True)) seen_codes = set() to_create, skipped_blank, skipped_duplicate, errors = [], [], [], [] for row in data_rows: if row is None or all(value is None for value in row): continue item_code = cell(row, 'item_code') if item_code is None or not str(item_code).strip(): skipped_blank.append(row) continue part_number = str(item_code).strip() if part_number in existing_codes or part_number in seen_codes: skipped_duplicate.append(part_number) continue buying_price = _to_decimal(cell(row, 'cost_price')) if buying_price is None: errors.append((part_number, 'missing/invalid buying price (UNITY PRICE KUNUNUA)')) continue quantity, _note = _parse_quantity(cell(row, 'quantity')) try: minimum_price, maximum_price = derive_prices( quantity, _to_decimal(cell(row, 'min_total')), _to_decimal(cell(row, 'max_total')), ) except ValueError as exc: errors.append((part_number, str(exc))) continue selling_price = minimum_price if minimum_price is not None else maximum_price if selling_price is None: errors.append((part_number, 'missing selling price (UNITY PRICE KUUZA / BEI YA MWISHO)')) continue if minimum_price is not None and maximum_price is not None and minimum_price > maximum_price: errors.append((part_number, 'minimum selling price is higher than the maximum')) continue name = str(cell(row, 'description') or '').strip() or part_number seen_codes.add(part_number) to_create.append({ 'part_number': part_number, 'name': name, 'buying_price': buying_price, 'selling_price': selling_price, 'minimum_price': minimum_price, 'maximum_price': maximum_price, 'quantity': quantity, }) return to_create, skipped_blank, skipped_duplicate, errors def _print_report(self, sheet_name, total_rows, to_create, skipped_blank, skipped_duplicate, errors): self.stdout.write(f"Sheet '{sheet_name}': {total_rows} data rows") self.stdout.write(f' valid to import : {len(to_create)}') self.stdout.write(f' blank (no code) : {len(skipped_blank)}') self.stdout.write(f' duplicate codes : {len(skipped_duplicate)}') self.stdout.write(f' errors : {len(errors)}') if errors: self.stdout.write(' -- sample errors --') for code, reason in errors[:25]: self.stdout.write(f' {code}: {reason}') if len(errors) > 25: self.stdout.write(f' ... and {len(errors) - 25} more') def _commit_import(self, to_create, skipped_blank, skipped_duplicate, errors, file_path, user): batch_id = f'BAT-{timezone.now():%Y%m%d-%H%M%S}' with transaction.atomic(): unit, _ = Unit.objects.get_or_create(name='Piece', defaults={'abbreviation': 'pc'}) created_parts = [] for row in to_create: part = SparePart.objects.create( name=row['name'], part_number=row['part_number'], category=None, unit=unit, buying_price=row['buying_price'], selling_price=row['selling_price'], minimum_price=row['minimum_price'], maximum_price=row['maximum_price'], quantity=row['quantity'], ) created_parts.append(part) if row['quantity']: StockMovement.objects.create( spare_part=part, movement_type=StockMovement.MovementType.ADJUSTMENT, quantity=row['quantity'], reference=batch_id, created_by=user, ) BatchImport.objects.create( batch_id=batch_id, source_filename=os.path.basename(file_path), imported_by=user, total_rows=len(to_create) + len(skipped_blank) + len(skipped_duplicate) + len(errors), created_count=len(created_parts), updated_count=0, skipped_count=len(skipped_blank) + len(skipped_duplicate), error_count=len(errors), status=BatchImport.Status.COMPLETED, notes=( f'{len(skipped_duplicate)} duplicate part numbers skipped; ' f'{len(errors)} rows had missing/invalid prices.' ), ) return batch_id, len(created_parts)