Mint campaign UTM links through the piha.li shortener (#10)

## Summary Closes #9. - Campaign composer auto-inserts a tracked homepage link (`utm_source=monica`, `utm_medium` = channel, `utm_campaign` = slug of the name) for email, SMS, and postcard QR — no manual UTM paste. - When `SHORTENER_BASE_URL` + `SHORTENER_API_TOKEN` are set, the app mints that long HTTPS URL via `POST /api/links/` on the shortener **API host** and puts the returned `piha.li` / `beta.piha.li` short URL in SMS, email hrefs, and QR codes. Empty env (local) falls back to the long UTM URL. - Live composer resolves shorts through a portal JSON endpoint (login + CSRF). Browser never calls the shortener. ## Secrets (control node, not git) **monica_site** (`~/Documents/secrets/monica_site/`): ``` # prod SHORTENER_BASE_URL=https://shortener.aimloperations.com SHORTENER_API_TOKEN=monica:<secret> # beta SHORTENER_BASE_URL=https://shortener-beta.aimloperations.com SHORTENER_API_TOKEN=monica:<beta-secret> ``` **url_shortening_service** (same secret, named token): ``` SHORTENER_API_TOKENS=monica:<secret> SHORT_ALLOWED_HOSTS=mkdrealtor.com,aimloperations.com ``` Prod public short host: `piha.li`. Beta: `beta.piha.li`. Generate with `python -c "import secrets; print(secrets.token_urlsafe(32))"`. Template port: westfarn/web_django_template#3 ## Test plan - [ ] `cd site && uv run python manage.py test messaging.tests.CampaignUtmLinkTests` - [ ] Composer: type a campaign name — email gets an HTML link, SMS gets a URL, postcard shows a QR - [ ] With shortener env set: SMS/QR show `piha.li` (or `beta.piha.li`); without it, long UTM URL still works - [ ] Copy/download QR into postcard designer - [ ] Secret files have `SHORTENER_*` on both caller and operator sides before beta/prod deploy Reviewed-on: #10
2026-08-30 13:24:40 -07:00
parent dc0f3b2908
commit 2b3f10e784
17 changed files with 1082 additions and 4 deletions
@@ -2,10 +2,13 @@

 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
@@ -13,6 +16,7 @@ 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
@@ -33,6 +37,21 @@ _MERGE_TAG_RE = re.compile(
    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,
@@ -66,6 +85,149 @@ def render_merge_tags(text: str, contact: Contact | None) -> str:
    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

@@ -287,6 +449,17 @@ def create_campaign_draft(
        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(
        [