` tag, and a freshly created page's header row is plain `
Experiment Name | ...|
|---|---|
Product Page Redesign | Completed | ...
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): """Extracts the header row and data rows of the first| cells. Only the first table is read, so a page with more than one table (for example, one a person added by hand) doesn't merge a second table's rows or header into the result.""" 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._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._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):
"""Realigns a row parsed against source_header onto the canonical HEADER
order, so a table whose columns were manually reordered in Confluence
doesn't misalign a column-position lookup like the Experiment Name key."""
if source_header == HEADER:
return row
values = dict(zip(source_header, row))
return [values.get(column) for column in HEADER]
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( "
| {render_cell(v)} | " for v in row) + "