Add campaign UTM links and piha.li shortener as a core helper.

Keep mint/UTM in always-on core so email_sms and directmail stay optional, and brand utm_source from SITE_NAME (or UTM_SOURCE) instead of a hardcoded client name.
Closes westfarn/web_django_template#3

site/core/campaign_utm.py

@@ -0,0 +1,238 @@
+"""Campaign tracked homepage links (UTM) plus optional piha.li short URLs.
+
+utm_source comes from UTM_SOURCE env, else a slug of SITE_NAME — never a
+hardcoded client name. Shared by email_sms and directmail.
+"""
+
+from __future__ import annotations
+
+import html
+import re
+from urllib.parse import urlencode
+
+from django.conf import settings
+from django.urls import NoReverseMatch, reverse
+
+from contacts.models import Channel
+from core.shortener import resolve_display_url
+
+UTM_CAMPAIGN_SLUG_MAX = 80
+_UTM_ANCHOR_RE = re.compile(r"(<a\b[^>]*>)(.*?)(</a>)", re.IGNORECASE | re.DOTALL)
+_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,
+)
+
+
+def campaign_utm_slug(name: str) -> str:
+    """Lowercase hyphenated slug (utm_campaign / utm_source)."""
+    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 campaign_utm_source() -> str:
+    """Brand slug for utm_source. Override with UTM_SOURCE env."""
+    explicit = (getattr(settings, "UTM_SOURCE", None) or "").strip()
+    if explicit:
+        return campaign_utm_slug(explicit)
+    return campaign_utm_slug(getattr(settings, "SITE_NAME", None) or "campaign")
+
+
+def public_site_base_url() -> str:
+    """Public homepage used as the UTM landing URL."""
+    return (getattr(settings, "PUBLIC_SITE_URL", None) or "").rstrip("/")
+
+
+def public_site_link_label() -> str:
+    """Visible email-link text (host), or SITE_NAME in local/dev."""
+    host = (
+        public_site_base_url()
+        .replace("https://", "")
+        .replace("http://", "")
+        .split("/")[0]
+    )
+    brand = (getattr(settings, "SITE_NAME", None) or "").strip() or "our website"
+    if (
+        not host
+        or host.startswith("127.")
+        or host.startswith("localhost")
+        or host.startswith("0.0.0.0")
+    ):
+        return brand
+    return host
+
+
+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": campaign_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 _utm_text_url_re() -> re.Pattern[str]:
+    return re.compile(
+        r"https?://[^\s<>\"]+utm_source="
+        + re.escape(campaign_utm_source())
+        + r"[^\s<>\"]*",
+        re.IGNORECASE,
+    )
+
+
+def _is_our_utm_anchor(attrs: str) -> bool:
+    lower = attrs.lower()
+    source = campaign_utm_source().lower()
+    return "data-campaign-utm" in lower or f"utm_source={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-campaign-utm="1">{inner}</a>'
+
+    out = _UTM_ANCHOR_RE.sub(_repl, body or "")
+    if replaced:
+        return out
+    anchor = f'<a href="{href}" data-campaign-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 ""
+    utm_re = _utm_text_url_re()
+    if utm_re.search(raw):
+        return utm_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 postcard_designer_url() -> str:
+    try:
+        return reverse("directmail:postcard_designer")
+    except NoReverseMatch:
+        return ""
+
+
+def utm_panel_context(
+    *,
+    campaign=None,
+    live: bool = False,
+    medium: str = "",
+) -> dict:
+    """Template context for `_utm_link_panel.html`."""
+    name = getattr(campaign, "name", "") or ""
+    ch = medium or getattr(campaign, "channel", "") or ""
+    display = ""
+    if campaign is not None:
+        _target, display = resolve_campaign_tracked_url(
+            name=name,
+            medium=ch,
+            campaign_id=getattr(campaign, "pk", None),
+            shorten=True,
+        )
+    try:
+        shorten_url = reverse("campaign_short_link")
+    except NoReverseMatch:
+        shorten_url = ""
+    return {
+        "utm_base_url": public_site_base_url(),
+        "utm_link_label": public_site_link_label(),
+        "utm_source": campaign_utm_source(),
+        "utm_shorten_url": shorten_url,
+        "utm_url": display,
+        "utm_live": live,
+        "utm_medium": ch,
+        "utm_campaign_name": name,
+        "utm_campaign_id": str(getattr(campaign, "pk", "") or ""),
+        "utm_postcard_designer_url": postcard_designer_url(),
+    }

site/core/shortener.py

