import csv import io import logging import tempfile import os from django.contrib import admin, messages from django.urls import path from django.shortcuts import render, redirect from django.template.response import TemplateResponse from .models import ( Tenant, TenantUserRole, InteractionType, DonationMethod, ElectionType, EventType, Voter, VotingRecord, Event, EventParticipation, Donation, Interaction, VoterLikelihood, CampaignSettings ) from .forms import VoterImportForm, EventImportForm logger = logging.getLogger(__name__) VOTER_MAPPABLE_FIELDS = [ ('voter_id', 'Voter ID'), ('first_name', 'First Name'), ('last_name', 'Last Name'), ('nickname', 'Nickname'), ('birthdate', 'Birthdate'), ('address_street', 'Street Address'), ('city', 'City'), ('state', 'State'), ('zip_code', 'Zip Code'), ('county', 'County'), ('phone', 'Phone'), ('email', 'Email'), ('district', 'District'), ('precinct', 'Precinct'), ('registration_date', 'Registration Date'), ('is_targeted', 'Is Targeted'), ('candidate_support', 'Candidate Support'), ('yard_sign', 'Yard Sign'), ('window_sticker', 'Window Sticker'), ('latitude', 'Latitude'), ('longitude', 'Longitude'), ] EVENT_MAPPABLE_FIELDS = [ ('date', 'Date'), ('event_type', 'Event Type (Name)'), ('description', 'Description'), ] class TenantUserRoleInline(admin.TabularInline): model = TenantUserRole extra = 1 class CampaignSettingsInline(admin.StackedInline): model = CampaignSettings can_delete = False @admin.register(Tenant) class TenantAdmin(admin.ModelAdmin): list_display = ('name', 'slug', 'created_at') search_fields = ('name',) inlines = [TenantUserRoleInline, CampaignSettingsInline] @admin.register(TenantUserRole) class TenantUserRoleAdmin(admin.ModelAdmin): list_display = ('user', 'tenant', 'role') list_filter = ('tenant', 'role') search_fields = ('user__username', 'tenant__name') @admin.register(InteractionType) class InteractionTypeAdmin(admin.ModelAdmin): list_display = ('name', 'tenant', 'is_active') list_filter = ('tenant', 'is_active') search_fields = ('name',) @admin.register(DonationMethod) class DonationMethodAdmin(admin.ModelAdmin): list_display = ('name', 'tenant', 'is_active') list_filter = ('tenant', 'is_active') search_fields = ('name',) @admin.register(ElectionType) class ElectionTypeAdmin(admin.ModelAdmin): list_display = ('name', 'tenant', 'is_active') list_filter = ('tenant', 'is_active') search_fields = ('name',) @admin.register(EventType) class EventTypeAdmin(admin.ModelAdmin): list_display = ('name', 'tenant', 'is_active') list_filter = ('tenant', 'is_active') search_fields = ('name',) class VotingRecordInline(admin.TabularInline): model = VotingRecord extra = 1 class DonationInline(admin.TabularInline): model = Donation extra = 1 class InteractionInline(admin.TabularInline): model = Interaction extra = 1 class VoterLikelihoodInline(admin.TabularInline): model = VoterLikelihood extra = 1 @admin.register(Voter) class VoterAdmin(admin.ModelAdmin): list_display = ('first_name', 'last_name', 'nickname', 'voter_id', 'tenant', 'district', 'candidate_support', 'is_targeted', 'city', 'state') list_filter = ('tenant', 'candidate_support', 'is_targeted', 'yard_sign', 'district', 'city', 'state') search_fields = ('first_name', 'last_name', 'nickname', 'voter_id', 'address', 'city', 'state', 'zip_code', 'county') inlines = [VotingRecordInline, DonationInline, InteractionInline, VoterLikelihoodInline] change_list_template = "admin/voter_change_list.html" def get_urls(self): urls = super().get_urls() my_urls = [ path('import-voters/', self.admin_site.admin_view(self.import_voters), name='import-voters'), ] return my_urls + urls def import_voters(self, request): if request.method == "POST": if "_import" in request.POST: file_path = request.POST.get('file_path') tenant_id = request.POST.get('tenant') tenant = Tenant.objects.get(id=tenant_id) mapping = {} for field_name, _ in VOTER_MAPPABLE_FIELDS: mapping[field_name] = request.POST.get(f'map_{field_name}') try: with open(file_path, 'r', encoding='UTF-8') as f: reader = csv.DictReader(f) count = 0 errors = 0 for row in reader: try: voter_data = {} for field_name, csv_col in mapping.items(): if csv_col: val = row.get(csv_col) if val is not None: if field_name == 'is_targeted': val = str(val).lower() in ['true', '1', 'yes'] voter_data[field_name] = val voter_id = voter_data.pop('voter_id', '') if 'candidate_support' in voter_data: if voter_data['candidate_support'] not in dict(Voter.SUPPORT_CHOICES): voter_data['candidate_support'] = 'unknown' if 'yard_sign' in voter_data: if voter_data['yard_sign'] not in dict(Voter.YARD_SIGN_CHOICES): voter_data['yard_sign'] = 'none' if 'window_sticker' in voter_data: if voter_data['window_sticker'] not in dict(Voter.WINDOW_STICKER_CHOICES): voter_data['window_sticker'] = 'none' for d_field in ['registration_date', 'birthdate', 'latitude', 'longitude']: if d_field in voter_data and not voter_data[d_field]: del voter_data[d_field] Voter.objects.update_or_create( tenant=tenant, voter_id=voter_id, defaults=voter_data ) count += 1 except Exception as e: logger.error(f"Error importing voter row: {e}") errors += 1 if os.path.exists(file_path): os.remove(file_path) self.message_user(request, f"Successfully imported {count} voters.") if errors > 0: self.message_user(request, f"Failed to import {errors} rows.", level=messages.WARNING) return redirect("..") except Exception as e: self.message_user(request, f"Error processing file: {e}", level=messages.ERROR) return redirect("..") else: form = VoterImportForm(request.POST, request.FILES) if form.is_valid(): csv_file = request.FILES['file'] tenant = form.cleaned_data['tenant'] if not csv_file.name.endswith('.csv'): self.message_user(request, "Please upload a CSV file.", level=messages.ERROR) return redirect("..") with tempfile.NamedTemporaryFile(delete=False, suffix='.csv') as tmp: for chunk in csv_file.chunks(): tmp.write(chunk) file_path = tmp.name with open(file_path, 'r', encoding='UTF-8') as f: reader = csv.reader(f) headers = next(reader) context = self.admin_site.each_context(request) context.update({ 'title': "Map Voter Fields", 'headers': headers, 'model_fields': VOTER_MAPPABLE_FIELDS, 'tenant_id': tenant.id, 'file_path': file_path, 'action_url': request.path, 'opts': self.model._meta, }) return render(request, "admin/import_mapping.html", context) else: form = VoterImportForm() context = self.admin_site.each_context(request) context['form'] = form context['title'] = "Import Voters" context['opts'] = self.model._meta return render(request, "admin/import_csv.html", context) @admin.register(Event) class EventAdmin(admin.ModelAdmin): list_display = ('event_type', 'date', 'tenant') list_filter = ('tenant', 'date', 'event_type') change_list_template = "admin/event_change_list.html" def get_urls(self): urls = super().get_urls() my_urls = [ path('import-events/', self.admin_site.admin_view(self.import_events), name='import-events'), ] return my_urls + urls def import_events(self, request): if request.method == "POST": if "_import" in request.POST: file_path = request.POST.get('file_path') tenant_id = request.POST.get('tenant') tenant = Tenant.objects.get(id=tenant_id) mapping = {} for field_name, _ in EVENT_MAPPABLE_FIELDS: mapping[field_name] = request.POST.get(f'map_{field_name}') try: with open(file_path, 'r', encoding='UTF-8') as f: reader = csv.DictReader(f) count = 0 errors = 0 for row in reader: try: date = row.get(mapping.get('date')) if mapping.get('date') else None event_type_name = row.get(mapping.get('event_type')) if mapping.get('event_type') else None description = row.get(mapping.get('description')) if mapping.get('description') else '' if not date or not event_type_name: errors += 1 continue event_type, _ = EventType.objects.get_or_create( tenant=tenant, name=event_type_name ) Event.objects.create( tenant=tenant, date=date, event_type=event_type, description=description ) count += 1 except Exception as e: logger.error(f"Error importing event row: {e}") errors += 1 if os.path.exists(file_path): os.remove(file_path) self.message_user(request, f"Successfully imported {count} events.") if errors > 0: self.message_user(request, f"Failed to import {errors} rows.", level=messages.WARNING) return redirect("..") except Exception as e: self.message_user(request, f"Error processing file: {e}", level=messages.ERROR) return redirect("..") else: form = EventImportForm(request.POST, request.FILES) if form.is_valid(): csv_file = request.FILES['file'] tenant = form.cleaned_data['tenant'] if not csv_file.name.endswith('.csv'): self.message_user(request, "Please upload a CSV file.", level=messages.ERROR) return redirect("..") with tempfile.NamedTemporaryFile(delete=False, suffix='.csv') as tmp: for chunk in csv_file.chunks(): tmp.write(chunk) file_path = tmp.name with open(file_path, 'r', encoding='UTF-8') as f: reader = csv.reader(f) headers = next(reader) context = self.admin_site.each_context(request) context.update({ 'title': "Map Event Fields", 'headers': headers, 'model_fields': EVENT_MAPPABLE_FIELDS, 'tenant_id': tenant.id, 'file_path': file_path, 'action_url': request.path, 'opts': self.model._meta, }) return render(request, "admin/import_mapping.html", context) else: form = EventImportForm() context = self.admin_site.each_context(request) context['form'] = form context['title'] = "Import Events" context['opts'] = self.model._meta return render(request, "admin/import_csv.html", context) @admin.register(EventParticipation) class EventParticipationAdmin(admin.ModelAdmin): list_display = ('voter', 'event', 'participation_type') list_filter = ('event__tenant', 'event', 'participation_type') @admin.register(CampaignSettings) class CampaignSettingsAdmin(admin.ModelAdmin): list_display = ('tenant', 'donation_goal') list_filter = ('tenant',)