Microsoft Clarity API w Pythonie: Kompletny poradnik

Microsoft Clarity udostępnia Data Export API, które pozwala pobierać dane analityczne programowo. W tym poradniku pokażę Ci, jak skonfigurować autoryzację, odpytać API w Pythonie, przetworzyć dane i zapisać je do SQLite — żebyś mógł budować własne dashboardy i raporty bez ręcznego przeglądania interfejsu Clarity.

Co oferuje Clarity API?

Clarity Data Export API to endpoint REST, który zwraca zagregowane dane o stronach Twojego projektu. Nie zwraca nagrań sesji ani surowych zdarzeń — zamiast tego dostarcza metryki na poziomie stron:

• Liczba sesji i unikalnych użytkowników
• Średnia głębokość scrolla
• Wskaźniki frustracji (rage clicks, dead clicks, quick backs)
• Czas na stronie i bounce rate
• Podział na urządzenia (desktop, mobile, tablet)

Ograniczenia API — ważne

Zanim napiszesz jakikolwiek kod, musisz znać limity:

Krok 1: Wygeneruj token API

1. Zaloguj się do clarity.microsoft.com
2. Otwórz projekt, z którego chcesz pobierać dane
3. Przejdź do Settings > Data Export
4. Kliknij Generate API Token
5. Skopiuj token i zapisz go bezpiecznie (np. w zmiennej środowiskowej)

Token wygląda jak długi ciąg znaków alfanumerycznych. Traktuj go jak hasło — nie commituj go do repozytorium.

# .env
CLARITY_API_TOKEN=eyJhbGciOiJSUzI1NiIsInR5cCI6...
CLARITY_PROJECT_ID=abc123xyz

Krok 2: Pierwsze zapytanie w Pythonie

Zainstaluj potrzebne biblioteki:

pip install requests python-dotenv

Podstawowy skrypt pobierający dane:

import os
import requests
from dotenv import load_dotenv

load_dotenv()

API_URL = "https://www.clarity.ms/export-data/api/v1/project-live-insights"
TOKEN = os.getenv("CLARITY_API_TOKEN")
PROJECT_ID = os.getenv("CLARITY_PROJECT_ID")

headers = {
    "Authorization": f"Bearer {TOKEN}",
    "Content-Type": "application/json"
}

params = {
    "projectId": PROJECT_ID,
    "numOfDays": 1  # 1, 2 lub 3
}

response = requests.get(API_URL, headers=headers, params=params)

if response.status_code == 200:
    data = response.json()
    print(f"Pobrano {len(data)} wierszy")
    for row in data[:3]:  # pokaż pierwsze 3
        print(row)
else:
    print(f"Błąd: {response.status_code}")
    print(response.text)

Krok 3: Struktura odpowiedzi API

API zwraca listę obiektów JSON. Każdy obiekt odpowiada jednej stronie (URL) w Twoim projekcie. Typowe pola:

{
  "url": "/produkty/buty-sportowe",
  "totalSessionCount": 342,
  "totalDistinctUserCount": 289,
  "pagesPerSession": 3.2,
  "scrollDepth": 0.68,
  "activeTime": 45.3,
  "deadClickCount": 12,
  "rageClickCount": 5,
  "quickBackCount": 8,
  "excessiveScrollCount": 3,
  "deviceBreakdown": {
    "desktop": 0.45,
    "mobile": 0.51,
    "tablet": 0.04
  }
}

Krok 4: Zapis do SQLite

Żeby budować trendy w czasie, zapisuj dane codziennie do lokalnej bazy SQLite:

import sqlite3
import json
from datetime import date

def init_db(db_path="clarity.db"):
    conn = sqlite3.connect(db_path)
    conn.execute("""
        CREATE TABLE IF NOT EXISTS daily_metrics (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            date TEXT NOT NULL,
            url TEXT NOT NULL,
            sessions INTEGER,
            users INTEGER,
            scroll_depth REAL,
            dead_clicks INTEGER,
            rage_clicks INTEGER,
            quick_backs INTEGER,
            raw_json TEXT,
            UNIQUE(date, url)
        )
    """)
    conn.commit()
    return conn