@@ -0,0 +1,87 @@
+"""Server-to-server client for url_shortening_service (piha.li).
+
+Call the API host (`SHORTENER_BASE_URL`), never the public short domain.
+Auth: Authorization: Bearer <SHORTENER_API_TOKEN> (name:secret).
+"""
+
+from __future__ import annotations
+
+import logging
+
+import requests
+from django.conf import settings
+
+logger = logging.getLogger(__name__)
+
+
+def is_configured() -> bool:
+    base = (getattr(settings, "SHORTENER_BASE_URL", None) or "").strip()
+    token = (getattr(settings, "SHORTENER_API_TOKEN", None) or "").strip()
+    return bool(base and token)
+
+
+def mint_short_url(
+    target_url: str,
+    *,
+    title: str = "",
+    external_ref: str = "",
+) -> str | None:
+    """POST /api/links/. Return short_url, or None if unconfigured / failed.
+
+    Does not raise. Compose must still work when the shortener is down or unset.
+    """
+    if not is_configured():
+        return None
+    target = (target_url or "").strip()
+    if not target.lower().startswith("https://"):
+        logger.info("shortener skip: target is not https")
+        return None
+
+    base = str(settings.SHORTENER_BASE_URL).rstrip("/")
+    token = str(settings.SHORTENER_API_TOKEN).strip()
+    timeout = int(getattr(settings, "SHORTENER_TIMEOUT_SECONDS", 10) or 10)
+    try:
+        response = requests.post(
+            f"{base}/api/links/",
+            headers={
+                "Authorization": f"Bearer {token}",
+                "Content-Type": "application/json",
+            },
+            json={
+                "target_url": target,
+                "title": (title or "")[:200],
+                "external_ref": (external_ref or "")[:64],
+            },
+            timeout=timeout,
+        )
+    except requests.RequestException:
+        logger.exception("shortener request failed")
+        return None
+
+    if response.status_code not in (200, 201):
+        logger.warning(
+            "shortener mint failed status=%s",
+            response.status_code,
+        )
+        return None
+    try:
+        data = response.json()
+    except ValueError:
+        logger.warning("shortener mint returned non-json")
+        return None
+    if not isinstance(data, dict):
+        return None
+    short = (data.get("short_url") or "").strip()
+    return short or None
+
+
+def resolve_display_url(
+    target_url: str,
+    *,
+    title: str = "",
+    external_ref: str = "",
+) -> str:
+    """Short URL when mint succeeds, otherwise the original target."""
+    return mint_short_url(
+        target_url, title=title, external_ref=external_ref
+    ) or target_url

site/core/templates/core/_utm_link_panel.html

@@ -0,0 +1,45 @@
+<div class="utm-panel" id="utm-link-panel"
+     data-base-url="{{ utm_base_url }}"
+     data-link-label="{{ utm_link_label }}"
+     data-utm-source="{{ utm_source }}"
+     data-medium="{{ utm_medium|default:'' }}"
+     data-campaign-name="{{ utm_campaign_name|default:'' }}"
+     data-shorten-url="{{ utm_shorten_url|default:'' }}"
+     data-campaign-id="{{ utm_campaign_id }}"
+     data-csrf="{{ csrf_token }}"
+     data-live="{% if utm_live %}1{% else %}0{% endif %}">
+  <div class="field" style="margin:0">
+    <label>Tracked site link</label>
+    <p class="hint" data-utm-email-hint hidden>
+      Auto-added to the email as a styled link. Updates as you type the campaign name.
+      <code>utm_medium=email</code> · <code>utm_campaign</code> matches the name.
+      Recipients tap a short <code>piha.li</code> link that redirects to this URL.
+    </p>
+    <p class="hint" data-utm-sms-hint hidden>
+      SMS is plain text — phones cannot show HTML links. A short <code>piha.li</code>
+      URL is in the message and phones make it tappable. Preview below shows it as a link.
+    </p>
+    <p class="hint" data-utm-postcard-hint hidden>
+      Copy the QR image into the postcard designer. Recipients scan it to open the
+      short tracked link (<code>piha.li</code>).
+    </p>
+    <code class="utm-url" data-utm-url>{{ utm_url|default:"" }}</code>
+    <div class="utm-actions">
+      <button type="button" class="btn btn-ghost btn-sm" data-utm-copy-url>Copy link</button>
+    </div>
+    <div class="utm-qr" data-utm-qr-wrap hidden>
+      <canvas data-utm-qr width="192" height="192" aria-label="Campaign QR code"></canvas>
+      <div class="utm-qr-actions">
+        <button type="button" class="btn btn-ghost btn-sm" data-utm-copy-qr>Copy QR image</button>
+        <button type="button" class="btn btn-ghost btn-sm" data-utm-download-qr>Download PNG</button>
+        {% if utm_postcard_designer_url %}
+        <p class="hint" style="margin-top:8px">
+          Paste or upload the PNG in
+          <a href="{{ utm_postcard_designer_url }}">Postcard design</a>.
+        </p>
+        {% endif %}
+      </div>
+    </div>
+    <p class="hint" data-utm-status hidden></p>
+  </div>
+</div>

