メールマーケティングの手動送信1000通が全自動になった

マーケティング担当の前田です。キャンペーンメールを顧客1000人に個別送信するため、Excel名簿を見ながらメールを一通ずつ編集して送るのに1日かかっていました。Claude CodeでCSV読み込み→HTML差し込み→一括送信→効果測定まで完全自動化できました。

STEP 1：CSVから個別化HTMLメールを自動生成する

顧客リストCSVを読み込み、各顧客に合わせたHTMLメールを自動生成します。

import pandas as pd
from string import Template

df = pd.read_csv("customers.csv")

HTML_TEMPLATE = Template("""




  
特別ご優待のご案内


  
${name}様
  
${company}でご活躍のことと存じます。
  
日頃のご愛顧に感謝し、${name}様限定の特別優待をご案内いたします。
  

    
🎁 ${industry}向け特別プラン
    
通常価格より 30%OFF
  
  
    詳細を見る →
  


  
  配信停止はこちら

""")

emails_to_send = []
for i, row in df.iterrows():
    html = HTML_TEMPLATE.substitute(
        name=row["名前"], company=row["会社名"],
        industry=row["業種"], customer_id=str(i+1001)
    )
    emails_to_send.append({"to": row["メールアドレス"], "name": row["名前"], "html": html})

print(f"{len(emails_to_send)}件のメールを準備完了")

STEP 2：レート制限付きで一括送信する

Gmailの送信制限に引っかからないよう、適切な間隔で一括送信します。

import time, base64
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from googleapiclient.errors import HttpError

RATE_LIMIT = 2  # 1秒あたり最大2通
RETRY_WAIT = 10

def send_email_gmail(service, to, subject, html_body):
    msg = MIMEMultipart("alternative")
    msg["To"] = to; msg["Subject"] = subject
    msg.attach(MIMEText(html_body, "html", "utf-8"))
    raw = base64.urlsafe_b64encode(msg.as_bytes()).decode("utf-8")
    
    for attempt in range(3):
        try:
            service.users().messages().send(
                userId="me", body={"raw": raw}).execute()
            return True
        except HttpError as e:
            if e.resp.status == 429:  # レート制限
                time.sleep(RETRY_WAIT * (attempt+1))
            else:
                print(f"送信エラー {to}: {e}"); return False
    return False

sent, failed = 0, []
subject = "【期間限定】特別ご優待のご案内"

for i, email_data in enumerate(emails_to_send):
    if i > 0 and i % RATE_LIMIT == 0:
        time.sleep(1)  # レート制限: 1秒に2通
    
    success = send_email_gmail(service, email_data["to"], subject, email_data["html"])
    if success:
        sent += 1
        if sent % 100 == 0: print(f"  進捗: {sent}/{len(emails_to_send)}件")
    else:
        failed.append(email_data["to"])

print(f"送信完了: {sent}件成功, {len(failed)}件失敗")

STEP 3：開封率・クリック率を自動トラッキングする

トラッキングピクセルとクリックトラッキングで開封率・クリック率を測定します。

import sqlite3, flask
from flask import Flask, request, send_file
import io

app = Flask(__name__)
DB_PATH = "email_tracking.db"

def init_db():
    with sqlite3.connect(DB_PATH) as conn:
        conn.execute("""CREATE TABLE IF NOT EXISTS events
            (customer_id TEXT, event_type TEXT, timestamp DATETIME DEFAULT CURRENT_TIMESTAMP, ip TEXT)""")

# 開封トラッキング（1x1ピクセル）
@app.route("/pixel/.gif")
def track_open(customer_id):
    with sqlite3.connect(DB_PATH) as conn:
        conn.execute("INSERT INTO events VALUES (?,?,CURRENT_TIMESTAMP,?)",
                     (customer_id, "open", request.remote_addr))
    # 1x1透明GIFを返す
    gif = b"GIF89a

€ÿÿÿ!ù,

D
;"
    return send_file(io.BytesIO(gif), mimetype="image/gif")

# 集計レポート生成
def generate_report():
    import pandas as pd, openpyxl
    with sqlite3.connect(DB_PATH) as conn:
        df = pd.read_sql("SELECT customer_id, event_type, COUNT(*) as count FROM events GROUP BY customer_id, event_type", conn)
    
    opens  = len(df[df["event_type"]=="open"]["customer_id"].unique())
    clicks = len(df[df["event_type"]=="click"]["customer_id"].unique())
    total  = len(emails_to_send)
    
    print(f"=== キャンペーン結果 ===")
    print(f"送信数:   {total}件")
    print(f"開封数:   {opens}件 ({opens/total*100:.1f}%)")
    print(f"クリック: {clicks}件 ({clicks/total*100:.1f}%)")
    
    report_df = pd.DataFrame({"指標":["送信数","開封数","開封率","クリック数","CTR"],
                               "値":[total, opens, f"{opens/total*100:.1f}%",clicks,f"{clicks/total*100:.1f}%"]})
    report_df.to_excel("campaign_report.xlsx", index=False)

