""" Management command to populate database with test data Usage: python manage.py populate_data [--clear] """ from django.core.management.base import BaseCommand from django.db import transaction from django.utils import timezone from django.contrib.auth import get_user_model from decimal import Decimal from datetime import timedelta, datetime import random from apps.business.models import Business, Location, BusinessMembership from apps.authentication.models import User, Role from apps.inventory.models import ( Brand, Category, InventoryItem, Accessory, Supplier, TransferRequest, TransferItem ) from apps.sales.models import ( Customer, Sale, SaleItem, PaymentMode, SalePayment, SaleReturn, SaleReturnItem, Commission, CreditTransaction, Quote, QuoteItem, QuoteFollowUp ) from apps.accounting.models import ( Account, AccountType, JournalEntry, JournalDetail, Transaction, CreditTransaction as AccountingCreditTransaction, CommissionStructure, SaleCommission ) User = get_user_model() class Command(BaseCommand): help = 'Populate database with test data' def add_arguments(self, parser): parser.add_argument( '--no-clear', action='store_true', help='Skip clearing existing data before populating', ) def handle(self, *args, **options): if not options['no_clear']: self.stdout.write(self.style.WARNING('Clearing existing data...')) self.clear_data() else: self.stdout.write(self.style.WARNING('Skipping data clearing...')) self.stdout.write(self.style.SUCCESS('Starting data population...')) with transaction.atomic(): # Create base data self.users = self.create_users() self.business = self.create_business() self.create_roles_and_memberships() self.locations = self.create_locations() # Inventory data self.brands = self.create_brands() self.categories = self.create_categories() self.suppliers = self.create_suppliers() self.inventory_items = self.create_inventory_items() self.accessories = self.create_accessories() # Sales data self.customers = self.create_customers() self.payment_modes = self.create_payment_modes() self.sales = self.create_sales() self.quotes = self.create_quotes() # Accounting data self.accounts = self.create_accounts() self.commission_structures = self.create_commission_structures() # Additional operations self.create_transfers() self.create_returns() self.create_credit_transactions() self.stdout.write(self.style.SUCCESS('Data population completed successfully!')) self.print_summary() def clear_data(self): """Clear all data from tables""" from apps.inventory.models import InventoryBatch from apps.authentication.models import Role models_to_clear = [ # Sales related SaleReturnItem, SaleReturn, SalePayment, SaleItem, Sale, QuoteFollowUp, QuoteItem, Quote, Commission, CreditTransaction, Customer, PaymentMode, # Inventory related TransferItem, TransferRequest, InventoryBatch, Accessory, InventoryItem, Supplier, Category, Brand, # Accounting related JournalDetail, JournalEntry, SaleCommission, CommissionStructure, AccountingCreditTransaction, Transaction, Account, # Auth related BusinessMembership, Role, # Business related (last) Location, Business ] for model in models_to_clear: try: count = model.objects.count() model.objects.all().delete() self.stdout.write(f' Deleted {count} {model.__name__} records') except Exception as e: self.stdout.write( self.style.WARNING(f' Error deleting {model.__name__}: {str(e)}') ) def create_users(self): """Create test users""" self.stdout.write('Creating users...') users = [] # Admin user admin, created = User.objects.get_or_create( email='admin@example.com', defaults={ 'username': 'admin1', 'first_name': 'Admin1', 'last_name': 'User', 'is_staff': True, 'is_superuser': True, 'is_email_verified': True } ) if created: admin.set_password('admin123') admin.save() users.append(admin) # Sales staff sales_names = [ ('John', 'Doe'), ('Jane', 'Smith'), ('Michael', 'Johnson'), ('Sarah', 'Williams'), ('David', 'Brown') ] for first, last in sales_names: username = f"{first.lower()}.{last.lower()}" email = f'{username}@example.com' user, created = User.objects.get_or_create( email=email, defaults={ 'username': username, 'first_name': first, 'last_name': last, 'is_staff': False, 'is_email_verified': True } ) if created: user.set_password('password123') user.save() users.append(user) self.stdout.write(self.style.SUCCESS(f'Created {len(users)} users')) return users def create_business(self): """Create business""" self.stdout.write('Creating business...') business, created = Business.objects.get_or_create( business_code='TECH001', defaults={ 'name': 'Tech Solutions Ltd', 'email': 'info@techsolutions.com', 'phone_number': '+254712345678', 'address_line1': '123 Kimathi Street', 'city': 'Nairobi', 'state': 'Nairobi County', 'country': 'Kenya', 'postal_code': '00100', 'tax_id': 'P051234567A', 'currency': 'KES', 'timezone': 'Africa/Nairobi', 'is_active': True, 'is_verified': True, # 'created_by': self.users[0] # Admin user } ) self.stdout.write(self.style.SUCCESS('Business created')) return business def create_roles_and_memberships(self): """Create roles and business memberships for users""" self.stdout.write('Creating roles and memberships...') # Create roles in authentication app (for RBAC permissions) role_data = [ ('OWNER', 'Business Owner', 'Full access to all business features'), ('ADMIN', 'Administrator', 'Administrative access with most permissions'), ('SALES_AGENT', 'Sales Agent', 'Can create sales and manage customers'), ('INVENTORY_CLERK', 'Inventory Clerk', 'Can manage inventory items'), ('MANAGER', 'Manager', 'Can manage staff and view reports'), ('ACCOUNTANT', 'Accountant', 'Can access financial records and reports'), ('VIEWER', 'Viewer', 'Read-only access to business data'), ] roles = {} for role_name, display_name, description in role_data: role, created = Role.objects.get_or_create( name=role_name, business=self.business, defaults={ 'display_name': display_name, 'description': description, 'is_system_role': False } ) roles[role_name] = role if created: self.stdout.write(f' Created role: {display_name}') # Create memberships try: # Admin as OWNER membership, created = BusinessMembership.objects.get_or_create( user=self.users[0], business=self.business, defaults={ 'role': roles['OWNER'], 'status': 'ACTIVE', 'is_primary': True, 'can_access_all_locations': True, 'invited_by': None, 'invitation_accepted_at': timezone.now(), } ) if created: self.stdout.write(f' ✓ Created OWNER membership for {self.users[0].email}') # Other users as sales agents for user in self.users[1:]: membership, created = BusinessMembership.objects.get_or_create( user=user, business=self.business, defaults={ 'role': roles['SALES_AGENT'], 'status': 'ACTIVE', 'is_primary': True, 'can_access_all_locations': True, 'invited_by': self.users[0], # Invited by admin 'invitation_accepted_at': timezone.now(), } ) if created: self.stdout.write(f' ✓ Created SALES_AGENT membership for {user.email}') except Exception as e: self.stdout.write(self.style.ERROR(f' ✗ Error creating memberships: {str(e)}')) raise self.stdout.write(self.style.SUCCESS(f'Created {len(role_data)} roles and {len(self.users)} memberships')) def create_locations(self): """Create business locations""" self.stdout.write('Creating locations...') location_data = [ { 'name': 'Main Store - Nairobi CBD', 'address_line1': '123 Kimathi Street, Nairobi', 'phone': '+254712345678', 'is_primary': True }, { 'name': 'Westlands Branch', 'address_line1': '456 Waiyaki Way, Westlands', 'phone': '+254723456789', 'is_primary': False }, { 'name': 'Mombasa Branch', 'address_line1': '789 Moi Avenue, Mombasa', 'phone': '+254734567890', 'is_primary': False } ] locations = [] for data in location_data: location, _ = Location.objects.get_or_create( business=self.business, name=data['name'], defaults=data ) locations.append(location) self.stdout.write(self.style.SUCCESS(f'Created {len(locations)} locations')) return locations def create_brands(self): """Create computer brands""" self.stdout.write('Creating brands...') brand_names = [ 'Dell', 'HP', 'Lenovo', 'Apple', 'ASUS', 'Acer', 'Microsoft', 'Samsung', 'Toshiba', 'MSI' ] brands = [] for name in brand_names: brand, _ = Brand.objects.get_or_create( name=name, defaults={'description': f'{name} computers and accessories'} ) brands.append(brand) self.stdout.write(self.style.SUCCESS(f'Created {len(brands)} brands')) return brands def create_categories(self): """Create product categories""" self.stdout.write('Creating categories...') category_data = [ 'Laptops', 'Desktops', 'Monitors', 'Keyboards', 'Mice', 'Printers', 'Scanners', 'Network Equipment', 'Storage Devices', 'Cables & Adapters' ] categories = [] for name in category_data: category, _ = Category.objects.get_or_create( name=name, defaults={'description': f'{name} category'} ) categories.append(category) self.stdout.write(self.style.SUCCESS(f'Created {len(categories)} categories')) return categories def create_suppliers(self): """Create suppliers""" self.stdout.write('Creating suppliers...') supplier_data = [ { 'name': 'Global Tech Supplies', 'contact_person': 'James Mwangi', 'email': 'james@globaltech.com', 'phone': '+254701234567', 'payment_terms': 'Net 30', 'credit_limit': Decimal('500000.00') }, { 'name': 'Africa Computer Distributors', 'contact_person': 'Mary Wanjiru', 'email': 'mary@africacomputers.com', 'phone': '+254702345678', 'payment_terms': 'Net 45', 'credit_limit': Decimal('750000.00') }, { 'name': 'Tech Import Ltd', 'contact_person': 'Peter Ouma', 'email': 'peter@techimport.com', 'phone': '+254703456789', 'payment_terms': 'Net 60', 'credit_limit': Decimal('1000000.00') } ] suppliers = [] for data in supplier_data: supplier, _ = Supplier.objects.get_or_create( business=self.business, name=data['name'], defaults={**data, 'business': self.business} ) suppliers.append(supplier) self.stdout.write(self.style.SUCCESS(f'Created {len(suppliers)} suppliers')) return suppliers def create_inventory_items(self): """Create inventory items""" self.stdout.write('Creating inventory items...') laptop_models = [ 'Latitude 7420', 'Inspiron 15 3000', 'XPS 13', 'EliteBook 840', 'Pavilion 15', 'ProBook 450', 'ThinkPad X1', 'IdeaPad 3', 'ThinkBook 15', 'MacBook Air M2', 'MacBook Pro 14', 'VivoBook 15' ] processors = ['Intel i5', 'Intel i7', 'Intel i9', 'AMD Ryzen 5', 'AMD Ryzen 7', 'Apple M1', 'Apple M2'] ram_options = ['8GB', '16GB', '32GB', '64GB'] storage_options = ['256GB SSD', '512GB SSD', '1TB SSD', '2TB SSD'] items = [] for i in range(50): brand = random.choice(self.brands) model = random.choice(laptop_models) location = random.choice(self.locations) supplier = random.choice(self.suppliers) purchase_price = Decimal(random.randint(30000, 150000)) selling_price = purchase_price * Decimal('1.3') # 30% markup item = InventoryItem.objects.create( business=self.business, asset_id=f'COMP-{i+1:05d}', brand=brand, model=model, category=self.categories[0], # Laptops condition=random.choice(['NEW', 'REFURBISHED', 'USED']), status=random.choice(['IN_STOCK', 'IN_STOCK', 'IN_STOCK', 'SOLD']), serial_number=f'SN{random.randint(100000, 999999)}', processor=random.choice(processors), ram=random.choice(ram_options), storage=random.choice(storage_options), screen_size=random.choice(['13"', '14"', '15.6"', '17"']), operating_system=random.choice(['Windows 11 Pro', 'Windows 11 Home', 'Ubuntu', 'macOS']), location=location, quantity=random.randint(1, 5) if i % 4 != 3 else 0, reorder_level=2, purchase_price=purchase_price, selling_price=selling_price, supplier=supplier, purchase_date=timezone.now().date() - timedelta(days=random.randint(1, 180)), warranty_start_date=timezone.now().date() - timedelta(days=random.randint(1, 90)), warranty_end_date=timezone.now().date() + timedelta(days=random.randint(90, 730)), warranty_provider=brand.name ) items.append(item) self.stdout.write(self.style.SUCCESS(f'Created {len(items)} inventory items')) return items def create_accessories(self): """Create accessories""" self.stdout.write('Creating accessories...') accessory_data = [ ('Wireless Mouse', 'Mice', 500, 750), ('USB Keyboard', 'Keyboards', 800, 1200), ('HDMI Cable', 'Cables & Adapters', 300, 500), ('USB-C Hub', 'Cables & Adapters', 1500, 2500), ('Laptop Bag', 'Accessories', 1200, 2000), ('Cooling Pad', 'Accessories', 1000, 1500), ] accessories = [] for name, category, purchase, selling in accessory_data: for i in range(3): brand = random.choice(self.brands) location = random.choice(self.locations) accessory = Accessory.objects.create( business=self.business, name=f'{brand.name} {name}', category=category, brand=brand, model=f'Model-{random.randint(100, 999)}', location=location, quantity=random.randint(10, 50), reorder_level=5, purchase_price=Decimal(purchase), selling_price=Decimal(selling), sku=f'ACC-{len(accessories)+1:05d}' ) accessories.append(accessory) self.stdout.write(self.style.SUCCESS(f'Created {len(accessories)} accessories')) return accessories def create_customers(self): """Create customers""" self.stdout.write('Creating customers...') customer_data = [ ('Alice Kimani', 'alice@email.com', '+254711111111'), ('Bob Ochieng', 'bob@email.com', '+254722222222'), ('Carol Wanjiru', 'carol@email.com', '+254733333333'), ('David Mwangi', 'david@email.com', '+254744444444'), ('Eve Akinyi', 'eve@email.com', '+254755555555'), ('Frank Kamau', 'frank@email.com', '+254766666666'), ('Grace Njeri', 'grace@email.com', '+254777777777'), ('Henry Otieno', 'henry@email.com', '+254788888888'), ('Irene Wangari', 'irene@email.com', '+254799999999'), ('John Mutua', 'john@email.com', '+254700000000'), ] customers = [] for name, email, phone in customer_data: customer = Customer.objects.create( business=self.business, name=name, email=email, phone=phone, address=f'{random.randint(1, 999)} Sample Street, Nairobi', id_number=f'{random.randint(10000000, 99999999)}', credit_limit=Decimal(random.choice([0, 50000, 100000, 200000])), current_credit_balance=Decimal('0.00') ) customers.append(customer) self.stdout.write(self.style.SUCCESS(f'Created {len(customers)} customers')) return customers def create_payment_modes(self): """Create payment modes""" self.stdout.write('Creating payment modes...') modes = ['CASH', 'MPESA', 'BANK', 'CREDIT', 'CHEQUE'] payment_modes = [] for mode in modes: pm = PaymentMode.objects.create( business=self.business, name=mode, is_active=True, requires_reference=mode in ['MPESA', 'BANK', 'CHEQUE'] ) payment_modes.append(pm) self.stdout.write(self.style.SUCCESS(f'Created {len(payment_modes)} payment modes')) return payment_modes def create_sales(self): """Create sales transactions""" self.stdout.write('Creating sales...') sales = [] for i in range(30): customer = random.choice(self.customers) salesperson = random.choice([u for u in self.users if not u.is_superuser]) location = random.choice(self.locations) sale_date = timezone.now() - timedelta(days=random.randint(1, 90)) # Get available inventory available_items = InventoryItem.objects.filter( business=self.business, location=location, status='IN_STOCK', quantity__gt=0 ) if not available_items.exists(): continue sale = Sale.objects.create( business=self.business, location=location, customer=customer, salesperson=salesperson, sale_date=sale_date, status='COMPLETED', subtotal=Decimal('0'), discount_amount=Decimal('0'), tax_amount=Decimal('0'), total_amount=Decimal('0'), commission_rate=Decimal('0.00007') ) # Add 1-3 items to sale num_items = random.randint(1, min(3, available_items.count())) selected_items = random.sample(list(available_items), num_items) subtotal = Decimal('0') for inventory_item in selected_items: quantity = 1 unit_price = inventory_item.selling_price discount = Decimal('0') total_price = unit_price * quantity - discount SaleItem.objects.create( sale=sale, inventory_item=inventory_item, item_name=f'{inventory_item.brand.name} {inventory_item.model}', quantity=quantity, unit_price=unit_price, discount=discount, total_price=total_price, purchase_cost=inventory_item.purchase_price, status='SOLD' ) subtotal += total_price inventory_item.reduce_quantity(quantity) sale.subtotal = subtotal sale.total_amount = subtotal sale.calculate_commission() sale.save() # Create payment payment_mode = random.choice(self.payment_modes) SalePayment.objects.create( sale=sale, payment_mode=payment_mode, amount=sale.total_amount, reference_number=f'REF{random.randint(100000, 999999)}' if payment_mode.requires_reference else '', payment_date=sale_date ) # Create commission record Commission.objects.create( salesperson=salesperson, business=self.business, sale=sale, amount=sale.commission_amount, status=random.choice(['PENDING', 'PAID']), net_commission=sale.commission_amount ) sales.append(sale) self.stdout.write(self.style.SUCCESS(f'Created {len(sales)} sales')) return sales def create_quotes(self): """Create quotes""" self.stdout.write('Creating quotes...') quotes = [] for i in range(15): customer = random.choice(self.customers) creator = random.choice([u for u in self.users if not u.is_superuser]) location = random.choice(self.locations) quote_date = timezone.now().date() - timedelta(days=random.randint(1, 60)) quote = Quote.objects.create( business=self.business, location=location, customer=customer, created_by=creator, quote_date=quote_date, valid_until=quote_date + timedelta(days=30), status=random.choice(['DRAFT', 'SENT', 'ACCEPTED', 'REJECTED', 'EXPIRED']), subtotal=Decimal('0'), total_amount=Decimal('0') ) # Add items to quote available_items = InventoryItem.objects.filter( business=self.business, status='IN_STOCK' )[:3] for inventory_item in available_items: QuoteItem.objects.create( quote=quote, inventory_item=inventory_item, item_name=f'{inventory_item.brand.name} {inventory_item.model}', quantity=1, unit_price=inventory_item.selling_price, discount=Decimal('0') ) quotes.append(quote) self.stdout.write(self.style.SUCCESS(f'Created {len(quotes)} quotes')) return quotes def create_accounts(self): """Create chart of accounts""" self.stdout.write('Creating accounts...') account_data = [ ('1000', 'Cash', AccountType.ASSET), ('1200', 'Accounts Receivable', AccountType.ASSET), ('1300', 'Inventory', AccountType.ASSET), ('2000', 'Accounts Payable', AccountType.LIABILITY), ('3000', 'Owner\'s Equity', AccountType.EQUITY), ('4000', 'Sales Revenue', AccountType.REVENUE), ('5000', 'Cost of Goods Sold', AccountType.COST_OF_GOODS_SOLD), ('6000', 'Operating Expenses', AccountType.EXPENSE), ('6100', 'Commission Expense', AccountType.EXPENSE), ] accounts = [] for code, name, acc_type in account_data: account, _ = Account.objects.get_or_create( business=self.business, account_code=code, defaults={ 'account_name': name, 'account_type': acc_type, 'balance': Decimal('0') } ) accounts.append(account) self.stdout.write(self.style.SUCCESS(f'Created {len(accounts)} accounts')) return accounts def create_commission_structures(self): """Create commission structures""" self.stdout.write('Creating commission structures...') structure = CommissionStructure.objects.create( business=self.business, name='Standard Sales Commission', commission_type='PERCENTAGE', percentage=Decimal('2.5'), is_active=True ) self.stdout.write(self.style.SUCCESS('Created commission structure')) return [structure] def create_transfers(self): """Create transfer requests""" self.stdout.write('Creating transfers...') if len(self.locations) < 2: return for i in range(5): source = random.choice(self.locations) destination = random.choice([loc for loc in self.locations if loc != source]) requester = random.choice(self.users) transfer = TransferRequest.objects.create( business=self.business, source_location=source, destination_location=destination, status=random.choice(['PENDING', 'APPROVED', 'COMPLETED']), requested_by=requester, transfer_number=f'TRF-{i+1:05d}' ) # Add items items = InventoryItem.objects.filter( business=self.business, location=source, status='IN_STOCK' )[:2] for item in items: TransferItem.objects.create( transfer=transfer, inventory_item=item, quantity=1 ) self.stdout.write(self.style.SUCCESS('Created transfers')) def create_returns(self): """Create sale returns""" self.stdout.write('Creating returns...') completed_sales = Sale.objects.filter(status='COMPLETED')[:5] for sale in completed_sales: return_obj = SaleReturn.objects.create( original_sale=sale, business=self.business, location=sale.location, processed_by=sale.salesperson, reason=random.choice(['DEFECTIVE', 'CUSTOMER_REQUEST', 'WRONG_ITEM']), total_refund_amount=Decimal('0') ) # Return first item from sale first_item = sale.items.first() if first_item: SaleReturnItem.objects.create( sale_return=return_obj, sale_item=first_item, inventory_item=first_item.inventory_item, quantity_returned=1, refund_amount=first_item.unit_price, restocked=True ) return_obj.total_refund_amount = first_item.unit_price return_obj.save() self.stdout.write(self.style.SUCCESS('Created returns')) def create_credit_transactions(self): """Create credit transactions""" self.stdout.write('Creating credit transactions...') # Create some credit sales credit_customers = [c for c in self.customers if c.credit_limit > 0][:3] for customer in credit_customers: sale = random.choice(self.sales) creator = random.choice(self.users) CreditTransaction.objects.create( customer=customer, business=self.business, sale=sale, transaction_type='SALE', amount=sale.total_amount, balance_after=sale.total_amount, transaction_date=sale.sale_date, created_by=creator ) customer.current_credit_balance += sale.total_amount customer.save() self.stdout.write(self.style.SUCCESS('Created credit transactions')) def print_summary(self): """Print summary of created data""" self.stdout.write('\n' + '='*50) self.stdout.write(self.style.SUCCESS('DATA POPULATION SUMMARY')) self.stdout.write('='*50) summary = [ ('Users', User.objects.count()), ('Businesses', Business.objects.count()), ('Locations', Location.objects.count()), ('Brands', Brand.objects.count()), ('Categories', Category.objects.count()), ('Suppliers', Supplier.objects.count()), ('Inventory Items', InventoryItem.objects.count()), ('Accessories', Accessory.objects.count()), ('Customers', Customer.objects.count()), ('Sales', Sale.objects.count()), ('Sale Items', SaleItem.objects.count()), ('Quotes', Quote.objects.count()), ('Accounts', Account.objects.count()), ('Transfers', TransferRequest.objects.count()), ('Returns', SaleReturn.objects.count()), ] for label, count in summary: self.stdout.write(f'{label:.<30} {count:>5}') self.stdout.write('='*50 + '\n')