Coverage for atlas_ecommerce/attribution.py: 98%
40 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-25 16:36 +0000
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-25 16:36 +0000
1from collections.abc import Iterable, Mapping
2from dataclasses import dataclass, field
3from typing import Any
5CLICK_ID_KEYS = ("gclid", "gbraid", "wbraid", "fbclid", "li_fat_id", "twclid")
6UTM_KEYS = ("utm_source", "utm_medium", "utm_campaign", "utm_term", "utm_content", "utm_id")
7CONSENT_ATTRIBUTE = "_consent"
8GPC_ATTRIBUTE = "_gpc"
9MIXPANEL_DISTINCT_ID_ATTRIBUTE = "_mp_distinct_id"
10MIXPANEL_DEVICE_ID_ATTRIBUTE = "_mp_device_id"
11FBP_ATTRIBUTE = "_fbp"
12FBC_ATTRIBUTE = "_fbc"
13USER_AGENT_ATTRIBUTE = "_ua"
14MAX_VALUE_LENGTH = 512
17@dataclass(frozen=True)
18class OrderAttribution:
19 click_ids: Mapping[str, str] = field(default_factory=dict)
20 utm: Mapping[str, str] = field(default_factory=dict)
21 mp_distinct_id: str | None = None
22 mp_device_id: str | None = None
23 fbp: str | None = None
24 fbc: str | None = None
25 user_agent: str | None = None
26 analytics: bool = False
27 marketing: bool = False
28 gpc: bool = False
30 @property
31 def present(self) -> bool:
32 return bool(self.click_ids or self.utm or self.mp_distinct_id or self.analytics or self.marketing)
34 @property
35 def mixpanel_properties(self) -> dict[str, str]:
36 # Flat names on purpose: GTM forwards the site's nested `attribution` object as flat event properties.
37 return {**self.utm, **self.click_ids}
40def _clean(value: Any) -> str | None:
41 if value is None:
42 return None
43 text = str(value).strip()
44 return text[:MAX_VALUE_LENGTH] if text else None
47def from_note_attributes(attributes: Iterable[tuple[str, Any]]) -> OrderAttribution:
48 values = {name: cleaned for name, raw in attributes if (cleaned := _clean(raw)) is not None}
49 consent = {part.strip() for part in values.get(CONSENT_ATTRIBUTE, "").split(",")}
50 return OrderAttribution(
51 click_ids={key: values[f"_{key}"] for key in CLICK_ID_KEYS if f"_{key}" in values},
52 utm={key: values[f"_{key}"] for key in UTM_KEYS if f"_{key}" in values},
53 mp_distinct_id=values.get(MIXPANEL_DISTINCT_ID_ATTRIBUTE),
54 mp_device_id=values.get(MIXPANEL_DEVICE_ID_ATTRIBUTE),
55 fbp=values.get(FBP_ATTRIBUTE),
56 fbc=values.get(FBC_ATTRIBUTE),
57 user_agent=values.get(USER_AGENT_ATTRIBUTE),
58 analytics="analytics" in consent,
59 marketing="marketing" in consent,
60 gpc=values.get(GPC_ATTRIBUTE) == "1",
61 )