Webスクレイピング 15 分で読めます

Twitter/Xデータをプロキシでスクレイピングする完全ガイド

2023年のAPI制限以降、多くの開発チームがWebスクレイピングに移行しています。本ガイドでは、住宅用プロキシを使用してXの公開データを収集する方法、レート制限への対処法、法的枠組みを解説します。

ProxyHat Team
Twitter/Xデータをプロキシでスクレイピングする完全ガイド

重要な免責事項： 本記事は、各プラットフォームの利用規約（Terms of Service）および適用法（米国のCFAA、EUのGDPRなど）を遵守した上で、正当な目的のために公開データにアクセスする開発者向けです。スクレイピングを行う前に、必ず対象サイトの規約を確認し、公式APIが提供されている場合は優先的に検討してください。

X/Twitter API制限後の状況：なぜスクレイピングが必要になったのか

2023年、X（旧Twitter）はAPIの提供体系を劇的に変更しました。かつて無料で利用できた検索エンドポイントや投稿取得機能は、現在では有料プランでのみ提供されています。月額100ドルのBasicプランでも、月間10,000投稿の取得制限があり、多くの開発チームやスタートアップにとって実用的ではありません。

この変更により、以下のような用途を持つチームがWebスクレイピングに移行を迫られています：

  • ソーシャルリスニング・センチメント分析ダッシュボード
  • ブランドモニタリング・評判管理ツール
  • トレンド追跡・市場調査プラットフォーム
  • ジャーナリズム・学術研究プロジェクト

しかし、Xはスクレイピングに対して非常に攻撃的な対策を講じています。データセンターIPのブロック、レート制限の強化、ログインなしのセッションへの厳しい制限などが実装されています。これらを回避し、安定してデータを収集するには、住宅用プロキシ（Residential Proxies）の使用が不可欠です。

公開Web経由でアクセス可能なデータとログイン必須のデータ

XのWebインターフェースは、ログインなしでも多くのデータにアクセスできます。ただし、ログイン状態と非ログイン状態では、アクセスできるデータとレート制限に大きな違いがあります。

ログインなしでアクセス可能なデータ

  • ユーザープロフィール：ユーザー名、バイオ、フォロワー数、アカウント作成日
  • 公開ツイート：投稿本文、画像、動画、リンク、リツイート数、いいね数
  • ツイートの返信スレッド：返信の一覧と内容
  • トレンド：地域ごとのトレンドトピック
  • 検索結果：キーワード検索、ハッシュタグ検索の結果（制限あり）

ログインが必要なデータ

  • 「For You」タイムライン：パーソナライズされたおすすめフィード
  • 通知：メンション、いいね、新しいフォロワー
  • ダイレクトメッセージ：DMの内容
  • 保護されたアカウントのツイート：鍵付きアカウントの投稿
  • 詳細なエンゲージメントデータ：インプレッション数、クリック数など

スクレイピングの観点からは、ログインなしでアクセス可能な公開データに焦点を当てるのが最も安全で実用的です。ログインを伴うスクレイピングは、アカウントの停止リスクが高く、法的・倫理的な問題もより複雑になります。

なぜ住宅用プロキシが不可欠なのか

Xは、データセンターからのアクセスを非常に厳しく監視しています。Amazon AWS、Google Cloud、DigitalOceanなどの主要クラウドプロバイダーのIPレンジは、ほぼ自動的にフラグが立てられ、レート制限が早期に適用されます。

データセンタープロキシの問題点

  • IPレンジが公開されており、X側で容易にブロック可能
  • 同一IPからの大量リクエストが検出されやすい
  • ボット検出システムによる指紋（フィンガープリント）分析の対象
  • 429エラー（Too Many Requests）の発生頻度が極めて高い

住宅用プロキシの利点

住宅用プロキシは、実際のISP（インターネットサービスプロバイダー）を通じて接続された家庭用IPアドレスを使用します。これにより、スクレイピングのリクエストは「通常のユーザー」として認識されやすくなります。

  • 信頼性の高いIPプール：数百万のIPアドレスからなるプールでローテーション可能
  • 地理的ターゲティング：特定の国や都市からのアクセスを模倣
  • 低ブロック率：データセンターIPと比較してはるかに低い検出率
  • セッション管理：スティッキーセッションで同一IPを維持し、ログイン状態を保持可能
