> ## Documentation Index
> Fetch the complete documentation index at: https://docs.kameleoon.com/llms.txt
> Use this file to discover all available pages before exploring further.
# 実験結果を Confluence にエクスポートする
> Automation API を使用して実験とその結果を取得し、ベイズ成功確率から最も成果の良いバリエーションを判断して、Python スクリプトまたは MCP に接続された AI アシスタントを用いて、レコードを Confluence ページにアップサートします。
## ゴール
このチュートリアルでは、[kameleoon\_to\_confluence.py](/assets/developer-docs/script/apis/automation-api-rest/tutorials/kameleoon_to_confluence.py.zip) スクリプトの動作をステップごとに説明します。Kameleoon の**実験 ID**、Confluence の**サイト**、Confluence の**スペース ID** を指定すると、スクリプトは実験のメタデータと統計結果を取得し、最も成果の良いバリエーションを判断して、Confluence ページ上のテーブルにレコードをアップサートします。同じ実験に対してスクリプトを再実行すると、重複を作成する代わりに既存の行を更新します。
ステップ 1～5 は、[Airtable エクスポートチュートリアル](./exporting-experiment-results-to-airtable) と同じリクエストとポーリングのフローを再利用します。変わるのは送信先だけです。
## 代替手段: スクリプトの代わりに AI アシスタントを使用してエクスポートする
スクリプトを実行するには Python 環境、4 つの保存済み認証情報、そしてエクスポートしたい実験ごとの手動での再実行が必要です。すでに Claude のような MCP 対応の AI アシスタントを使用している場合は、Kameleoon と Atlassian のそれぞれのリモート MCP（Model Context Protocol）サーバーに、コーディングツール経由または Claude アプリで直接接続することで、代わりに会話から同じエクスポートを実行できます。Claude を使って結果をエクスポートする方法については、[ガイド](/ja/mcp/mcp-confluence-guide) に従ってください。
Atlassian MCP サーバーは Jira、Confluence、Bitbucket を対象とするツールを公開しており、Confluence ページへの書き込みアクセスを含みます。Kameleoon MCP サーバーは実験やフィーチャーフラグの開始・一時停止・停止・削除も行えます。いずれのコネクタが提案する操作も、承認する前に確認してください。
## 要件
* **Kameleoon API の認証情報。** Automation API にはアクセストークンが必要です。スクリプトは `client_credentials` グラントを使用して、`client_id` と `client_secret` からプログラムでアクセストークンを取得します。[アクセストークンを取得する](/ja/developer-docs/apis/automation-api-rest/get-started/get-started#1-アクセストークンを取得する) を参照してください。
* **Confluence API トークン**（トークンが属するアカウントのメールアドレスとペア）。[id.atlassian.com/manage-profile/security/api-tokens](https://id.atlassian.com/manage-profile/security/api-tokens) でトークンを作成してください。スクリプトは両者を HTTP Basic 認証として、すべての Confluence リクエストに送信します。
* **Confluence スペースの数値スペース ID**（ページをホストするスペース用）。Notion のデータベース ID や Airtable のベース ID と異なり、スペースの数値 ID はその URL に表示されず（URL には代わりにスペースキーが表示される）、Confluence ウェブ UI のどこにも表示されません。MCP に接続した AI アシスタントに検索させるか、Confluence REST API の [Get spaces](https://developer.atlassian.com/cloud/confluence/rest/v2/api-group-space/#api-spaces-get) エンドポイントから自分で取得してください。
* **Python 3.9 以上**、`requests` ライブラリを含む（`pip install requests`）。
Confluence は Notion データベースや Airtable テーブルのようなあらかじめ構築する機能を持ちません。スクリプトは送信先ページをそれ自体で作成し、テーブルと共にそのページを初回実行時に生成します。以降の実行では、タイトルでそのページを検出して更新します。
すべての認証情報は環境変数に保存してください。スクリプトにシークレットをハードコードしないでください。
```bash theme={null}
export KAMELEOON_CLIENT_ID="..."
export KAMELEOON_CLIENT_SECRET="..."
export CONFLUENCE_EMAIL="..."
export CONFLUENCE_API_TOKEN="..."
```
このチュートリアルでは、例として実験 **Product Page Redesign**（ID `188308`）を使用します。オリジナルバリエーションに加えて、2 つのバリエーション `828220` と `828221` があります。
## 1. Automation API で認証する
**エンドポイント:** トークンエンドポイントに POST リクエストを送信して、アクセストークンを取得します。
```
POST https://api.kameleoon.com/oauth/token
```
| フィールド | 型 | 説明 |
| -------------- | ------ | ----------------------------- |
| grant\_type | String | `client_credentials` に設定します。 |
| client\_id | String | Automation API のクライアント ID。 |
| client\_secret | String | Automation API のクライアントシークレット。 |
**例:**
```python theme={null}
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"]
```
**レスポンス:**
```json theme={null}
{ "access_token": "eyJhbGciOiJIUzUxMiIsInR5cCI6IkpXVCJ9..." }
```
以降のすべての Automation API リクエストで、返された `access_token` を `Bearer` トークンとして送信してください。アクセストークンはデフォルトで 2 時間有効です。
## 2. 実験を取得する
**エンドポイント:** [Get an experiment](/api-reference/experiment/get-an-experiment) エンドポイントに GET リクエストを送信して、実験のメタデータを取得します。
```
GET https://api.kameleoon.com/experiments/{experimentId}
```
| フィールド | 型 | 説明 |
| ------------ | ------- | ---------------------- |
| experimentId | Integer | 必須のパスパラメータ。取得する実験の ID。 |
**例:**
```python theme={null}
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()
```
**レスポンス（一部省略）:**
```json theme={null}
{
"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]
}
```
スクリプトは、Confluence 行用に `name`、`status`、`dateStarted`、`dateEnded`、`description` を読み取り、次のステップで結果リクエストの範囲を絞り込むために `mainGoalId` を読み取ります。
API はデフォルトで `mainGoalId` を返すため、これを読み取るために `optionalFields` パラメータは必要ありません。Automation API は `status` フィールドの固定された列挙値を公開していませんが、API 全体の他のステータス系フィールドは一貫して大文字のトークン（たとえば `STOPPED`、`ACTIVE`、`DRAFT`）を使用します。[ステップ 6](#6-データを-confluence-列にマッピングする) では、この前提のもとで `status` フィールドをマッピングします。マッピングに依存する前に、実際のリクエストでアカウントが返す正確なトークンを確認してください。
## 3. 実験の結果をリクエストする
**エンドポイント:** [Request experiment's results](/api-reference/experiment/request-experiments-results) エンドポイントに POST リクエストを送信して、結果レポートの生成をトリガーします。
```
POST https://api.kameleoon.com/experiments/{experimentId}/results
```
| フィールド | 型 | 説明 |
| -------------------- | ------- | ------------------------------------------------------------------------ |
| experimentId | Integer | 必須のパスパラメータ。 |
| goalsIds | Array | レポートを指定したゴール ID に限定します。スクリプトは実験の `mainGoalId` を渡します。 |
| referenceVariationId | String | 比較の基準として使用するバリエーション。`"0"` はオリジナルページを使用します。 |
| visitorData | Boolean | 訪問ベースのデータには `false`、訪問者ベースのデータには `true` を指定します。 |
| sequentialTesting | Boolean | ベイズ成功確率の代わりに、信頼区間に逐次検定を使用する場合は `true` に設定します。いずれか一方の手法のみを有効にしてください。 |
| bayesian | Boolean | レポートにベイズ成功確率を含める場合は `true` に設定します。スクリプトはこの値を読み取って *Probability* 列に設定します。 |
| conversionType | String | `ALL_CONVERSION` または `CONVERTED_VISITS`。 |
**例:**
```python theme={null}
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"]
```
**レスポンス:**
```json theme={null}
{ "dataCode": "14443931880924098266207585267983330260134079899081739889989435434342588016765" }
```
Kameleoon はレポートを非同期で生成します。エンドポイントは `dataCode` を返し、[ステップ 4](#4-結果をポーリングする) ではこれを使用して結果をポーリングします。
このスクリプトには `bayesian: true` が必要で、`sequentialTesting: false` を設定します。`bayesian` と `sequentialTesting` は有意性を算出するための代替手法であり、このチュートリアルはベイズ成功確率をレポートします。ベイズ推定を有効にすると、レポートの `reliability` の値にベイズ成功確率（あるバリエーションが参照バリエーションに勝る確率）が反映され、スクリプトはこの値を *Probability* 列にマッピングします。アカウントが別のデフォルトの統計手法を使用している場合は、Kameleoon アプリ内の同じレポートと照らして値を確認してください。
## 4. 結果をポーリングする
**エンドポイント:** [Poll results](/api-reference/data/poll-results) エンドポイントに GET リクエストを送信し、レポートが準備できるまで取得を繰り返します。
```
GET https://api.kameleoon.com/results?dataCode={dataCode}
```
| フィールド | 型 | 説明 |
| -------- | ------ | ------------------------------------------------------------------------------- |
| dataCode | String | 必須のクエリパラメータ。ステップ 3 で `POST /experiments/{experimentId}/results` が返したハッシュを使用します。 |
レスポンスの `status` は、Kameleoon がレポートを計算している間は `WAITING`、データが利用可能になると `READY`、失敗時は `ERROR` または `TIMEOUT` になります。ステータスが `ERROR` または `TIMEOUT` の場合、レスポンスにはトップレベルの `errorDescription` が含まれます。スクリプトは、ステータスが `READY` になるまで一定の間隔でポーリングします。
**例:**
```python theme={null}
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.")
```
**レスポンス（一部省略）:**
```json theme={null}
{
"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. 最も成果の良いバリエーションを選択する
結果には、`variationData` の下にバリエーションごとに 1 つのエントリと、オリジナルページ用の `_reference` の行が含まれます。各バリエーションについて、リクエストしたゴールの指標は `breakdownData._reference.generalData.goalsData[goalId]` の下にあります。
スクリプトは `_reference` エントリをスキップし、各バリエーションの `improvementRate` と `reliability`（ベイズ成功確率）を読み取り、改善率が最も高いバリエーションを最も成果の良いものとして選択します。次のステップでマッピングされる *Result* 列には、そのバリエーションが実際に勝利したかどうか、つまり十分に高い成功確率と正の上昇率を達成したかどうかが記録されます。
**例:**
```python theme={null}
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
```
この例では、両方のバリエーションが 100% のベイズ成功確率に達していますが、バリエーション `828220` は +211.48% の改善を示しているのに対し、バリエーション `828221` は -43.33% です。したがって、バリエーション `828220` が最も成果の良いバリエーションであり、95% を超える確率と正の上昇率を伴う正真正銘の勝者です。
## 6. データを Confluence 列にマッピングする
スクリプトは、実験のメタデータと最も成果の良いバリエーションの指標を Confluence テーブル行に変換し、Notion と Airtable エクスポートと同じ 8 列を使用します。
| Confluence 列 | ソース | 変換 |
| --------------- | --------------------------------- | ---------------------------------------------------------------------------------------------------------------------------- |
| Experiment Name | `experiment.name` | そのまま使用。アップサートのキーとして使用します。 |
| Status | `experiment.status` | 大文字・小文字を区別せずにマッピング: `active → Running`、`draft`/`planned → Implementing`、`stopped`/`diverted → Completed`、`paused → Defunct`。 |
| Start date | `experiment.dateStarted` | 日時を ISO 形式の日付（`YYYY-MM-DD`）に切り詰めたもの。 |
| End date | `experiment.dateEnded` | 日時を ISO 形式の日付に切り詰めたもの。 |
| Notes | `experiment.description` | そのまま使用。 |
| Actual | 最も成果の良いバリエーションの `improvementRate` | そのまま使用（計測された上昇率、%）。 |
| Probability | 最も成果の良いバリエーションのベイズ成功確率 | 区分: `≥95` → `80% - High`、`≥80` → `50% - Medium`、それ以外は `20% - Low`。 |
| Result | ベイズ成功確率 + `improvementRate` | 確率が `95` 以上かつ上昇率が正 → `Success`。`95` 以上かつ上昇率が負 → `Failure`。それ以外は `Inconclusive`。 |
**例:**
```python theme={null}
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]
```
Automation API は実験の `status` フィールドに固定された列挙値を公開しておらず、トークンは今後変わる可能性があります。`map_status` は大文字・小文字を区別せずにマッチングし、認識できないステータスに対しては `None` を返すため、誤った値を書き込む代わりに *Status* セルへの書き込みを省略します。1 回の `GET /experiments/{experimentId}` でアカウントが返すトークンを確認し、必要に応じて `STATUS_MAP` を拡張してください。
## 7. 既存の Experiments テーブルをパースする
Confluence は Notion データベースや Airtable テーブルオブジェクトを持ちません。代わりに、スクリプトは 1 つの専用ページに 1 つの HTML テーブルを保持し、バリエーションごとに 1 行ずつ置き、*Experiment Name* 列を Notion が Title プロパティを扱う方法や Airtable がマージフィールドを扱う方法と同じ方法で扱います—アップサートキーとして。スクリプトはテーブルを `full-width` レイアウトで構築します。Confluence のデフォルトの狭いページ幅では、8 列の適度な長さのヘッダーが単語の途中で折り返されてしまうためです。
Confluence のストレージフォーマット（ページの本体が保存されるマークアップ）ではすべてのセルのテキストが `` タグでラップされ、新たに作成されたページのヘッダー行は `` で囲まれていない単純な `| ... |
|---|
` セルです。以下のパーサーは Python 標準の `html.parser.HTMLParser` に基づいており、ベアヘッダーケースと `` ラップケースの両方に対応し、空のセルを Confluence が `` でレンダリングするか自己クローズ `` でレンダリングするかに関わらず同じものとして扱います。また最初の `` で停止するため、ページの他の場所に手で追加した 2 番目のテーブルはパースされた結果にマージされません。また、不正な編集によって開かれたままになっている行またはセルをクローズし、次のタグが新しい行を開いても、テーブル自体が終わっても対応するため、開かれたままのタグが誤って行をドロップすることはありません。
**例:**
```python theme={null}
from html.parser import HTMLParser
class _TableParser(HTMLParser):
"""Confluence ストレージフォーマットから HTML テーブルをパースして行を抽出します。"""
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):
"""source_header に対してパースされた行を正規の HEADER 順序に再整列します。
Confluence でカラムが手動で並べ替えられたテーブルが、Experiment Name キー
のようなカラム位置参照を誤整列させないようにします。真の並べ替え（同じ 8 つ
のラベル、異なる順序）のみを再整列します。ヘッダーが名前変更または認識不可
なカラムを持つ場合、既存の順序にフォールバックし、推測の代わりに警告を
出力します。"""
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` は、Confluence エディタで列をドラッグするなど、列が手動で並び替えられたテーブルに対応します。これがなければ、位置による参照は新しい行の *Experiment Name* を既存テーブルの現在その位置に座っている任意の列と比較し、誤った行をマッチングするか何もマッチングしません。真の並べ替え（同じ 8 つのラベル、異なる順序）のみを再整列します。ヘッダーセルのテキストが編集された場合（たとえば *Notes* を *Comments* に改題）、ラベルはもはや `HEADER` と照合されません。その場合、関数は既存の列の順序にフォールバックし、推測の代わりに警告を出力し、そのカラムのデータをすべての行で無言でブランク化することはありません。
## 8. Confluence ページを検出、作成、または更新する
**検出:** `title` と `space-id` クエリパラメータを [Get pages](https://developer.atlassian.com/cloud/confluence/rest/v2/api-group-page/#api-pages-get) エンドポイントで使用してスペース内のページをタイトルで検索します。`body-format=storage` を渡すと、同じ呼び出しで現在のテーブル内容が返されます。
```
GET https://{site}/wiki/api/v2/pages
```
**作成:** ページが一致しない場合、[Create page](https://developer.atlassian.com/cloud/confluence/rest/v2/api-group-page/#api-pages-post) エンドポイントで新しいページを作成します。テーブルは 1 行から構築されます。
```
POST https://{site}/wiki/api/v2/pages
```
**更新:** ページが一致する場合、既存の行と新しくアップサートされたものからテーブル全体を再構築し、[Update page](https://developer.atlassian.com/cloud/confluence/rest/v2/api-group-page/#api-pages-id-put) エンドポイントでページを上書きします。
```
PUT https://{site}/wiki/api/v2/pages/{id}
```
Confluence には行単位または フィールド単位のアップデートエンドポイントがありません。ページを更新するとその本体全体が置き換わるため、スクリプトは常に現在のテーブルを読み取り、1 行をメモリ内でアップサートし、テーブル全体を書き戻します。この全体的な置き換えは Airtable エクスポート（単一レコードをパッチ）や Notion エクスポート（単一ページのプロパティをパッチ）と異なります。
Notion および Airtable API と異なり、Confluence は呼び出し元に対して更新ごとに `version.number` をインクリメント（現在のバージョン番号に 1 を加えたもの）するよう要求しています。現在の番号を再度送信するか、`version` を省略すると、リクエストは失敗します。MCP に接続した AI アシスタントが Atlassian 独自のツールを使用すると、これは自動的に処理されます。本スクリプトのような直接 REST 呼び出しには適用されません。
**例:**
```python theme={null}
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"
```
**レスポンス（一部省略）:**
```json theme={null}
{
"id": "557057",
"status": "current",
"title": "Experiments",
"spaceId": "327682",
"version": { "number": 3 },
"body": {
"storage": {
"value": "Experiment Name | ...
|---|
Product Page Redesign | Completed | ...
"
}
}
}
```
## 9. スクリプトを実行する
実験 ID、Confluence サイト、スペース ID を引数として渡します:
```bash theme={null}
python kameleoon_to_confluence.py \
--experiment-id 188308 \
--site your-domain.atlassian.net \
--space-id 327682
```
スクリプトは各ステップを出力します—認証、取得した実験、最も成果の良いバリエーション、マッピングされた行、そして Confluence ページを作成したか更新したか。`--page-title` でデフォルトの `Experiments` 以外のページ名を指定します。
## スクリプト全文
以下の完全なスクリプトは、[kameleoon\_to\_confluence.py](/assets/developer-docs/script/apis/automation-api-rest/tutorials/kameleoon_to_confluence.py.zip) と関数単位で一致しています。そのままコピーするか、リンク先からファイルをダウンロードしてください。
```python theme={null}
#!/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
# Automation API のステータストークンを Confluence ページ用の Status
# 値にマップします。*Status* ラベルが Running / Implementing / Completed /
# Defunct と異なる場合はターゲット値を調整し、マッピングに依存する前に
# 実際の GET /experiments/{experimentId} でアカウントが返すトークンを確認してください。
STATUS_MAP = {
"ACTIVE": "Running",
"DRAFT": "Implementing",
"PLANNED": "Implementing",
"PAUSED": "Defunct",
"STOPPED": "Completed",
"DIVERTED": "Completed",
}
# Experiments テーブルのカラム順です。スクリプトは新しいページを構築するときも
# 既存のページをパースするときも、このヘッダーに対して列を位置的にキーイングします。
HEADER = [
"Experiment Name",
"Status",
"Start date",
"End date",
"Notes",
"Actual",
"Probability",
"Result",
]
TABLE_INTRO = (
"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
# --------------------------------------------------------------------------- #
from html.parser import HTMLParser
class _TableParser(HTMLParser):
"""Confluence ストレージフォーマットから HTML テーブルをパースして行を抽出します。"""
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):
"""source_header に対してパースされた行を正規の HEADER 順序に再整列します。
Confluence でカラムが手動で並べ替えられたテーブルが、Experiment Name キー
のようなカラム位置参照を誤整列させないようにします。真の並べ替え（同じ 8 つ
のラベル、異なる順序）のみを再整列します。ヘッダーが名前変更または認識不可
なカラムを持つ場合、既存の順序にフォールバックし、推測の代わりに警告を
出力します。"""
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()
```
## カスタマイズに関する注意事項
* **ステータスマッピング** は `STATUS_MAP` 定数にあり、API が返す大文字のステータストークンをキーとします。*Status* 列が `Running` / `Implementing` / `Completed` / `Defunct` と異なる場合はターゲットの値を調整し、マッピングに依存する前に実際の `GET /experiments/{experimentId}` リクエストでアカウントが返すトークンを確認してください。
* **Probability** は測定されたベイズ成功確率から取得され、結果リクエストで `bayesian: true` が必要です。事前実験の見積もりが必要な場合は、`build_row` から `Probability` の行を削除してください。
* **ゴールの選択** には実験の `mainGoalId` を使用します。別のゴールについてレポートするには、そのゴール ID を `request_results` と `pick_best_variation` に渡してください。
* **アップサートキー。** *Experiment Name* との正確な一致により、大文字・小文字や空白の違いは既存の行を更新する代わりに新しい行を作成してしまいます。既存の行を誤って更新したり、重複を作成したりするのを避けるために、このページでは各実験の名前を一意に保ってください。
* **ページ全体の更新。** Confluence には行単位の書き込みがないため、すべての更新はテーブル全体を再構築します。スクリプトは常に現在のテーブルを読み取り、メモリ内で 1 行をアップサートし、テーブル全体を書き戻します。スクリプトの実行の合間にページを手動で編集する場合は、編集内容をスクリプトが構築するテーブル内に収めてください。スクリプトは現在、そのテーブル外のコンテンツを保持しません。
* **バージョン競合。** スクリプトは常に書き込む直前にページの現在の `version.number` を読み取ります。読み取りと書き込みの間に手動編集があると、次の `PUT` がバージョン競合で失敗します。このような場合はスクリプトを再実行してください。
* **レート制限。** Automation API は 10 秒あたり最大 50 リクエスト、1 時間あたり最大 1,000 リクエストまで許可していますが、Kameleoon はアカウントごとに 1 分あたり 12 コール未満に抑えることを推奨します。Confluence Cloud は独自のレート制限を実施しており、計画によって異なります。詳細は [Atlassian のレート制限ドキュメント](https://developer.atlassian.com/cloud/confluence/rate-limiting/) を参照してください。複数の実験をバッチ処理するには、トークンをキャッシュし、リクエストをスロットルし、大容量のニーズについては [Data API](/ja/developer-docs/apis/data-api-rest/overview) を検討してください。