site/core/tests_campaign_utm.py

@@ -0,0 +1,169 @@
+"""Campaign UTM + shortener helpers (always-on core)."""
+
+from unittest.mock import MagicMock, patch
+
+from django.test import SimpleTestCase, TestCase, override_settings
+from django.urls import reverse
+
+from core.campaign_utm import (
+    build_campaign_utm_url,
+    campaign_utm_slug,
+    campaign_utm_source,
+    ensure_campaign_utm_in_text,
+    ensure_campaign_utm_link,
+)
+from core.shortener import mint_short_url, resolve_display_url
+
+
+class CampaignUtmHelperTests(SimpleTestCase):
+    def test_slug(self):
+        self.assertEqual(campaign_utm_slug("Spring seller tips"), "spring-seller-tips")
+
+    @override_settings(SITE_NAME="Acme HVAC", UTM_SOURCE="")
+    def test_source_from_site_name(self):
+        self.assertEqual(campaign_utm_source(), "acme-hvac")
+
+    @override_settings(SITE_NAME="Acme HVAC", UTM_SOURCE="acme")
+    def test_source_env_override(self):
+        self.assertEqual(campaign_utm_source(), "acme")
+
+    @override_settings(
+        SITE_NAME="Acme HVAC",
+        UTM_SOURCE="",
+        PUBLIC_SITE_URL="https://acmehvac.com",
+    )
+    def test_url_shape(self):
+        url = build_campaign_utm_url(name="Spring seller tips", medium="email")
+        self.assertEqual(
+            url,
+            "https://acmehvac.com/?utm_source=acme-hvac"
+            "&utm_medium=email&utm_campaign=spring-seller-tips",
+        )
+
+    def test_sms_replaces_short_url(self):
+        updated = ensure_campaign_utm_in_text(
+            "Hi\n\nhttps://piha.li/a3k9xm",
+            "https://piha.li/zzzzzz",
+        )
+        self.assertEqual(updated, "Hi\n\nhttps://piha.li/zzzzzz")
+
+
+class ShortenerClientTests(SimpleTestCase):
+    @override_settings(SHORTENER_BASE_URL="", SHORTENER_API_TOKEN="")
+    def test_unconfigured_returns_none(self):
+        self.assertIsNone(mint_short_url("https://acmehvac.com/"))
+        self.assertEqual(
+            resolve_display_url("https://acmehvac.com/x"),
+            "https://acmehvac.com/x",
+        )
+
+    @override_settings(
+        SHORTENER_BASE_URL="https://shortener.aimloperations.com",
+        SHORTENER_API_TOKEN="acme:test-token",
+    )
+    def test_mint_posts_bearer(self):
+        mock = MagicMock()
+        mock.status_code = 201
+        mock.json.return_value = {
+            "code": "a3k9xm",
+            "short_url": "https://piha.li/a3k9xm",
+        }
+        with patch("core.shortener.requests.post", return_value=mock) as post:
+            short = mint_short_url(
+                "https://acmehvac.com/?utm_source=acme",
+                title="Spring",
+                external_ref="campaign-uuid",
+            )
+        self.assertEqual(short, "https://piha.li/a3k9xm")
+        post.assert_called_once()
+        headers = post.call_args.kwargs["headers"]
+        self.assertEqual(headers["Authorization"], "Bearer acme:test-token")
+        payload = post.call_args.kwargs["json"]
+        self.assertEqual(payload["target_url"], "https://acmehvac.com/?utm_source=acme")
+        self.assertEqual(payload["external_ref"], "campaign-uuid")
+
+    @override_settings(
+        SHORTENER_BASE_URL="https://shortener.aimloperations.com",
+        SHORTENER_API_TOKEN="acme:test-token",
+        SITE_NAME="Acme HVAC",
+        UTM_SOURCE="",
+        PUBLIC_SITE_URL="https://acmehvac.com",
+    )
+    def test_ensure_link_falls_back_on_401(self):
+        mock = MagicMock()
+        mock.status_code = 401
+        mock.text = '{"detail":"Unauthorized"}'
+        with patch("core.shortener.requests.post", return_value=mock):
+            body = ensure_campaign_utm_link(
+                "See you Saturday",
+                name="Open house",
+                medium="sms",
+                html=False,
+                campaign_id="11111111-1111-1111-1111-111111111111",
+            )
+        self.assertIn("utm_medium=sms", body)
+        self.assertIn("utm_campaign=open-house", body)
+
+
+class CampaignShortLinkViewTests(TestCase):
+    def setUp(self):
+        from django.contrib.auth import get_user_model
+
+        User = get_user_model()
+        self.user = User.objects.create_user(
+            username="utm-composer", password="test-pass-123"
+        )
+        self.client.login(username="utm-composer", password="test-pass-123")
+
+    def test_requires_login(self):
+        from django.test import Client
+
+        response = Client().post(
+            reverse("campaign_short_link"),
+            data='{"name":"Open house","medium":"sms"}',
+            content_type="application/json",
+        )
+        self.assertEqual(response.status_code, 302)
+
+    @override_settings(
+        PUBLIC_SITE_URL="https://acmehvac.com",
+        SITE_NAME="Acme HVAC",
+        UTM_SOURCE="",
+        SHORTENER_BASE_URL="",
+        SHORTENER_API_TOKEN="",
+    )
+    def test_unconfigured_returns_long_url(self):
+        response = self.client.post(
+            reverse("campaign_short_link"),
+            data='{"name":"Open house","medium":"sms"}',
+            content_type="application/json",
+        )
+        self.assertEqual(response.status_code, 200)
+        data = response.json()
+        self.assertFalse(data["shortened"])
+        self.assertIn("utm_medium=sms", data["display_url"])
+        self.assertIn("utm_source=acme-hvac", data["display_url"])
+
+    @override_settings(
+        PUBLIC_SITE_URL="https://acmehvac.com",
+        SITE_NAME="Acme HVAC",
+        SHORTENER_BASE_URL="https://shortener.aimloperations.com",
+        SHORTENER_API_TOKEN="acme:test-token",
+    )
+    def test_mints_short_url(self):
+        mock = MagicMock()
+        mock.status_code = 201
+        mock.json.return_value = {
+            "code": "a3k9xm",
+            "short_url": "https://piha.li/a3k9xm",
+        }
+        with patch("core.shortener.requests.post", return_value=mock):
+            response = self.client.post(
+                reverse("campaign_short_link"),
+                data='{"name":"Open house","medium":"sms"}',
+                content_type="application/json",
+            )
+        self.assertEqual(response.status_code, 200)
+        data = response.json()
+        self.assertTrue(data["shortened"])
+        self.assertEqual(data["display_url"], "https://piha.li/a3k9xm")

