"""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"(]*>)(.*?)()", 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 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'{inner}' out = _UTM_ANCHOR_RE.sub(_repl, body or "") if replaced: return out anchor = f'{label_html}' text = (body or "").rstrip() if text: return f"{text}\n
{anchor}
" return f"{anchor}
" 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(), }