Experimentergebnisse nach Confluence exportieren

Ziel

Dieses Tutorial beschreibt Schritt für Schritt, wie das Skript kameleoon_to_confluence.py funktioniert. Anhand einer Kameleoon-Experiment-ID, einer Confluence-Site und einer Confluence-Space-ID ruft das Skript die Experimentmetadaten und die statistischen Ergebnisse ab, ermittelt die leistungsstärkste Variation, bildet die Daten auf das Confluence-Seitenschema ab und schreibt einen Datensatz per Upsert auf eine Confluence-Seite. Ein erneuter Lauf des Skripts für dasselbe Experiment aktualisiert die bestehende Zeile, statt ein Duplikat zu erstellen. Die Schritte 1–5 verwenden denselben Anfrage- und Abfrage-Ablauf wie das Airtable-Export-Tutorial. Nur das Ziel ist unterschiedlich.

Alternative: Export mit einem KI-Assistenten statt mit dem Skript

Das Ausführen des Skripts erfordert eine Python-Umgebung, vier hinterlegte Anmeldedaten und einen manuellen erneuten Lauf für jedes Experiment, das Sie exportieren möchten. Wenn Sie bereits einen MCP-kompatiblen KI-Assistenten wie Claude verwenden, können Sie denselben Export stattdessen aus einer Konversation heraus durchführen, indem Sie ihn mit den Remote-MCP-Servern (Model Context Protocol) von Kameleoon und Atlassian verbinden – entweder über ein Coding-Tool oder direkt in der Claude-App. Folgen Sie unserem Leitfaden, um Ergebnisse mit Claude zu exportieren.

Der Atlassian MCP Server stellt Tools über Jira, Confluence und Bitbucket bereit, einschließlich Schreibzugriff auf Confluence-Seiten. Der Kameleoon MCP Server kann auch Experimente und Feature Flags starten, pausieren, stoppen oder löschen. Überprüfen Sie jede Aktion, die einer der beiden Connectors vorschlägt, bevor Sie sie genehmigen.

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 Confluence-API-Token, gepaart mit der E-Mail-Adresse des Kontos, zu dem es gehört. Erstellen Sie ein Token auf id.atlassian.com/manage-profile/security/api-tokens. Das Skript sendet beide als HTTP-Basic-Authentifizierung auf jeder Confluence-Anfrage.
Die numerische Confluence-Space-ID für den Space, der die Seite enthält. Im Gegensatz zu einer Notion-Datenbank-ID oder einer Airtable-Base-ID erscheint die numerische Space-ID nicht in ihrer URL (diese URL zeigt stattdessen den Space-Schlüssel), und sie wird auch nirgendwo in der Confluence-Benutzeroberfläche angezeigt. Bitten Sie einen MCP-verbundenen KI-Assistenten, sie für Sie nachzuschlagen, oder rufen Sie sie selbst vom Confluence-REST-API-Endpoint Get spaces ab.
Python 3.9+ mit der Bibliothek requests (pip install requests).

Confluence hat kein Äquivalent zu einer Notion-Datenbank oder einer Airtable-Tabelle, die Sie im Voraus erstellen. Das Skript erstellt die Zielseite selbst mit ihrer Tabelle beim ersten Lauf. Bei jedem späteren Lauf findet es diese Seite nach Titel und aktualisiert sie.

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

export KAMELEOON_CLIENT_ID="..."
export KAMELEOON_CLIENT_SECRET="..."
export CONFLUENCE_EMAIL="..."
export CONFLUENCE_API_TOKEN="..."

Das Tutorial verwendet das Beispielexperiment Product Page Redesign (ID 188308) mit zwei Variationen zusätzlich zur Originalversion: Variation 828220 und Variation 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 die Confluence-Zeile 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 anstelle der Bayesianischen Erfolgswahrscheinlichkeit zu verwenden. Aktivieren Sie nur eine der beiden Methoden.
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": False,
        "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 und setzt sequentialTesting: false. bayesian und sequentialTesting sind alternative Methoden zur Berechnung der Signifikanz, und dieses Tutorial gibt die Bayesianische Erfolgswahrscheinlichkeit aus. 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.

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 Variation 828220 zeigt eine Verbesserung von +211,48 % gegenüber -43,33 % bei Variation 828221. Variation 828220 ist daher die leistungsstärkste Variation und, mit einer Wahrscheinlichkeit über 95 % und einer positiven Verbesserung, ein echter Gewinner.

