westfarn
8680c082fe
Add merge tags for email/SMS, paginated removable recipients, PCM event panels for postcard campaigns, and a portal modal when phone/address matches an existing contact.
240 lines
7.4 KiB
Python
240 lines
7.4 KiB
Python
"""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,
|
|
match_email: bool = True,
|
|
match_phone: bool = True,
|
|
match_address: bool = True,
|
|
) -> 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 match_email and email_norm:
|
|
hit = Contact.objects.filter(email__iexact=email_norm).first()
|
|
if hit:
|
|
return hit, "email"
|
|
|
|
if match_phone:
|
|
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 match_address and 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 _apply_contact_fields(
|
|
existing: Contact,
|
|
*,
|
|
email_norm: str,
|
|
first_name: str,
|
|
last_name: str,
|
|
phone: str,
|
|
postal: dict,
|
|
has_postal: bool,
|
|
source: str,
|
|
notes_append: str,
|
|
) -> None:
|
|
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)
|
|
|
|
|
|
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 = "",
|
|
merge_phone_address: bool = True,
|
|
merge_into: Contact | None = None,
|
|
) -> tuple[Contact, bool, str]:
|
|
"""
|
|
Find or create a contact, merging on email / phone / address.
|
|
|
|
Returns (contact, created, match_reason).
|
|
|
|
``merge_phone_address=False`` only merges on exact email.
|
|
``merge_into`` forces merge into that contact row.
|
|
"""
|
|
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)
|
|
|
|
if merge_into is not None:
|
|
existing, reason = merge_into, "manual"
|
|
else:
|
|
existing, reason = find_matching_contact(
|
|
email=email_norm,
|
|
phone=phone,
|
|
postal_address=postal if has_postal else None,
|
|
match_phone=merge_phone_address,
|
|
match_address=merge_phone_address,
|
|
)
|
|
|
|
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, ""
|
|
|
|
_apply_contact_fields(
|
|
existing,
|
|
email_norm=email_norm,
|
|
first_name=first_name,
|
|
last_name=last_name,
|
|
phone=phone,
|
|
postal=postal,
|
|
has_postal=has_postal,
|
|
source=source,
|
|
notes_append=notes_append,
|
|
)
|
|
return existing, False, reason
|