web_django_template/site/contacts/consent.py

"""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