def save_metrics(conn, data):
    today = date.today().isoformat()
    saved = 0

    for row in data:
        try:
            conn.execute("""
                INSERT OR REPLACE INTO daily_metrics
                (date, url, sessions, users, scroll_depth,
                 dead_clicks, rage_clicks, quick_backs, raw_json)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
            """, (
                today,
                row.get("url", ""),
                row.get("totalSessionCount", 0),
                row.get("totalDistinctUserCount", 0),
                row.get("scrollDepth", 0),
                row.get("deadClickCount", 0),
                row.get("rageClickCount", 0),
                row.get("quickBackCount", 0),
                json.dumps(row)
            ))
            saved += 1
        except Exception as e:
            print(f"Błąd zapisu: {e}")

    conn.commit()
    print(f"Zapisano {saved} wierszy za {today}")

# Użycie:
conn = init_db()
save_metrics(conn, data)  # data z poprzedniego kroku
conn.close()

Krok 5: Automatyzacja przez cron

Dodaj skrypt do crona, żeby uruchamiał się codziennie:

# Edytuj crontab
crontab -e

# Dodaj linię (uruchamianie o 7:00 rano):
0 7 * * * cd /sciezka/do/projektu && /usr/bin/python3 collect.py >> logs/collect.log 2>&1

Krok 6: Prosty raport z danych

Mając dane w SQLite, możesz generować raporty. Przykład — top 5 stron z największą liczbą rage clicks w ostatnim tygodniu:

import sqlite3
from datetime import date, timedelta

conn = sqlite3.connect("clarity.db")
week_ago = (date.today() - timedelta(days=7)).isoformat()

query = """
    SELECT url,
           SUM(rage_clicks) as total_rage,
           SUM(dead_clicks) as total_dead,
           AVG(scroll_depth) as avg_scroll,
           SUM(sessions) as total_sessions
    FROM daily_metrics
    WHERE date >= ?
    GROUP BY url
    ORDER BY total_rage DESC
    LIMIT 5
"""

rows = conn.execute(query, (week_ago,)).fetchall()
print("Top 5 stron z rage clicks (ostatnie 7 dni):")
print(f"{'URL':<50} {'Rage':>6} {'Dead':>6} {'Scroll':>7} {'Sesje':>7}")
print("-" * 80)
for row in rows:
    print(f"{row[0]:<50} {row[1]:>6} {row[2]:>6} {row[3]:>7.0%} {row[4]:>7}")

conn.close()

Obsługa błędów i retry

API Clarity może zwracać błędy — zwłaszcza gdy przekroczysz limit zapytań. Dodaj obsługę retry:

import time

def fetch_clarity_data(token, project_id, num_days=1, max_retries=3):
    headers = {"Authorization": f"Bearer {token}"}
    params = {"projectId": project_id, "numOfDays": num_days}

    for attempt in range(max_retries):
        response = requests.get(API_URL, headers=headers, params=params)

        if response.status_code == 200:
            return response.json()
        elif response.status_code == 429:  # rate limit
            wait = 60 * (attempt + 1)
            print(f"Rate limit, czekam {wait}s...")
            time.sleep(wait)
        else:
            print(f"Błąd {response.status_code}: {response.text}")
            return None

    print("Przekroczono liczbę prób")
    return None

Najlepsze praktyki

• Pobieraj dane codziennie z parametrem numOfDays=1 — oszczędzasz zapytania i unikasz duplikatów
• Zapisuj surowy JSON w osobnej kolumnie — na wypadek, gdybyś potrzebował pól, których nie sparsowałeś
• Używaj UNIQUE constraint na (date, url) — zapobiega duplikatom przy ponownym uruchomieniu
• Loguj wszystkie zapytania — przy limicie 10/dzień chcesz wiedzieć, ile zostało
• Nie hardcoduj tokenów — używaj zmiennych środowiskowych lub pliku .env