site/core/tests_features.py

@@ -50,6 +50,8 @@ class AlwaysOnImportIsolationTests(SimpleTestCase):
         "dashboard/views.py",
         "public/views.py",
         "core/management/commands/dispatch_due.py",
+        "core/shortener.py",
+        "core/campaign_utm.py",
         "client_site/urls.py",
     ]
 

site/core/views.py

@@ -1,9 +1,12 @@
 import io
+import json
 
+from django.contrib.auth.decorators import login_required
 from django.http import FileResponse, JsonResponse
 from django.shortcuts import get_object_or_404
-from django.views.decorators.http import require_GET
+from django.views.decorators.http import require_GET, require_POST
 
+from core.campaign_utm import resolve_campaign_tracked_url
 from core.models import StoredFile
 
 
@@ -24,3 +27,40 @@ def stored_file(request, pk):
         response["Content-Disposition"] = f'inline; filename="{stored.filename}"'
     response["Cache-Control"] = "public, max-age=86400"
     return response
+
+
+@login_required
+@require_POST
+def campaign_short_link(request):
+    """Mint (or reuse) a short URL for the campaign tracked landing link.
+
+    Browser talks to this portal endpoint only. The shortener is server-to-server.
+    Does not import optional campaign apps — campaign_id is only an external_ref.
+    """
+    try:
+        payload = json.loads(request.body.decode() or "{}")
+    except (json.JSONDecodeError, UnicodeDecodeError):
+        payload = {}
+    if not isinstance(payload, dict):
+        payload = {}
+    name = (payload.get("name") or request.POST.get("name") or "").strip()
+    medium = (payload.get("medium") or request.POST.get("medium") or "").strip()
+    campaign_id = (
+        payload.get("campaign_id") or request.POST.get("campaign_id") or ""
+    )
+    campaign_id = str(campaign_id).strip() or None
+    target, display = resolve_campaign_tracked_url(
+        name=name or "campaign",
+        medium=medium,
+        campaign_id=campaign_id,
+        user_id=getattr(request.user, "pk", None),
+        shorten=True,
+    )
+    return JsonResponse(
+        {
+            "target_url": target,
+            "short_url": display if display != target else "",
+            "display_url": display,
+            "shortened": display != target,
+        }
+    )