Mint campaign UTM links through the piha.li shortener.
CI / test (pull_request) Successful in 21s

Closes #9.

Auto-insert tracked homepage links in campaign compose, register them with
url_shortening_service, and put short piha.li URLs in SMS, email hrefs, and
postcard QR codes when SHORTENER_* is configured.
This commit is contained in:
2026-08-30 15:19:57 -05:00
parent dc0f3b2908
commit ef0655e8d5
17 changed files with 1082 additions and 4 deletions
+78
View File
@@ -1,6 +1,7 @@
import hashlib
import hmac
import io
import json
import logging
from django.conf import settings
@@ -26,12 +27,17 @@ from messaging.providers.postcard.pcm import (
list_designs,
)
from messaging.services import (
channel_for_audience,
create_campaign_draft,
enqueue_campaign_send,
ensure_campaign_utm_link,
message_is_removable,
opted_in_contacts,
parse_scheduled_for,
public_site_base_url,
public_site_link_label,
record_sms_stop,
resolve_campaign_tracked_url,
send_campaign_test_email,
)
from messaging.webhooks import (
@@ -462,6 +468,16 @@ def campaign_list(request):
if not body:
body = "Postcard mailing"
else:
if audience in Campaign.Audience.values:
channel = channel_for_audience(audience)
if channel in (Channel.EMAIL, Channel.SMS):
body = ensure_campaign_utm_link(
body,
name=name or "campaign",
medium=channel,
html=(channel == Channel.EMAIL),
)
form_data["body"] = body
if not body:
form_errors.append("Body is required.")
if (
@@ -506,10 +522,62 @@ def campaign_list(request):
"form_data": form_data,
"form_errors": form_errors,
"image_upload_url": reverse("messaging:campaign_image_upload"),
"utm_base_url": public_site_base_url(),
"utm_link_label": public_site_link_label(),
"utm_shorten_url": reverse("messaging:campaign_short_link"),
"utm_url": "",
},
)
@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.
"""
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()
if campaign_id:
try:
campaign = Campaign.objects.filter(pk=campaign_id).first()
except (ValidationError, ValueError):
campaign = None
if campaign is None:
return JsonResponse({"detail": "Campaign not found."}, status=404)
name = name or campaign.name
medium = medium or campaign.channel
campaign_id = str(campaign.pk)
else:
campaign_id = 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,
}
)
@login_required
def campaign_detail(request, pk):
campaign = get_object_or_404(Campaign, pk=pk)
@@ -518,6 +586,12 @@ def campaign_detail(request, pk):
except (TypeError, ValueError):
page = 1
ctx = _campaign_report(campaign, page=page)
_target, display_url = resolve_campaign_tracked_url(
name=campaign.name,
medium=campaign.channel,
campaign_id=campaign.pk,
shorten=True,
)
return render(
request,
"messaging/campaign_detail.html",
@@ -529,6 +603,10 @@ def campaign_detail(request, pk):
"recent_events": ctx["recent_events"],
"events_title": ctx["events_title"],
"events_empty": ctx["events_empty"],
"utm_url": display_url,
"utm_base_url": public_site_base_url(),
"utm_link_label": public_site_link_label(),
"utm_shorten_url": reverse("messaging:campaign_short_link"),
"can_send": campaign.status
in {
Campaign.Status.DRAFT,