Experimentergebnisse nach Airtable exportieren

Ziel

Dieses Tutorial beschreibt Schritt für Schritt, wie das Skript kameleoon_to_airtable.py funktioniert. Anhand einer Kameleoon-Experiment-ID, einer Airtable-Base-ID und einer Airtable-Table-ID ruft das Skript die Experimentmetadaten und die statistischen Ergebnisse ab, ermittelt die leistungsstärkste Variation, bildet die Daten auf das Airtable-Schema Experiments ab und schreibt den Datensatz zurück nach Airtable. Ein erneuter Lauf des Skripts für dasselbe Experiment aktualisiert die bestehende Zeile, statt ein Duplikat zu erstellen. Das Tutorial baut auf dem vorherigen Tutorial zum Abrufen von Experimentergebnissen mit der Automation API auf und verwendet denselben Ablauf für Anfrage und Abfrage.

Voraussetzungen

Kameleoon-API-Anmeldedaten. Die Automation API erfordert ein Access Token. Das Skript ruft dieses programmgesteuert anhand einer client_id und eines client_secret über den client_credentials-Grant ab. Siehe Access Token abrufen.
Ein persönliches Airtable-Zugriffstoken mit dem Scope data.records:write für die Ziel-Base.
Die Airtable-Base-ID und Table-ID. Beide finden Sie in der URL der Tabelle oder in der API-Dokumentation der Base. Die Base-ID beginnt mit app, die Table-ID mit tbl.
Eine Airtable-Tabelle mit bereits erstelltem Schema Experiments. Das Skript schreibt in diese Felder: Experiment Name, Status, Start date, End date, Notes, Actual, Probability und Result. Zusätzlich erwartet es, dass die manuell zu befüllenden Felder Assignee, Category, Prediction, Mkt Est, Eng Est und Attachments existieren, auch wenn es sie nie setzt. Airtable lehnt das Schreiben in einen Feldnamen ab, der in der Tabelle noch nicht existiert, und typecast wandelt Werttypen nur für bereits vorhandene Felder um (es erstellt keine fehlenden Felder oder Auswahloptionen). Legen Sie die Tabelle mit diesen Feldern und den passenden Status-Auswahloptionen an, bevor Sie das Skript ausführen. Den Wert, den jedes Feld erhält, finden Sie in Schritt 6.
Python 3.9+ mit der Bibliothek requests (pip install requests).

Speichern Sie alle Anmeldedaten in Umgebungsvariablen. Codieren Sie Geheimnisse niemals fest im Skript.

export KAMELEOON_CLIENT_ID="..."
export KAMELEOON_CLIENT_SECRET="..."
export AIRTABLE_TOKEN="..."

Das Tutorial verwendet das Beispielexperiment Product Page Redesign (ID 188308) mit zwei Variationen zusätzlich zur Originalversion: Redesign 1 (ID 828220) und Redesign 2 (ID 828221).

1. Bei der Automation API authentifizieren

Endpoint: Rufen Sie ein Access Token ab, indem Sie eine POST-Anfrage an den Token-Endpoint senden.

POST https://api.kameleoon.com/oauth/token

Feld	Typ	Beschreibung
grant_type	String	Auf `client_credentials` setzen.
client_id	String	Ihre Automation-API-Client-ID.
client_secret	String	Ihr Automation-API-Client-Secret.

Beispiel:

def kameleoon_token(client_id, client_secret):
    resp = requests.post(
        "https://api.kameleoon.com/oauth/token",
        headers={"Content-Type": "application/x-www-form-urlencoded"},
        data={
            "grant_type": "client_credentials",
            "client_id": client_id,
            "client_secret": client_secret,
        },
    )
    resp.raise_for_status()
    return resp.json()["access_token"]

Antwort:

{ "access_token": "eyJhbGciOiJIUzUxMiIsInR5cCI6IkpXVCJ9..." }

Senden Sie den zurückgegebenen access_token als Bearer-Token bei jeder weiteren Anfrage an die Automation API. Access Tokens sind standardmäßig 2 Stunden lang gültig.

2. Das Experiment abrufen

Endpoint: Rufen Sie die Experimentmetadaten ab, indem Sie eine GET-Anfrage an den Endpoint Get an experiment senden.