特徴 データセンタープロキシ 住宅用プロキシ
IPプールサイズ 数万〜数十万 数百万〜数千万
検出リスク 高い 低い
レート制限の厳しさ 非常に厳しい 標準的
コスト 低〜中 中〜高
Xスクレイピングへの適合性 不推奨 推奨

ProxyHatの住宅用プロキシを使用する場合、以下の接続設定でアクセスできます：

# HTTP プロキシ（デフォルト）
http://USERNAME:PASSWORD@gate.proxyhat.com:8080

# 地理ターゲティング例（米国）
http://user-country-US:PASSWORD@gate.proxyhat.com:8080

# スティッキーセッション（同一IPを維持）
http://user-session-myquery123:PASSWORD@gate.proxyhat.com:8080

Python + Playwrightでの実装例

Xは現在、シングルページアプリケーション（SPA）として構築されており、データはGraphQLエンドポイントからJSON形式で取得され、JavaScriptによって動的にレンダリングされます。以下の例では、Playwrightを使用してブラウザを自動化し、住宅用プロキシ経由でデータを取得します。

基本的なセットアップ

import asyncio
from playwright.async_api import async_playwright
import json

# ProxyHatの住宅用プロキシ設定
PROXY_CONFIG = {
    "server": "http://gate.proxyhat.com:8080",
    "username": "user-country-US",  # 米国の住宅用IPを使用
    "password": "YOUR_PASSWORD"
}

async def scrape_user_profile(username: str):
    """ユーザープロフィールをスクレイピング"""
    async with async_playwright() as p:
        browser = await p.chromium.launch(
            proxy=PROXY_CONFIG,
            headless=True
        )
        context = await browser.new_context(
            user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
        )
        page = await context.new_page()
        
        # GraphQLリクエストをインターセプト
        graphql_responses = []
        
        async def capture_graphql(response):
            if "graphql" in response.url and "UserBy" in response.url:
                try:
                    data = await response.json()
                    graphql_responses.append(data)
                except:
                    pass
        
        page.on("response", capture_graphql)
        
        # ユーザーページに移動
        url = f"https://x.com/{username}"
        await page.goto(url, wait_until="networkidle")
        
        # データが読み込まれるまで待機
        await page.wait_for_timeout(3000)
        
        await browser.close()
        
        if graphql_responses:
            return graphql_responses[0]
        return None

# 実行
result = asyncio.run(scrape_user_profile("elonmusk"))
print(json.dumps(result, indent=2))

検索結果のスクレイピング

import asyncio
from playwright.async_api import async_playwright
import json
import random

PROXY_CONFIG = {
    "server": "http://gate.proxyhat.com:8080",
    "username": "user-country-US",
    "password": "YOUR_PASSWORD"
}

async def scrape_search(query: str, max_scroll: int = 5):
    """検索結果をスクレイピング"""
    async with async_playwright() as p:
        browser = await p.chromium.launch(
            proxy=PROXY_CONFIG,
            headless=True
        )
        context = await browser.new_context(
            viewport={"width": 1280, "height": 720},
            user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
        )
        page = await context.new_page()
        
        tweets = []
        
        async def capture_tweets(response):
            if "SearchTimeline" in response.url:
                try:
                    data = await response.json()
                    # JSONからツイートデータを抽出
                    entries = data.get("data", {}).get("search_by_query", {}).get("timeline", {}).get("instructions", [{}])[0].get("entries", [])
                    for entry in entries:
                        if entry.get("entryId", "").startswith("tweet-"):
                            tweet_data = entry.get("content", {}).get("itemContent", {}).get("tweet_results", {}).get("result", {})
                            if tweet_data:
                                tweets.append({
                                    "id": tweet_data.get("rest_id"),
                                    "text": tweet_data.get("legacy", {}).get("full_text"),
                                    "created_at": tweet_data.get("legacy", {}).get("created_at"),
                                    "likes": tweet_data.get("legacy", {}).get("favorite_count"),
                                    "retweets": tweet_data.get("legacy", {}).get("retweet_count")
                                })
                except Exception as e:
                    print(f"Error capturing tweets: {e}")
        
        page.on("response", capture_tweets)
        
        # 検索ページに移動
        search_url = f"https://x.com/search?q={query}&src=typed_query"
        await page.goto(search_url, wait_until="networkidle")
        
        # スクロールで追加データを読み込み
        for i in range(max_scroll):
            await page.evaluate("window.scrollTo(0, document.body.scrollHeight)")
            await page.wait_for_timeout(random.randint(2000, 4000))
        
        await browser.close()
        return tweets

