Exportar los resultados de experimentos a Confluence

Objetivo

Este tutorial describe cómo funciona el script kameleoon_to_confluence.py, paso a paso. Dado un ID de experimento de Kameleoon, un sitio de Confluence y un ID de espacio de Confluence, el script recupera los metadatos y los resultados estadísticos del experimento, identifica la variación con mejor rendimiento, e inserta o actualiza una fila para ese experimento en una tabla en una página de Confluence. Volver a ejecutar el script para el mismo experimento actualiza su fila existente en lugar de añadir un duplicado. Los pasos 1–5 reutilizan el mismo flujo de solicitud y consulta que el tutorial de exportación a Airtable. Solo cambia el destino.

Alternativa: exportar con un asistente de IA en lugar del script

Ejecutar el script requiere un entorno de Python, cuatro credenciales almacenadas, y una ejecución manual por cada experimento que quiera exportar. Si ya utiliza un asistente de IA compatible con MCP como Claude, puede realizar la misma exportación desde una conversación, conectándolo a los servidores MCP (Model Context Protocol) remotos propios de Kameleoon y de Atlassian, ya sea a través de una herramienta de código o directamente en la aplicación Claude. Siga la guía para exportar los resultados usando Claude.

El servidor MCP de Atlassian expone herramientas a través de Jira, Confluence y Bitbucket, incluyendo acceso de escritura a páginas de Confluence. El servidor MCP de Kameleoon también puede iniciar, pausar, detener o eliminar experimentos y feature flags. Revise cualquier acción que proponga cualquiera de los conectores antes de aprobarla.

Requisitos

Credenciales de la API de Kameleoon. La Automation API requiere un token de acceso. El script obtiene uno de forma programática a partir de un client_id y un client_secret mediante el grant client_credentials. Consulte Obtener un token de acceso.
Un token de API de Confluence, emparejado con la dirección de correo electrónico de la cuenta a la que pertenece. Cree un token en id.atlassian.com/manage-profile/security/api-tokens. El script envía ambos como autenticación HTTP básica en cada solicitud a Confluence.
El ID de espacio numérico de Confluence para el espacio que aloja la página. A diferencia de un ID de base de datos de Notion o un ID de base de Airtable, el ID numérico de un espacio no aparece en su URL (esa URL muestra la clave del espacio en su lugar), y tampoco se muestra en ningún lugar de la interfaz de usuario web de Confluence. Pida a un asistente de IA conectado por MCP que lo busque por usted, o consígalo usted mismo del endpoint Get spaces de la API REST de Confluence.
Python 3.9+ con la biblioteca requests (pip install requests).

Confluence no tiene equivalente a una base de datos de Notion o una tabla de Airtable que construya por adelantado. El script crea la página de destino a sí mismo, con su tabla, la primera vez que se ejecuta. En cada ejecución posterior, encuentra esa página por título y la actualiza.

Almacene todas las credenciales en variables de entorno. Nunca codifique secretos directamente en el script.

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

El tutorial utiliza el experimento de ejemplo Product Page Redesign (ID 188308), con dos variaciones además de la original: variación 828220 y variación 828221.

1. Autenticarse con la Automation API

Endpoint: obtenga un token de acceso enviando una solicitud POST al endpoint de token.

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

Campo	Tipo	Descripción
grant_type	String	Establézcalo en `client_credentials`.
client_id	String	Su ID de cliente de la Automation API.
client_secret	String	Su secreto de cliente de la Automation API.

Ejemplo:

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"]

Respuesta:

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

Envíe el access_token devuelto como token Bearer en cada solicitud posterior a la Automation API. Los tokens de acceso son válidos durante 2 horas de forma predeterminada.

2. Recuperar el experimento

Endpoint: obtenga los metadatos del experimento enviando una solicitud GET al endpoint Get an experiment.

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

Campo	Tipo	Descripción
experimentId	Integer	Parámetro de ruta obligatorio. El ID del experimento que se va a recuperar.