GET https://api.kameleoon.com/experiments/{experimentId}

Feld	Typ	Beschreibung
experimentId	Integer	Obligatorischer Pfadparameter. Die ID des abzurufenden Experiments.

Beispiel:

def get_experiment(token, experiment_id):
    resp = requests.get(
        f"https://api.kameleoon.com/experiments/{experiment_id}",
        headers={"Authorization": f"Bearer {token}"},
    )
    resp.raise_for_status()
    return resp.json()

Antwort (gekürzt):

{
  "id": 188308,
  "name": "Product Page Redesign",
  "status": "STOPPED",
  "dateStarted": "2025-01-15T09:00:00Z",
  "dateEnded": "2025-02-12T18:00:00Z",
  "description": "Testing two redesigns of the product page.",
  "mainGoalId": 279599,
  "variations": [828220, 828221]
}

Das Skript liest name, status, dateStarted, dateEnded und description für den Airtable-Datensatz sowie mainGoalId, um die Ergebnisanfrage im nächsten Schritt einzugrenzen.

Die API gibt mainGoalId standardmäßig zurück, sodass Sie keinen optionalFields-Parameter benötigen, um sie zu lesen. Die Automation API veröffentlicht kein festes Enum für das Feld status, aber andere statusartige Felder in der gesamten API verwenden durchgängig Token in Großbuchstaben (zum Beispiel STOPPED, ACTIVE, DRAFT). Schritt 6 bildet das Feld status unter dieser Annahme ab. Bestätigen Sie die genauen Token, die Ihr Konto zurückgibt, mit einer Live-Anfrage, bevor Sie sich auf dieses Mapping verlassen.

3. Die Ergebnisse des Experiments anfordern

Endpoint: Lösen Sie die Erstellung des Ergebnisberichts aus, indem Sie eine POST-Anfrage an den Endpoint Request experiment’s results senden.

POST https://api.kameleoon.com/experiments/{experimentId}/results

Feld	Typ	Beschreibung
experimentId	Integer	Obligatorischer Pfadparameter.
goalsIds	Array	Beschränkt den Bericht auf die angegebenen Ziel-IDs. Das Skript übergibt die `mainGoalId` des Experiments.
referenceVariationId	String	Die Variation, die als Referenz für den Vergleich verwendet wird. `"0"` verwendet die Originalseite.
visitorData	Boolean	`false` für besuchsbasierte Daten, `true` für besucherbasierte Daten.
sequentialTesting	Boolean	Auf `true` setzen, um Sequential Testing für Konfidenzintervalle zu verwenden.
bayesian	Boolean	Auf `true` setzen, um die Bayesianische Erfolgswahrscheinlichkeit in den Bericht aufzunehmen. Das Skript liest diesen Wert aus, um das Feld Probability zu befüllen.
conversionType	String	`ALL_CONVERSION` oder `CONVERTED_VISITS`.

Beispiel:

def request_results(token, experiment_id, goal_id):
    body = {
        "visitorData": False,
        "sequentialTesting": True,
        "bayesian": True,
        "referenceVariationId": "0",
        "conversionType": "ALL_CONVERSION",
        "goalsIds": [goal_id] if goal_id else None,
    }
    resp = requests.post(
        f"https://api.kameleoon.com/experiments/{experiment_id}/results",
        headers={
            "Authorization": f"Bearer {token}",
            "Content-Type": "application/json",
            "Accept": "*/*",
        },
        json=body,
    )
    resp.raise_for_status()
    return resp.json()["dataCode"]

Antwort:

{ "dataCode": "14443931880924098266207585267983330260134079899081739889989435434342588016765" }

Kameleoon erstellt den Bericht asynchron. Der Endpoint gibt einen dataCode zurück, den Schritt 4 verwendet, um das Ergebnis abzufragen.

