snowplow-python-tracker/examples/tracker_api_example/app.py at master · erdal-pb/snowplow-python-tracker · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
from distutils.log import error
from snowplow_tracker import (
    Tracker,
    Emitter,
    Subject,
    SelfDescribingJson,
    PageView,
    PagePing,
    SelfDescribing,
    ScreenView,
    StructuredEvent,
)
import sys


def get_url_from_args():
    if len(sys.argv) != 2:
        raise ValueError("Collector Endpoint is required")
    return sys.argv[1]


def main():
    collector_url = get_url_from_args()

    e = Emitter(collector_url)

    s = Subject().set_platform("pc")
    s.set_lang("en").set_user_id("test_user")

    t = Tracker(namespace="snowplow_tracker", emitters=e, subject=s)

    print("Sending events to " + e.endpoint)

    event_subject = Subject()
    event_subject.set_color_depth(10)

    page_view = PageView(
        page_url="https://www.snowplow.io",
        page_title="Homepage",
        event_subject=event_subject,
    )
    t.track(page_view)

    page_ping = PagePing(
        page_url="https://www.snowplow.io",
        page_title="Homepage",
        event_subject=t.subject,
    )
    t.track(page_ping)

    link_click = SelfDescribing(
        SelfDescribingJson(
            "iglu:com.snowplowanalytics.snowplow/link_click/jsonschema/1-0-1",
            {"targetUrl": "https://www.snowplow.io"},
        ),
        event_subject=t.subject,
    )
    t.track(link_click)

    id = t.get_uuid()
    screen_view = ScreenView(id_=id, name="name", event_subject=t.subject)
    t.track(screen_view)

    struct_event = StructuredEvent(
        category="shop",
        action="add-to-basket",
        property_="pcs",
        value=2,
        event_subject=t.subject,
    )
    t.track(struct_event)
    t.flush()


if __name__ == "__main__":
    main()