generated from westfarn/web_django_template
107 lines
3.3 KiB
Python
107 lines
3.3 KiB
Python
import logging
|
|
|
|
from django.utils.crypto import get_random_string
|
|
|
|
from analytics.models import Attribution, PageView, UTMVisit
|
|
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
CORRELATION_COOKIE = "ms_cid"
|
|
|
|
# Portal, auth, health, assets, and transactional public endpoints.
|
|
_SKIP_PREFIXES = (
|
|
"/portal/",
|
|
"/admin/",
|
|
"/accounts/",
|
|
"/api/",
|
|
"/healthz",
|
|
"/static/",
|
|
"/media/",
|
|
"/unsubscribe/",
|
|
)
|
|
_SKIP_PATHS = frozenset({"/robots.txt", "/sitemap.xml", "/favicon.ico"})
|
|
|
|
|
|
class PublicPageViewMiddleware:
|
|
"""Record successful GET hits on public marketing pages."""
|
|
|
|
def __init__(self, get_response):
|
|
self.get_response = get_response
|
|
|
|
def __call__(self, request):
|
|
response = self.get_response(request)
|
|
if _should_record_page_view(request, response):
|
|
try:
|
|
PageView.objects.create(path=request.path[:512])
|
|
except Exception:
|
|
logger.exception("Failed to record page view for %s", request.path)
|
|
return response
|
|
|
|
|
|
def _should_record_page_view(request, response) -> bool:
|
|
if request.method != "GET":
|
|
return False
|
|
if getattr(response, "status_code", 0) != 200:
|
|
return False
|
|
path = request.path
|
|
if path in _SKIP_PATHS:
|
|
return False
|
|
return not any(path.startswith(prefix) for prefix in _SKIP_PREFIXES)
|
|
|
|
|
|
class UTMTrackingMiddleware:
|
|
"""Capture UTM params into UTMVisit and stash a correlation id cookie."""
|
|
|
|
def __init__(self, get_response):
|
|
self.get_response = get_response
|
|
|
|
def __call__(self, request):
|
|
cid = request.COOKIES.get(CORRELATION_COOKIE) or get_random_string(32)
|
|
request.utm_correlation_id = cid
|
|
|
|
params = request.GET
|
|
has_utm = any(params.get(k) for k in (
|
|
"utm_source", "utm_medium", "utm_campaign", "utm_term", "utm_content"
|
|
))
|
|
if has_utm or params.get("utm_source"):
|
|
UTMVisit.objects.create(
|
|
correlation_id=cid,
|
|
path=request.path[:512],
|
|
referrer=(request.META.get("HTTP_REFERER") or "")[:1024],
|
|
utm_source=params.get("utm_source", "")[:128],
|
|
utm_medium=params.get("utm_medium", "")[:128],
|
|
utm_campaign=params.get("utm_campaign", "")[:128],
|
|
utm_term=params.get("utm_term", "")[:128],
|
|
utm_content=params.get("utm_content", "")[:128],
|
|
user_agent=(request.META.get("HTTP_USER_AGENT") or "")[:512],
|
|
)
|
|
|
|
response = self.get_response(request)
|
|
if CORRELATION_COOKIE not in request.COOKIES:
|
|
response.set_cookie(
|
|
CORRELATION_COOKIE,
|
|
cid,
|
|
max_age=60 * 60 * 24 * 30,
|
|
samesite="Lax",
|
|
)
|
|
return response
|
|
|
|
|
|
def attribute_lead_from_request(request, lead) -> Attribution | None:
|
|
cid = getattr(request, "utm_correlation_id", None) or request.COOKIES.get(
|
|
CORRELATION_COOKIE
|
|
)
|
|
visit = None
|
|
if cid:
|
|
visit = (
|
|
UTMVisit.objects.filter(correlation_id=cid).order_by("-created_at").first()
|
|
)
|
|
return Attribution.objects.create(
|
|
lead=lead,
|
|
visit=visit,
|
|
utm_source=visit.utm_source if visit else "",
|
|
utm_medium=visit.utm_medium if visit else "",
|
|
utm_campaign=visit.utm_campaign if visit else "",
|
|
)
|