Dieses Skript erfordert bayesian: true. Bei aktiviertem Bayesian-Modus trägt der Wert reliability des Berichts die Bayesianische Erfolgswahrscheinlichkeit (die Wahrscheinlichkeit, dass eine Variation die Referenz übertrifft), die das Skript auf das Feld Probability abbildet. Gleichen Sie den Wert mit demselben Bericht in der Kameleoon-App ab, wenn Ihr Konto eine andere Standardstatistikmethode verwendet. Die Spezifikation der Automation API kennzeichnet dateIntervals als erforderlich, aber das Beispiel oben lässt es aus und liefert dennoch einen gültigen Bericht. Die Spezifikation dokumentiert nicht, worauf ein ausgelassenes dateIntervals standardmäßig zurückgreift. Dieses Tutorial geht davon aus, dass es den gesamten Laufzeitraum des Experiments abdeckt. Bestätigen Sie dieses Verhalten anhand Ihres eigenen Kontos, bevor Sie sich darauf verlassen. Übergeben Sie stattdessen ein dateIntervals-Array, um den Bericht auf einen bestimmten Zeitraum einzugrenzen.

4. Die Ergebnisse abfragen

Endpoint: Rufen Sie den Bericht ab, indem Sie GET-Anfragen an den Endpoint Poll results senden, bis er bereit ist.

GET https://api.kameleoon.com/results?dataCode={dataCode}

Feld	Typ	Beschreibung
dataCode	String	Obligatorischer Query-Parameter. Verwenden Sie den Hash, den `POST /experiments/{experimentId}/results` in Schritt 3 zurückgegeben hat.

Der status der Antwort lautet WAITING, solange Kameleoon den Bericht berechnet, READY, sobald die Daten verfügbar sind, oder ERROR beziehungsweise TIMEOUT bei einem Fehler. Wenn der Status ERROR oder TIMEOUT lautet, enthält die Antwort ein errorDescription-Feld auf oberster Ebene. Das Skript fragt in einem festen Intervall ab, bis der Status READY lautet. Beispiel:

def poll_results(token, data_code, max_attempts=30, delay=2.0):
    for _ in range(max_attempts):
        resp = requests.get(
            "https://api.kameleoon.com/results",
            headers={"Authorization": f"Bearer {token}"},
            params={"dataCode": data_code},
        )
        resp.raise_for_status()
        payload = resp.json()
        status = payload.get("status")
        if status == "READY":
            return payload["data"]
        if status in ("ERROR", "TIMEOUT"):
            raise RuntimeError(payload.get("errorDescription") or status)
        time.sleep(delay)  # status == "WAITING"
    raise TimeoutError("Timed out waiting for results.")

Antwort (gekürzt):

{
  "status": "READY",
  "data": {
    "variationData": {
      "_reference": { "breakdownData": { "_reference": { "generalData": {
        "goalsData": { "279599": { "conversionRate": 0.0149 } } } } } },
      "828220": { "breakdownData": { "_reference": { "generalData": {
        "goalsData": { "279599": {
          "reliability": 100.0,
          "improvementRate": 211.48,
          "conversionRate": 0.0466
        } } } } } },
      "828221": { "breakdownData": { "_reference": { "generalData": {
        "goalsData": { "279599": {
          "reliability": 100.0,
          "improvementRate": -43.33,
          "conversionRate": 0.0085
        } } } } } }
    }
  }
}

5. Die leistungsstärkste Variation auswählen

Die Ergebnisse enthalten unter variationData einen Eintrag pro Variation sowie die Zeile _reference für die Originalseite. Für jede Variation liegen die Metriken für das angeforderte Ziel unter breakdownData._reference.generalData.goalsData[goalId]. Das Skript überspringt den Eintrag _reference, liest für jede Variation improvementRate und reliability (die Bayesianische Erfolgswahrscheinlichkeit) und wählt die Variation mit der höchsten Verbesserungsrate als leistungsstärkste aus. Das im nächsten Schritt abgebildete Feld Result erfasst, ob diese Variation tatsächlich gewonnen hat: ob sie eine ausreichend hohe Erfolgswahrscheinlichkeit bei einer positiven Verbesserung erreicht hat. Beispiel:

def pick_best_variation(result_data, goal_id):
    best, best_improvement = {}, float("-inf")
    for variation_id, vdata in result_data["variationData"].items():
        if variation_id == "_reference":
            continue
        general = vdata["breakdownData"]["_reference"]["generalData"]
        goals_data = general.get("goalsData", {})
        if not goals_data:
            continue
        key = str(goal_id) if str(goal_id) in goals_data else next(iter(goals_data))
        metrics = goals_data[key]
        improvement = metrics.get("improvementRate")
        if improvement is not None and improvement > best_improvement:
            best_improvement = improvement
            best = {
                "variation_id": variation_id,
                "improvement_rate": improvement,
                "bayesian_probability": metrics.get("reliability"),
            }
    return best

