""" Data Integrity Check Management Command This command checks for various data integrity issues in the accounting system: - Unbalanced journal entries (debits != credits) - Orphaned ledger entries (no journal entry) - Accounts with negative balances where inappropriate Requirements: 14.4, 19.7 """ from django.core.management.base import BaseCommand from django.db.models import Sum, Q, F from decimal import Decimal from accounting.models import JournalEntry, JournalEntryLine, GeneralLedger, Account class Command(BaseCommand): help = 'Check accounting data integrity and generate a report of issues found' def add_arguments(self, parser): parser.add_argument( '--fix', action='store_true', help='Attempt to automatically fix issues (use with caution)', ) parser.add_argument( '--verbose', action='store_true', help='Show detailed output', ) def handle(self, *args, **options): self.fix_mode = options['fix'] self.verbose = options['verbose'] self.issues_found = [] self.stdout.write(self.style.SUCCESS('Starting data integrity check...\n')) # Run all checks self.check_unbalanced_journal_entries() self.check_orphaned_ledger_entries() self.check_negative_balances() # Generate report self.generate_report() def check_unbalanced_journal_entries(self): """ Check for journal entries where total debits != total credits. Requirements: 14.4 """ self.stdout.write('\n1. Checking for unbalanced journal entries...') # Get all journal entries with their line totals journal_entries = JournalEntry.objects.annotate( total_debits=Sum('lines__debit_amount'), total_credits=Sum('lines__credit_amount') ).filter( # Find entries where debits don't equal credits ~Q(total_debits=F('total_credits')) ) for entry in journal_entries: difference = entry.total_debits - entry.total_credits issue = { 'type': 'Unbalanced Journal Entry', 'severity': 'HIGH', 'reference': entry.reference_number, 'description': ( f'Journal entry {entry.reference_number} is unbalanced. ' f'Debits: {entry.total_debits}, Credits: {entry.total_credits}, ' f'Difference: {difference}' ), 'status': entry.status, 'date': entry.transaction_date } self.issues_found.append(issue) if self.verbose: self.stdout.write( self.style.WARNING(f' - {issue["description"]}') ) if not journal_entries: self.stdout.write(self.style.SUCCESS(' ✓ No unbalanced journal entries found')) else: self.stdout.write( self.style.ERROR(f' ✗ Found {len(journal_entries)} unbalanced journal entries') ) def check_orphaned_ledger_entries(self): """ Check for general ledger entries that reference non-existent journal entries. Requirements: 14.4 """ self.stdout.write('\n2. Checking for orphaned ledger entries...') # This shouldn't happen due to PROTECT on_delete, but check anyway orphaned_entries = [] for ledger_entry in GeneralLedger.objects.select_related('journal_entry', 'journal_entry_line').all(): # Check if journal entry exists if not ledger_entry.journal_entry: orphaned_entries.append(ledger_entry) issue = { 'type': 'Orphaned Ledger Entry', 'severity': 'HIGH', 'reference': f'Ledger ID {ledger_entry.id}', 'description': ( f'General ledger entry {ledger_entry.id} has no associated journal entry. ' f'Account: {ledger_entry.account.code}, Date: {ledger_entry.transaction_date}' ), 'account': ledger_entry.account.code, 'date': ledger_entry.transaction_date } self.issues_found.append(issue) if self.verbose: self.stdout.write( self.style.WARNING(f' - {issue["description"]}') ) # Check if journal entry line exists if not ledger_entry.journal_entry_line: issue = { 'type': 'Orphaned Ledger Entry', 'severity': 'HIGH', 'reference': f'Ledger ID {ledger_entry.id}', 'description': ( f'General ledger entry {ledger_entry.id} has no associated journal entry line. ' f'Account: {ledger_entry.account.code}, Date: {ledger_entry.transaction_date}' ), 'account': ledger_entry.account.code, 'date': ledger_entry.transaction_date } self.issues_found.append(issue) if self.verbose: self.stdout.write( self.style.WARNING(f' - {issue["description"]}') ) if not orphaned_entries: self.stdout.write(self.style.SUCCESS(' ✓ No orphaned ledger entries found')) else: self.stdout.write( self.style.ERROR(f' ✗ Found {len(orphaned_entries)} orphaned ledger entries') ) def check_negative_balances(self): """ Check for accounts with negative balances where inappropriate. Asset and Expense accounts should generally have positive balances (debits). Liability, Equity, and Income accounts should generally have positive balances (credits). Requirements: 14.4 """ self.stdout.write('\n3. Checking for inappropriate negative balances...') negative_balance_issues = [] # Get all accounts with ledger entries accounts = Account.objects.filter(ledger_entries__isnull=False).distinct() for account in accounts: # Get the most recent ledger entry for this account last_entry = GeneralLedger.objects.filter( account=account ).order_by('-transaction_date', '-posted_at', '-id').first() if last_entry: balance = last_entry.balance # Check for inappropriate negative balances # For asset/expense accounts: balance should be >= 0 (debit balance) # For liability/equity/income accounts: balance should be >= 0 (credit balance) # The balance field stores positive for the account's natural side # Flag if balance is significantly negative (more than 0.01 for rounding) if balance < Decimal('-0.01'): # This could be a data entry error or a business rule violation issue = { 'type': 'Negative Balance', 'severity': 'MEDIUM', 'reference': account.code, 'description': ( f'Account {account.code} - {account.name} has a negative balance: {balance}. ' f'Account type: {account.account_type}. ' f'This may indicate a data entry error or overdraft.' ), 'account': account.code, 'balance': balance, 'account_type': account.account_type } self.issues_found.append(issue) negative_balance_issues.append(issue) if self.verbose: self.stdout.write( self.style.WARNING(f' - {issue["description"]}') ) if not negative_balance_issues: self.stdout.write(self.style.SUCCESS(' ✓ No inappropriate negative balances found')) else: self.stdout.write( self.style.WARNING( f' ! Found {len(negative_balance_issues)} accounts with negative balances' ) ) def generate_report(self): """ Generate a comprehensive report of all integrity issues found. Requirements: 14.4 """ self.stdout.write('\n' + '=' * 80) self.stdout.write(self.style.SUCCESS('DATA INTEGRITY CHECK REPORT')) self.stdout.write('=' * 80 + '\n') if not self.issues_found: self.stdout.write( self.style.SUCCESS('✓ No data integrity issues found. System is healthy!\n') ) return # Group issues by severity high_severity = [i for i in self.issues_found if i['severity'] == 'HIGH'] medium_severity = [i for i in self.issues_found if i['severity'] == 'MEDIUM'] self.stdout.write(f'Total Issues Found: {len(self.issues_found)}\n') if high_severity: self.stdout.write(self.style.ERROR(f'\nHIGH SEVERITY ISSUES: {len(high_severity)}')) self.stdout.write(self.style.ERROR('-' * 80)) for issue in high_severity: self.stdout.write(f'\n{issue["type"]}: {issue["reference"]}') self.stdout.write(f' {issue["description"]}') if medium_severity: self.stdout.write(self.style.WARNING(f'\n\nMEDIUM SEVERITY ISSUES: {len(medium_severity)}')) self.stdout.write(self.style.WARNING('-' * 80)) for issue in medium_severity: self.stdout.write(f'\n{issue["type"]}: {issue["reference"]}') self.stdout.write(f' {issue["description"]}') self.stdout.write('\n' + '=' * 80) self.stdout.write( self.style.WARNING( '\nRECOMMENDATION: Review and correct these issues to ensure data integrity.' ) ) self.stdout.write('=' * 80 + '\n')