Ejemplo:

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()

Respuesta (truncada):

{
  "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]
}

El script lee name, status, dateStarted, dateEnded y description para la fila de Confluence, y mainGoalId para acotar la solicitud de resultados en el siguiente paso.

La API devuelve mainGoalId de forma predeterminada, por lo que no necesita un parámetro optionalFields para leerlo. La Automation API no publica una enumeración fija para el campo status, pero otros campos de tipo estado en toda la API utilizan sistemáticamente tokens en mayúsculas (por ejemplo, STOPPED, ACTIVE, DRAFT). El paso 6 asigna el campo status partiendo de ese supuesto. Confirme los tokens exactos que devuelve su cuenta con una solicitud real antes de confiar en la asignación.

3. Solicitar los resultados del experimento

Endpoint: active la generación del informe de resultados enviando una solicitud POST al endpoint Request experiment’s results.

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

Campo	Tipo	Descripción
experimentId	Integer	Parámetro de ruta obligatorio.
goalsIds	Array	Restringe el informe a los IDs de objetivo indicados. El script pasa el `mainGoalId` del experimento.
referenceVariationId	String	La variación utilizada como referencia para la comparación. `"0"` utiliza la página original.
visitorData	Boolean	`false` para datos basados en visitas, `true` para datos basados en visitantes.
sequentialTesting	Boolean	Establézcalo en `true` para usar pruebas secuenciales en los intervalos de confianza en lugar de la probabilidad de éxito bayesiana. Active un método u otro, no ambos.
bayesian	Boolean	Establézcalo en `true` para incluir la probabilidad de éxito bayesiana en el informe. El script lee este valor para completar el campo Probability.
conversionType	String	`ALL_CONVERSION` o `CONVERTED_VISITS`.

Ejemplo:

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"]

Respuesta:

{ "dataCode": "14443931880924098266207585267983330260134079899081739889989435434342588016765" }

Kameleoon genera el informe de forma asíncrona. El endpoint devuelve un dataCode, que paso 4 utiliza para consultar el resultado.

Este script requiere bayesian: true y establece sequentialTesting: false. bayesian y sequentialTesting son métodos alternativos para calcular la significancia, y este tutorial reporta la probabilidad de éxito bayesiana. Con Bayesian habilitado, el valor reliability del informe representa la probabilidad de éxito bayesiana (la probabilidad de que una variación supere a la referencia), que el script asigna al campo Probability. Confirme el valor con el mismo informe en la aplicación Kameleoon si su cuenta utiliza un método estadístico predeterminado diferente.

4. Consultar los resultados

Endpoint: recupere el informe enviando solicitudes GET al endpoint Poll results hasta que esté listo.

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

Campo	Tipo	Descripción
dataCode	String	Parámetro de consulta obligatorio. Use el hash que devolvió `POST /experiments/{experimentId}/results` en paso 3.

El status de la respuesta es WAITING mientras Kameleoon calcula el informe, READY cuando los datos están disponibles, o ERROR / TIMEOUT en caso de fallo. Cuando el estado es ERROR o TIMEOUT, la respuesta incluye un errorDescription de nivel superior. El script consulta a intervalos fijos hasta que el estado es READY. Ejemplo:

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.")

Respuesta (truncada):