Im Beispiel erreichen beide Variationen eine Bayesianische Erfolgswahrscheinlichkeit von 100 %, aber Redesign 1 (828220) zeigt eine Verbesserung von +211,48 % gegenüber -43,33 % bei Redesign 2. Redesign 1 ist daher die leistungsstärkste Variation und, mit einer Wahrscheinlichkeit über 95 % und einer positiven Verbesserung, ein echter Gewinner.

6. Die Daten auf Airtable-Felder abbilden

Das Skript wandelt die Experimentmetadaten und die Metriken der leistungsstärksten Variation in das Airtable-Schema Experiments um.

Airtable-Feld	Quelle	Transformation
Experiment Name	`experiment.name`	Direkt. Wird als Upsert-Schlüssel verwendet.
Status	`experiment.status`	Zugeordnet, ohne Berücksichtigung der Groß-/Kleinschreibung: `ACTIVE → Running`, `DRAFT`/`PLANNED → Implementing`, `STOPPED`/`DIVERTED → Completed`, `PAUSED → Defunct`.
Start date	`experiment.dateStarted`	Datum/Uhrzeit, gekürzt auf ein ISO-Datum (`YYYY-MM-DD`).
End date	`experiment.dateEnded`	Datum/Uhrzeit, gekürzt auf ein ISO-Datum.
Notes	`experiment.description`	Direkt.
Actual	`improvementRate` der besten Variation	Direkt (gemessene Verbesserung, %).
Probability	Bayesianische Erfolgswahrscheinlichkeit der besten Variation	In Stufen eingeteilt: `≥95 → 80% - High`, `≥80 → 50% - Medium`, sonst `20% - Low`.
Result	Bayesianische Erfolgswahrscheinlichkeit + `improvementRate`	Wahrscheinlichkeit `≥ 95` und Verbesserung > 0 → `Success`; `≥ 95` und Verbesserung < 0 → `Failure`; andernfalls `Inconclusive`.

Das Skript setzt keine Felder ohne Kameleoon-Quelle: Assignee, Category, Prediction, Mkt Est, Eng Est und Attachments. Diese Felder bleiben für die manuelle Eingabe in Airtable verfügbar. Das Skript lässt außerdem leere Werte aus, sodass es niemals eine bestehende Zelle mit einem leeren Wert überschreibt. Beispiel:

STATUS_MAP = {
    "ACTIVE": "Running",
    "DRAFT": "Implementing",
    "PLANNED": "Implementing",
    "PAUSED": "Defunct",
    "STOPPED": "Completed",
    "DIVERTED": "Completed",
}


def map_status(status):
    return STATUS_MAP.get((status or "").upper())


def to_iso_date(value):
    return value[:10] if value else None


def bayesian_to_probability(probability):
    if probability is None:
        return None
    if probability >= 95:
        return "80% - High"
    if probability >= 80:
        return "50% - Medium"
    return "20% - Low"


def derive_result(probability, improvement):
    if probability is None or improvement is None:
        return "Inconclusive"
    if probability >= 95 and improvement > 0:
        return "Success"
    if probability >= 95 and improvement < 0:
        return "Failure"
    return "Inconclusive"


def build_airtable_fields(experiment, best):
    probability = best.get("bayesian_probability")
    improvement = best.get("improvement_rate")
    fields = {
        "Experiment Name": experiment.get("name"),
        "Status": map_status(experiment.get("status")),
        "Start date": to_iso_date(experiment.get("dateStarted")),
        "End date": to_iso_date(experiment.get("dateEnded")),
        "Notes": experiment.get("description"),
        "Actual": improvement,
        "Probability": bayesian_to_probability(probability),
        "Result": derive_result(probability, improvement),
    }
    return {k: v for k, v in fields.items() if v is not None}

