monica_site/site/messaging/services.py

"""Consent checks and unsubscribe helpers."""

from __future__ import annotations

import html
import re
from datetime import datetime
from typing import TYPE_CHECKING
from urllib.parse import urlencode

from django.conf import settings
from django.core import signing
from django.db.models import QuerySet
from django.urls import reverse
from django.utils import timezone

from contacts.models import Channel, ConsentRecord, Contact, Suppression
from messaging.models import Campaign, Message, MessageTemplate
from messaging.shortener import resolve_display_url

if TYPE_CHECKING:
    from django.contrib.auth.models import AbstractBaseUser

AUDIENCE_CHANNEL = {
    Campaign.Audience.EMAIL_OPT_IN: Channel.EMAIL,
    Campaign.Audience.SMS_OPT_IN: Channel.SMS,
    Campaign.Audience.POSTCARD_OPT_IN: Channel.POSTCARD,
}

UNSUB_SALT = "monica-site-unsubscribe"
UNSUB_MAX_AGE = 60 * 60 * 24 * 365  # 1 year

# {{first_name}} preferred; {first_name} also accepted (composer hint legacy).
_MERGE_TAG_RE = re.compile(
    r"\{\{\s*(first_name|last_name|email|phone|full_name)\s*\}\}"
    r"|\{\s*(first_name|last_name|email|phone|full_name)\s*\}",
    re.IGNORECASE,
)

# Campaign tracked links: utm_source is always the brand; medium = channel.
UTM_SOURCE = "monica"
UTM_CAMPAIGN_SLUG_MAX = 80
_UTM_ANCHOR_RE = re.compile(r"(<a\b[^>]*>)(.*?)(</a>)", re.IGNORECASE | re.DOTALL)
_UTM_TEXT_URL_RE = re.compile(
    r"https?://[^\s<>\"]+utm_source=" + re.escape(UTM_SOURCE) + r"[^\s<>\"]*",
    re.IGNORECASE,
)
# Minted short links in SMS (piha.li prod, beta.piha.li, local compose).
_SHORT_TEXT_URL_RE = re.compile(
    r"https?://(?:(?:www\.)?(?:beta\.)?piha\.li|(?:127\.0\.0\.1|localhost):\d+)"
    r"/[a-z0-9]{4,8}",
    re.IGNORECASE,
)

REMOVABLE_MESSAGE_STATUSES = frozenset(
    {
        Message.Status.DRAFT,
        Message.Status.SCHEDULED,
        Message.Status.FAILED,
    }
)


def render_merge_tags(text: str, contact: Contact | None) -> str:
    """Replace personalization tags with contact field values."""
    if not text:
        return text or ""
    first = (getattr(contact, "first_name", None) or "").strip() if contact else ""
    last = (getattr(contact, "last_name", None) or "").strip() if contact else ""
    email = (getattr(contact, "email", None) or "").strip() if contact else ""
    phone = (getattr(contact, "phone", None) or "").strip() if contact else ""
    full = f"{first} {last}".strip()
    values = {
        "first_name": first,
        "last_name": last,
        "email": email,
        "phone": phone,
        "full_name": full,
    }

    def _replace(match: re.Match[str]) -> str:
        key = (match.group(1) or match.group(2) or "").lower()
        return values.get(key, "")

    return _MERGE_TAG_RE.sub(_replace, text)


def public_site_base_url() -> str:
    """Public homepage used as the UTM landing URL."""
    base = (getattr(settings, "PUBLIC_SITE_URL", None) or "").rstrip("/")
    if not base:
        base = "https://mkdrealtor.com"
    return base


def public_site_link_label() -> str:
    """Visible email-link text (host), with a stable label in local/dev."""
    host = (
        public_site_base_url()
        .replace("https://", "")
        .replace("http://", "")
        .split("/")[0]
    )
    if host.startswith("127.") or host.startswith("localhost") or host.startswith("0.0.0.0"):
        return "MKDRealtor.com"
    return host or "MKDRealtor.com"


def campaign_utm_slug(name: str) -> str:
    """Lowercase hyphenated utm_campaign from the campaign name."""
    slug = re.sub(r"[^a-z0-9]+", "-", (name or "").strip().lower())
    slug = re.sub(r"-{2,}", "-", slug).strip("-")
    return (slug or "campaign")[:UTM_CAMPAIGN_SLUG_MAX]


