blastengine 配信後調査を自動化する小技：配信ログAPI＋メール解析レポートCSVジョブの使いどころ

blastengineを使ったメール配信において、「送った後に何が起きたか」を人力で追わず、APIで"調査データ"を定期回収するための実装ポイントに絞って解説します。

前提：配信後に欲しいのは"可視化"より"再現可能な調査データ"

メール配信後の調査で陥りがちなのが「管理画面のスクショを撮って報告」というパターンです。その場では確認できても、後から検証したいときに再現できません。

本記事で目指すのは、ログやレポートをファイルとして残し、後からいつでも検証できる状態を作ることです。管理画面は確認用として使い、調査の証跡はAPIで取得したデータを保存しておく——この方針を前提に進めます。

小技1：配信ログはmaillog単位で回収して"証跡"にする

blastengineの配信ログAPIでは、1通ごとに maillog_id が振られます。このIDをキーにすれば、以下の情報を追跡できます。

• 送信成否（status）
• 最終応答コード・メッセージ
• 配信履歴（送信試行の時系列）

「いつ・誰に・どういう結果で送ったか」を後から説明できる形で保存しておくことで、トラブル発生時の調査工数を大幅に削減できます。

実装メモ

配信ログ一覧APIは anchor パラメータでページングします。anchor に指定した配信ログID未満のデータが返却される仕組みです。基本的な回収フローは以下のとおりです。

1. 配信ログ一覧を取得（GET /logs/mails/results）
2. 取得したデータの末尾（最小）の maillog_id を次の anchor として使用
3. データがなくなるまで繰り返し

import requests
import json
import os
from datetime import datetime

API_BASE = "https://app.engn.jp/api/v1"
HEADERS = {"Authorization": f"Bearer {API_TOKEN}"}

def fetch_all_mail_logs(delivery_id: int) -> list:
    """配信ログ一覧をanchorでページングしながら全件取得"""
    all_logs = []
    anchor = None

    while True:
        params = {"delivery_id": delivery_id, "count": 100}
        if anchor:
            params["anchor"] = anchor

        resp = requests.get(
            f"{API_BASE}/logs/mails/results",
            headers=HEADERS,
            params=params
        )
        data = resp.json()

        logs = data.get("data", [])
        if not logs:
            break

        all_logs.extend(logs)

        # 取得したデータの末尾（最小）のmaillog_idを次のanchorに使用
        # anchorに指定した値未満のデータが返却される
        anchor = min(log["maillog_id"] for log in logs)

    return all_logs

def save_logs(delivery_id: int, logs: list):
    """日付＋delivery_idで冪等にファイル保存"""
    os.makedirs("logs", exist_ok=True)
    date_str = datetime.now().strftime("%Y-%m-%d")
    filename = f"logs/{date_str}_delivery_{delivery_id}.json"

    with open(filename, "w") as f:
        json.dump(logs, f, ensure_ascii=False, indent=2)

保存時のファイル名は「日付＋delivery_id」で冪等性を確保します。同じデータを重複保存しない仕組みにしておくと、リトライ時も安心です。

小技2：配信ログ詳細は"原因特定用"に絞って引く

配信ログ詳細API（GET /logs/mails/{maillog_id}）は1件ずつ取得する必要があります。大量配信後に全件の詳細を取ると、API呼び出し回数が膨大になり現実的ではありません。

そこで、一覧取得時点で「失敗」や「特定ステータス」のものだけ詳細を取得するという割り切りが有効です。

• 一覧で status が HARDERROR や SOFTERROR のものをフィルタ
• フィルタ結果の maillog_id だけ詳細APIを叩く

成功した配信は一覧の情報で十分なケースがほとんどです。詳細が必要になるのは「なぜ失敗したか」を調べるときだけ、と割り切りましょう。

実装メモ

詳細レスポンスから保存対象にすべき項目は以下の3つに絞れます。

項目	用途
last_response_code	SMTPの応答コード（550, 421など）で失敗原因を分類
last_response_message	相手サーバーからのエラーメッセージ本文
sent_history	送信試行の時系列ログ（リトライ状況の確認）

def filter_failed_logs(logs: list) -> list:
    """失敗ステータスのログだけ抽出"""
    return [
        log for log in logs
        if log.get("status") in ("HARDERROR", "SOFTERROR", "DROP")
    ]

def fetch_log_detail(maillog_id: int) -> dict:
    """配信ログ詳細を取得"""
    resp = requests.get(
        f"{API_BASE}/logs/mails/{maillog_id}",
        headers=HEADERS
    )
    return resp.json()

def extract_failure_info(detail: dict) -> dict:
    """調査に必要な3項目だけ抽出"""
    return {
        "maillog_id": detail.get("maillog_id"),
        "email": detail.get("email"),
        "last_response_code": detail.get("last_response_code"),
        "last_response_message": detail.get("last_response_message"),
        "sent_history": detail.get("sent_history", []),
    }

def collect_failure_details(logs: list) -> list:
    """失敗分だけ詳細を取得して保存用データを作成"""
    failed_logs = filter_failed_logs(logs)
    details = []

    for log in failed_logs:
        maillog_id = log.get("maillog_id")
        detail = fetch_log_detail(maillog_id)
        details.append(extract_failure_info(detail))

    return details

この3項目を保存しておけば、「なぜ届かなかったか」の調査は大半カバーできます。それ以外のフィールドは必要になったときに取り直せばよいでしょう。

小技3：メール解析レポートは"ジョブ生成→DL→保管"をバッチ化する

