Kameleoon experiment results, kept current by "
"kameleoon_to_confluence.py. Each row is upserted by matching the "
"Experiment Name column; re-running the script for the "
"same experiment updates its row instead of adding a duplicate.
"
)
# --------------------------------------------------------------------------- #
# 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
en un documento en formato de almacenamiento de Confluence. Confluence
envuelve el texto de las celdas en etiquetas y no siempre emite
; una fila cuenta como encabezado la primera vez que está
compuesta de celdas | . 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 |
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 | / |
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 |
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"{text}
"
def build_table_storage(rows):
head_cells = "".join(f"{render_cell(c)} | " for c in HEADER)
body_rows = "".join(
"
" + "".join(f"| {render_cell(v)} | " for v in row) + "
"
for row in rows
)
return f'{TABLE_INTRO}'
# --------------------------------------------------------------------------- #
# 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](https://developer.atlassian.com/cloud/confluence/rate-limiting/) 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](/es/developer-docs/apis/data-api-rest/overview) para necesidades de gran volumen.