Claude Codeに頼んだらGmailのメールテンプレートが全自動になった話

import csv

def load_contacts_from_csv(filepath: str) -> list[dict]:
    """CSVから顧客情報を読み込む"""
    contacts = []
    with open(filepath, 'r', encoding='utf-8') as f:
        reader = csv.DictReader(f)
        for row in reader:
            if not row.get('email'):
                print(f"⚠️  メールアドレスなし: {row.get('company', '不明')}")
                continue
            contacts.append({
                'company':    row.get('company', ''),
                'name':       row.get('name', ''),
                'title':      row.get('title', ''),
                'email':      row['email'],
                'pain_point': row.get('pain_point', ''),
                'industry':   row.get('industry', ''),
                'last_contact': row.get('last_contact', ''),
            })
    print(f"📂 {len(contacts)}件の連絡先を読み込みました")
    return contacts

import anthropic
import json

client = anthropic.Anthropic()

def generate_personalized_email(
    contact: dict,
    campaign: dict,
    tone: str = "丁寧"
) -> dict:
    """顧客情報をもとにClaudeがパーソナライズメールを生成する"""

    prompt = f"""以下の顧客情報をもとに営業メールを作成してください。

【顧客情報】
- 会社名: {contact['company']}
- 担当者名: {contact['name']}様
- 役職: {contact['title']}
- 業種: {contact['industry']}
- 課題・関心: {contact['pain_point']}
- 最終接触日: {contact.get('last_contact', '初回')}

【キャンペーン情報】
- テーマ: {campaign['theme']}
- 訴求事例: {campaign['case_study']}
- CTA: {campaign['cta']}

【トーン】{tone}
【文字数】件名15文字以内、本文200〜350文字

JSON形式で返答:
{{
  "subject": "件名",
  "body": "本文（改行\\n使用）"
}}"""

    message = client.messages.create(
        model="claude-opus-4-5",
        max_tokens=1024,
        messages=[{"role": "user", "content": prompt}]
    )
    return json.loads(message.content[0].text)

import time

def create_batch_drafts(
    contacts: list,
    campaign: dict,
    rate_limit: int = 3
) -> list:
    """全顧客のメールを生成しGmailの下書きとして保存する"""

    service = build_gmail_service()
    results = []

    for i, contact in enumerate(contacts):
        try:
            email_data = generate_personalized_email(contact, campaign)
            draft = create_gmail_draft(
                service,
                contact['email'],
                email_data['subject'],
                email_data['body']
            )
            results.append({
                'contact':  contact,
                'draft_id': draft['id'],
                'subject':  email_data['subject'],
                'status':   'success'
            })
            print(f"✅ [{i+1}/{len(contacts)}] {contact['company']} {contact['name']}様")
            time.sleep(1 / rate_limit)
        except Exception as e:
            print(f"❌ エラー: {contact['company']} — {e}")
            results.append({'contact': contact, 'status': 'error', 'error': str(e)})

    success = sum(1 for r in results if r['status'] == 'success')
    print(f"\n📊 完了: {success}/{len(contacts)}件成功")
    return results

def generate_ab_variants(contact: dict, campaign: dict) -> dict:
    """同一顧客向けにA/Bパターンを生成する"""
    variant_a = generate_personalized_email(
        contact, campaign, tone="課題訴求・共感型"
    )
    variant_b = generate_personalized_email(
        contact, campaign, tone="実績訴求・数値強調型"
    )
    return {'variant_a': variant_a, 'variant_b': variant_b}


def track_reply_rates(sent_log: list) -> dict:
    """送信ログから返信率を集計する"""
    service = build_gmail_service()
    stats = {'A': {'sent': 0, 'replied': 0},
             'B': {'sent': 0, 'replied': 0}}

    for log in sent_log:
        variant   = log['variant']
        thread_id = log['thread_id']
        stats[variant]['sent'] += 1
        thread = service.users().threads().get(
            userId='me', id=thread_id
        ).execute()
        if len(thread.get('messages', [])) > 1:
            stats[variant]['replied'] += 1

    for v in ['A', 'B']:
        rate = stats[v]['replied'] / max(stats[v]['sent'], 1) * 100
        print(f"パターン{v}: 返信率 {rate:.1f}% ({stats[v]['replied']}/{stats[v]['sent']}件)")
    return stats

import schedule, time, requests, datetime

SLACK_WEBHOOK = "https://hooks.slack.com/services/YOUR/SLACK/WEBHOOK"
CSV_PATH      = "contacts.csv"

def notify_slack(message: str) -> None:
    """Slackに完了通知を送る"""
    requests.post(SLACK_WEBHOOK, json={"text": message})

def weekly_batch_job() -> None:
    """毎週月曜日に自動実行されるジョブ"""
    print(f"[{datetime.datetime.now()}] 週次バッチ開始")

    contacts = load_contacts_from_csv(CSV_PATH)
    campaign = {
        'theme':      "Claude Code業務自動化",
        'case_study': "在庫管理の工数を月30時間削減した事例",
        'cta':        "まず30分のデモをご覧ください",
    }

    results = create_batch_drafts(contacts, campaign)
    success = sum(1 for r in results if r['status'] == 'success')

    notify_slack(
        f"✅ 週次メール生成完了\n"
        f"件数: {success}/{len(contacts)}件\n"
        f"実行日時: {datetime.datetime.now().strftime('%Y-%m-%d %H:%M')}"
    )

# 毎週月曜 9:00 に実行
schedule.every().monday.at("09:00").do(weekly_batch_job)

if __name__ == "__main__":
    print("⏰ スケジューラ起動中（毎週月曜 09:00 に実行）")
    while True:
        schedule.run_pending()
        time.sleep(60)