Die Automation API veröffentlicht kein festes Enum für das Feld status des Experiments, und die Token können sich weiterentwickeln. map_status vergleicht ohne Berücksichtigung der Groß-/Kleinschreibung und gibt für einen nicht erkannten Status None zurück, wodurch die Zelle Status verworfen wird, statt einen falschen Wert zu schreiben. Bestätigen Sie die Token, die Ihr Konto zurückgibt, mit einer einzelnen GET /experiments/{experimentId}-Anfrage, und erweitern Sie STATUS_MAP bei Bedarf.

7. Den Datensatz per Upsert in Airtable schreiben

Der Airtable-Endpoint Update table (PATCH /v0/meta/bases/{baseId}/tables/{tableId}) ändert nur den Namen und die Beschreibung einer Tabelle; er kann keine Daten in Zeilen schreiben. Verwenden Sie zum Befüllen eines Datensatzes den records-Endpoint mit der Option performUpsert.

Endpoint: Erstellen oder aktualisieren Sie den Datensatz, indem Sie eine PATCH-Anfrage an den Records-Endpoint senden.

PATCH https://api.airtable.com/v0/{baseId}/{tableId}

Feld	Typ	Beschreibung
performUpsert.fieldsToMergeOn	Array	Feld(er), die zum Abgleich mit einem bestehenden Datensatz verwendet werden. Das Skript gleicht anhand von `Experiment Name` ab.
records	Array	Eine Liste von Datensätzen (maximal 10 pro Anfrage), jeweils mit einem `fields`-Objekt.
typecast	Boolean	`true` lässt Airtable Zeichenketten in Auswahloptionen umwandeln und Daten parsen.

Beispiel:

def upsert_record(airtable_token, base_id, table_id, fields):
    resp = requests.patch(
        f"https://api.airtable.com/v0/{base_id}/{table_id}",
        headers={
            "Authorization": f"Bearer {airtable_token}",
            "Content-Type": "application/json",
        },
        json={
            "performUpsert": {"fieldsToMergeOn": ["Experiment Name"]},
            "typecast": True,
            "records": [{"fields": fields}],
        },
    )
    resp.raise_for_status()
    return resp.json()

Antwort (gekürzt):

{
  "records": [
    {
      "id": "rec0Vh8KLmNoPqRsT",
      "fields": {
        "Experiment Name": "Product Page Redesign",
        "Status": "Completed",
        "Start date": "2025-01-15",
        "End date": "2025-02-12",
        "Actual": 211.48,
        "Probability": "80% - High",
        "Result": "Success"
      }
    }
  ],
  "createdRecords": [],
  "updatedRecords": ["rec0Vh8KLmNoPqRsT"]
}

Die Antwort meldet das Ergebnis pro Datensatz: Eine zurückgegebene ID unter createdRecords bedeutet, dass Airtable eine neue Zeile erstellt hat, während eine ID unter updatedRecords bedeutet, dass Airtable eine bestehende Zeile aktualisiert hat.

8. Das Skript ausführen

Übergeben Sie die Experiment-ID sowie die Airtable-Base- und -Table-ID als Argumente:

python kameleoon_to_airtable.py \
  --experiment-id 188308 \
  --base-id appXXXXXXXXXXXXXX \
  --table-id tbll9adSjdedH5t3f

Das Skript gibt jeden Schritt aus: die Authentifizierung, das abgerufene Experiment, die leistungsstärkste Variation, die zugeordneten Felder sowie, ob es den Airtable-Datensatz erstellt oder aktualisiert hat.

Vollständiges Skript

Das vollständige Skript unten entspricht Funktion für Funktion kameleoon_to_airtable.py. Kopieren Sie es direkt, oder laden Sie die Datei über diesen Link herunter.

#!/usr/bin/env python3
"""Export a Kameleoon experiment's results to an Airtable *Experiments* table.

Given a Kameleoon experiment ID, an Airtable base ID, and an Airtable table ID,
this script:

  1. Authenticates with the Automation API (client_credentials grant).
  2. Retrieves the experiment metadata.
  3. Requests the experiment's results (with the Bayesian success probability).
  4. Polls until the report is ready.
  5. Selects the best-performing variation.
  6. Maps the data onto the Airtable *Experiments* schema.
  7. Upserts the record into Airtable, keyed on "Experiment Name".

Credentials are read from environment variables (never hard-code secrets):

    export KAMELEOON_CLIENT_ID="..."
    export KAMELEOON_CLIENT_SECRET="..."
    export AIRTABLE_TOKEN="..."

Usage:

    python kameleoon_to_airtable.py \
        --experiment-id 188308 \
        --base-id appXXXXXXXXXXXXXX \
        --table-id tbll9adSjdedH5t3f
"""