def build_campaign_utm_url(
    *,
    name: str,
    medium: str,
    base_url: str = "",
) -> str:
    """Homepage URL with utm_source / utm_medium / utm_campaign."""
    base = (base_url or public_site_base_url()).rstrip("/")
    channel = (medium or "").strip().lower()
    if channel not in Channel.values:
        channel = Channel.EMAIL
    query = urlencode(
        {
            "utm_source": UTM_SOURCE,
            "utm_medium": channel,
            "utm_campaign": campaign_utm_slug(name),
        }
    )
    return f"{base}/?{query}"


def campaign_shortener_ref(*, campaign_id=None, user_id=None, medium: str = "", name: str = "") -> str:
    """external_ref for the shortener (max 64). Prefer campaign UUID."""
    if campaign_id:
        return str(campaign_id)[:64]
    slug = campaign_utm_slug(name)[:40]
    ch = (medium or "email").strip().lower()[:8]
    uid = "" if user_id is None else str(user_id)
    return f"p{uid}-{ch}-{slug}"[:64]


def resolve_campaign_tracked_url(
    *,
    name: str,
    medium: str,
    campaign_id=None,
    user_id=None,
    shorten: bool = True,
) -> tuple[str, str]:
    """Return (long UTM URL, display URL). Display is short when mint works."""
    target = build_campaign_utm_url(name=name, medium=medium)
    if not shorten:
        return target, target
    display = resolve_display_url(
        target,
        title=f"{(name or 'campaign').strip()} ({medium})"[:200],
        external_ref=campaign_shortener_ref(
            campaign_id=campaign_id, user_id=user_id, medium=medium, name=name
        ),
    )
    return target, display


def _is_our_utm_anchor(attrs: str) -> bool:
    lower = attrs.lower()
    return "data-monica-utm" in lower or f"utm_source={UTM_SOURCE}" in lower


def ensure_campaign_utm_in_html(body: str, url: str, label: str) -> str:
    """Insert or refresh the tracked <a> in an HTML (email) body."""
    href = html.escape(url, quote=True)
    label_html = html.escape(label)
    replaced = False

    def _repl(match: re.Match[str]) -> str:
        nonlocal replaced
        if replaced or not _is_our_utm_anchor(match.group(1)):
            return match.group(0)
        replaced = True
        inner = match.group(2) if (match.group(2) or "").strip() else label_html
        return f'<a href="{href}" data-monica-utm="1">{inner}</a>'

    out = _UTM_ANCHOR_RE.sub(_repl, body or "")
    if replaced:
        return out
    anchor = f'<a href="{href}" data-monica-utm="1">{label_html}</a>'
    text = (body or "").rstrip()
    if text:
        return f"{text}\n<p>{anchor}</p>"
    return f"<p>{anchor}</p>"


def ensure_campaign_utm_in_text(body: str, url: str) -> str:
    """Insert or refresh the tracked URL in a plain-text (SMS) body."""
    raw = body or ""
    if _UTM_TEXT_URL_RE.search(raw):
        return _UTM_TEXT_URL_RE.sub(url, raw, count=1)
    if _SHORT_TEXT_URL_RE.search(raw):
        return _SHORT_TEXT_URL_RE.sub(url, raw, count=1)
    text = raw.rstrip()
    return f"{text}\n\n{url}" if text else url


def ensure_campaign_utm_link(
    body: str,
    *,
    name: str,
    medium: str,
    html: bool,
    campaign_id=None,
    shorten: bool | None = None,
) -> str:
    """Keep a tracked site link in the body, matching campaign name + channel."""
    do_shorten = bool(campaign_id) if shorten is None else shorten
    _target, url = resolve_campaign_tracked_url(
        name=name,
        medium=medium,
        campaign_id=campaign_id,
        shorten=do_shorten,
    )
    if html:
        return ensure_campaign_utm_in_html(body, url, public_site_link_label())
    return ensure_campaign_utm_in_text(body, url)


def message_is_removable(message: Message) -> bool:
    return message.status in REMOVABLE_MESSAGE_STATUSES


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:
        # Address on file defaults to postcard-eligible until explicit opt-out.
        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:
    """Write ConsentRecord + Suppression for one channel."""
    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:
    """Update consent for each provided channel key."""
    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]:
    """Return (contact, channel) or (None, '') on bad/expired token."""
    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:
    """One-click opt-out for the channel encoded in the token."""
    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 channel_for_audience(audience: str) -> str:
    try:
        return AUDIENCE_CHANNEL[audience]
    except KeyError as exc:
        raise ValueError(f"Unknown audience: {audience}") from exc


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:
        # Explicit opt-in, or address on file with no postcard consent row yet.
        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


