60 lines
2.6 KiB
Python
60 lines
2.6 KiB
Python
import os
|
|
from django.contrib.auth.decorators import login_required
|
|
from django.contrib.auth.forms import PasswordChangeForm
|
|
from django.utils.dateparse import parse_date
|
|
from datetime import datetime, time, timedelta
|
|
import base64
|
|
import re
|
|
import urllib.parse
|
|
import urllib.request
|
|
import csv
|
|
import io
|
|
import json
|
|
from django.http import JsonResponse, HttpResponse
|
|
from django.urls import reverse
|
|
from django.shortcuts import render, redirect, get_object_or_404
|
|
from django.db import transaction
|
|
from django.db.models import Q, Sum, Value, DecimalField
|
|
from django.contrib import messages
|
|
from django.core.paginator import Paginator
|
|
from django.conf import settings
|
|
from django.db.models.functions import Coalesce
|
|
from .models import format_phone_number, Voter, Tenant, Interaction, Donation, VoterLikelihood, EventParticipation, Event, EventType, InteractionType, DonationMethod, ElectionType, CampaignSettings, Volunteer, ParticipationStatus, VolunteerEvent, Interest, VolunteerRole, ScheduledCall, BulkTask
|
|
from .filter_helper import get_filtered_voter_queryset, get_phone_search_filters
|
|
from .forms import VoterForm, InteractionForm, DonationForm, VoterLikelihoodForm, EventParticipationForm, VoterImportForm, AdvancedVoterSearchForm, EventParticipantAddForm, EventForm, VolunteerForm, VolunteerEventForm, VolunteerEventAddForm, DoorVisitLogForm, ScheduledCallForm, UserUpdateForm, EventParticipationImportForm, ParticipantMappingForm
|
|
from django.core.mail import get_connection, EmailMessage
|
|
import logging
|
|
import zoneinfo
|
|
from django.utils import timezone
|
|
|
|
from .permissions import role_required, can_view_donations, can_edit_voter, can_view_volunteers, can_edit_volunteer, can_view_voters, get_user_role, STAFF_ROLES, can_access_call_queue
|
|
logger = logging.getLogger(__name__)
|
|
|
|
from .task_runners import start_bulk_sms_task
|
|
|
|
def _robust_decode(content):
|
|
if not content:
|
|
return ""
|
|
for enc in ["utf-8-sig", "utf-8", "iso-8859-1", "windows-1252"]:
|
|
try:
|
|
return content.decode(enc)
|
|
except UnicodeDecodeError:
|
|
continue
|
|
return content.decode("utf-8", errors="replace")
|
|
|
|
|
|
def _handle_uploaded_file(uploaded_file):
|
|
"""
|
|
Handles uploaded CSV files, saves them to a temporary file, and extracts headers.
|
|
Returns (headers, temp_file_path) or (None, None) if an error occurs.
|
|
"""
|
|
import tempfile
|
|
try:
|
|
with tempfile.NamedTemporaryFile(delete=False, suffix=".csv") as tmp:
|
|
for chunk in uploaded_file.chunks():
|
|
tmp.write(chunk)
|
|
return tmp.name
|
|
except Exception as e:
|
|
logger.error(f"Error handling uploaded file: {e}")
|
|
return None
|