6. Die Daten auf Confluence-Spalten abbilden

Das Skript wandelt die Experimentmetadaten und die Metriken der leistungsstärksten Variation in eine Zeile für die Confluence-Tabelle um und verwendet dabei dieselben acht Spalten wie die Notion- und Airtable-Exporte.

Confluence-Spalte	Quelle	Transformation
Experiment Name	`experiment.name`	Direkt. Wird als Upsert-Schlüssel verwendet.
Status	`experiment.status`	Zugeordnet, unabhängig von Groß-/Kleinschreibung: `active → Running`, `draft`/`planned → Implementing`, `stopped`/`diverted → Completed`, `paused → Defunct`.
Start date	`experiment.dateStarted`	Datums-Uhrzeit gekürzt auf ein ISO-Datum (`YYYY-MM-DD`).
End date	`experiment.dateEnded`	Datums-Uhrzeit gekürzt auf ein ISO-Datum.
Notes	`experiment.description`	Direkt.
Actual	`improvementRate` der leistungsstärksten Variation	Direkt (gemessene Verbesserung, %).
Probability	Bayesianische Erfolgswahrscheinlichkeit der leistungsstärksten 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`.

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_row(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 [fields.get(column) for column in HEADER]

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. Die vorhandene Experiments-Tabelle analysieren

Confluence hat kein Datenbank- oder Tabellenobjekt an sich. Stattdessen verwaltet das Skript eine dedizierte Seite mit einer einzelnen HTML-Tabelle darauf, eine Zeile pro Experiment, und behandelt die Spalte Experiment Name genauso wie Notion eine Titeleigenschaft oder Airtable ein Merge-Feld behandelt: als den Schlüssel, auf den es einen Upsert durchführt. Das Skript erstellt diese Tabelle mit einem full-width-Layout, statt des schmaleren Standardlayouts von Confluence, da acht Spalten mit mittellangen Kopfzeilen bei der Standardseitenbreite umgebrochen werden. Confluences Speicherformat (das HTML-ähnliche Markup, in dem der Inhalt einer Seite gespeichert ist) wickelt den Text jeder Zelle in ein <p>-Tag, und eine neu erstellte Seitenkopfzeile ist einfaches <tr><th>...</th></tr> ohne umgebendes <thead>. Der untenstehende Parser, basierend auf Pythons Standard html.parser.HTMLParser, verarbeitet sowohl diesen bloßen Header-Fall als auch einen von <thead> umhüllten, und behandelt eine leere Zelle gleich, ob Confluence sie als leeres <p></p> oder selbstschließendes <p /> rendert. Er stoppt auch beim ersten </table>, sodass eine zweite Tabelle an anderer Stelle auf der Seite (eine, die Sie von Hand hinzugefügt haben, zum Beispiel) niemals in das analysierte Ergebnis verschmilzt, und schließt jede Zeile oder Zelle, die eine fehlerhafte Bearbeitung offen gelassen hat, ob das nächste Tag eine neue Zeile öffnet oder die Tabelle selbst endet, damit ein verirrtes nicht geschlossenes Tag keine Zeile verlieren kann. Beispiel:

from html.parser import HTMLParser


class _TableParser(HTMLParser):
    """Extrahiert die Kopfzeile und Datenzeilen der ersten <table> in einem
    Confluence-Speicherformat-Dokument. Confluence wickelt Zelltext in <p>-Tags
    und gibt nicht immer <thead> aus; eine Zeile zählt als Header, wenn sie zum
    ersten Mal aus <th>-Zellen besteht. Nur die erste Tabelle wird gelesen,
    daher verschmilzt eine Seite mit mehr als einer Tabelle (zum Beispiel eine,
    die eine Person von Hand hinzugefügt hat) die Zeilen einer zweiten Tabelle
    oder den Header nicht in das Ergebnis."""

    def __init__(self):
        super().__init__()
        self.header = None
        self.rows = []
        self._row = None
        self._cell = None
        self._row_is_header = False
        self._in_table = False
        self._table_done = False

    def _close_cell(self):
        if self._cell is not None:
            self._row.append("".join(self._cell).strip())
            self._cell = None

    def _close_row(self):
        if self._row is not None:
            self._close_cell()
            if self.header is None and self._row_is_header:
                self.header = self._row
            else:
                self.rows.append(self._row)
            self._row = None

    def handle_starttag(self, tag, attrs):
        if self._table_done:
            return
        if tag == "table" and not self._in_table:
            self._in_table = True
        elif tag == "tr" and self._in_table:
            self._close_row()  # eine noch offene Zeile schließen, die ein nicht geschlossenes <tr> hinterlassen hat
            self._row = []
            self._row_is_header = False
        elif tag in ("td", "th") and self._row is not None:
            self._close_cell()  # eine noch offene Zelle schließen, die ein nicht geschlossenes <td>/<th> hinterlassen hat
            self._cell = []
            if tag == "th":
                self._row_is_header = True

    def handle_data(self, data):
        if self._cell is not None:
            self._cell.append(data)

    def handle_endtag(self, tag):
        if self._table_done:
            return
        if tag in ("td", "th"):
            self._close_cell()
        elif tag == "tr":
            self._close_row()
        elif tag == "table":
            self._close_row()  # eine noch offene Zeile/Zelle schließen, die ein nicht geschlossenes finales <tr> hinterlassen hat
            self._in_table = False
            self._table_done = True


def parse_experiments_table(storage_value):
    parser = _TableParser()
    parser.feed(storage_value or "")
    return parser.header or HEADER, parser.rows


def reorder_row(source_header, row):
    """Richtet eine Zeile, die gegen source_header analysiert wurde, auf die
    kanonische HEADER-Reihenfolge aus, sodass eine Tabelle, deren Spalten
    in Confluence manuell neu angeordnet wurden (zum Beispiel durch eine
    Person, die eine Spalte im Editor zieht), die Nachschlageverfolgung nach
    Position nicht falsch ausrichtet. Richtet nur eine echte Neuanordnung
    aus (dieselben acht Beschriftungen, unterschiedliche Reihenfolge); ein
    Header mit einem umbenannten oder nicht erkannten Label fällt auf die
    vorhandene Reihenfolge zurück und gibt eine Warnung aus, statt zu raten."""
    if source_header == HEADER:
        return row
    if sorted(source_header) == sorted(HEADER):
        values = dict(zip(source_header, row))
        return [values.get(column) for column in HEADER]
    print(
        f"Warning: table header {source_header} doesn't match the expected "
        f"columns {HEADER}; keeping existing values by position.",
        file=sys.stderr,
    )
    return row


def upsert_row(source_header, rows, key_column, new_row):
    rows = [reorder_row(source_header, row) for row in rows]
    key_index = HEADER.index(key_column)
    key = new_row[key_index]
    for i, row in enumerate(rows):
        if row[key_index] == key:
            rows[i] = new_row
            return rows
    rows.append(new_row)
    return rows

reorder_row schützt vor einer Tabelle, deren Spalten in Confluence manuell neu angeordnet wurden, zum Beispiel wenn eine Person eine Spalte im Editor zieht. Ohne es würde eine Nachschlageverfolgung nach Position die Spalte Experiment Name der neuen Zeile gegen alles vergleichen, was jetzt an dieser Position in der vorhandenen Tabelle sitzt, wodurch die falsche Zeile stillschweigend abgeglichen oder gar keine wird. Sie richtet nur eine echte Neuanordnung aus, wenn der Header immer noch dieselben acht Beschriftungen in einer anderen Reihenfolge hat. Wenn der Text einer Kopfzelle selbst bearbeitet wurde, zum Beispiel wenn Notes in Comments umbenannt wurde, stimmen die Beschriftungen nicht länger mit HEADER überein, also fällt die Funktion auf die vorhandene Spaltenreihenfolge zurück und gibt eine Warnung aus, statt zu raten, statt stumm diese Spalte Daten auf jeder Zeile zu löschen.

8. Die Confluence-Seite finden, erstellen oder aktualisieren

Suchen: Suchen Sie die Seite nach Titel innerhalb des Spaces unter Verwendung der Query-Parameter title und space-id auf dem Endpoint Get pages. Das Übergeben von body-format=storage gibt den aktuellen Tabelleninhalt im selben Aufruf zurück.

GET https://{site}/wiki/api/v2/pages

Erstellen: Wenn keine Seite übereinstimmt, erstellen Sie eine mit dem Endpoint Create page, wobei die Tabelle bereits aus einer einzelnen Zeile erstellt ist.

POST https://{site}/wiki/api/v2/pages

Aktualisieren: Wenn eine Seite übereinstimmt, erstellen Sie die gesamte Tabelle aus ihren vorhandenen Zeilen plus der upgeserteten neu auf und überschreiben Sie die Seite mit dem Endpoint Update page.

PUT https://{site}/wiki/api/v2/pages/{id}

Confluence hat keinen Schreibvorgang pro Zeile oder pro Feld. Das Aktualisieren einer Seite ersetzt ihren gesamten Body, daher liest das Skript immer die aktuelle Tabelle, upsert eine Zeile im Speicher und schreibt die gesamte Tabelle zurück. Diese ganze Body-Ersetzung unterscheidet sich vom Airtable-Export, der einen einzelnen Datensatz patcht, und vom Notion-Export, der die Eigenschaften einer einzelnen Seite patcht.

Im Gegensatz zu den Notion- und Airtable-APIs erfordert Confluence vom Aufrufer, version.number bei jedem Update auf die aktuelle Versionsnummer plus eins zu inkrementieren. Das erneute Senden der aktuellen Nummer oder das Auslassen von version lässt die Anfrage fehlschlagen. Ein MCP-verbundener KI-Assistent mit Atlassian-eigenen Tools verarbeitet dies automatisch; ein direkter REST-Aufruf, wie in diesem Skript, nicht.

Beispiel:

def find_page(base_url, auth, space_id, title):
    resp = requests.get(
        f"{base_url}/pages",
        auth=auth,
        params={"space-id": space_id, "title": title, "body-format": "storage"},
    )
    resp.raise_for_status()
    results = resp.json().get("results", [])
    return results[0] if results else None


def create_page(base_url, auth, space_id, title, storage_value):
    resp = requests.post(
        f"{base_url}/pages",
        auth=auth,
        headers={"Content-Type": "application/json"},
        json={
            "spaceId": space_id,
            "status": "current",
            "title": title,
            "body": {"representation": "storage", "value": storage_value},
        },
    )
    resp.raise_for_status()
    return resp.json()


def update_page(base_url, auth, page, storage_value):
    resp = requests.put(
        f"{base_url}/pages/{page['id']}",
        auth=auth,
        headers={"Content-Type": "application/json"},
        json={
            "id": page["id"],
            "status": "current",
            "title": page["title"],
            "version": {"number": page["version"]["number"] + 1},
            "body": {"representation": "storage", "value": storage_value},
        },
    )
    resp.raise_for_status()
    return resp.json()


def upsert_page(base_url, auth, space_id, title, new_row):
    page = find_page(base_url, auth, space_id, title)
    if page is None:
        storage_value = build_table_storage([new_row])
        return create_page(base_url, auth, space_id, title, storage_value), "Created"

    existing_header, rows = parse_experiments_table(page["body"]["storage"]["value"])
    rows = upsert_row(existing_header, rows, "Experiment Name", new_row)
    storage_value = build_table_storage(rows)
    return update_page(base_url, auth, page, storage_value), "Updated"

Antwort (gekürzt):

{
  "id": "557057",
  "status": "current",
  "title": "Experiments",
  "spaceId": "327682",
  "version": { "number": 3 },
  "body": {
    "storage": {
      "value": "<table data-layout=\"full-width\"><tbody><tr><th><p>Experiment Name</p></th>...</tr><tr><td><p>Product Page Redesign</p></td><td><p>Completed</p></td>...</tr></tbody></table>"
    }
  }
}

9. Das Skript ausführen

Übergeben Sie die Experiment-ID, die Confluence-Site und die Space-ID als Argumente:

python kameleoon_to_confluence.py \
  --experiment-id 188308 \
  --site your-domain.atlassian.net \
  --space-id 327682

Das Skript gibt jeden Schritt aus: Authentifizierung, das abgerufene Experiment, die leistungsstärkste Variation, die zugeordnete Zeile sowie, ob es die Confluence-Seite erstellt oder aktualisiert hat. Übergeben Sie --page-title, um eine Seitennamen andere als den Standard Experiments anzugeben.

Vollständiges Skript

Das vollständige Skript unten entspricht Funktion für Funktion kameleoon_to_confluence.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 a Confluence page.

Given a Kameleoon experiment ID and a Confluence site, space ID, and page
title, 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 Confluence's Experiments table columns.
  7. Finds the Confluence page by title, or creates it if it doesn't exist.
  8. Parses the page's existing table and upserts a row, keyed on
     "Experiment Name".
  9. Writes the updated table back to Confluence.

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

    export KAMELEOON_CLIENT_ID="..."
    export KAMELEOON_CLIENT_SECRET="..."
    export CONFLUENCE_EMAIL="..."
    export CONFLUENCE_API_TOKEN="..."

Usage:

    python kameleoon_to_confluence.py \
        --experiment-id 188308 \
        --site your-domain.atlassian.net \
        --space-id 327682 \
        --page-title Experiments
"""

import argparse
import html
import os
import sys
import time
from html.parser import HTMLParser

import requests
from requests.auth import HTTPBasicAuth

# Bildet die von der Automation API zurückgegebenen Status-Token in
# Großbuchstaben auf die Werte der Status-Spalte ab, die nach Confluence
# geschrieben werden. Passen Sie die Zielwerte an, wenn Ihre Status-Beschriftungen
# von Running/Implementing/Completed/Defunct abweichen, und bestätigen Sie die
# Token, die Ihr Konto zurückgibt, mit einer einzelnen GET /experiments/{experimentId}-Anfrage.
STATUS_MAP = {
    "ACTIVE": "Running",
    "DRAFT": "Implementing",
    "PLANNED": "Implementing",
    "PAUSED": "Defunct",
    "STOPPED": "Completed",
    "DIVERTED": "Completed",
}

# Spaltenreihenfolge für die Experiments-Tabelle. Das Skript richtet Spalten
# positionell gegen diesen Header aus, sowohl wenn es eine neue Seite erstellt
# als auch wenn es eine vorhandene analysiert.
HEADER = [
    "Experiment Name",
    "Status",
    "Start date",
    "End date",
    "Notes",
    "Actual",
    "Probability",
    "Result",
]

TABLE_INTRO = (
    "<p>Kameleoon experiment results, kept current by "
    "kameleoon_to_confluence.py. Each row is upserted by matching the "
    "<strong>Experiment Name</strong> column; re-running the script for the "
    "same experiment updates its row instead of adding a duplicate.</p>"
)


# --------------------------------------------------------------------------- #
# 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": False,
        "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 Confluence columns
# --------------------------------------------------------------------------- #
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_row(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 [fields.get(column) for column in HEADER]


# --------------------------------------------------------------------------- #
# 7. Parse the existing Experiments table
# --------------------------------------------------------------------------- #
class _TableParser(HTMLParser):
    """Extrahiert die Kopfzeile und Datenzeilen der ersten <table> in einem
    Confluence-Speicherformat-Dokument. Confluence wickelt Zelltext in <p>-Tags
    und gibt nicht immer <thead> aus; eine Zeile zählt als Header, wenn sie zum
    ersten Mal aus <th>-Zellen besteht. Nur die erste Tabelle wird gelesen,
    daher verschmilzt eine Seite mit mehr als einer Tabelle (zum Beispiel eine,
    die eine Person von Hand hinzugefügt hat) die Zeilen einer zweiten Tabelle
    oder den Header nicht in das Ergebnis."""

    def __init__(self):
        super().__init__()
        self.header = None
        self.rows = []
        self._row = None
        self._cell = None
        self._row_is_header = False
        self._in_table = False
        self._table_done = False

    def _close_cell(self):
        if self._cell is not None:
            self._row.append("".join(self._cell).strip())
            self._cell = None

    def _close_row(self):
        if self._row is not None:
            self._close_cell()
            if self.header is None and self._row_is_header:
                self.header = self._row
            else:
                self.rows.append(self._row)
            self._row = None

    def handle_starttag(self, tag, attrs):
        if self._table_done:
            return
        if tag == "table" and not self._in_table:
            self._in_table = True
        elif tag == "tr" and self._in_table:
            self._close_row()  # eine noch offene Zeile schließen, die ein nicht geschlossenes <tr> hinterlassen hat
            self._row = []
            self._row_is_header = False
        elif tag in ("td", "th") and self._row is not None:
            self._close_cell()  # eine noch offene Zelle schließen, die ein nicht geschlossenes <td>/<th> hinterlassen hat
            self._cell = []
            if tag == "th":
                self._row_is_header = True

    def handle_data(self, data):
        if self._cell is not None:
            self._cell.append(data)

    def handle_endtag(self, tag):
        if self._table_done:
            return
        if tag in ("td", "th"):
            self._close_cell()
        elif tag == "tr":
            self._close_row()
        elif tag == "table":
            self._close_row()  # eine noch offene Zeile/Zelle schließen, die ein nicht geschlossenes finales <tr> hinterlassen hat
            self._in_table = False
            self._table_done = True


def parse_experiments_table(storage_value):
    parser = _TableParser()
    parser.feed(storage_value or "")
    return parser.header or HEADER, parser.rows


def reorder_row(source_header, row):
    """Richtet eine Zeile, die gegen source_header analysiert wurde, auf die
    kanonische HEADER-Reihenfolge aus, sodass eine Tabelle, deren Spalten
    in Confluence manuell neu angeordnet wurden (zum Beispiel durch eine
    Person, die eine Spalte im Editor zieht), die Nachschlageverfolgung nach
    Position nicht falsch ausrichtet. Richtet nur eine echte Neuanordnung
    aus (dieselben acht Beschriftungen, unterschiedliche Reihenfolge); ein
    Header mit einem umbenannten oder nicht erkannten Label fällt auf die
    vorhandene Reihenfolge zurück und gibt eine Warnung aus, statt zu raten."""
    if source_header == HEADER:
        return row
    if sorted(source_header) == sorted(HEADER):
        values = dict(zip(source_header, row))
        return [values.get(column) for column in HEADER]
    print(
        f"Warning: table header {source_header} doesn't match the expected "
        f"columns {HEADER}; keeping existing values by position.",
        file=sys.stderr,
    )
    return row


def upsert_row(source_header, rows, key_column, new_row):
    rows = [reorder_row(source_header, row) for row in rows]
    key_index = HEADER.index(key_column)
    key = new_row[key_index]
    for i, row in enumerate(rows):
        if row[key_index] == key:
            rows[i] = new_row
            return rows
    rows.append(new_row)
    return rows


def render_cell(value):
    text = "" if value in (None, "") else html.escape(str(value))
    return f"<p>{text}</p>"


def build_table_storage(rows):
    head_cells = "".join(f"<th>{render_cell(c)}</th>" for c in HEADER)
    body_rows = "".join(
        "<tr>" + "".join(f"<td>{render_cell(v)}</td>" for v in row) + "</tr>"
        for row in rows
    )
    return f'{TABLE_INTRO}<table data-layout="full-width"><tbody><tr>{head_cells}</tr>{body_rows}</tbody></table>'


# --------------------------------------------------------------------------- #
# 8. Find, create, or update the Confluence page
# --------------------------------------------------------------------------- #
def find_page(base_url, auth, space_id, title):
    resp = requests.get(
        f"{base_url}/pages",
        auth=auth,
        params={"space-id": space_id, "title": title, "body-format": "storage"},
    )
    resp.raise_for_status()
    results = resp.json().get("results", [])
    return results[0] if results else None


def create_page(base_url, auth, space_id, title, storage_value):
    resp = requests.post(
        f"{base_url}/pages",
        auth=auth,
        headers={"Content-Type": "application/json"},
        json={
            "spaceId": space_id,
            "status": "current",
            "title": title,
            "body": {"representation": "storage", "value": storage_value},
        },
    )
    resp.raise_for_status()
    return resp.json()


def update_page(base_url, auth, page, storage_value):
    resp = requests.put(
        f"{base_url}/pages/{page['id']}",
        auth=auth,
        headers={"Content-Type": "application/json"},
        json={
            "id": page["id"],
            "status": "current",
            "title": page["title"],
            "version": {"number": page["version"]["number"] + 1},
            "body": {"representation": "storage", "value": storage_value},
        },
    )
    resp.raise_for_status()
    return resp.json()


def upsert_page(base_url, auth, space_id, title, new_row):
    page = find_page(base_url, auth, space_id, title)
    if page is None:
        storage_value = build_table_storage([new_row])
        return create_page(base_url, auth, space_id, title, storage_value), "Created"

    existing_header, rows = parse_experiments_table(page["body"]["storage"]["value"])
    rows = upsert_row(existing_header, rows, "Experiment Name", new_row)
    storage_value = build_table_storage(rows)
    return update_page(base_url, auth, page, storage_value), "Updated"


# --------------------------------------------------------------------------- #
# 9. Orchestration
# --------------------------------------------------------------------------- #
def run(experiment_id, site, space_id, page_title):
    client_id = os.environ.get("KAMELEOON_CLIENT_ID")
    client_secret = os.environ.get("KAMELEOON_CLIENT_SECRET")
    confluence_email = os.environ.get("CONFLUENCE_EMAIL")
    confluence_token = os.environ.get("CONFLUENCE_API_TOKEN")

    missing = [
        name
        for name, value in (
            ("KAMELEOON_CLIENT_ID", client_id),
            ("KAMELEOON_CLIENT_SECRET", client_secret),
            ("CONFLUENCE_EMAIL", confluence_email),
            ("CONFLUENCE_API_TOKEN", confluence_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 a table row.
    row = build_row(experiment, best)
    print("Mapped row:")
    for column, value in zip(HEADER, row):
        print(f"  {column}: {value}")

    # 7-8. Find, create, or update the Confluence page.
    base_url = f"https://{site}/wiki/api/v2"
    auth = HTTPBasicAuth(confluence_email, confluence_token)
    page, action = upsert_page(base_url, auth, space_id, page_title, row)
    print(f"{action} Confluence page {page['id']} ({page_title!r}).")

    return page


def main():
    parser = argparse.ArgumentParser(
        description="Export a Kameleoon experiment's results to Confluence."
    )
    parser.add_argument(
        "--experiment-id",
        required=True,
        help="ID of the Kameleoon experiment to export.",
    )
    parser.add_argument(
        "--site",
        required=True,
        help="Confluence site hostname (for example, your-domain.atlassian.net).",
    )
    parser.add_argument(
        "--space-id",
        required=True,
        help="Numeric Confluence space ID that hosts the Experiments page.",
    )
    parser.add_argument(
        "--page-title",
        default="Experiments",
        help="Title of the page to create or update (default: Experiments).",
    )
    args = parser.parse_args()

    try:
        run(args.experiment_id, args.site, args.space_id, args.page_title)
    except requests.exceptions.RequestException 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-Beschriftungen 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 Sie eine Vor-Experiment-Schätzung statt einer gemessenen bevorzugen, entfernen Sie die Zeile Probability aus build_row.
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. Der Abgleich nach Experiment Name ist exakt, daher erstellen Unterschiede bei Groß-/Kleinschreibung oder Leerzeichen eine neue Zeile, statt die bestehende zu aktualisieren. Weil der Find-then-Write-Ablauf nicht atomar ist, vermeiden Sie das gleichzeitige Ausführen zweier Exporte für dasselbe Experiment, und vermeiden Sie das Umbenennen eines Experiments zwischen Läufen, wenn Sie nicht auch eine neue Zeile dafür möchten.
Ganz-Seiten-Updates. Jede Aktualisierung schreibt die gesamte Tabelle neu, da Confluence keinen Schreibvorgang pro Zeile hat. Wenn Sie die Seite zwischen Skriptläufen von Hand verwalten, halten Sie Ihre Bearbeitungen innerhalb der Tabelle, die das Skript erstellt. Das Skript bewahrt derzeit Inhalte außerhalb dieser Tabelle nicht.
Versionskonflikte. Das Skript liest immer die aktuelle version.number der Seite unmittelbar vor dem Schreiben, daher lässt eine zwischen dem Lesen und dem Schreiben vorgenommene manuelle Bearbeitung die nächste PUT mit einem Versionskonflikt fehlschlagen. Führen Sie das Skript erneut aus, wenn dies geschieht.
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. Confluence Cloud erzwingt seine eigenen Rate Limits, die je nach Plan variieren; siehe Atlassian-Dokumentation zu Rate Limiting für aktuelle Werte. 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

​Alternative: Export mit einem KI-Assistenten statt mit dem Skript

​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 Confluence-Spalten abbilden

​7. Die vorhandene Experiments-Tabelle analysieren

​8. Die Confluence-Seite finden, erstellen oder aktualisieren

​9. Das Skript ausführen

​Vollständiges Skript

​Anpassungshinweise

Ziel

Alternative: Export mit einem KI-Assistenten statt mit dem Skript

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 Confluence-Spalten abbilden

7. Die vorhandene Experiments-Tabelle analysieren

8. Die Confluence-Seite finden, erstellen oder aktualisieren

9. Das Skript ausführen

Vollständiges Skript

Anpassungshinweise