def parse_scheduled_for(raw: str | None):
    """Parse optional ``datetime-local`` value into an aware datetime."""
    value = (raw or "").strip()
    if not value:
        return None
    try:
        parsed = datetime.fromisoformat(value)
    except ValueError as exc:
        raise ValueError("Invalid schedule datetime.") from exc
    if timezone.is_naive(parsed):
        return timezone.make_aware(parsed, timezone.get_current_timezone())
    return parsed


def create_campaign_draft(
    *,
    name: str,
    audience: str,
    subject: str = "",
    body: str = "",
    scheduled_for=None,
    created_by: AbstractBaseUser | None = None,
    template: MessageTemplate | None = None,
) -> Campaign:
    """Persist a draft campaign and per-recipient Message stubs."""
    channel = channel_for_audience(audience)
    campaign = Campaign.objects.create(
        name=name,
        channel=channel,
        audience=audience,
        status=Campaign.Status.DRAFT,
        scheduled_for=scheduled_for,
        subject_override=subject,
        body_override=body,
        created_by=created_by,
        template=template,
    )
    if channel in (Channel.EMAIL, Channel.SMS):
        body = ensure_campaign_utm_link(
            body,
            name=name,
            medium=channel,
            html=(channel == Channel.EMAIL),
            campaign_id=campaign.pk,
        )
        if body != campaign.body_override:
            campaign.body_override = body
            campaign.save(update_fields=["body_override", "updated_at"])
    contacts = list(opted_in_contacts(channel))
    Message.objects.bulk_create(
        [
            Message(
                campaign=campaign,
                contact=contact,
                channel=channel,
                status=Message.Status.DRAFT,
                scheduled_for=scheduled_for,
                body_snapshot=body,
            )
            for contact in contacts
        ]
    )
    return campaign


def campaign_notify_recipient(campaign: Campaign) -> str:
    """Email address for the realtor summary (created_by, else CONTACT_EMAIL)."""
    from django.conf import settings

    user = campaign.created_by
    if user is not None:
        email = (getattr(user, "email", None) or "").strip()
        if email:
            return email
    return (settings.CONTACT_EMAIL or "").strip()


def send_campaign_completion_notify(campaign: Campaign) -> bool:
    """
    One-shot summary email when a campaign finishes sending.

    Returns True if mail was sent (or already sent earlier).
    """
    from django.conf import settings
    from django.core.mail import EmailMultiAlternatives
    from django.db.models import Count, Q
    from django.template.loader import get_template
    from django.urls import reverse

    from public.email_branding import email_brand_context

    campaign.refresh_from_db()
    if campaign.notify_sent_at:
        return True
    if campaign.status != Campaign.Status.COMPLETED:
        return False

    to_email = campaign_notify_recipient(campaign)
    if not to_email:
        return False

    # Claim the notify slot atomically so concurrent refresh calls only send once.
    now = timezone.now()
    claimed = Campaign.objects.filter(
        pk=campaign.pk,
        status=Campaign.Status.COMPLETED,
        notify_sent_at__isnull=True,
    ).update(notify_sent_at=now)
    if not claimed:
        return True
    campaign.notify_sent_at = now

    counts = campaign.messages.aggregate(
        sent=Count(
            "id",
            filter=Q(
                status__in=[
                    Message.Status.SENT,
                    Message.Status.DELIVERED,
                    Message.Status.OPENED,
                    Message.Status.CLICKED,
                ]
            ),
        ),
        delivered=Count(
            "id",
            filter=Q(
                status__in=[
                    Message.Status.DELIVERED,
                    Message.Status.OPENED,
                    Message.Status.CLICKED,
                ]
            ),
        ),
        failed=Count(
            "id",
            filter=Q(
                status__in=[
                    Message.Status.FAILED,
                    Message.Status.BOUNCED,
                ]
            ),
        ),
        suppressed=Count("id", filter=Q(status=Message.Status.SUPPRESSED)),
        total=Count("id"),
    )
    report_path = reverse("messaging:campaign_detail", kwargs={"pk": campaign.pk})
    public = (settings.PUBLIC_SITE_URL or "").rstrip("/")
    report_url = f"{public}{report_path}" if public else report_path

    subject = f"Campaign sent: {campaign.name}"
    ctx = email_brand_context(
        subject=subject,
        campaign_name=campaign.name,
        channel_display=campaign.get_channel_display(),
        total=counts["total"],
        sent=counts["sent"],
        delivered=counts["delivered"],
        failed=counts["failed"],
        suppressed=counts["suppressed"],
        report_url=report_url,
    )
    text_content = get_template("emails/campaign_complete.txt").render(ctx)
    html_content = get_template("emails/campaign_complete.html").render(ctx)
    email = EmailMultiAlternatives(
        subject=subject,
        body=text_content,
        from_email=settings.DEFAULT_FROM_EMAIL,
        to=[to_email],
    )
    email.attach_alternative(html_content, "text/html")
    try:
        email.send(fail_silently=False)
    except Exception:  # noqa: BLE001 — don't block completion on mail errors
        import logging

        logging.getLogger(__name__).exception(
            "Campaign completion notify failed for %s", campaign.pk
        )
        Campaign.objects.filter(pk=campaign.pk).update(notify_sent_at=None)
        campaign.notify_sent_at = None
        return False

    return True