import argparse
import os
import sys
import time

import requests

# Maps the uppercase status tokens returned by the Automation API to the
# options of the Airtable *Status* field. Adjust the target values if your
# *Status* options differ, and confirm the tokens your account returns with a
# single GET /experiments/{experimentId}.
STATUS_MAP = {
    "ACTIVE": "Running",
    "DRAFT": "Implementing",
    "PLANNED": "Implementing",
    "PAUSED": "Defunct",
    "STOPPED": "Completed",
    "DIVERTED": "Completed",
}


# --------------------------------------------------------------------------- #
# 1. Authenticate with the Automation API
# --------------------------------------------------------------------------- #
def kameleoon_token(client_id, client_secret):
    resp = requests.post(
        "https://api.kameleoon.com/oauth/token",
        headers={"Content-Type": "application/x-www-form-urlencoded"},
        data={
            "grant_type": "client_credentials",
            "client_id": client_id,
            "client_secret": client_secret,
        },
    )
    resp.raise_for_status()
    return resp.json()["access_token"]


# --------------------------------------------------------------------------- #
# 2. Retrieve the experiment
# --------------------------------------------------------------------------- #
def get_experiment(token, experiment_id):
    resp = requests.get(
        f"https://api.kameleoon.com/experiments/{experiment_id}",
        headers={"Authorization": f"Bearer {token}"},
    )
    resp.raise_for_status()
    return resp.json()


# --------------------------------------------------------------------------- #
# 3. Request the experiment's results
# --------------------------------------------------------------------------- #
def request_results(token, experiment_id, goal_id):
    body = {
        "visitorData": False,
        "sequentialTesting": True,
        "bayesian": True,
        "referenceVariationId": "0",
        "conversionType": "ALL_CONVERSION",
        "goalsIds": [goal_id] if goal_id else None,
    }
    resp = requests.post(
        f"https://api.kameleoon.com/experiments/{experiment_id}/results",
        headers={
            "Authorization": f"Bearer {token}",
            "Content-Type": "application/json",
            "Accept": "*/*",
        },
        json=body,
    )
    resp.raise_for_status()
    return resp.json()["dataCode"]


# --------------------------------------------------------------------------- #
# 4. Poll for the results
# --------------------------------------------------------------------------- #
def poll_results(token, data_code, max_attempts=30, delay=2.0):
    for _ in range(max_attempts):
        resp = requests.get(
            "https://api.kameleoon.com/results",
            headers={"Authorization": f"Bearer {token}"},
            params={"dataCode": data_code},
        )
        resp.raise_for_status()
        payload = resp.json()
        status = payload.get("status")
        if status == "READY":
            return payload["data"]
        if status in ("ERROR", "TIMEOUT"):
            raise RuntimeError(payload.get("errorDescription") or status)
        time.sleep(delay)  # status == "WAITING"
    raise TimeoutError("Timed out waiting for results.")


# --------------------------------------------------------------------------- #
# 5. Select the best-performing variation
# --------------------------------------------------------------------------- #
def pick_best_variation(result_data, goal_id):
    best, best_improvement = {}, float("-inf")
    for variation_id, vdata in result_data["variationData"].items():
        if variation_id == "_reference":
            continue
        general = vdata["breakdownData"]["_reference"]["generalData"]
        goals_data = general.get("goalsData", {})
        if not goals_data:
            continue
        key = str(goal_id) if str(goal_id) in goals_data else next(iter(goals_data))
        metrics = goals_data[key]
        improvement = metrics.get("improvementRate")
        if improvement is not None and improvement > best_improvement:
            best_improvement = improvement
            best = {
                "variation_id": variation_id,
                "improvement_rate": improvement,
                "bayesian_probability": metrics.get("reliability"),
            }
    return best


