実験結果を Airtable にエクスポートする

ゴール

このチュートリアルでは、kameleoon_to_airtable.py スクリプトの動作をステップごとに説明します。Kameleoon の実験 ID、Airtable のベース ID、Airtable のテーブル ID を指定すると、スクリプトは実験のメタデータと統計結果を取得し、最も成果の良いバリエーションを判断して、データを Airtable の Experiments スキーマにマッピングし、レコードを Airtable に書き戻します。同じ実験に対してスクリプトを再実行すると、重複を作成する代わりに既存の行を更新します。本チュートリアルは、Automation API を使用して実験結果を取得するチュートリアルに続くものであり、同じリクエストとポーリングのフローを再利用します。

要件

Kameleoon API の認証情報。 Automation API にはアクセストークンが必要です。スクリプトは client_credentials グラントを使用して、client_id と client_secret からプログラムでアクセストークンを取得します。アクセストークンを取得するを参照してください。
対象のベースに対して data.records:write スコープを持つ Airtable の個人アクセストークン。
Airtable のベース ID とテーブル ID。 どちらもテーブルの URL、またはベースの API ドキュメントに表示されます。ベース ID は app で始まり、テーブル ID は tbl で始まります。
Experiments スキーマがすでに構築された Airtable テーブル。スクリプトは次のフィールドに書き込みます: Experiment Name、Status、Start date、End date、Notes、Actual、Probability、Result。また、手動入力用のフィールド Assignee、Category、Prediction、Mkt Est、Eng Est、Attachments が存在することを前提としていますが、スクリプトがこれらの値を設定することはありません。Airtable はテーブルにまだ存在しないフィールド名への書き込みを拒否し、typecast は既存のフィールドに対して値の型を変換するだけです（不足しているフィールドや選択肢を作成することはありません）。スクリプトを実行する前に、これらのフィールドと対応する Status の選択肢を持つテーブルを作成してください。各フィールドに設定される値については、ステップ 6 を参照してください。
Python 3.9 以上、requests ライブラリを含む（pip install requests）。

すべての認証情報は環境変数に保存してください。スクリプトにシークレットをハードコードしないでください。

export KAMELEOON_CLIENT_ID="..."
export KAMELEOON_CLIENT_SECRET="..."
export AIRTABLE_TOKEN="..."

このチュートリアルでは、例として実験 Product Page Redesign（ID 188308）を使用します。オリジナルバリエーションに加えて、Redesign 1（ID 828220）と Redesign 2（ID 828221）の 2 つのバリエーションがあります。

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 のクライアントシークレット。

例:

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

レスポンス:

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

以降のすべての Automation API リクエストで、返された access_token を Bearer トークンとして送信してください。アクセストークンはデフォルトで 2 時間有効です。

2. 実験を取得する

エンドポイント: Get an experiment エンドポイントに GET リクエストを送信して、実験のメタデータを取得します。

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

フィールド	型	説明
experimentId	Integer	必須のパスパラメータ。取得する実験の ID。

例:

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

レスポンス（一部省略）:

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

スクリプトは、Airtable レコード用に name、status、dateStarted、dateEnded、description を読み取り、次のステップで結果リクエストの範囲を絞り込むために mainGoalId を読み取ります。

API はデフォルトで mainGoalId を返すため、これを読み取るために optionalFields パラメータは必要ありません。Automation API は status フィールドの固定された列挙値を公開していませんが、API 全体の他のステータス系フィールドは一貫して大文字のトークン（たとえば STOPPED、ACTIVE、DRAFT）を使用します。ステップ 6 では、この前提のもとで status フィールドをマッピングします。マッピングに依存する前に、実際のリクエストでアカウントが返す正確なトークンを確認してください。

3. 実験の結果をリクエストする

エンドポイント: Request experiment’s 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`。

例:

def request_results(token, experiment_id, goal_id):
    body = {
        "visitorData": False,
        "sequentialTesting": True,
        "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"]

レスポンス:

{ "dataCode": "14443931880924098266207585267983330260134079899081739889989435434342588016765" }

Kameleoon はレポートを非同期で生成します。エンドポイントは dataCode を返し、ステップ 4 ではこれを使用して結果をポーリングします。

このスクリプトには bayesian: true が必要です。ベイズ推定を有効にすると、レポートの reliability の値にベイズ成功確率（あるバリエーションが参照バリエーションに勝る確率）が反映され、スクリプトはこの値を Probability フィールドにマッピングします。アカウントが別のデフォルトの統計手法を使用している場合は、Kameleoon アプリ内の同じレポートと照らして値を確認してください。Automation API の仕様では dateIntervals は必須と記載されていますが、上記の例ではこれを省略しても有効なレポートが返されます。仕様には、dateIntervals を省略した場合のデフォルト動作は記載されていません。本チュートリアルでは、実験の全期間をカバーすると仮定しているため、この動作に依存する前に、自分のアカウントで確認してください。特定の期間にレポートを限定するには、代わりに dateIntervals 配列を渡します。

4. 結果をポーリングする

エンドポイント: 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 になるまで一定の間隔でポーリングします。例:

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

レスポンス（一部省略）:

{
  "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 フィールドには、そのバリエーションが実際に勝利したかどうか、つまり十分に高い成功確率と正の上昇率を達成したかどうかが記録されます。例:

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% のベイズ成功確率に達していますが、Redesign 1（828220）は +211.48% の改善を示しているのに対し、Redesign 2 は -43.33% です。したがって、Redesign 1 が最も成果の良いバリエーションであり、95% を超える確率と正の上昇率を伴う正真正銘の勝者です。

6. データを Airtable フィールドにマッピングする

スクリプトは、実験のメタデータと最も成果の良いバリエーションの指標を、Airtable の Experiments スキーマに変換します。

Airtable フィールド	ソース	変換
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`。