開封情報などの解析レポートは、配信ログとは別のAPIで取得します。blastengineでは「ジョブを生成→完了を待つ→CSVをダウンロード」という非同期フローになっています。ダウンロードできる項目は「配信ID」「配信日時」「メールアドレス」「開封日時」です。

注意: 開封履歴はHTMLメールのみ対応しています。テキストメールのみの配信では取得できません。

手動でやると面倒ですが、この一連の流れをバッチ化してしまえば定期的にレポートを蓄積できます。

実装メモ

1. レポート生成ジョブ起動（POST /deliveries/{delivery_id}/analysis/report）
2. 完了待ち（GET /deliveries/-/analysis/report/{job_id}）
3. CSVダウンロード（GET /deliveries/-/analysis/report/{job_id}/download）
4. ストレージへ保存

import time
import os

def create_report_job(delivery_id: int) -> int:
    """メール解析レポートCSV生成ジョブを開始"""
    resp = requests.post(
        f"{API_BASE}/deliveries/{delivery_id}/analysis/report",
        headers=HEADERS
    )
    data = resp.json()
    return data["job_id"]

def wait_for_job(job_id: int, interval: int = 5, max_wait: int = 300) -> dict:
    """ジョブ完了をポーリングで待機"""
    elapsed = 0

    while elapsed < max_wait:
        resp = requests.get(
            f"{API_BASE}/deliveries/-/analysis/report/{job_id}",
            headers=HEADERS
        )
        data = resp.json()

        if data.get("status") == "FINISHED":
            return data
        if data.get("status") in ("FAILED", "SYSTEM_ERROR", "TIMEOUT"):
            raise Exception(f"Report job failed: {data}")

        time.sleep(interval)
        elapsed += interval

    raise TimeoutError(f"Job {job_id} did not complete within {max_wait}s")

def download_and_save_report(delivery_id: int):
    """ジョブ生成→待機→DL→保存を一括実行"""
    # 1. ジョブ生成
    job_id = create_report_job(delivery_id)
    print(f"Created job: {job_id}")

    # 2. 完了待ち
    job_result = wait_for_job(job_id)
    csv_url = job_result.get("mail_open_file_url")

    # 3. CSVダウンロード（zip形式で返却される）
    csv_resp = requests.get(csv_url, headers=HEADERS)

    # 4. ファイル保存（年月/配信ID_opens.zip）
    year_month = datetime.now().strftime("%Y-%m")
    os.makedirs(f"reports/{year_month}", exist_ok=True)
    filename = f"reports/{year_month}/delivery_{delivery_id}_opens.zip"

    with open(filename, "wb") as f:
        f.write(csv_resp.content)

    print(f"Saved: {filename}")
    return filename

ファイル名に「年月」「配信ID」「レポート種別」を含めておくと、後から探しやすくなります。

使い分け：配信ログAPIと解析レポートCSV

配信ログと解析レポートは目的が異なるので、混ぜずに管理することをおすすめします。

種別	目的	保存形式の例
配信ログ	送達/失敗の調査、応答コードの確認	JSON（maillog_id単位）
解析レポート	開封情報の集計	CSV（配信ID単位）

保存先ディレクトリやスキーマを分けておくと、「送達調査」と「効果測定」で参照するデータが明確になります。

data/
  logs/           # 配信ログ（JSON）
  reports/        # 解析レポート（CSV）

注意点：62日で消えるので"回収しないと調べられない"

blastengineでは、配信情報・宛先情報・ログが一定期間（62日）で削除されます。

つまり、回収しないままにしておくと、後から「あの配信どうなった？」と聞かれても調べようがなくなります。

定期バッチで外部ストレージへ退避しておく理由はここにあります。「調査が必要になってから取りに行く」では間に合わないケースがある、と認識しておきましょう。

統合コード（バッチ実行用）

上記の関数を組み合わせた、バッチ実行用のメイン処理です。

def collect_delivery_data(delivery_ids: list):
    """配信IDリストに対してログ回収＋レポート保存を実行"""
    for delivery_id in delivery_ids:
        print(f"Processing delivery_id: {delivery_id}")

        # 1) 配信ログ一覧を回収
        logs = fetch_all_mail_logs(delivery_id)
        save_logs(delivery_id, logs)
        print(f"  Collected {len(logs)} mail logs")

        # 2) 失敗分だけ詳細を回収
        failure_details = collect_failure_details(logs)
        if failure_details:
            date_str = datetime.now().strftime("%Y-%m-%d")
            filename = f"logs/{date_str}_delivery_{delivery_id}_failures.json"
            with open(filename, "w") as f:
                json.dump(failure_details, f, ensure_ascii=False, indent=2)
            print(f"  Collected {len(failure_details)} failure details")

        # 3) 解析レポートを保存（開封情報）
        try:
            download_and_save_report(delivery_id)
        except Exception as e:
            print(f"  Warning: analysis report failed - {e}")

if __name__ == "__main__":
    # 例：直近の配信IDを指定して実行
    target_ids = [12345, 12346, 12347]
    collect_delivery_data(target_ids)

このスクリプトを日次や週次のcronで動かせば、配信データを自動で蓄積できます。

まとめ

配信後の調査を自動化するポイントを3つの小技として紹介しました。

1. 配信ログは証跡として回収：maillog_id単位で保存し、後から追跡可能に
2. 詳細は失敗に絞る：全件取得は重いので、原因調査が必要なものだけ
3. 解析レポートはジョブでCSV化して保管：非同期フローをバッチ化

これで「後から調べられない」という状況を潰せます。管理画面に頼らず、APIで取れるデータは早めに外部へ退避しておく——この習慣が、トラブル対応時の工数を大きく減らしてくれます。