# 実行例
results = asyncio.run(scrape_search("AI%20artificial%20intelligence"))
print(f"取得したツイート数: {len(results)}")
for tweet in results[:5]:
    print(f"\n@{tweet['id']}: {tweet['text'][:100]}...")

IPローテーション付きの実装

import asyncio
from playwright.async_api import async_playwright
import random

def get_rotating_proxy():
    """リクエストごとに新しいIPを取得"""
    session_id = f"session-{random.randint(100000, 999999)}"
    return {
        "server": "http://gate.proxyhat.com:8080",
        "username": f"user-country-US-session-{session_id}",
        "password": "YOUR_PASSWORD"
    }

async def scrape_with_rotation(usernames: list):
    """複数ユーザーをIPローテーション付きでスクレイピング"""
    results = []
    
    for username in usernames:
        proxy = get_rotating_proxy()
        
        async with async_playwright() as p:
            browser = await p.chromium.launch(proxy=proxy, headless=True)
            page = await browser.new_page()
            
            try:
                await page.goto(f"https://x.com/{username}", timeout=30000)
                await page.wait_for_selector("[data-testid='UserName']", timeout=10000)
                
                # プロフィール情報を抽出
                name = await page.locator("[data-testid='UserName']").inner_text()
                bio = await page.locator("[data-testid='UserDescription']").inner_text() if await page.locator("[data-testid='UserDescription']").count() > 0 else ""
                
                results.append({
                    "username": username,
                    "name": name,
                    "bio": bio
                })
                
                print(f"✓ {username}: 成功")
                
            except Exception as e:
                print(f"✗ {username}: エラー - {e}")
            
            finally:
                await browser.close()
        
        # 礼儀的な遅延
        await asyncio.sleep(random.uniform(2, 5))
    
    return results

# 実行
usernames = ["nasa", "spacex", "openai", "google", "microsoft"]
results = asyncio.run(scrape_with_rotation(usernames))

レート制限への対処法

Xは、スクレイピングを検出・防止するために複数のレート制限メカニズムを実装しています。これらを理解し、適切に対処することが重要です。

429エラー（Too Many Requests）

最も一般的なエラーは、HTTP 429ステータスコードです。これは、特定のIPアドレスまたはアカウントが短時間に過剰なリクエストを行ったことを示しています。

import asyncio
from playwright.async_api import async_playwright

async def scrape_with_retry(url: str, max_retries: int = 3):
    """リトライロジック付きスクレイピング"""
    proxy = {
        "server": "http://gate.proxyhat.com:8080",
        "username": "user-country-US",
        "password": "YOUR_PASSWORD"
    }
    
    for attempt in range(max_retries):
        async with async_playwright() as p:
            browser = await p.chromium.launch(proxy=proxy, headless=True)
            page = await browser.new_page()
            
            response = await page.goto(url, wait_until="domcontentloaded")
            
            if response.status == 429:
                # レート制限に達した場合
                retry_after = int(response.headers.get("retry-after", 60))
                print(f"レート制限検出。{retry_after}秒待機...")
                await browser.close()
                await asyncio.sleep(retry_after)
                continue
            
            if response.status == 200:
                content = await page.content()
                await browser.close()
                return content
            
            await browser.close()
    
    raise Exception("最大リトライ回数を超過")

スライディングウィンドウ検出

Xは、固定時間ウィンドウではなく、スライディングウィンドウアルゴリズムを使用してレート制限を適用する場合があります。これは、過去N分間のリクエスト数を継続的に監視することを意味します。

  • 戦略：リクエスト間にランダムな遅延を追加（2〜5秒）
  • 戦略：1時間あたりのリクエスト数を事前に制限
  • 戦略：複数のIPアドレスで負荷を分散

IPレベル vs アカウントレベルの制限

ログインなしでスクレイピングする場合、制限は主にIPアドレスレベルで適用されます。しかし、ログインを伴う場合は、アカウントレベルの制限も考慮する必要があります。