スクリプトは、Kameleoon 側にソースがないフィールド（Assignee、Category、Prediction、Mkt Est、Eng Est、Attachments）を設定しません。これらのフィールドは、Airtable 内で手動入力用として利用できる状態のままです。また、スクリプトは空の値を省略するため、既存のセルを空欄で上書きすることはありません。例:

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_airtable_fields(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 {k: v for k, v in fields.items() if v is not None}

Automation API は実験の status フィールドに固定された列挙値を公開しておらず、トークンは今後変わる可能性があります。map_status は大文字・小文字を区別せずにマッチングし、認識できないステータスに対しては None を返すため、誤った値を書き込む代わりに Status セルへの書き込みを省略します。1 回の GET /experiments/{experimentId} でアカウントが返すトークンを確認し、必要に応じて STATUS_MAP を拡張してください。

7. レコードを Airtable にアップサートする

Airtable の Update table エンドポイント（PATCH /v0/meta/bases/{baseId}/tables/{tableId}）は、テーブルの名前と説明のみを変更するものであり、行にデータを書き込むことはできません。レコードにデータを入力するには、performUpsert オプションを指定して records エンドポイントを使用してください。

エンドポイント: records エンドポイントに PATCH リクエストを送信して、レコードを作成または更新します。

PATCH https://api.airtable.com/v0/{baseId}/{tableId}

フィールド	型	説明
performUpsert.fieldsToMergeOn	Array	既存のレコードとの照合に使用するフィールド。スクリプトは `Experiment Name` でマージします。
records	Array	レコードのリスト（1 リクエストあたり最大 10 件）。各レコードは `fields` オブジェクトを持ちます。
typecast	Boolean	`true` にすると、Airtable が文字列を選択肢に変換し、日付を解析できるようになります。

例:

def upsert_record(airtable_token, base_id, table_id, fields):
    resp = requests.patch(
        f"https://api.airtable.com/v0/{base_id}/{table_id}",
        headers={
            "Authorization": f"Bearer {airtable_token}",
            "Content-Type": "application/json",
        },
        json={
            "performUpsert": {"fieldsToMergeOn": ["Experiment Name"]},
            "typecast": True,
            "records": [{"fields": fields}],
        },
    )
    resp.raise_for_status()
    return resp.json()

レスポンス（一部省略）:

{
  "records": [
    {
      "id": "rec0Vh8KLmNoPqRsT",
      "fields": {
        "Experiment Name": "Product Page Redesign",
        "Status": "Completed",
        "Start date": "2025-01-15",
        "End date": "2025-02-12",
        "Actual": 211.48,
        "Probability": "80% - High",
        "Result": "Success"
      }
    }
  ],
  "createdRecords": [],
  "updatedRecords": ["rec0Vh8KLmNoPqRsT"]
}

レスポンスは、レコードごとに結果を報告します。createdRecords の下に ID が返される場合は Airtable が新しい行を作成したことを意味し、updatedRecords の下に ID がある場合は Airtable が既存の行を更新したことを意味します。

8. スクリプトを実行する

実験 ID と Airtable のベース ID、テーブル ID を引数として渡します:

python kameleoon_to_airtable.py \
  --experiment-id 188308 \
  --base-id appXXXXXXXXXXXXXX \
  --table-id tbll9adSjdedH5t3f

スクリプトは、認証、取得した実験、最も成果の良いバリエーション、マッピングされたフィールド、そして Airtable レコードを作成したか更新したかという各ステップを出力します。

スクリプト全文

以下の完全なスクリプトは、kameleoon_to_airtable.py と関数単位で一致しています。そのままコピーするか、リンク先からファイルをダウンロードしてください。

#!/usr/bin/env python3
"""Export a Kameleoon experiment's results to an Airtable *Experiments* table.

Given a Kameleoon experiment ID, an Airtable base ID, and an Airtable table ID,
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 the Airtable *Experiments* schema.
  7. Upserts the record into Airtable, keyed on "Experiment Name".

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

    export KAMELEOON_CLIENT_ID="..."
    export KAMELEOON_CLIENT_SECRET="..."
    export AIRTABLE_TOKEN="..."

Usage:

    python kameleoon_to_airtable.py \
        --experiment-id 188308 \
        --base-id appXXXXXXXXXXXXXX \
        --table-id tbll9adSjdedH5t3f
"""

import argparse
import os
import sys
import time

import requests

# Maps the uppercase status tokens returned by the Automation API to the
# options of the Airtable *Status* field. Adjust the target values if your
# *Status* options differ, 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",
}


