444 lines
19 KiB
Python
444 lines
19 KiB
Python
import zoneinfo
|
|
from django.db import models
|
|
from django.contrib.auth.models import User
|
|
import json
|
|
import urllib.parse
|
|
import urllib.request
|
|
import logging
|
|
from decimal import Decimal
|
|
from django.conf import settings
|
|
import re
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
def format_phone_number(phone):
|
|
"""Formats a phone number to (xxx) xxx-xxxx if it has 10 digits or 11 starting with 1."""
|
|
if not phone:
|
|
return phone
|
|
digits = re.sub(r'\D', '', str(phone))
|
|
if len(digits) == 10:
|
|
return f"({digits[:3]}) {digits[3:6]}-{digits[6:]}"
|
|
elif len(digits) == 11 and digits.startswith('1'):
|
|
return f"({digits[1:4]}) {digits[4:7]}-{digits[7:]}"
|
|
return phone
|
|
|
|
class Tenant(models.Model):
|
|
name = models.CharField(max_length=100)
|
|
created_at = models.DateTimeField(auto_now_add=True)
|
|
|
|
def __str__(self):
|
|
return self.name
|
|
|
|
class TenantUserRole(models.Model):
|
|
ROLE_CHOICES = [
|
|
('system_admin', 'System Administrator'),
|
|
('campaign_admin', 'Campaign Administrator'),
|
|
('campaign_staff', 'Campaign Staff'),
|
|
]
|
|
user = models.ForeignKey(User, on_delete=models.CASCADE, related_name='tenant_roles')
|
|
tenant = models.ForeignKey(Tenant, on_delete=models.CASCADE, related_name='user_roles')
|
|
role = models.CharField(max_length=20, choices=ROLE_CHOICES)
|
|
|
|
class Meta:
|
|
unique_together = ('user', 'tenant', 'role')
|
|
|
|
def __str__(self):
|
|
return f"{self.user.username} - {self.tenant.name} ({self.role})"
|
|
|
|
class InteractionType(models.Model):
|
|
tenant = models.ForeignKey(Tenant, on_delete=models.CASCADE, related_name='interaction_types')
|
|
name = models.CharField(max_length=100)
|
|
is_active = models.BooleanField(default=True)
|
|
|
|
class Meta:
|
|
unique_together = ('tenant', 'name')
|
|
|
|
def __str__(self):
|
|
return self.name
|
|
|
|
class DonationMethod(models.Model):
|
|
tenant = models.ForeignKey(Tenant, on_delete=models.CASCADE, related_name='donation_methods')
|
|
name = models.CharField(max_length=100)
|
|
is_active = models.BooleanField(default=True)
|
|
|
|
class Meta:
|
|
unique_together = ('tenant', 'name')
|
|
|
|
def __str__(self):
|
|
return self.name
|
|
|
|
class ElectionType(models.Model):
|
|
tenant = models.ForeignKey(Tenant, on_delete=models.CASCADE, related_name='election_types')
|
|
name = models.CharField(max_length=100)
|
|
is_active = models.BooleanField(default=True)
|
|
|
|
class Meta:
|
|
unique_together = ('tenant', 'name')
|
|
|
|
def __str__(self):
|
|
return self.name
|
|
|
|
class ParticipationStatus(models.Model):
|
|
tenant = models.ForeignKey(Tenant, on_delete=models.CASCADE, related_name='participation_statuses')
|
|
name = models.CharField(max_length=100)
|
|
is_active = models.BooleanField(default=True)
|
|
|
|
class Meta:
|
|
unique_together = ('tenant', 'name')
|
|
verbose_name_plural = 'Participation Statuses'
|
|
|
|
def __str__(self):
|
|
return self.name
|
|
|
|
class VolunteerRole(models.Model):
|
|
tenant = models.ForeignKey(Tenant, on_delete=models.CASCADE, related_name='volunteer_roles')
|
|
name = models.CharField(max_length=100)
|
|
is_active = models.BooleanField(default=True)
|
|
|
|
class Meta:
|
|
unique_together = ('tenant', 'name')
|
|
|
|
def __str__(self):
|
|
return self.name
|
|
|
|
class EventType(models.Model):
|
|
tenant = models.ForeignKey(Tenant, on_delete=models.CASCADE, related_name='event_types')
|
|
name = models.CharField(max_length=100)
|
|
available_roles = models.ManyToManyField(VolunteerRole, blank=True, related_name='event_types')
|
|
default_volunteer_role = models.ForeignKey(VolunteerRole, on_delete=models.SET_NULL, null=True, blank=True, related_name="default_for_event_types")
|
|
is_active = models.BooleanField(default=True)
|
|
|
|
class Meta:
|
|
unique_together = ('tenant', 'name')
|
|
|
|
def __str__(self):
|
|
return self.name
|
|
|
|
class Interest(models.Model):
|
|
tenant = models.ForeignKey(Tenant, on_delete=models.CASCADE, related_name='interests')
|
|
name = models.CharField(max_length=100)
|
|
|
|
class Meta:
|
|
unique_together = ('tenant', 'name')
|
|
|
|
def __str__(self):
|
|
return self.name
|
|
|
|
class Voter(models.Model):
|
|
SUPPORT_CHOICES = [
|
|
('unknown', 'Unknown'),
|
|
('supporting', 'Supporting'),
|
|
('not_supporting', 'Not Supporting'),
|
|
]
|
|
YARD_SIGN_CHOICES = [
|
|
('none', 'None'),
|
|
('wants', 'Wants a yard sign'),
|
|
('has', 'Has a yard sign'),
|
|
]
|
|
WINDOW_STICKER_CHOICES = [
|
|
('none', 'None'),
|
|
('wants', 'Wants Sticker'),
|
|
('has', 'Has Sticker'),
|
|
]
|
|
PHONE_TYPE_CHOICES = [
|
|
('home', 'Home Phone'),
|
|
('cell', 'Cell Phone'),
|
|
('work', 'Work Phone'),
|
|
]
|
|
|
|
tenant = models.ForeignKey(Tenant, on_delete=models.CASCADE, related_name='voters')
|
|
voter_id = models.CharField(max_length=50, blank=True, db_index=True)
|
|
first_name = models.CharField(max_length=100, db_index=True)
|
|
last_name = models.CharField(max_length=100, db_index=True)
|
|
nickname = models.CharField(max_length=100, blank=True)
|
|
birthdate = models.DateField(null=True, blank=True, db_index=True)
|
|
address = models.TextField(blank=True)
|
|
address_street = models.CharField(max_length=255, blank=True, db_index=True)
|
|
city = models.CharField(max_length=100, blank=True, db_index=True)
|
|
state = models.CharField(max_length=2, blank=True, db_index=True)
|
|
prior_state = models.CharField(max_length=2, blank=True)
|
|
zip_code = models.CharField(max_length=20, blank=True, db_index=True)
|
|
county = models.CharField(max_length=100, blank=True)
|
|
latitude = models.DecimalField(max_digits=12, decimal_places=9, null=True, blank=True)
|
|
longitude = models.DecimalField(max_digits=12, decimal_places=9, null=True, blank=True)
|
|
phone = models.CharField(max_length=20, blank=True)
|
|
phone_type = models.CharField(max_length=10, choices=PHONE_TYPE_CHOICES, default='cell')
|
|
secondary_phone = models.CharField(max_length=20, blank=True)
|
|
secondary_phone_type = models.CharField(max_length=10, choices=PHONE_TYPE_CHOICES, default="cell")
|
|
email = models.EmailField(blank=True)
|
|
district = models.CharField(max_length=100, blank=True, db_index=True)
|
|
precinct = models.CharField(max_length=100, blank=True, db_index=True)
|
|
registration_date = models.DateField(null=True, blank=True)
|
|
is_targeted = models.BooleanField(default=False, db_index=True)
|
|
candidate_support = models.CharField(max_length=20, choices=SUPPORT_CHOICES, default='unknown', db_index=True)
|
|
yard_sign = models.CharField(max_length=20, choices=YARD_SIGN_CHOICES, default='none', db_index=True)
|
|
window_sticker = models.CharField(max_length=20, choices=WINDOW_STICKER_CHOICES, default='none', verbose_name='Window Sticker Status', db_index=True)
|
|
notes = models.TextField(blank=True)
|
|
door_visit = models.BooleanField(default=False, db_index=True)
|
|
neighborhood = models.CharField(max_length=100, blank=True, db_index=True)
|
|
|
|
created_at = models.DateTimeField(auto_now_add=True)
|
|
|
|
def geocode_address(self, use_fallback=True):
|
|
"""
|
|
Attempts to geocode the voter's address using Google Maps API.
|
|
Returns (success, error_message).
|
|
"""
|
|
if not self.address:
|
|
return False, "No address provided."
|
|
|
|
api_key = getattr(settings, 'GOOGLE_MAPS_API_KEY', None)
|
|
if not api_key:
|
|
return False, "Google Maps API Key not configured."
|
|
|
|
def _fetch(addr):
|
|
try:
|
|
query = urllib.parse.quote(addr)
|
|
url = f"https://maps.googleapis.com/maps/api/geocode/json?address={query}&key={api_key}"
|
|
req = urllib.request.Request(url)
|
|
with urllib.request.urlopen(req, timeout=10) as response:
|
|
data = json.loads(response.read().decode())
|
|
if data.get('status') == 'OK':
|
|
result = data['results'][0]
|
|
return result['geometry']['location']['lat'], result['geometry']['location']['lng'], None
|
|
elif data.get('status') == 'ZERO_RESULTS':
|
|
return None, None, "No results found."
|
|
elif data.get('status') == 'OVER_QUERY_LIMIT':
|
|
return None, None, "Query limit exceeded."
|
|
elif data.get('status') == 'REQUEST_DENIED':
|
|
return None, None, f"Request denied: {data.get('error_message', 'No message')}"
|
|
elif data.get('status') == 'INVALID_REQUEST':
|
|
return None, None, "Invalid request."
|
|
else:
|
|
return None, None, f"Google Maps Error: {data.get('status')}"
|
|
except Exception as e:
|
|
return None, None, str(e)
|
|
|
|
logger.info(f"Geocoding with Google Maps: {self.address}")
|
|
lat, lon, err = _fetch(self.address)
|
|
|
|
if not lat and use_fallback:
|
|
# Try fallback: City, State, Zip
|
|
fallback_parts = [self.city, self.state, self.zip_code]
|
|
fallback_addr = ", ".join([p for p in fallback_parts if p])
|
|
if fallback_addr and fallback_addr != self.address:
|
|
logger.info(f"Geocoding fallback: {fallback_addr}")
|
|
lat, lon, fallback_err = _fetch(fallback_addr)
|
|
if lat:
|
|
err = None # Clear previous error if fallback works
|
|
|
|
if lat and lon:
|
|
# Truncate coordinates to 12 characters as requested
|
|
self.latitude = Decimal(str(lat)[:12])
|
|
self.longitude = Decimal(str(lon)[:12])
|
|
logger.info(f"Geocoding success: {self.latitude}, {self.longitude}")
|
|
return True, None
|
|
|
|
logger.warning(f"Geocoding failed for {self.address}: {err}")
|
|
return False, err
|
|
|
|
def save(self, *args, **kwargs):
|
|
# Auto-format phone number
|
|
self.phone = format_phone_number(self.phone)
|
|
self.secondary_phone = format_phone_number(self.secondary_phone)
|
|
|
|
# Ensure coordinates are truncated to 12 characters before saving
|
|
if self.latitude:
|
|
self.latitude = Decimal(str(self.latitude)[:12])
|
|
if self.longitude:
|
|
self.longitude = Decimal(str(self.longitude)[:12])
|
|
|
|
# Auto concatenation: address street, city, state, zip
|
|
parts = [self.address_street, self.city, self.state, self.zip_code]
|
|
self.address = ", ".join([p for p in parts if p])
|
|
|
|
# Change detection
|
|
should_geocode = False
|
|
if not self.pk:
|
|
# New record
|
|
# Only auto-geocode if coordinates were not already provided
|
|
if self.latitude is None or self.longitude is None:
|
|
should_geocode = True
|
|
else:
|
|
orig = Voter.objects.get(pk=self.pk)
|
|
# Detect if address components changed
|
|
address_changed = (self.address_street != orig.address_street or
|
|
self.city != orig.city or
|
|
self.state != orig.state or
|
|
self.zip_code != orig.zip_code)
|
|
|
|
coords_provided = (self.latitude != orig.latitude or self.longitude != orig.longitude)
|
|
|
|
# If specifically provided in import, treat as provided even if same as DB
|
|
if getattr(self, "_coords_provided_in_import", False):
|
|
coords_provided = True
|
|
|
|
# Auto-geocode if address changed AND coordinates were NOT manually updated
|
|
if address_changed and not coords_provided:
|
|
should_geocode = True
|
|
|
|
# Auto-geocode if coordinates are still missing and were not just provided
|
|
if (self.latitude is None or self.longitude is None) and not coords_provided:
|
|
should_geocode = True
|
|
|
|
if should_geocode and self.address:
|
|
# We don't want to block save if geocoding fails, so we just call it
|
|
self.geocode_address()
|
|
|
|
super().save(*args, **kwargs)
|
|
|
|
def __str__(self):
|
|
return f"{self.first_name} {self.last_name}"
|
|
|
|
class VotingRecord(models.Model):
|
|
voter = models.ForeignKey(Voter, on_delete=models.CASCADE, related_name='voting_records')
|
|
election_date = models.DateField()
|
|
election_description = models.CharField(max_length=255)
|
|
primary_party = models.CharField(max_length=100, blank=True)
|
|
|
|
def __str__(self):
|
|
return f"{self.voter} - {self.election_description}"
|
|
|
|
class Event(models.Model):
|
|
tenant = models.ForeignKey(Tenant, on_delete=models.CASCADE, related_name='events')
|
|
name = models.CharField(max_length=255, db_index=True)
|
|
date = models.DateField()
|
|
start_time = models.TimeField(null=True, blank=True)
|
|
end_time = models.TimeField(null=True, blank=True)
|
|
event_type = models.ForeignKey(EventType, on_delete=models.PROTECT, null=True)
|
|
default_volunteer_role = models.ForeignKey(VolunteerRole, on_delete=models.SET_NULL, null=True, blank=True, related_name='default_for_events')
|
|
description = models.TextField(blank=True)
|
|
location_name = models.CharField(max_length=255, blank=True)
|
|
address = models.CharField(max_length=255, blank=True)
|
|
city = models.CharField(max_length=100, blank=True)
|
|
state = models.CharField(max_length=2, blank=True)
|
|
zip_code = models.CharField(max_length=20, blank=True)
|
|
latitude = models.DecimalField(max_digits=12, decimal_places=9, null=True, blank=True)
|
|
longitude = models.DecimalField(max_digits=12, decimal_places=9, null=True, blank=True)
|
|
|
|
class Meta:
|
|
unique_together = ('tenant', 'name')
|
|
|
|
def save(self, *args, **kwargs):
|
|
# Ensure coordinates are truncated to 12 characters before saving
|
|
if self.latitude:
|
|
self.latitude = Decimal(str(self.latitude)[:12])
|
|
if self.longitude:
|
|
self.longitude = Decimal(str(self.longitude)[:12])
|
|
super().save(*args, **kwargs)
|
|
|
|
def __str__(self):
|
|
if self.name:
|
|
return f"{self.name} ({self.date})"
|
|
return f"{self.event_type} on {self.date}"
|
|
|
|
class Volunteer(models.Model):
|
|
class Meta:
|
|
ordering = ("last_name", "first_name")
|
|
tenant = models.ForeignKey(Tenant, on_delete=models.CASCADE, related_name='volunteers')
|
|
user = models.OneToOneField(User, on_delete=models.SET_NULL, null=True, blank=True, related_name='volunteer_profile')
|
|
first_name = models.CharField(max_length=100, blank=True)
|
|
last_name = models.CharField(max_length=100, blank=True)
|
|
email = models.EmailField()
|
|
phone = models.CharField(max_length=20, blank=True)
|
|
interests = models.ManyToManyField(Interest, blank=True, related_name='volunteers')
|
|
assigned_events = models.ManyToManyField(Event, through='VolunteerEvent', related_name='assigned_volunteers')
|
|
is_default_caller = models.BooleanField(default=False)
|
|
notes = models.TextField(blank=True)
|
|
|
|
def save(self, *args, **kwargs):
|
|
# Auto-format phone number
|
|
self.phone = format_phone_number(self.phone)
|
|
|
|
if self.is_default_caller:
|
|
# Only one default caller per tenant
|
|
Volunteer.objects.filter(tenant=self.tenant, is_default_caller=True).exclude(pk=self.pk).update(is_default_caller=False)
|
|
|
|
super().save(*args, **kwargs)
|
|
|
|
def __str__(self):
|
|
return f"{self.first_name} {self.last_name}".strip() or self.email
|
|
|
|
class VolunteerEvent(models.Model):
|
|
volunteer = models.ForeignKey(Volunteer, on_delete=models.CASCADE, related_name="event_assignments")
|
|
event = models.ForeignKey(Event, on_delete=models.CASCADE, related_name="volunteer_assignments")
|
|
role_type = models.ForeignKey(VolunteerRole, on_delete=models.SET_NULL, null=True, blank=True, related_name="volunteer_assignments")
|
|
|
|
def __str__(self):
|
|
return f"{self.volunteer} at {self.event} as {self.role_type or 'Assigned'}"
|
|
|
|
class EventParticipation(models.Model):
|
|
event = models.ForeignKey(Event, on_delete=models.CASCADE, related_name='participations')
|
|
voter = models.ForeignKey(Voter, on_delete=models.CASCADE, related_name='event_participations')
|
|
participation_status = models.ForeignKey(ParticipationStatus, on_delete=models.PROTECT, null=True)
|
|
|
|
def __str__(self):
|
|
return f"{self.voter} at {self.event} ({self.participation_status})"
|
|
|
|
class Donation(models.Model):
|
|
voter = models.ForeignKey(Voter, on_delete=models.CASCADE, related_name='donations')
|
|
date = models.DateField()
|
|
method = models.ForeignKey(DonationMethod, on_delete=models.SET_NULL, null=True)
|
|
amount = models.DecimalField(max_digits=10, decimal_places=2)
|
|
|
|
def __str__(self):
|
|
return f"{self.voter} - {self.amount} on {self.date}"
|
|
|
|
class Interaction(models.Model):
|
|
voter = models.ForeignKey(Voter, on_delete=models.CASCADE, related_name='interactions')
|
|
volunteer = models.ForeignKey(Volunteer, on_delete=models.SET_NULL, null=True, blank=True, related_name='interactions')
|
|
type = models.ForeignKey(InteractionType, on_delete=models.SET_NULL, null=True)
|
|
date = models.DateTimeField()
|
|
description = models.CharField(max_length=255)
|
|
notes = models.TextField(blank=True)
|
|
|
|
def __str__(self):
|
|
return f"{self.voter} - {self.type} on {self.date}"
|
|
|
|
class VoterLikelihood(models.Model):
|
|
LIKELIHOOD_CHOICES = [
|
|
('not_likely', 'Not Likely'),
|
|
('somewhat_likely', 'Somewhat Likely'),
|
|
('very_likely', 'Very Likely'),
|
|
]
|
|
voter = models.ForeignKey(Voter, on_delete=models.CASCADE, related_name='likelihoods')
|
|
election_type = models.ForeignKey(ElectionType, on_delete=models.CASCADE)
|
|
likelihood = models.CharField(max_length=20, choices=LIKELIHOOD_CHOICES)
|
|
|
|
class Meta:
|
|
unique_together = ('voter', 'election_type')
|
|
|
|
def __str__(self):
|
|
return f"{self.voter} - {self.election_type}: {self.get_likelihood_display()}"
|
|
|
|
class ScheduledCall(models.Model):
|
|
STATUS_CHOICES = [
|
|
('pending', 'Pending'),
|
|
('completed', 'Completed'),
|
|
('cancelled', 'Cancelled'),
|
|
]
|
|
tenant = models.ForeignKey(Tenant, on_delete=models.CASCADE, related_name='scheduled_calls')
|
|
voter = models.ForeignKey(Voter, on_delete=models.CASCADE, related_name='scheduled_calls')
|
|
volunteer = models.ForeignKey(Volunteer, on_delete=models.SET_NULL, null=True, blank=True, related_name='assigned_calls')
|
|
comments = models.TextField(blank=True)
|
|
status = models.CharField(max_length=20, choices=STATUS_CHOICES, default='pending')
|
|
created_at = models.DateTimeField(auto_now_add=True)
|
|
updated_at = models.DateTimeField(auto_now=True)
|
|
|
|
def __str__(self):
|
|
return f"Call for {self.voter} assigned to {self.volunteer}"
|
|
|
|
class CampaignSettings(models.Model):
|
|
tenant = models.OneToOneField(Tenant, on_delete=models.CASCADE, related_name='settings')
|
|
donation_goal = models.DecimalField(max_digits=12, decimal_places=2, default=170000.00)
|
|
twilio_account_sid = models.CharField(max_length=100, blank=True, default='ACcd11acb5095cec6477245d385a2bf127')
|
|
twilio_auth_token = models.CharField(max_length=100, blank=True, default='89ec830d0fa02ab0afa6c76084865713')
|
|
twilio_from_number = models.CharField(max_length=20, blank=True, default='+18556945903')
|
|
timezone = models.CharField(max_length=100, default="America/Chicago", choices=[(tz, tz) for tz in sorted(zoneinfo.available_timezones())])
|
|
|
|
class Meta:
|
|
verbose_name = 'Campaign Settings'
|
|
verbose_name_plural = 'Campaign Settings'
|
|
|
|
def __str__(self):
|
|
return f'Settings for {self.tenant.name}' |