# apps/accounting/management/commands/generate_accounting_data.py from django.core.management.base import BaseCommand from django.db import transaction from django.utils import timezone from django.db.models import Sum from decimal import Decimal from datetime import date, timedelta import random from faker import Faker from apps.authentication.models import User from apps.business.models import Business, BusinessMembership from apps.authentication.models import Role from apps.accounting.models import ( Account, AccountType, JournalEntry, JournalDetail, CreditTransaction, CreditPayment, CommissionStructure, SaleCommission ) from apps.accounting.services import AccountingService, CommissionService, CreditManagementService from apps.sales.models import Sale, Customer from apps.inventory.models import InventoryItem, Brand, Category from apps.business.models import Location fake = Faker() class Command(BaseCommand): help = 'Generate dummy accounting data for testing' def add_arguments(self, parser): parser.add_argument( '--business-id', type=str, help='Business ID (UUID) to generate data for', ) parser.add_argument( '--months', type=int, default=6, help='Number of months of historical data to generate (default: 6)', ) parser.add_argument( '--transactions-per-month', type=int, default=50, help='Number of transactions per month (default: 50)', ) parser.add_argument( '--clear-existing', action='store_true', help='Clear existing accounting data before generating', ) def handle(self, *args, **options): business_id = options.get('business_id') months = options['months'] transactions_per_month = options['transactions_per_month'] clear_existing = options['clear_existing'] if not business_id: self.stdout.write(self.style.ERROR('Please provide --business-id')) return try: business = Business.objects.get(id=business_id) except Business.DoesNotExist: self.stdout.write(self.style.ERROR(f'Business with ID {business_id} not found')) return self.stdout.write(self.style.SUCCESS(f'Generating data for business: {business.name}')) with transaction.atomic(): if clear_existing: self.clear_existing_data(business_id) # Generate data self.generate_chart_of_accounts(business_id) customers = self.generate_customers(business_id) agents = self.generate_sales_agents(business) inventory_items = self.generate_inventory_items(business_id) commission_structure = self.generate_commission_structure(business_id) # Generate historical transactions self.generate_historical_transactions( business_id, customers, agents, inventory_items, commission_structure, months, transactions_per_month ) self.stdout.write(self.style.SUCCESS('✅ Dummy data generation completed!')) self.print_summary(business_id) def clear_existing_data(self, business_id): """Clear existing accounting data""" self.stdout.write('Clearing existing data...') Account.objects.filter(business_id=business_id).delete() JournalEntry.objects.filter(business_id=business_id).delete() CreditTransaction.objects.filter(business_id=business_id).delete() SaleCommission.objects.filter(business_id=business_id).delete() CommissionStructure.objects.filter(business_id=business_id).delete() self.stdout.write(self.style.WARNING('Existing data cleared')) def generate_chart_of_accounts(self, business_id): """Generate complete chart of accounts""" self.stdout.write('Generating chart of accounts...') accounting_service = AccountingService(business_id) # Create all default accounts default_accounts = [ 'CASH', 'PETTY_CASH', 'ACCOUNTS_RECEIVABLE', 'INVENTORY', 'FIXED_ASSETS', 'ACCOUNTS_PAYABLE', 'CREDIT_CARD_PAYABLE', 'SALES_TAX_PAYABLE', 'OWNERS_EQUITY', 'RETAINED_EARNINGS', 'SALES_REVENUE', 'SERVICE_REVENUE', 'COST_OF_GOODS_SOLD', 'RENT_EXPENSE', 'COMMISSION_EXPENSE', 'UTILITIES_EXPENSE', 'SALARY_EXPENSE', 'OFFICE_SUPPLIES', 'INSURANCE_EXPENSE', 'ADVERTISING_EXPENSE', 'DEPRECIATION_EXPENSE', 'BANK_CHARGES', 'MISCELLANEOUS_EXPENSE' ] for account_name in default_accounts: accounting_service.get_account(account_name) # Create some custom revenue accounts custom_accounts = [ ('4100', 'Product Sales - Electronics', AccountType.REVENUE), ('4200', 'Product Sales - Accessories', AccountType.REVENUE), ('4300', 'Service Revenue - Repairs', AccountType.REVENUE), ('6350', 'Marketing Expense', AccountType.EXPENSE), ('6450', 'Travel Expense', AccountType.EXPENSE), ('6550', 'Training & Development', AccountType.EXPENSE), ] for code, name, acc_type in custom_accounts: Account.objects.get_or_create( business_id=business_id, account_code=code, defaults={ 'account_name': name, 'account_type': acc_type, 'is_active': True } ) account_count = Account.objects.filter(business_id=business_id).count() self.stdout.write(self.style.SUCCESS(f'✓ Created {account_count} accounts')) def generate_customers(self, business_id, count=30): """Generate customer records""" self.stdout.write(f'Generating {count} customers...') customers = [] for i in range(count): customer = Customer.objects.create( business_id=business_id, name=fake.company(), email=fake.email(), phone=fake.phone_number()[:20], address=fake.street_address(), credit_limit=Decimal(random.choice([50000, 100000, 150000, 200000])), notes=f'Generated customer for testing' ) customers.append(customer) self.stdout.write(self.style.SUCCESS(f'✓ Created {len(customers)} customers')) return customers def generate_sales_agents(self, business, count=5): """Generate sales agent users""" self.stdout.write(f'Generating {count} sales agents...') # Get or create the SALES_AGENT role sales_agent_role, _ = Role.objects.get_or_create( name='SALES_AGENT', defaults={ 'description': 'Sales Agent Role', 'permissions': {} } ) agents = [] for i in range(count): username = f'agent_{fake.user_name()}_{i}_{random.randint(1000, 9999)}' email = f'agent{i}_{random.randint(1000, 9999)}@{business.name.lower().replace(" ", "").replace("'", "")}.com' agent, created = User.objects.get_or_create( username=username, defaults={ 'email': email, 'first_name': fake.first_name(), 'last_name': fake.last_name(), 'is_active': True } ) if created: agent.set_password('testpass123') agent.save() # Create a business for the agent # Note: Business model doesn't have a 'code' field try: agent_business = Business.objects.create( name=f"{agent.first_name} {agent.last_name}'s Business", owner=agent, email=email, phone_number=fake.phone_number()[:20], address_line1=fake.street_address(), city=fake.city(), state=fake.state(), postal_code=fake.postcode(), business_code=f'AG-{random.randint(10000, 99999)}' ) self.stdout.write(self.style.SUCCESS( f'✓ Created business "{agent_business.name}" for {email}' )) except Exception as e: self.stdout.write(self.style.WARNING( f'✗ Error creating business for {email}: {str(e)}' )) # Add to main business with proper Role instance BusinessMembership.objects.get_or_create( user=agent, business=business, defaults={ 'role': sales_agent_role, # Use Role instance, not string 'is_active': True, 'invitation_accepted': True } ) agents.append(agent) self.stdout.write(self.style.SUCCESS(f'✓ Created {len(agents)} sales agents')) return agents def generate_inventory_items(self, business_id, count=50): """Generate inventory items""" self.stdout.write(f'Generating {count} inventory items...') # Get or create a location location, _ = Location.objects.get_or_create( business_id=business_id, name='Main Warehouse', defaults={'location_type': 'WAREHOUSE'} ) # Get or create brands (Brand model is NOT tenant-aware, no business_id) brand_names = ['Apple', 'Samsung', 'Dell', 'HP', 'Lenovo', 'Asus', 'Acer', 'Microsoft'] brands = {} for brand_name in brand_names: brand, _ = Brand.objects.get_or_create( name=brand_name, defaults={'description': f'{brand_name} products'} ) brands[brand_name] = brand # Get or create categories (Category model is also NOT tenant-aware) from apps.inventory.models import Category category_names = ['Laptop', 'Desktop', 'Tablet', 'Phone', 'Monitor', 'Keyboard', 'Mouse', 'Headset'] categories = {} for category_name in category_names: category, _ = Category.objects.get_or_create( name=category_name, defaults={'description': f'{category_name} products'} ) categories[category_name] = category items = [] for i in range(count): brand_name = random.choice(brand_names) brand = brands[brand_name] category_name = random.choice(category_names) category = categories[category_name] purchase_price = Decimal(random.randint(200, 2000)) markup = Decimal(random.uniform(1.2, 1.8)) selling_price = purchase_price * markup item = InventoryItem.objects.create( business_id=business_id, location=location, brand=brand, # Use Brand instance, not string model=f'{category_name} {fake.bothify(text="??-####")}', category=category, # Use Category instance, not string serial_number=fake.bothify(text='SN-##########'), asset_id=fake.bothify(text='ASSET-######'), purchase_price=purchase_price, selling_price=selling_price.quantize(Decimal('0.01')), quantity=random.randint(5, 50), condition='NEW', status='IN_STOCK', processor=f'Intel Core i{random.choice([5, 7, 9])}' if category_name in ['Laptop', 'Desktop'] else 9, ram=f'{random.choice([8, 16, 32])}GB' if category_name in ['Laptop', 'Desktop'] else '8 GB', storage=f'{random.choice([256, 512, 1024])}GB SSD' if category_name in ['Laptop', 'Desktop'] else '256 GB SSD' ) items.append(item) self.stdout.write(self.style.SUCCESS(f'✓ Created {len(items)} inventory items')) return items def generate_commission_structure(self, business_id): """Generate commission structure""" self.stdout.write('Generating commission structure...') structure = CommissionStructure.objects.create( business_id=business_id, name='Standard Sales Commission', commission_type='PERCENTAGE', percentage=Decimal('5.00'), is_active=True ) self.stdout.write(self.style.SUCCESS('✓ Created commission structure')) return structure def generate_historical_transactions(self, business_id, customers, agents, inventory_items, commission_structure, months, transactions_per_month): """Generate historical transactions""" self.stdout.write(f'Generating {months} months of transactions...') accounting_service = AccountingService(business_id) commission_service = CommissionService(business_id) credit_service = CreditManagementService(business_id) end_date = date.today() start_date = end_date - timedelta(days=months * 30) total_transactions = 0 current_date = start_date while current_date <= end_date: # Generate transactions for this day daily_transactions = random.randint( transactions_per_month // 30 - 2, transactions_per_month // 30 + 2 ) for _ in range(daily_transactions): transaction_type = random.choices( ['sale', 'expense', 'payment'], weights=[0.6, 0.25, 0.15] )[0] if transaction_type == 'sale': self.generate_sale_transaction( business_id, accounting_service, commission_service, current_date, customers, agents, inventory_items, commission_structure ) elif transaction_type == 'expense': self.generate_expense_transaction( business_id, accounting_service, current_date, agents ) elif transaction_type == 'payment': self.generate_payment_transaction( business_id, accounting_service, credit_service, current_date ) total_transactions += 1 current_date += timedelta(days=1) self.stdout.write(self.style.SUCCESS( f'✓ Generated {total_transactions} transactions over {months} months' )) def generate_sale_transaction(self, business_id, accounting_service, commission_service, transaction_date, customers, agents, inventory_items, commission_structure): """Generate a sale transaction""" from apps.sales.models import SaleItem customer = random.choice(customers) agent = random.choice(agents) payment_method = random.choices( ['CASH', 'CREDIT', 'MOBILE_MONEY'], weights=[0.4, 0.4, 0.2] )[0] # Get business and location business = Business.objects.get(id=business_id) location = inventory_items[0].location if inventory_items else None if not location: return # Create sale items num_items = random.randint(1, 5) sale_items_data = random.sample(inventory_items, min(num_items, len(inventory_items))) total_amount = Decimal('0') total_cost = Decimal('0') sale_items_list = [] for item in sale_items_data: quantity = random.randint(1, 3) unit_price = item.selling_price item_total = unit_price * quantity total_amount += item_total total_cost += item.purchase_price * quantity sale_items_list.append({ 'item': item, 'quantity': quantity, 'unit_price': unit_price, 'total': item_total }) # Create actual Sale record sale = Sale.objects.create( business=business, location=location, customer=customer, salesperson=agent, sale_date=timezone.make_aware(timezone.datetime.combine(transaction_date, timezone.datetime.min.time())), total_amount=total_amount, subtotal=total_amount, status='COMPLETED' ) # Create SaleItem records for item_data in sale_items_list: SaleItem.objects.create( sale=sale, inventory_item=item_data['item'], item_name=f"{item_data['item'].brand.name} {item_data['item'].model}", quantity=item_data['quantity'], unit_price=item_data['unit_price'], total_price=item_data['total'], purchase_cost=item_data['item'].purchase_price, item_type='inventory_item' ) # Create journal entry journal_entry = accounting_service.create_sale_entry(sale) # Create commission commission_amount = total_amount * (commission_structure.percentage / 100) commission = SaleCommission.objects.create( business_id=business_id, sale=sale, agent=agent, commission_structure=commission_structure, sale_amount=total_amount, commission_amount=commission_amount, status='PENDING' ) # Randomly approve and pay some commissions if random.random() > 0.3: commission.status = 'APPROVED' commission.approved_by = agent commission.approved_at = timezone.make_aware(timezone.datetime.combine(transaction_date, timezone.datetime.min.time())) + timedelta(days=random.randint(1, 7)) commission.save() if random.random() > 0.4 and commission.approved_at <= timezone.now(): commission.status = 'PAID' commission.paid_at = commission.approved_at + timedelta(days=random.randint(1, 14)) commission.payment_reference = f'PAY-{random.randint(1000, 9999)}' commission.save() def generate_expense_transaction(self, business_id, accounting_service, transaction_date, agents): """Generate an expense transaction""" expense_types = [ ('6000', 'RENT_EXPENSE', 'Monthly rent payment'), ('6200', 'UTILITIES_EXPENSE', 'Electricity and water'), ('6300', 'SALARY_EXPENSE', 'Employee salaries'), ('6400', 'OFFICE_SUPPLIES', 'Office supplies purchase'), ('6500', 'INSURANCE_EXPENSE', 'Insurance premium'), ('6600', 'ADVERTISING_EXPENSE', 'Marketing campaign'), ('6800', 'BANK_CHARGES', 'Bank service fees'), ('6350', 'Marketing Expense', 'Social media ads'), ('6450', 'Travel Expense', 'Business travel'), ] expense_code, expense_type, description = random.choice(expense_types) amount = Decimal(random.randint(500, 10000)) # Use an actual agent user instead of ID 1 created_by = random.choice(agents) expense_data = { 'expense_date': transaction_date, 'account_code': expense_code, 'description': description, 'amount': amount, 'reference_number': f'EXP-{random.randint(1000, 9999)}', 'created_by_id': created_by.id } accounting_service.create_expense_entry(expense_data) def generate_payment_transaction(self, business_id, accounting_service, credit_service, transaction_date): """Generate a payment transaction for existing credit""" # Get unpaid credits before this date unpaid_credits = CreditTransaction.objects.filter( business_id=business_id, status__in=['PENDING', 'PARTIAL'], created_at__date__lte=transaction_date ) if unpaid_credits.exists(): credit = random.choice(unpaid_credits) payment_amount = credit.balance * Decimal(random.uniform(0.2, 1.0)) try: credit_service.record_payment( credit.id, payment_amount, payment_date=transaction_date, payment_method='BANK_TRANSFER', reference_number=f'PMT-{random.randint(1000, 9999)}' ) except: pass # Skip if payment fails def print_summary(self, business_id): """Print summary of generated data""" self.stdout.write('\n' + '='*60) self.stdout.write(self.style.SUCCESS('DATA GENERATION SUMMARY')) self.stdout.write('='*60 + '\n') accounts = Account.objects.filter(business_id=business_id).count() journal_entries = JournalEntry.objects.filter(business_id=business_id).count() credits = CreditTransaction.objects.filter(business_id=business_id).count() commissions = SaleCommission.objects.filter(business_id=business_id).count() customers = Customer.objects.filter(business_id=business_id).count() # Calculate balances accounting_service = AccountingService(business_id) trial_balance = accounting_service.get_trial_balance() self.stdout.write(f'Accounts Created: {accounts}') self.stdout.write(f'Journal Entries: {journal_entries}') self.stdout.write(f'Credit Transactions: {credits}') self.stdout.write(f'Commissions: {commissions}') self.stdout.write(f'Customers: {customers}') self.stdout.write(f'\nTotal Debits: ${trial_balance["total_debits"]:,.2f}') self.stdout.write(f'Total Credits: ${trial_balance["total_credits"]:,.2f}') self.stdout.write(f'Balanced: {"✓ Yes" if trial_balance["is_balanced"] else "✗ No"}') # Credit summary total_receivables = CreditTransaction.objects.filter( business_id=business_id, status__in=['PENDING', 'PARTIAL', 'OVERDUE'] ).aggregate(total=Sum('balance'))['total'] or Decimal('0') self.stdout.write(f'\nTotal Accounts Receivable: ${total_receivables:,.2f}') # Commission summary pending_commissions = SaleCommission.objects.filter( business_id=business_id, status='PENDING' ).aggregate(total=Sum('commission_amount'))['total'] or Decimal('0') self.stdout.write(f'Pending Commissions: ${pending_commissions:,.2f}') self.stdout.write('\n' + '='*60 + '\n')