# --------------------------------------------------------------------------- #
# 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": True,
        "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 Airtable fields
# --------------------------------------------------------------------------- #
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_airtable_fields(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 {k: v for k, v in fields.items() if v is not None}


# --------------------------------------------------------------------------- #
# 7. Upsert the record into Airtable
# --------------------------------------------------------------------------- #
def upsert_record(airtable_token, base_id, table_id, fields):
    resp = requests.patch(
        f"https://api.airtable.com/v0/{base_id}/{table_id}",
        headers={
            "Authorization": f"Bearer {airtable_token}",
            "Content-Type": "application/json",
        },
        json={
            "performUpsert": {"fieldsToMergeOn": ["Experiment Name"]},
            "typecast": True,
            "records": [{"fields": fields}],
        },
    )
    resp.raise_for_status()
    return resp.json()


# --------------------------------------------------------------------------- #
# 8. Orchestration
# --------------------------------------------------------------------------- #
def run(experiment_id, base_id, table_id):
    client_id = os.environ.get("KAMELEOON_CLIENT_ID")
    client_secret = os.environ.get("KAMELEOON_CLIENT_SECRET")
    airtable_token = os.environ.get("AIRTABLE_TOKEN")

    missing = [
        name
        for name, value in (
            ("KAMELEOON_CLIENT_ID", client_id),
            ("KAMELEOON_CLIENT_SECRET", client_secret),
            ("AIRTABLE_TOKEN", airtable_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 Airtable fields.
    fields = build_airtable_fields(experiment, best)
    print("Mapped Airtable fields:")
    for key, value in fields.items():
        print(f"  {key}: {value}")

    # 7. Upsert the record.
    response = upsert_record(airtable_token, base_id, table_id, fields)
    created = response.get("createdRecords") or []
    record_id = response["records"][0]["id"] if response.get("records") else None
    action = "Created" if record_id in created else "Updated"
    print(f"{action} Airtable record {record_id}.")

    return response


def main():
    parser = argparse.ArgumentParser(
        description="Export a Kameleoon experiment's results to Airtable."
    )
    parser.add_argument(
        "--experiment-id",
        required=True,
        help="ID of the Kameleoon experiment to export.",
    )
    parser.add_argument(
        "--base-id",
        required=True,
        help="Airtable base ID (starts with 'app').",
    )
    parser.add_argument(
        "--table-id",
        required=True,
        help="Airtable table ID (starts with 'tbl') or table name.",
    )
    args = parser.parse_args()

    try:
        run(args.experiment_id, args.base_id, args.table_id)
    except requests.HTTPError 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 が必要です。Probability フィールドが実験前の見積もりを手動で入力するものである場合は、build_airtable_fields から Probability の行を削除してください。
ゴールの選択 には実験の mainGoalId を使用します。別のゴールについてレポートするには、そのゴール ID を request_results と pick_best_variation に渡してください。
アップサートキー。 Airtable はマージフィールドを厳密に一致させるため、Experiment Name の大文字・小文字や空白の違いは、既存の行を更新する代わりに新しい行を作成してしまいます。実験名を安定させるか、専用の安定した識別子フィールドでマージしてください。
Actual フィールドの形式。 スクリプトは、生の improvementRate の値（たとえば 211.48）を Actual に書き込みます。Actual が Airtable の Percent フィールドである場合は、分数ではなく通常の数値を受け取るように設定するか、分数ベースの Percent フィールドに合わせて build_airtable_fields 内でその値を 100 で割ってください。
レート制限。 Automation API は 10 秒あたり最大 50 リクエスト、1 時間あたり最大 1,000 リクエストまで許可していますが、Kameleoon はアカウントごとに 1 分あたり 12 コール未満に抑えることを推奨しており、高頻度のトラッキングに Automation API を使用しないよう推奨しています。多数の実験をまとめて処理する場合は、トークンをキャッシュし、リクエストをスロットリングし、大量のデータが必要な場合は Data API の使用を検討してください。

​ゴール

​要件

​1. Automation API で認証する

​2. 実験を取得する

​3. 実験の結果をリクエストする

​4. 結果をポーリングする

​5. 最も成果の良いバリエーションを選択する

​6. データを Airtable フィールドにマッピングする

​7. レコードを Airtable にアップサートする

​8. スクリプトを実行する

​スクリプト全文

​カスタマイズに関する注意事項

ゴール

要件

1. Automation API で認証する

2. 実験を取得する

3. 実験の結果をリクエストする

4. 結果をポーリングする

5. 最も成果の良いバリエーションを選択する

6. データを Airtable フィールドにマッピングする

7. レコードを Airtable にアップサートする

8. スクリプトを実行する

スクリプト全文

カスタマイズに関する注意事項