# --------------------------------------------------------------------------- #
# 6. Map the data to Airtable fields
# --------------------------------------------------------------------------- #
def map_status(status):
    return STATUS_MAP.get((status or "").upper())


def to_iso_date(value):
    return value[:10] if value else None


def bayesian_to_probability(probability):
    if probability is None:
        return None
    if probability >= 95:
        return "80% - High"
    if probability >= 80:
        return "50% - Medium"
    return "20% - Low"


def derive_result(probability, improvement):
    if probability is None or improvement is None:
        return "Inconclusive"
    if probability >= 95 and improvement > 0:
        return "Success"
    if probability >= 95 and improvement < 0:
        return "Failure"
    return "Inconclusive"


def build_airtable_fields(experiment, best):
    probability = best.get("bayesian_probability")
    improvement = best.get("improvement_rate")
    fields = {
        "Experiment Name": experiment.get("name"),
        "Status": map_status(experiment.get("status")),
        "Start date": to_iso_date(experiment.get("dateStarted")),
        "End date": to_iso_date(experiment.get("dateEnded")),
        "Notes": experiment.get("description"),
        "Actual": improvement,
        "Probability": bayesian_to_probability(probability),
        "Result": derive_result(probability, improvement),
    }
    return {k: v for k, v in fields.items() if v is not None}


# --------------------------------------------------------------------------- #
# 7. Upsert the record into Airtable
# --------------------------------------------------------------------------- #
def upsert_record(airtable_token, base_id, table_id, fields):
    resp = requests.patch(
        f"https://api.airtable.com/v0/{base_id}/{table_id}",
        headers={
            "Authorization": f"Bearer {airtable_token}",
            "Content-Type": "application/json",
        },
        json={
            "performUpsert": {"fieldsToMergeOn": ["Experiment Name"]},
            "typecast": True,
            "records": [{"fields": fields}],
        },
    )
    resp.raise_for_status()
    return resp.json()


# --------------------------------------------------------------------------- #
# 8. Orchestration
# --------------------------------------------------------------------------- #
def run(experiment_id, base_id, table_id):
    client_id = os.environ.get("KAMELEOON_CLIENT_ID")
    client_secret = os.environ.get("KAMELEOON_CLIENT_SECRET")
    airtable_token = os.environ.get("AIRTABLE_TOKEN")

    missing = [
        name
        for name, value in (
            ("KAMELEOON_CLIENT_ID", client_id),
            ("KAMELEOON_CLIENT_SECRET", client_secret),
            ("AIRTABLE_TOKEN", airtable_token),
        )
        if not value
    ]
    if missing:
        raise SystemExit(
            "Missing required environment variable(s): " + ", ".join(missing)
        )

    # 1. Authenticate.
    token = kameleoon_token(client_id, client_secret)
    print("Authenticated with the Automation API.")

    # 2. Retrieve the experiment.
    experiment = get_experiment(token, experiment_id)
    goal_id = experiment.get("mainGoalId")
    print(
        f"Fetched experiment {experiment.get('id')}: "
        f"{experiment.get('name')!r} (status: {experiment.get('status')})."
    )

    # 3-4. Request and poll for the results.
    data_code = request_results(token, experiment_id, goal_id)
    result_data = poll_results(token, data_code)

    # 5. Select the best-performing variation.
    best = pick_best_variation(result_data, goal_id)
    if best:
        print(
            f"Best-performing variation: {best['variation_id']} "
            f"(uplift {best['improvement_rate']}%, "
            f"Bayesian probability {best.get('bayesian_probability')}%)."
        )
    else:
        print("No variation data available for the requested goal.")

    # 6. Map the data to Airtable fields.
    fields = build_airtable_fields(experiment, best)
    print("Mapped Airtable fields:")
    for key, value in fields.items():
        print(f"  {key}: {value}")

    # 7. Upsert the record.
    response = upsert_record(airtable_token, base_id, table_id, fields)
    created = response.get("createdRecords") or []
    record_id = response["records"][0]["id"] if response.get("records") else None
    action = "Created" if record_id in created else "Updated"
    print(f"{action} Airtable record {record_id}.")

    return response


