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 "", )