Improve outreach compose, contact merge, and email assets.
Add a Quill email editor with DB-backed image storage, selectable PCM designs, postcard defaults for addressed contacts, and merge-by-phone/address on the contact form.
This commit is contained in:
@@ -0,0 +1,202 @@
|
||||
"""Contact identity matching and merge helpers."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
|
||||
from contacts.models import Contact
|
||||
|
||||
_PHONE_MIN_DIGITS = 10
|
||||
|
||||
|
||||
def normalize_phone_digits(value: str) -> str:
|
||||
return "".join(ch for ch in (value or "") if ch.isdigit())
|
||||
|
||||
|
||||
def phones_match(a: str, b: str) -> bool:
|
||||
"""True when both phones have ≥10 digits and last-10 match."""
|
||||
da = normalize_phone_digits(a)
|
||||
db = normalize_phone_digits(b)
|
||||
if len(da) < _PHONE_MIN_DIGITS or len(db) < _PHONE_MIN_DIGITS:
|
||||
return False
|
||||
return da[-10:] == db[-10:]
|
||||
|
||||
|
||||
def _norm_addr_part(value: str) -> str:
|
||||
text = (value or "").strip().lower()
|
||||
text = re.sub(r"[.,#]", " ", text)
|
||||
return re.sub(r"\s+", " ", text).strip()
|
||||
|
||||
|
||||
def addresses_match(a: dict | None, b: dict | None) -> bool:
|
||||
"""
|
||||
Strong postal match: same street line1 + ZIP, or line1 + city + state.
|
||||
|
||||
Empty / partial addresses never match.
|
||||
"""
|
||||
if not a or not b:
|
||||
return False
|
||||
line1_a = _norm_addr_part(a.get("line1") or "")
|
||||
line1_b = _norm_addr_part(b.get("line1") or "")
|
||||
if not line1_a or not line1_b or line1_a != line1_b:
|
||||
return False
|
||||
|
||||
zip_a = _norm_addr_part(a.get("zip") or "")
|
||||
zip_b = _norm_addr_part(b.get("zip") or "")
|
||||
if zip_a and zip_b:
|
||||
return zip_a[:5] == zip_b[:5]
|
||||
|
||||
city_a = _norm_addr_part(a.get("city") or "")
|
||||
city_b = _norm_addr_part(b.get("city") or "")
|
||||
state_a = _norm_addr_part(a.get("state") or "")
|
||||
state_b = _norm_addr_part(b.get("state") or "")
|
||||
if city_a and city_b and state_a and state_b:
|
||||
return city_a == city_b and state_a == state_b
|
||||
return False
|
||||
|
||||
|
||||
def find_matching_contact(
|
||||
*,
|
||||
email: str = "",
|
||||
phone: str = "",
|
||||
postal_address: dict | None = None,
|
||||
) -> tuple[Contact | None, str]:
|
||||
"""
|
||||
Resolve an existing contact.
|
||||
|
||||
Priority: email → phone (last 10) → mailing address.
|
||||
Returns (contact, reason) where reason is email|phone|address|"".
|
||||
"""
|
||||
email_norm = (email or "").strip().lower()
|
||||
if email_norm:
|
||||
hit = Contact.objects.filter(email__iexact=email_norm).first()
|
||||
if hit:
|
||||
return hit, "email"
|
||||
|
||||
phone_digits = normalize_phone_digits(phone)
|
||||
if len(phone_digits) >= _PHONE_MIN_DIGITS:
|
||||
tail = phone_digits[-10:]
|
||||
for row in Contact.objects.exclude(phone="").only("id", "phone").iterator():
|
||||
other = normalize_phone_digits(row.phone)
|
||||
if len(other) >= _PHONE_MIN_DIGITS and other[-10:] == tail:
|
||||
return Contact.objects.get(pk=row.pk), "phone"
|
||||
|
||||
if Contact.postal_address_has_content(postal_address):
|
||||
line1 = (postal_address.get("line1") or "").strip()
|
||||
zip_code = (postal_address.get("zip") or "").strip()
|
||||
candidates = Contact.objects.exclude(postal_address={})
|
||||
if zip_code:
|
||||
candidates = candidates.filter(
|
||||
postal_address__zip__istartswith=zip_code[:5]
|
||||
)
|
||||
elif line1:
|
||||
candidates = candidates.filter(postal_address__line1__iexact=line1)
|
||||
for row in candidates.iterator():
|
||||
if addresses_match(postal_address, row.postal_address):
|
||||
return row, "address"
|
||||
|
||||
return None, ""
|
||||
|
||||
|
||||
def _merge_notes(existing: str, addition: str) -> str:
|
||||
existing = (existing or "").strip()
|
||||
addition = (addition or "").strip()
|
||||
if not addition:
|
||||
return existing
|
||||
if not existing:
|
||||
return addition
|
||||
if addition in existing:
|
||||
return existing
|
||||
return f"{existing}\n{addition}".strip()
|
||||
|
||||
|
||||
def upsert_contact(
|
||||
*,
|
||||
email: str,
|
||||
first_name: str = "",
|
||||
last_name: str = "",
|
||||
phone: str = "",
|
||||
postal_address: dict | None = None,
|
||||
source: str = Contact.Source.CONTACT_FORM,
|
||||
notes_append: str = "",
|
||||
) -> tuple[Contact, bool, str]:
|
||||
"""
|
||||
Find or create a contact, merging on email / phone / address.
|
||||
|
||||
Returns (contact, created, match_reason).
|
||||
When merged onto a different email, the submitted email is noted in ``notes``.
|
||||
"""
|
||||
email_norm = (email or "").strip().lower()
|
||||
phone = (phone or "").strip()
|
||||
postal = postal_address if isinstance(postal_address, dict) else {}
|
||||
has_postal = Contact.postal_address_has_content(postal)
|
||||
|
||||
existing, reason = find_matching_contact(
|
||||
email=email_norm,
|
||||
phone=phone,
|
||||
postal_address=postal if has_postal else None,
|
||||
)
|
||||
|
||||
if existing is None:
|
||||
contact = Contact.objects.create(
|
||||
email=email_norm or None,
|
||||
first_name=(first_name or "").strip(),
|
||||
last_name=(last_name or "").strip(),
|
||||
phone=phone,
|
||||
postal_address=postal if has_postal else {},
|
||||
source=source,
|
||||
notes=(notes_append or "").strip(),
|
||||
)
|
||||
return contact, True, ""
|
||||
|
||||
changed: list[str] = []
|
||||
if first_name and first_name.strip():
|
||||
existing.first_name = first_name.strip()
|
||||
changed.append("first_name")
|
||||
if last_name is not None and str(last_name).strip() != "":
|
||||
existing.last_name = last_name.strip()
|
||||
changed.append("last_name")
|
||||
if phone:
|
||||
existing.phone = phone
|
||||
changed.append("phone")
|
||||
if has_postal:
|
||||
existing.postal_address = postal
|
||||
changed.append("postal_address")
|
||||
|
||||
existing_email = (existing.email or "").strip().lower()
|
||||
if email_norm and email_norm != existing_email:
|
||||
if not existing_email:
|
||||
taken = (
|
||||
Contact.objects.filter(email__iexact=email_norm)
|
||||
.exclude(pk=existing.pk)
|
||||
.exists()
|
||||
)
|
||||
if not taken:
|
||||
existing.email = email_norm
|
||||
changed.append("email")
|
||||
else:
|
||||
existing.notes = _merge_notes(
|
||||
existing.notes,
|
||||
f"Alternate email from form (owned elsewhere): {email_norm}",
|
||||
)
|
||||
changed.append("notes")
|
||||
else:
|
||||
existing.notes = _merge_notes(
|
||||
existing.notes,
|
||||
f"Alternate email from form: {email_norm}",
|
||||
)
|
||||
changed.append("notes")
|
||||
|
||||
if notes_append:
|
||||
existing.notes = _merge_notes(existing.notes, notes_append)
|
||||
changed.append("notes")
|
||||
|
||||
if source and existing.source == Contact.Source.OTHER:
|
||||
existing.source = source
|
||||
changed.append("source")
|
||||
|
||||
if changed:
|
||||
fields = sorted(set(changed) | {"updated_at"})
|
||||
existing.save(update_fields=fields)
|
||||
|
||||
return existing, False, reason
|
||||
@@ -74,7 +74,7 @@
|
||||
<div class="field">
|
||||
<label class="check-row"><input type="checkbox" name="consent_postcard" value="1" {% if prefs.postcard %}checked{% endif %}> Postcard mailings</label>
|
||||
</div>
|
||||
<p class="hint-block" style="margin-top:16px">Postcard campaigns need a street address and postcard consent. Opt-outs also write a suppression so campaigns skip this contact.</p>
|
||||
<p class="hint-block" style="margin-top:16px">Postcard mailings default on when a street address is on file. Uncheck to opt out. Opt-outs also write a suppression so campaigns skip this contact.</p>
|
||||
</div>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
@@ -0,0 +1,154 @@
|
||||
from django.test import Client, TestCase
|
||||
from django.urls import reverse
|
||||
|
||||
from contacts.models import Contact
|
||||
from contacts.services import (
|
||||
addresses_match,
|
||||
find_matching_contact,
|
||||
phones_match,
|
||||
upsert_contact,
|
||||
)
|
||||
from leads.models import Lead
|
||||
|
||||
|
||||
class ContactMatchHelpersTests(TestCase):
|
||||
def test_phones_match_ignores_formatting(self):
|
||||
self.assertTrue(phones_match("(630) 452-4443", "+1-630-452-4443"))
|
||||
self.assertFalse(phones_match("6304524443", "6304524444"))
|
||||
self.assertFalse(phones_match("4524443", "6304524443")) # too short
|
||||
|
||||
def test_addresses_match_by_line1_and_zip(self):
|
||||
a = Contact.make_postal_address(
|
||||
line1="123 Main St.",
|
||||
city="Naperville",
|
||||
state="IL",
|
||||
zip_code="60540-1234",
|
||||
)
|
||||
b = Contact.make_postal_address(
|
||||
line1="123 Main St",
|
||||
city="Elsewhere",
|
||||
state="IL",
|
||||
zip_code="60540",
|
||||
)
|
||||
self.assertTrue(addresses_match(a, b))
|
||||
|
||||
def test_find_by_phone_then_address(self):
|
||||
existing = Contact.objects.create(
|
||||
email="one@example.com",
|
||||
phone="6305551212",
|
||||
first_name="Pat",
|
||||
postal_address=Contact.make_postal_address(
|
||||
line1="9 Oak Ave",
|
||||
city="Wheaton",
|
||||
state="IL",
|
||||
zip_code="60187",
|
||||
),
|
||||
)
|
||||
by_phone, reason = find_matching_contact(
|
||||
email="other@example.com",
|
||||
phone="(630) 555-1212",
|
||||
)
|
||||
self.assertEqual(by_phone, existing)
|
||||
self.assertEqual(reason, "phone")
|
||||
|
||||
by_addr, reason = find_matching_contact(
|
||||
email="third@example.com",
|
||||
phone="9995550000",
|
||||
postal_address=Contact.make_postal_address(
|
||||
line1="9 Oak Ave",
|
||||
city="Wheaton",
|
||||
state="IL",
|
||||
zip_code="60187",
|
||||
),
|
||||
)
|
||||
self.assertEqual(by_addr, existing)
|
||||
self.assertEqual(reason, "address")
|
||||
|
||||
|
||||
class UpsertContactMergeTests(TestCase):
|
||||
def test_merge_by_phone_keeps_one_contact(self):
|
||||
original = Contact.objects.create(
|
||||
email="ryan@example.com",
|
||||
phone="6305559999",
|
||||
first_name="Ryan",
|
||||
)
|
||||
contact, created, reason = upsert_contact(
|
||||
email="ryan.alt@example.com",
|
||||
first_name="Ryan",
|
||||
last_name="Westfall",
|
||||
phone="630-555-9999",
|
||||
)
|
||||
self.assertFalse(created)
|
||||
self.assertEqual(reason, "phone")
|
||||
self.assertEqual(contact.pk, original.pk)
|
||||
self.assertEqual(Contact.objects.count(), 1)
|
||||
contact.refresh_from_db()
|
||||
self.assertEqual(contact.email, "ryan@example.com")
|
||||
self.assertIn("ryan.alt@example.com", contact.notes)
|
||||
self.assertEqual(contact.last_name, "Westfall")
|
||||
|
||||
def test_merge_by_address(self):
|
||||
original = Contact.objects.create(
|
||||
email="home@example.com",
|
||||
first_name="Sam",
|
||||
postal_address=Contact.make_postal_address(
|
||||
line1="100 Lake St",
|
||||
city="Naperville",
|
||||
state="IL",
|
||||
zip_code="60540",
|
||||
),
|
||||
)
|
||||
contact, created, reason = upsert_contact(
|
||||
email="new@example.com",
|
||||
first_name="Sam",
|
||||
postal_address=Contact.make_postal_address(
|
||||
line1="100 Lake St",
|
||||
city="Naperville",
|
||||
state="IL",
|
||||
zip_code="60540",
|
||||
),
|
||||
)
|
||||
self.assertFalse(created)
|
||||
self.assertEqual(reason, "address")
|
||||
self.assertEqual(contact.pk, original.pk)
|
||||
self.assertEqual(Contact.objects.count(), 1)
|
||||
|
||||
|
||||
class ContactFormMergeTests(TestCase):
|
||||
def setUp(self):
|
||||
self.client = Client()
|
||||
self.existing = Contact.objects.create(
|
||||
email="primary@example.com",
|
||||
phone="6301112222",
|
||||
first_name="Alex",
|
||||
postal_address=Contact.make_postal_address(
|
||||
line1="55 River Rd",
|
||||
city="Aurora",
|
||||
state="IL",
|
||||
zip_code="60505",
|
||||
),
|
||||
)
|
||||
|
||||
def test_contact_form_merges_on_phone(self):
|
||||
url = reverse("public:contact")
|
||||
response = self.client.post(
|
||||
url,
|
||||
{
|
||||
"first_name": "Alex",
|
||||
"last_name": "Lee",
|
||||
"email": "alt@example.com",
|
||||
"phone": "(630) 111-2222",
|
||||
"address_line1": "55 River Rd",
|
||||
"address_city": "Aurora",
|
||||
"address_state": "IL",
|
||||
"address_zip": "60505",
|
||||
"interest": "buying",
|
||||
"message": "Looking to buy",
|
||||
},
|
||||
)
|
||||
self.assertEqual(response.status_code, 302)
|
||||
self.assertEqual(Contact.objects.count(), 1)
|
||||
lead = Lead.objects.get()
|
||||
self.assertEqual(lead.contact_id, self.existing.pk)
|
||||
self.assertIn("alt@example.com", lead.message)
|
||||
self.assertIn("merged by phone", lead.message)
|
||||
Reference in New Issue
Block a user