{
  "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. Seleccionar la variación con mejor rendimiento

Los resultados contienen una entrada por variación bajo variationData, además de la línea _reference para la página original. Para cada variación, las métricas del objetivo solicitado se encuentran bajo breakdownData._reference.generalData.goalsData[goalId]. El script omite la entrada _reference, lee el improvementRate y el reliability (la probabilidad de éxito bayesiana) de cada variación, y selecciona como mejor variación la que tiene la tasa de mejora más alta. El campo Result, asignado en el siguiente paso, registra si esa variación realmente ganó: si alcanzó una probabilidad de éxito suficientemente alta con una mejora positiva. Ejemplo:

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

En este ejemplo, ambas variaciones alcanzan una probabilidad de éxito bayesiana del 100%, pero la variación 828220 muestra una mejora del +211,48% frente al -43,33% de la variación 828221. Variación 828220 tiene por lo tanto la tasa de mejora más alta y, con una probabilidad superior al 95% y una mejora positiva, es la ganadora genuina.

6. Asignar los datos a las columnas de Confluence

El script transforma los metadatos del experimento y las métricas de la variación con mejor rendimiento en una fila para la tabla de Confluence, utilizando las mismas ocho columnas que las exportaciones a Notion y Airtable.

Columna de Confluence	Origen	Transformación
Experiment Name	`experiment.name`	Directo. Se utiliza como clave del upsert.
Status	`experiment.status`	Asignado sin distinguir mayúsculas y minúsculas: `active → Running`, `draft`/`planned → Implementing`, `stopped`/`diverted → Completed`, `paused → Defunct`.
Start date	`experiment.dateStarted`	Fecha y hora truncadas a una fecha ISO (`YYYY-MM-DD`).
End date	`experiment.dateEnded`	Fecha y hora truncadas a una fecha ISO.
Notes	`experiment.description`	Directo.
Actual	`improvementRate` de la variación ganadora	Directo (mejora medida, %).
Probability	probabilidad de éxito bayesiana de la variación ganadora	Agrupada por rangos: `≥95 → 80% - High`, `≥80 → 50% - Medium`, en otro caso `20% - Low`.
Result	probabilidad de éxito bayesiana + `improvementRate`	probabilidad `≥ 95` y mejora > 0 → `Success`; `≥ 95` y mejora < 0 → `Failure`; en cualquier otro caso → `Inconclusive`.

Ejemplo:

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]

La Automation API no publica una enumeración fija para el campo status del experimento, y los tokens pueden evolucionar. map_status compara sin distinguir mayúsculas y minúsculas y devuelve None para un estado no reconocido, lo que omite la celda Status en lugar de escribir un valor incorrecto. Confirme los tokens que devuelve su cuenta con una única solicitud GET /experiments/{experimentId} y amplíe STATUS_MAP si es necesario.

7. Analizar la tabla de Experimentos existente

Confluence no tiene un objeto de base de datos o tabla propio. En su lugar, el script mantiene una página dedicada con una única tabla HTML, una fila por experimento, y trata la columna Experiment Name de la misma manera que Notion trata una propiedad de título o Airtable trata un campo de fusión: como la clave en la que hace un upsert. El script construye esa tabla con una disposición full-width en lugar de la predeterminada más estrecha de Confluence, ya que ocho columnas de encabezados moderadamente largos se envuelven en mitad de palabra en el ancho de página predeterminado. El formato de almacenamiento de Confluence (el marcado similar a HTML en el que se almacena el cuerpo de una página) envuelve el texto de cada celda en una etiqueta <p>, y la fila de encabezado de una página recién creada son celdas <tr><th>...</th></tr> planas sin un <thead> circundante. El analizador de abajo, construido en el html.parser.HTMLParser estándar de Python, maneja tanto ese caso de encabezado plano como uno envuelto en <thead>, y trata una celda vacía igual si Confluence la renderiza como <p></p> vacío o como <p /> auto-cerrado. También se detiene en el primer </table>, así que una segunda tabla en otro lugar de la página (una que haya añadido manualmente, por ejemplo) nunca se fusiona con el resultado analizado, y cierra cualquier fila o celda que una edición mal formada dejó abierta, independientemente de si la siguiente etiqueta abre una fila nueva o la tabla misma termina, por lo que una etiqueta sin cerrar extraviada no puede eliminar silenciosamente una fila. Ejemplo:

from html.parser import HTMLParser