def refresh_campaign_status(campaign: Campaign) -> Campaign:
    """Set campaign to completed when no messages remain pending."""
    campaign.refresh_from_db()
    pending = campaign.messages.filter(
        status__in=[
            Message.Status.DRAFT,
            Message.Status.SCHEDULED,
            Message.Status.QUEUED,
        ]
    ).exists()
    if pending:
        return campaign
    if campaign.status == Campaign.Status.SENDING:
        campaign.status = Campaign.Status.COMPLETED
        campaign.save(update_fields=["status", "updated_at"])
        send_campaign_completion_notify(campaign)
    return campaign


def enqueue_campaign_send(campaign: Campaign) -> int:
    """
    Queue draft/scheduled/failed messages for send.

    Dev uses ImmediateBackend → each enqueue runs inline via SMTP/console.
    """
    from messaging.tasks import send_campaign_message

    sendable = list(
        campaign.messages.filter(
            status__in=[
                Message.Status.DRAFT,
                Message.Status.SCHEDULED,
                Message.Status.FAILED,
            ]
        )
    )
    if not sendable:
        return 0

    campaign.status = Campaign.Status.SENDING
    campaign.save(update_fields=["status", "updated_at"])

    enqueued = 0
    for message in sendable:
        message.status = Message.Status.QUEUED
        message.save(update_fields=["status", "updated_at"])
        try:
            send_campaign_message.enqueue(message_id=str(message.pk))
        except Exception:  # noqa: BLE001 — task already persisted FAILED
            pass
        enqueued += 1

    refresh_campaign_status(campaign)
    return enqueued


def send_campaign_test_email(campaign: Campaign, to_email: str) -> None:
    """Send one preview copy to ``to_email`` without touching recipient rows."""
    from django.conf import settings
    from django.core.mail import EmailMultiAlternatives
    from django.template.loader import get_template

    from public.email_branding import (
        campaign_body_to_email_html,
        campaign_body_to_plain_text,
        email_brand_context,
    )

    if campaign.channel != Channel.EMAIL:
        raise ValueError("Test send is only available for email campaigns.")
    subject = campaign.subject_override or (
        campaign.template.subject if campaign.template_id else "Message from Monica"
    )
    body = campaign.body_override or (
        campaign.template.body if campaign.template_id else ""
    )
    if not subject.strip():
        raise ValueError("Campaign has no subject.")
    if not body.strip():
        raise ValueError("Campaign has no body.")

    # Preview merge tags using first recipient when available.
    sample = (
        campaign.messages.select_related("contact")
        .order_by("created_at")
        .first()
    )
    sample_contact = sample.contact if sample else None
    subject = render_merge_tags(subject, sample_contact)
    body = render_merge_tags(body, sample_contact)

    notice = (
        "This is a test send from the Monica portal. "
        "Recipient list was not notified."
    )
    ctx = email_brand_context(
        title=f"[TEST] {subject}",
        content=f"{campaign_body_to_plain_text(body)}\n\n{notice}",
        content_html=(
            f"{campaign_body_to_email_html(body)}"
            f'<p style="margin:24px 0 0;color:#6b7280;font-size:13px;">{notice}</p>'
        ),
    )
    text_content = get_template("emails/marketing_email.txt").render(ctx)
    html_content = get_template("emails/marketing_email.html").render(ctx)
    email = EmailMultiAlternatives(
        subject=f"[TEST] {subject}",
        body=text_content,
        from_email=settings.DEFAULT_FROM_EMAIL,
        to=[to_email],
    )
    email.attach_alternative(html_content, "text/html")
    email.send(fail_silently=False)