from django.contrib import admin from django.urls import path from django.shortcuts import render, redirect from django.contrib import messages import csv import io import traceback from django.utils import timezone from django.contrib.auth.models import User from .models import RegisteredUser, Attendance, Event, EventAssignment, RegistrationEventAssignment class RegisteredUserAdmin(admin.ModelAdmin): list_display = ('registration_code', 'full_name', 'email', 'event_name', 'payment_status', 'has_attended', 'timestamp') list_filter = ('event_name', 'payment_status', 'has_attended') search_fields = ('registration_code', 'full_name', 'email') readonly_fields = ('timestamp',) change_list_template = 'admin/registered_user_changelist.html' def get_urls(self): urls = super().get_urls() my_urls = [ path('upload-csv/', self.upload_csv, name='upload_csv'), ] return my_urls + urls # Helper function to get value with multiple possible column names def get_field_value(self, row, field_name, default=''): variations = [ field_name, # Registration_Code field_name.lower(), # registration_code field_name.upper(), # REGISTRATION_CODE field_name.replace('_', ' '), # Registration Code field_name.replace('_', ' ').lower(), # registration code ] for var in variations: if var in row and row[var]: return row[var] # If we can't find the field using the variations, try a direct key lookup for key in row.keys(): if key.lower().replace(' ', '_').replace('-', '_') == field_name.lower(): return row[key] return default def upload_csv(self, request): if request.method == 'POST': csv_file = request.FILES.get('csv_file') if not csv_file: self.message_user(request, 'No file uploaded', level=messages.ERROR) return redirect('..') if not csv_file.name.endswith('.csv'): self.message_user(request, 'File is not a CSV', level=messages.ERROR) return redirect('..') # Process CSV file try: # Read the file content file_data = csv_file.read().decode('utf-8') # Parse CSV csv_data = io.StringIO(file_data) reader = csv.DictReader(csv_data) # Clean up field names to handle BOM and whitespace if reader.fieldnames: reader.fieldnames = [fn.strip().replace('\ufeff', '') for fn in reader.fieldnames] else: self.message_user(request, 'Warning: No columns found in CSV', level=messages.WARNING) return redirect('..') # Process rows new_count = 0 # Count of new users created skipped_count = 0 # Count of users skipped (already exist) rows_list = list(reader) # Convert to list to see if it's empty if not rows_list: self.message_user(request, 'Warning: CSV file has no data rows', level=messages.WARNING) return redirect('..') for row in rows_list: # Get registration code registration_code = self.get_field_value(row, 'Registration_Code') if not registration_code: self.message_user(request, f'Skipping row, no registration code found: {row}', level=messages.WARNING) continue # Get other fields try: # Extract data from row email = self.get_field_value(row, 'Email') full_name = self.get_field_value(row, 'Full_Name') event_name = self.get_field_value(row, 'Event_Name') status = self.get_field_value(row, 'Status') payment_status = self.get_field_value(row, 'Payment_Status') package_name = self.get_field_value(row, 'Package_Name') package_description = self.get_field_value(row, 'Package_Description') activity_price_str = self.get_field_value(row, 'Activity_Price', '0') # Processing user data # Convert price to float try: activity_price = float(activity_price_str) except ValueError: activity_price = 0.0 self.message_user( request, f'Warning: Could not convert price "{activity_price_str}" to number for {registration_code}', level=messages.WARNING ) # Check if user already exists existing_user = RegisteredUser.objects.filter(registration_code=registration_code).first() if existing_user: # Skip existing user skipped_count += 1 # self.message_user( # request, # f'Skipped existing user: {registration_code} - {full_name}', # level=messages.INFO # ) else: # Create new user user = RegisteredUser.objects.create( registration_code=registration_code, email=email, full_name=full_name, event_name=event_name, status=status, payment_status=payment_status, package_name=package_name, package_description=package_description, activity_price=activity_price ) new_count += 1 # self.message_user( # request, # f'Created new user: {registration_code} - {full_name}', # level=messages.SUCCESS # ) except Exception as row_error: self.message_user( request, f'Error processing row with code {registration_code}: {str(row_error)}', level=messages.ERROR ) # Show summary message if new_count > 0 and skipped_count > 0: self.message_user( request, f'CSV upload complete: {new_count} new users created, {skipped_count} existing users skipped', level=messages.SUCCESS ) elif new_count > 0: self.message_user( request, f'CSV upload complete: {new_count} new users created', level=messages.SUCCESS ) elif skipped_count > 0: self.message_user( request, f'CSV upload complete: All {skipped_count} users already exist (no new users created)', level=messages.INFO ) else: self.message_user( request, 'CSV upload complete: No users processed', level=messages.WARNING ) return redirect('..') except Exception as e: self.message_user(request, f'Error processing CSV: {str(e)}', level=messages.ERROR) return redirect('..') return render(request, 'admin/csv_upload.html') class AttendanceAdmin(admin.ModelAdmin): list_display = ('user', 'event_name', 'check_in_time') list_filter = ('event_name', 'check_in_time') search_fields = ('user__full_name', 'user__registration_code', 'event_name') date_hierarchy = 'check_in_time' class EventAdmin(admin.ModelAdmin): list_display = ('name', 'date', 'is_active', 'created_at') list_filter = ('is_active', 'date') search_fields = ('name', 'description') date_hierarchy = 'date' class EventAssignmentAdmin(admin.ModelAdmin): list_display = ('user', 'event', 'assigned_at') list_filter = ('event', 'assigned_at') search_fields = ('user__username', 'event__name') date_hierarchy = 'assigned_at' autocomplete_fields = ['user', 'event'] def get_queryset(self, request): # Super users can see all assignments if request.user.is_superuser: return super().get_queryset(request) # Regular admin users can only see their own assignments return super().get_queryset(request).filter(user=request.user) def has_change_permission(self, request, obj=None): # Only superusers can change assignments return request.user.is_superuser def has_delete_permission(self, request, obj=None): # Only superusers can delete assignments return request.user.is_superuser def has_add_permission(self, request): # Only superusers can add assignments return request.user.is_superuser class RegistrationEventAssignmentAdmin(admin.ModelAdmin): list_display = ('user', 'registration_event', 'assigned_at', 'is_active') list_filter = ('registration_event', 'is_active', 'assigned_at') search_fields = ('user__username', 'user__first_name', 'user__last_name', 'registration_event__name') date_hierarchy = 'assigned_at' autocomplete_fields = ['user'] def get_queryset(self, request): # Super users can see all assignments if request.user.is_superuser: return super().get_queryset(request) # Regular admin users can only see their own assignments return super().get_queryset(request).filter(user=request.user) def has_change_permission(self, request, obj=None): # Only superusers can change assignments return request.user.is_superuser def has_delete_permission(self, request, obj=None): # Only superusers can delete assignments return request.user.is_superuser def has_add_permission(self, request): # Only superusers can add assignments return request.user.is_superuser def formfield_for_foreignkey(self, db_field, request, **kwargs): if db_field.name == "registration_event": # Only show active registration events from registration.models import Event as RegistrationEvent kwargs["queryset"] = RegistrationEvent.objects.filter(is_active=True) return super().formfield_for_foreignkey(db_field, request, **kwargs) admin.site.register(RegisteredUser, RegisteredUserAdmin) admin.site.register(Attendance, AttendanceAdmin) admin.site.register(Event, EventAdmin) admin.site.register(EventAssignment, EventAssignmentAdmin) admin.site.register(RegistrationEventAssignment, RegistrationEventAssignmentAdmin)