class _TableParser(HTMLParser):
    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()  # close a still-open row left by an unclosed <tr>
            self._row = []
            self._row_is_header = False
        elif tag in ("td", "th") and self._row is not None:
            self._close_cell()  # close a still-open cell left by an unclosed <td>/<th>
            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()  # close a still-open row/cell left by an unclosed final <tr>
            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):
    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 protege contra una tabla cuyas columnas fueron reordenadas manualmente en Confluence, por ejemplo por una persona arrastrando una columna en el editor. Sin ella, una búsqueda por posición compararía el Experiment Name de la fila nueva contra cualquier columna que ahora esté en esa posición en la tabla existente, coincidiendo silenciosamente con la fila equivocada o ninguna en absoluto. Solo realinea un verdadero reordenamiento, donde el encabezado aún tiene las mismas ocho etiquetas en un orden diferente. Si el texto de una celda de encabezado fue editado en sí mismo, por ejemplo retitulando Notes a Comments, las etiquetas ya no coinciden HEADER en absoluto, así que la función cae de vuelta al orden de columna existente e imprime una advertencia en lugar de adivinar, en lugar de silenciosamente blanquear los datos de esa columna en cada fila.

8. Buscar, crear o actualizar la página de Confluence

Buscar: busque la página por título dentro del espacio utilizando los parámetros de consulta title y space-id en el endpoint Get pages. Pasando body-format=storage devuelve el contenido de la tabla actual en la misma llamada.

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

Crear: si ninguna página coincide, cree una con el endpoint Create page, con la tabla ya construida a partir de una única fila.

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

Actualizar: si una página coincide, reconstruya la tabla completa a partir de sus filas existentes más la insertada o actualizada, y sobrescriba la página con el endpoint Update page.

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

Confluence no tiene un endpoint de actualización por fila o por campo. La actualización de una página reemplaza su cuerpo completo, así que el script siempre lee la tabla actual, inserta o actualiza una fila en memoria, y escribe la tabla completa de vuelta. Ese reemplazo de cuerpo completo difiere de la exportación a Airtable, que parchea un único registro, y de la exportación a Notion, que parchea las propiedades de una única página.

A diferencia de las APIs de Notion y Airtable, Confluence requiere que la persona que hace la llamada incremente version.number en cada actualización, al número de versión actual más uno. Enviar el número actual de nuevo, u omitir version, hace que la solicitud falle. Un asistente de IA conectado por MCP usando las herramientas propias de Atlassian maneja esto automáticamente; una llamada REST directa, como en este script, no.

Ejemplo:

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"

Respuesta (truncada):

{
  "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. Ejecutar el script

Pase el ID del experimento, el sitio de Confluence y el ID del espacio como argumentos:

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

El script imprime cada paso: la autenticación, el experimento obtenido, la variación con mejor rendimiento, la fila asignada, y si creó o actualizó la página de Confluence. Pase --page-title para dirigirse a un nombre de página distinto del predeterminado Experiments.

Script completo

El script completo de abajo coincide función por función con kameleoon_to_confluence.py. Cópielo directamente, o descargue el archivo desde ese enlace.

#!/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

# Maps the uppercase status tokens returned by the Automation API to the
# Status column values written to Confluence. Adjust the target values if
# you want different labels, 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",
}

# Column order for the Experiments table. The script keys columns
# positionally against this header, both when it builds a new page and when
# it parses an existing one.
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):
    """Extrae la fila de encabezado y las filas de datos de la primera <table>
    en un documento en formato de almacenamiento de Confluence. Confluence
    envuelve el texto de las celdas en etiquetas <p> y no siempre emite
    <thead>; una fila cuenta como encabezado la primera vez que está
    compuesta de celdas <th>. Solo se lee la primera tabla, así que una
    página con más de una tabla (por ejemplo, una que una persona añadió a
    mano) no fusiona las filas ni el encabezado de una segunda tabla en el
    resultado."""

    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()  # close a still-open row left by an unclosed <tr>
            self._row = []
            self._row_is_header = False
        elif tag in ("td", "th") and self._row is not None:
            self._close_cell()  # close a still-open cell left by an unclosed <td>/<th>
            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()  # close a still-open row/cell left by an unclosed final <tr>
            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):
    """Realinea una fila analizada contra source_header al orden canónico de
    HEADER, de modo que una tabla cuyas columnas fueron reordenadas
    manualmente en Confluence no desalinee una búsqueda por posición de
    columna como la clave Experiment Name. Solo realinea un verdadero
    reordenamiento (las mismas ocho etiquetas, en un orden diferente); un
    encabezado con una columna renombrada o no reconocida vuelve al orden
    existente y advierte en lugar de adivinar."""
    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()