def main():
    parser = argparse.ArgumentParser(
        description="Export a Kameleoon experiment's results to Airtable."
    )
    parser.add_argument(
        "--experiment-id",
        required=True,
        help="ID of the Kameleoon experiment to export.",
    )
    parser.add_argument(
        "--base-id",
        required=True,
        help="Airtable base ID (starts with 'app').",
    )
    parser.add_argument(
        "--table-id",
        required=True,
        help="Airtable table ID (starts with 'tbl') or table name.",
    )
    args = parser.parse_args()

    try:
        run(args.experiment_id, args.base_id, args.table_id)
    except requests.HTTPError as exc:
        detail = ""
        if exc.response is not None:
            detail = f" — {exc.response.status_code}: {exc.response.text}"
        print(f"HTTP error: {exc}{detail}", file=sys.stderr)
        sys.exit(1)
    except (RuntimeError, TimeoutError) as exc:
        print(f"Error: {exc}", file=sys.stderr)
        sys.exit(1)


if __name__ == "__main__":
    main()

Anpassungshinweise

Status-Mapping befindet sich in der Konstante STATUS_MAP, die anhand der von der API zurückgegebenen Status-Token in Großbuchstaben indiziert ist. Passen Sie die Zielwerte an, wenn Ihre Status-Optionen von Running / Implementing / Completed / Defunct abweichen, und bestätigen Sie die Token, die Ihr Konto zurückgibt, mit einer Live-Anfrage GET /experiments/{experimentId}, bevor Sie sich auf das Mapping verlassen.
Probability stammt aus der gemessenen Bayesianischen Erfolgswahrscheinlichkeit, wofür bayesian: true in der Ergebnisanfrage erforderlich ist. Wenn Ihr Feld Probability stattdessen eine manuell eingetragene Schätzung vor dem Experiment ist, entfernen Sie die Zeile Probability aus build_airtable_fields.
Zielauswahl verwendet die mainGoalId des Experiments. Um über ein anderes Ziel zu berichten, übergeben Sie dessen ID an request_results und pick_best_variation.
Upsert-Schlüssel. Airtable gleicht das Merge-Feld exakt ab, sodass Unterschiede bei Groß-/Kleinschreibung oder Leerzeichen in Experiment Name eine neue Zeile erstellen, statt die bestehende zu aktualisieren. Halten Sie Experimentnamen stabil, oder gleichen Sie anhand eines dedizierten, stabilen Kennungsfelds ab.
Format des Felds Actual. Das Skript schreibt den rohen Wert improvementRate (zum Beispiel 211.48) in Actual. Wenn Actual ein Airtable-Percent-Feld ist, stellen Sie es so ein, dass es eine reine Zahl statt eines Bruchwerts erwartet, oder teilen Sie den Wert in build_airtable_fields durch 100, um ihn an ein bruchbasiertes Percent-Feld anzupassen.
Rate Limits. Die Automation API erlaubt bis zu 50 Anfragen pro 10 Sekunden und 1.000 pro Stunde, aber Kameleoon empfiehlt, pro Konto unter 12 Aufrufen pro Minute zu bleiben, und rät von der Verwendung der Automation API für hochfrequentes Tracking ab. Wenn Sie viele Experimente in Batches verarbeiten, cachen Sie Tokens, drosseln Sie Anfragen, und ziehen Sie für Anwendungsfälle mit hohem Volumen die Data API in Betracht.

​Ziel

​Voraussetzungen

​1. Bei der Automation API authentifizieren

​2. Das Experiment abrufen

​3. Die Ergebnisse des Experiments anfordern

​4. Die Ergebnisse abfragen

​5. Die leistungsstärkste Variation auswählen

​6. Die Daten auf Airtable-Felder abbilden

​7. Den Datensatz per Upsert in Airtable schreiben

​8. Das Skript ausführen

​Vollständiges Skript

​Anpassungshinweise

Ziel

Voraussetzungen

1. Bei der Automation API authentifizieren

2. Das Experiment abrufen

3. Die Ergebnisse des Experiments anfordern

4. Die Ergebnisse abfragen

5. Die leistungsstärkste Variation auswählen

6. Die Daten auf Airtable-Felder abbilden

7. Den Datensatz per Upsert in Airtable schreiben

8. Das Skript ausführen

Vollständiges Skript

Anpassungshinweise