Template
Populate the client website template with catalog feature flags.
Extract always-on public/portal/UTM plus optional email_sms, directmail, blog, payments, social, and social_ai so new client sites can be bootstrapped from this seed. Refs #1 Refs #2 Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
@@ -0,0 +1,179 @@
|
||||
"""Consent, suppression, and unsubscribe helpers (always-on)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from django.core import signing
|
||||
from django.db.models import QuerySet
|
||||
from django.urls import reverse
|
||||
|
||||
from contacts.models import Channel, ConsentRecord, Contact, Suppression
|
||||
|
||||
UNSUB_SALT = "client-site-unsubscribe"
|
||||
UNSUB_MAX_AGE = 60 * 60 * 24 * 365 # 1 year
|
||||
|
||||
|
||||
def contact_may_receive(contact: Contact, channel: str) -> bool:
|
||||
if Suppression.objects.filter(
|
||||
contact=contact, channel=channel, active=True
|
||||
).exists():
|
||||
return False
|
||||
consent = ConsentRecord.objects.filter(contact=contact, channel=channel).first()
|
||||
if channel == Channel.POSTCARD:
|
||||
if consent is None:
|
||||
return Contact.postal_address_has_content(contact.postal_address)
|
||||
return bool(consent.opted_in)
|
||||
return bool(consent and consent.opted_in)
|
||||
|
||||
|
||||
def channel_preferences(contact: Contact) -> dict[str, bool]:
|
||||
"""Current opt-in flags for every channel (missing record = False).
|
||||
|
||||
Postcard: missing consent + postal address → shown as opted in (default).
|
||||
"""
|
||||
flags = {c.value: False for c in Channel}
|
||||
seen: set[str] = set()
|
||||
for record in contact.consents.all():
|
||||
flags[record.channel] = record.opted_in
|
||||
seen.add(record.channel)
|
||||
if (
|
||||
Channel.POSTCARD not in seen
|
||||
and Contact.postal_address_has_content(contact.postal_address)
|
||||
):
|
||||
flags[Channel.POSTCARD] = True
|
||||
return flags
|
||||
|
||||
|
||||
def set_channel_consent(
|
||||
contact: Contact,
|
||||
channel: str,
|
||||
*,
|
||||
opted_in: bool,
|
||||
reason: str = "",
|
||||
) -> None:
|
||||
if channel not in Channel.values:
|
||||
raise ValueError(f"Unknown channel: {channel}")
|
||||
ConsentRecord.objects.update_or_create(
|
||||
contact=contact,
|
||||
channel=channel,
|
||||
defaults={"opted_in": opted_in, "reason": reason},
|
||||
)
|
||||
Suppression.objects.update_or_create(
|
||||
contact=contact,
|
||||
channel=channel,
|
||||
defaults={
|
||||
"active": not opted_in,
|
||||
"reason": reason if not opted_in else "",
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
def set_channel_preferences(
|
||||
contact: Contact,
|
||||
preferences: dict[str, bool],
|
||||
*,
|
||||
reason: str = "",
|
||||
) -> None:
|
||||
for channel, opted_in in preferences.items():
|
||||
if channel not in Channel.values:
|
||||
continue
|
||||
set_channel_consent(
|
||||
contact, channel, opted_in=bool(opted_in), reason=reason
|
||||
)
|
||||
|
||||
|
||||
def unsubscribe_all(contact: Contact, *, reason: str = "unsubscribe_all") -> None:
|
||||
for channel in Channel:
|
||||
set_channel_consent(
|
||||
contact, channel.value, opted_in=False, reason=reason
|
||||
)
|
||||
|
||||
|
||||
def make_unsubscribe_token(contact_id: str, channel: str = Channel.EMAIL) -> str:
|
||||
return signing.dumps({"c": str(contact_id), "ch": channel}, salt=UNSUB_SALT)
|
||||
|
||||
|
||||
def parse_unsubscribe_token(token: str) -> tuple[Contact | None, str]:
|
||||
try:
|
||||
data = signing.loads(token, salt=UNSUB_SALT, max_age=UNSUB_MAX_AGE)
|
||||
except signing.BadSignature:
|
||||
return None, ""
|
||||
contact = (
|
||||
Contact.objects.filter(pk=data.get("c"))
|
||||
.prefetch_related("consents")
|
||||
.first()
|
||||
)
|
||||
if not contact:
|
||||
return None, ""
|
||||
channel = data.get("ch") or Channel.EMAIL
|
||||
if channel not in Channel.values:
|
||||
channel = Channel.EMAIL
|
||||
return contact, channel
|
||||
|
||||
|
||||
def process_unsubscribe_token(token: str) -> bool:
|
||||
contact, channel = parse_unsubscribe_token(token)
|
||||
if not contact:
|
||||
return False
|
||||
set_channel_consent(
|
||||
contact, channel, opted_in=False, reason="unsubscribe_link"
|
||||
)
|
||||
return True
|
||||
|
||||
|
||||
def preferences_url(contact_id: str, channel: str = Channel.EMAIL) -> str:
|
||||
token = make_unsubscribe_token(contact_id, channel)
|
||||
return reverse("public:unsubscribe", kwargs={"token": token})
|
||||
|
||||
|
||||
def one_click_unsubscribe_url(contact_id: str, channel: str = Channel.EMAIL) -> str:
|
||||
token = make_unsubscribe_token(contact_id, channel)
|
||||
return reverse("public:unsubscribe_one_click", kwargs={"token": token})
|
||||
|
||||
|
||||
def record_sms_stop(phone: str) -> bool:
|
||||
digits = "".join(ch for ch in (phone or "") if ch.isdigit())
|
||||
if len(digits) < 7:
|
||||
return False
|
||||
tail = digits[-10:]
|
||||
contact = None
|
||||
for row in Contact.objects.exclude(phone="").iterator():
|
||||
stored = "".join(ch for ch in row.phone if ch.isdigit())
|
||||
if stored.endswith(tail) or tail.endswith(stored[-10:]):
|
||||
contact = row
|
||||
break
|
||||
if not contact:
|
||||
return False
|
||||
set_channel_consent(
|
||||
contact, Channel.SMS, opted_in=False, reason="sms_stop"
|
||||
)
|
||||
return True
|
||||
|
||||
|
||||
def opted_in_contacts(channel: str) -> QuerySet[Contact]:
|
||||
"""Contacts opted in for channel and not actively suppressed."""
|
||||
suppressed = Suppression.objects.filter(
|
||||
channel=channel, active=True
|
||||
).values_list("contact_id", flat=True)
|
||||
|
||||
if channel == Channel.POSTCARD:
|
||||
with_address = Contact.objects.filter(
|
||||
postal_address__has_key="line1",
|
||||
).exclude(postal_address__line1="")
|
||||
explicit = with_address.filter(
|
||||
consents__channel=Channel.POSTCARD,
|
||||
consents__opted_in=True,
|
||||
)
|
||||
implicit = with_address.exclude(consents__channel=Channel.POSTCARD)
|
||||
qs = (explicit | implicit).exclude(pk__in=suppressed).distinct()
|
||||
return qs.order_by("first_name", "last_name", "email")
|
||||
|
||||
qs = (
|
||||
Contact.objects.filter(
|
||||
consents__channel=channel,
|
||||
consents__opted_in=True,
|
||||
)
|
||||
.exclude(pk__in=suppressed)
|
||||
.distinct()
|
||||
.order_by("first_name", "last_name", "email")
|
||||
)
|
||||
return qs
|
||||
Reference in New Issue
Block a user