generated from westfarn/web_django_template
275 lines
9.8 KiB
Python
275 lines
9.8 KiB
Python
from django.contrib import messages
|
|
from django.contrib.auth.decorators import login_required
|
|
from django.core.exceptions import ValidationError
|
|
from django.core.validators import validate_email
|
|
from django.db.models import Prefetch, Q
|
|
from django.http import JsonResponse
|
|
from django.shortcuts import get_object_or_404, redirect, render
|
|
from django.views.decorators.http import require_GET, require_http_methods
|
|
|
|
from contacts.models import Channel, ConsentRecord, Contact
|
|
from contacts.nominatim import NominatimError, suggest_addresses
|
|
from contacts.services import find_matching_contact, upsert_contact
|
|
from contacts.consent import channel_preferences, set_channel_preferences
|
|
|
|
|
|
def _consent_flags(contact: Contact) -> dict[str, bool]:
|
|
return channel_preferences(contact)
|
|
|
|
|
|
def _postal_from_post(post) -> dict:
|
|
return Contact.make_postal_address(
|
|
line1=post.get("address_line1", ""),
|
|
line2=post.get("address_line2", ""),
|
|
city=post.get("address_city", ""),
|
|
state=post.get("address_state", ""),
|
|
zip_code=post.get("address_zip", ""),
|
|
country=post.get("address_country", "US"),
|
|
)
|
|
|
|
|
|
@login_required
|
|
def contact_list(request):
|
|
contacts = Contact.objects.prefetch_related(
|
|
Prefetch("consents", queryset=ConsentRecord.objects.all())
|
|
).all()
|
|
q = (request.GET.get("q") or "").strip()
|
|
if q:
|
|
contacts = contacts.filter(
|
|
Q(first_name__icontains=q)
|
|
| Q(last_name__icontains=q)
|
|
| Q(email__icontains=q)
|
|
| Q(phone__icontains=q)
|
|
)
|
|
rows = list(contacts[:200])
|
|
for contact in rows:
|
|
contact.consent_flags = _consent_flags(contact)
|
|
return render(
|
|
request,
|
|
"contacts/list.html",
|
|
{"contacts": rows, "q": q},
|
|
)
|
|
|
|
|
|
@login_required
|
|
@require_http_methods(["GET", "POST"])
|
|
def contact_create(request):
|
|
form = {
|
|
"first_name": "",
|
|
"last_name": "",
|
|
"email": "",
|
|
"phone": "",
|
|
"address_line1": "",
|
|
"address_line2": "",
|
|
"address_city": "",
|
|
"address_state": "",
|
|
"address_zip": "",
|
|
"address_country": "US",
|
|
"notes": "",
|
|
"consent_email": True,
|
|
"consent_sms": False,
|
|
"consent_postcard": True,
|
|
}
|
|
match_prompt = None
|
|
if request.method == "POST":
|
|
for key in list(form.keys()):
|
|
if key.startswith("consent_"):
|
|
form[key] = key in request.POST
|
|
else:
|
|
form[key] = (request.POST.get(key) or "").strip()
|
|
email = form["email"].lower()
|
|
resolve = (request.POST.get("resolve_match") or "").strip()
|
|
match_id = (request.POST.get("match_id") or "").strip()
|
|
errors: list[str] = []
|
|
if not form["first_name"]:
|
|
errors.append("First name is required.")
|
|
if not email:
|
|
errors.append("Email is required.")
|
|
else:
|
|
try:
|
|
validate_email(email)
|
|
except ValidationError:
|
|
errors.append("Enter a valid email address.")
|
|
postal = _postal_from_post(request.POST)
|
|
has_postal = Contact.postal_address_has_content(postal)
|
|
if form["consent_sms"] and not form["phone"]:
|
|
errors.append("Phone is required for SMS consent.")
|
|
if form["consent_postcard"] and not has_postal:
|
|
form["consent_postcard"] = False
|
|
|
|
if not errors:
|
|
existing, reason = find_matching_contact(
|
|
email=email,
|
|
phone=form["phone"],
|
|
postal_address=postal if has_postal else None,
|
|
)
|
|
# Phone/address collision (different email): ask user unless they chose.
|
|
if (
|
|
existing
|
|
and reason in {"phone", "address"}
|
|
and resolve not in {"update", "create"}
|
|
):
|
|
match_prompt = {
|
|
"contact": existing,
|
|
"reason": reason,
|
|
"reason_label": "phone number"
|
|
if reason == "phone"
|
|
else "mailing address",
|
|
}
|
|
else:
|
|
merge_into = None
|
|
merge_phone_address = True
|
|
if resolve == "update" and match_id:
|
|
merge_into = Contact.objects.filter(pk=match_id).first()
|
|
if merge_into is None:
|
|
errors.append("Matched contact no longer exists.")
|
|
elif resolve == "create":
|
|
merge_phone_address = False
|
|
elif reason == "email" and existing:
|
|
merge_into = existing
|
|
|
|
if not errors:
|
|
contact, created, used_reason = upsert_contact(
|
|
email=email,
|
|
first_name=form["first_name"],
|
|
last_name=form["last_name"],
|
|
phone=form["phone"],
|
|
postal_address=postal if has_postal else None,
|
|
source=Contact.Source.MANUAL,
|
|
notes_append=form["notes"],
|
|
merge_phone_address=merge_phone_address,
|
|
merge_into=merge_into,
|
|
)
|
|
set_channel_preferences(
|
|
contact,
|
|
{
|
|
Channel.EMAIL: form["consent_email"],
|
|
Channel.SMS: form["consent_sms"],
|
|
Channel.POSTCARD: form["consent_postcard"],
|
|
},
|
|
reason="portal_manual",
|
|
)
|
|
if created:
|
|
verb = "Added"
|
|
elif used_reason == "email":
|
|
verb = "Updated (same email)"
|
|
elif resolve == "update":
|
|
verb = f"Updated (matched by {reason or used_reason})"
|
|
else:
|
|
verb = f"Updated (matched by {used_reason or 'email'})"
|
|
messages.success(request, f"{verb} {contact}.")
|
|
return redirect("contacts:detail", pk=contact.pk)
|
|
for err in errors:
|
|
messages.error(request, err)
|
|
return render(
|
|
request,
|
|
"contacts/create.html",
|
|
{"form": form, "match_prompt": match_prompt},
|
|
)
|
|
|
|
@login_required
|
|
@require_http_methods(["GET", "POST"])
|
|
def contact_detail(request, pk):
|
|
contact = get_object_or_404(
|
|
Contact.objects.prefetch_related("consents"), pk=pk
|
|
)
|
|
if request.method == "POST":
|
|
first_name = (request.POST.get("first_name") or "").strip()
|
|
last_name = (request.POST.get("last_name") or "").strip()
|
|
email = (request.POST.get("email") or "").strip().lower()
|
|
phone = (request.POST.get("phone") or "").strip()
|
|
errors: list[str] = []
|
|
if not first_name:
|
|
errors.append("First name is required.")
|
|
if not email:
|
|
errors.append("Email is required.")
|
|
else:
|
|
try:
|
|
validate_email(email)
|
|
except ValidationError:
|
|
errors.append("Enter a valid email address.")
|
|
else:
|
|
taken = (
|
|
Contact.objects.filter(email__iexact=email)
|
|
.exclude(pk=contact.pk)
|
|
.exists()
|
|
)
|
|
if taken:
|
|
errors.append("Another contact already uses that email.")
|
|
if errors:
|
|
for err in errors:
|
|
messages.error(request, err)
|
|
prefs = _consent_flags(contact)
|
|
# Reflect submitted values so the user can fix them.
|
|
contact.first_name = first_name
|
|
contact.last_name = last_name
|
|
contact.email = email
|
|
contact.phone = phone
|
|
contact.postal_address = _postal_from_post(request.POST)
|
|
contact.notes = (request.POST.get("notes") or "").strip()
|
|
return render(
|
|
request,
|
|
"contacts/detail.html",
|
|
{"contact": contact, "prefs": prefs},
|
|
)
|
|
|
|
contact.first_name = first_name
|
|
contact.last_name = last_name
|
|
contact.email = email
|
|
contact.phone = phone
|
|
contact.postal_address = _postal_from_post(request.POST)
|
|
contact.notes = (request.POST.get("notes") or "").strip()
|
|
contact.save(
|
|
update_fields=[
|
|
"first_name",
|
|
"last_name",
|
|
"email",
|
|
"phone",
|
|
"postal_address",
|
|
"notes",
|
|
"updated_at",
|
|
]
|
|
)
|
|
set_channel_preferences(
|
|
contact,
|
|
{
|
|
Channel.EMAIL: "consent_email" in request.POST,
|
|
Channel.SMS: "consent_sms" in request.POST,
|
|
Channel.POSTCARD: "consent_postcard" in request.POST,
|
|
},
|
|
reason="portal_manual",
|
|
)
|
|
messages.success(request, "Contact updated.")
|
|
return redirect("contacts:detail", pk=contact.pk)
|
|
prefs = _consent_flags(contact)
|
|
return render(
|
|
request,
|
|
"contacts/detail.html",
|
|
{"contact": contact, "prefs": prefs},
|
|
)
|
|
|
|
|
|
@login_required
|
|
def contact_import(request):
|
|
return render(request, "contacts/import.html")
|
|
|
|
|
|
@require_GET
|
|
def address_suggest(request):
|
|
"""
|
|
Backend proxy for Nominatim search. Browser JS must call this URL only —
|
|
never Nominatim directly.
|
|
"""
|
|
q = (request.GET.get("q") or "").strip()
|
|
if len(q) < 3:
|
|
return JsonResponse({"results": []})
|
|
try:
|
|
limit = int(request.GET.get("limit") or 5)
|
|
except (TypeError, ValueError):
|
|
limit = 5
|
|
try:
|
|
results = suggest_addresses(q, limit=limit)
|
|
except NominatimError as exc:
|
|
return JsonResponse({"error": str(exc), "results": []}, status=502)
|
|
return JsonResponse({"results": results})
|