制限タイプ 影響範囲 対処法
IPレベル ログインなしセッション 住宅用プロキシでローテーション
アカウントレベル ログイン済みセッション 複数アカウントの使用（リスクあり）
フィンガープリント ブラウザ/デバイス Playwrightのステルス設定
行動分析 リクエストパターン 人間らしい遅延と間隔

法的枠組みと倫理的考慮事項

Xの利用規約（Terms of Service）

Xの利用規約では、明示的に「自動化された手段によるデータ収集」を制限しています。ただし、利用規約違反は必ずしも違法ではなく、民事契約上の問題にとどまる場合もあります。しかし、以下の点に注意が必要です：

  • CFAA（Computer Fraud and Abuse Act）：米国の連邦法。認証されていないシステムへのアクセスを禁止。ただし、公開データへのアクセスについては解釈が分かれる。
  • GDPR：EUの個人情報保護規則。個人データの収集・処理には同意が必要な場合がある。
  • hiQ Labs v. LinkedIn：米国連邦控訴裁判所の判決。公開データのスクレイピングはCFAA違反にならない可能性を示唆。

スクレイピングが適切でないケース

  • 公式APIが利用可能：必要なデータがAPIで取得できる場合、スクレイピングは不要
  • 大量のデータ収集：サーバーに過度な負荷をかける行為は法的リスクが高い
  • 個人データの収集：GDPRやCCPAの対象となる可能性
  • 競合他社のデータ：商業的秘密や著作権の問題が生じる可能性

公式APIを使用すべきケース

Xは現在、以下の有料APIプランを提供しています：

  • Basic（月額$100）：月間10,000投稿の読み取り
  • Pro（月額$5,000）：月間100,000投稿の読み取り
  • Enterprise：カスタム価格、無制限に近いアクセス

予算が許容し、公式APIで要件を満たせる場合は、APIの使用を強くお勧めします。安定性、法的リスクの低減、データの品質という点で優れています。

Node.jsでの実装例

const { chromium } = require('playwright');

const proxyConfig = {
  server: 'http://gate.proxyhat.com:8080',
  username: 'user-country-US',
  password: 'YOUR_PASSWORD'
};

async function scrapeUserProfile(username) {
  const browser = await chromium.launch({
    proxy: proxyConfig,
    headless: true
  });
  
  const context = await browser.newContext({
    userAgent: 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
  });
  
  const page = await context.newPage();
  
  const tweets = [];
  
  // GraphQLレスポンスをキャプチャ
  page.on('response', async (response) => {
    if (response.url().includes('UserTweets')) {
      try {
        const data = await response.json();
        // ツイートデータを抽出
        console.log('GraphQL response captured');
      } catch (e) {}
    }
  });
  
  await page.goto(`https://x.com/${username}`, { waitUntil: 'networkidle' });
  await page.waitForTimeout(3000);
  
  await browser.close();
  return tweets;
}

// 実行
scrapeUserProfile('nasa').then(console.log).catch(console.error);

Key Takeaways：重要なポイント

要点まとめ：

  • XのAPI制限により、多くのチームがWebスクレイピングに移行している
  • データセンタープロキシはXの検出システムにより高確率でブロックされる
  • 住宅用プロキシは実際のISP IPを使用し、検出リスクを大幅に低減
  • ログインなしでアクセス可能な公開データに焦点を当てるのが最も安全
  • 429エラーにはリトライロジックとIPローテーションで対処
  • 利用規約と適用法を理解し、公式APIの使用も検討する

まとめと次のステップ

X/Twitterの公開データをスクレイピングするには、住宅用プロキシの使用がほぼ必須です。データセンタープロキシでは、高いブロック率と頻繁な429エラーに直面し、効率的なデータ収集は困難です。

ProxyHatの住宅用プロキシを使用することで、以下のメリットが得られます：

  • 数百万の住宅用IPプールへのアクセス
  • 地理的ターゲティングによる地域固有のデータ取得
  • スティッキーセッションによるログイン状態の維持
  • 高い成功率と低いブロック率

次のステップとして、ProxyHatの料金プランを確認し、プロジェクトに適したプロキシパッケージを選択してください。また、Webスクレイピングのユースケースページで、他のスクレイピングプロジェクトの事例もご覧ください。

始める準備はできましたか？

AIフィルタリングで148か国以上、5,000万以上のレジデンシャルIPにアクセス。

料金を見るレジデンシャルプロキシ
← ブログに戻る