Notas de personalización

La asignación de estado vive en la constante STATUS_MAP, indexada por los tokens de estado en mayúsculas que devuelve la API. Ajuste los valores de destino si sus opciones de Status difieren de Running / Implementing / Completed / Defunct, y confirme los tokens que devuelve su cuenta con una solicitud real GET /experiments/{experimentId} antes de confiar en la asignación.
Probability procede de la probabilidad de éxito bayesiana medida, que requiere bayesian: true en la solicitud de resultados. Si desea una estimación previa al experimento en su lugar, elimine la línea Probability de build_row.
La selección de objetivo usa el mainGoalId del experimento. Para generar el informe sobre un objetivo diferente, pase su ID a request_results y pick_best_variation.
Clave de upsert. El coincidencia en Experiment Name es exacta, por lo que las diferencias de mayúsculas/minúsculas o espacios crean una fila nueva en lugar de actualizar la existente. Porque el flujo de búsqueda-entonces-escritura no es atómico, evite ejecutar dos exportaciones para el mismo experimento simultáneamente, y evite renombrar un experimento entre ejecuciones a menos que también quiera una fila nueva para él.
Actualizaciones de página completa. Cada actualización reescribe la tabla completa, ya que Confluence no tiene una escritura por fila. Si mantiene la página manualmente entre ejecuciones del script, mantenga sus ediciones dentro de la tabla que construye el script. El script actualmente no conserva contenido fuera de esa tabla.
Conflictos de versión. El script siempre lee el version.number actual de la página inmediatamente antes de escribir, por lo que una edición manual hecha entre la lectura y la escritura hace que el siguiente PUT falle con un conflicto de versión. Ejecute el script de nuevo si eso sucede.
Límites de frecuencia. La Automation API permite hasta 50 solicitudes cada 10 segundos y 1000 por hora, pero Kameleoon recomienda mantenerse por debajo de 12 llamadas por minuto por cuenta. Confluence Cloud aplica sus propios límites de frecuencia, que varían según el plan; consulte la documentación de limitación de frecuencia de Atlassian para valores actuales. Si procesa muchos experimentos por lotes, almacene en caché los tokens, limite la frecuencia de las solicitudes, y considere la Data API para necesidades de gran volumen.

​Objetivo

​Alternativa: exportar con un asistente de IA en lugar del script

​Requisitos

​1. Autenticarse con la Automation API

​2. Recuperar el experimento

​3. Solicitar los resultados del experimento

​4. Consultar los resultados

​5. Seleccionar la variación con mejor rendimiento

​6. Asignar los datos a las columnas de Confluence

​7. Analizar la tabla de Experimentos existente

​8. Buscar, crear o actualizar la página de Confluence

​9. Ejecutar el script

​Script completo

​Notas de personalización

Objetivo

Alternativa: exportar con un asistente de IA en lugar del script

Requisitos

1. Autenticarse con la Automation API

2. Recuperar el experimento

3. Solicitar los resultados del experimento

4. Consultar los resultados

5. Seleccionar la variación con mejor rendimiento

6. Asignar los datos a las columnas de Confluence

7. Analizar la tabla de Experimentos existente

8. Buscar, crear o actualizar la página de Confluence

9. Ejecutar el script

Script completo

Notas de personalización