STEP 4：A/Bテストを自動実行する

件名のA/Bテストを自動で設定し、効果の高い件名を自動判定します。

import random, time
from datetime import datetime, timedelta

def ab_test_send(emails, subject_a, subject_b):
    random.shuffle(emails)
    group_a = emails[:len(emails)//2]
    group_b = emails[len(emails)//2:]
    
    test_id = datetime.now().strftime("%Y%m%d%H%M")
    
    print(f"A版送信開始: {len(group_a)}件 | 件名: {subject_a}")
    for e in group_a:
        send_email_gmail(service, e["to"], subject_a + f"?test={test_id}&group=A", e["html"])
        time.sleep(0.5)
    
    print(f"B版送信開始: {len(group_b)}件 | 件名: {subject_b}")
    for e in group_b:
        send_email_gmail(service, e["to"], subject_b + f"?test={test_id}&group=B", e["html"])
        time.sleep(0.5)
    
    return test_id

def evaluate_ab_test(test_id, wait_hours=24):
    print(f"{wait_hours}時間後にA/Bテスト評価...")
    time.sleep(wait_hours * 3600)  # 実際は schedule で実行
    
    with sqlite3.connect(DB_PATH) as conn:
        results = pd.read_sql(
            "SELECT group_name, COUNT(DISTINCT customer_id) as opens FROM events WHERE test_id=? AND event_type='open' GROUP BY group_name",
            conn, params=(test_id,))
    
    opens_a = results[results["group_name"]=="A"]["opens"].values[0]
    opens_b = results[results["group_name"]=="B"]["opens"].values[0]
    rate_a  = opens_a / 500 * 100
    rate_b  = opens_b / 500 * 100
    winner  = "A版" if rate_a > rate_b else "B版"
    
    print(f"A版開封率: {rate_a:.1f}%  B版開封率: {rate_b:.1f}%")
    print(f"🏆 勝者: {winner}")
    return winner

subject_a = "【期間限定】30%OFFの特別ご優待"
subject_b = "3日間限定！あなたへの特別プランをご案内"
test_id = ab_test_send(emails_to_send, subject_a, subject_b)

STEP 5：バウンスメールを自動検出してリストをクリーニングする

送信エラーや配信不能アドレスを自動検出してリストから除外します。

def detect_bounces(service):
    results = service.users().messages().list(
        userId="me",
        q='from:(mailer-daemon@* OR postmaster@*) subject:(undelivered OR failure OR bounce) newer_than:1d'
    ).execute()
    
    bounce_emails = []
    for msg_ref in results.get("messages", []):
        detail = service.users().messages().get(userId="me", id=msg_ref["id"], format="full").execute()
        headers = {h["name"]: h["value"] for h in detail["payload"]["headers"]}
        
        # 本文から宛先メールアドレスを抽出
        body = ""
        if "parts" in detail["payload"]:
            for part in detail["payload"]["parts"]:
                if part["mimeType"] == "text/plain":
                    body = base64.urlsafe_b64decode(part["body"].get("data","")).decode(errors="ignore")
        
        # メールアドレスを正規表現で抽出
        import re
        found = re.findall(r"Final-Recipient:.*?<([^>]+)>|Original-Recipient:.*?<([^>]+)>", body)
        for match in found:
            email_addr = match[0] or match[1]
            if email_addr: bounce_emails.append(email_addr.lower())
    
    return list(set(bounce_emails))

bounces = detect_bounces(service)
print(f"バウンス検出: {len(bounces)}件")

# リストから除外
df = pd.read_csv("customers.csv")
df_clean = df[~df["メールアドレス"].str.lower().isin(bounces)]
df_clean.to_csv("customers_clean.csv", index=False)
print(f"リストクリーニング完了: {len(df)-len(df_clean)}件除外 → {len(df_clean)}件のリストに更新")

制作実績サンプル

このクオリティの資料が、あなたの業務でも作れます

Aurant Technologies が実際に制作したコンサルティング資料（全82枚）のサンプルです。Claude Code × AI自動化により、従来の1/5のコスト・時間で同等品質を実現します。

タイトル・ダーク背景

3カラム比較・分析

AI活用フロー図解

Before/After比較

構造化コンテンツ

→

全10枚を
見る

御社のテーマでこのレベルの資料を作成できます

制作費・納期・自動化プランは無料相談でご確認いただけます

📊 サンプル全10枚を見る →