大量リクエストによるブロックからの保護: テクニックとツール

アカウントやIPアドレスのブロックは、パース、自動化、ソーシャルメディアでの大量操作における主な問題です。現代のアンチボットシステムは、リクエストの頻度からブラウザのフィンガープリンティングまで、数十のパラメータを分析します。このガイドでは、自動化検出の具体的なメカニズムとそれを回避するための実用的な方法を解説します。

自動化検出メカニズム

現代の保護システムは、ボットを特定するための多層分析を使用しています。これらのメカニズムを理解することは、正しい回避戦略を選択するために非常に重要です。

分析の主要パラメータ

リクエストの頻度 (Request Rate): 人間は物理的に1分間に100リクエストを送信することはできません。システムは、全体の数だけでなく、時間の分布も分析します — リクエスト間の均等な間隔はボットを示します。

行動パターン: 行動のシーケンス、スクロールの深さ、マウスの動き、ページ上の滞在時間。リンクを遅延なしに瞬時に移動するボットは、簡単に認識されます。

技術的フィンガープリンティング: User-Agent、HTTPヘッダー、ヘッダーの順序、TLSフィンガープリンティング、Canvas/WebGLフィンガープリンティング。これらのパラメータの不一致は、アンチボットシステムにとって赤信号です。

レート制限とリクエスト頻度の管理

リクエストの送信速度の管理は、ブロックからの最初の防御線です。プロキシのローテーションを使用しても、あまりにも攻撃的なパースはバンにつながります。

動的遅延

固定間隔（例えば、リクエスト間にちょうど2秒）では簡単に認識されます。正規分布を使用したランダムな遅延を使用してください:

import time
import random
import numpy as np

def human_delay(min_delay=1.5, max_delay=4.0, mean=2.5, std=0.8):
    """
    人間の行動を模倣する
    正規分布に基づく遅延の生成
    """
    delay = np.random.normal(mean, std)
    # 範囲を制限
    delay = max(min_delay, min(delay, max_delay))
    
    # 現実味を加えるためにマイクロ遅延を追加
    delay += random.uniform(0, 0.3)
    
    time.sleep(delay)

# 使用例
for url in urls:
    response = session.get(url)
    human_delay(min_delay=2, max_delay=5, mean=3, std=1)

アダプティブレート制限

より高度なアプローチは、サーバーの応答に基づいて速度を調整することです。429（Too Many Requests）や503のコードを受け取った場合は、自動的にペースを落とします:

class AdaptiveRateLimiter:
    def __init__(self, initial_delay=2.0):
        self.current_delay = initial_delay
        self.min_delay = 1.0
        self.max_delay = 30.0
        self.error_count = 0
        
    def wait(self):
        time.sleep(self.current_delay + random.uniform(0, 0.5))
        
    def on_success(self):
        # 成功したリクエストの際に徐々に加速
        self.current_delay = max(
            self.min_delay, 
            self.current_delay * 0.95
        )
        self.error_count = 0
        
    def on_rate_limit(self):
        # ブロック時に急激に減速
        self.error_count += 1
        self.current_delay = min(
            self.max_delay,
            self.current_delay * (1.5 + self.error_count * 0.5)
        )
        print(f"レート制限に達しました。新しい遅延: {self.current_delay:.2f}s")

# 使用例
limiter = AdaptiveRateLimiter(initial_delay=2.0)

for url in urls:
    limiter.wait()
    response = session.get(url)
    
    if response.status_code == 429:
        limiter.on_rate_limit()
        time.sleep(60)  # 再試行前の待機
    elif response.status_code == 200:
        limiter.on_success()
    else:
        # その他のエラー処理
        pass

プロキシのローテーションとIPアドレスの管理

大量リクエストに1つのIPアドレスを使用すると、ブロックが保証されます。プロキシのローテーションは負荷を分散させ、検出リスクを低下させます。

ローテーション戦略

1. リクエストごとのローテーション: 各リクエストまたはNリクエストごとにIPを変更します。これは、各リクエストの匿名性が重要な検索エンジンのパースに適しています。

2. 時間ごとのローテーション: 5-15分ごとにIPを変更します。これは、セッションの安定性が重要なソーシャルメディアで効果的です。

3. スティッキーセッション: ユーザーセッション全体（認証、行動のシーケンス）に1つのIPを使用します。これは、CSRFからの保護があるサイトにとって重要です。

import requests
from itertools import cycle

class ProxyRotator:
    def __init__(self, proxy_list, rotation_type='request', rotation_interval=10):
        """
        rotation_type: 'request' (各リクエスト) または 'time' (時間ごと)
        rotation_interval: リクエスト数または秒数
        """
        self.proxies = cycle(proxy_list)
        self.current_proxy = next(self.proxies)
        self.rotation_type = rotation_type
        self.rotation_interval = rotation_interval
        self.request_count = 0
        self.last_rotation = time.time()
        
    def get_proxy(self):
        if self.rotation_type == 'request':
            self.request_count += 1
            if self.request_count >= self.rotation_interval:
                self.current_proxy = next(self.proxies)
                self.request_count = 0
                print(f"ローテーション: {self.current_proxy}")
                
        elif self.rotation_type == 'time':
            if time.time() - self.last_rotation >= self.rotation_interval:
                self.current_proxy = next(self.proxies)
                self.last_rotation = time.time()
                print(f"ローテーション: {self.current_proxy}")
                
        return {'http': self.current_proxy, 'https': self.current_proxy}

# 使用例
proxy_list = [
    'http://user:pass@proxy1.example.com:8000',
    'http://user:pass@proxy2.example.com:8000',
    'http://user:pass@proxy3.example.com:8000',
]

rotator = ProxyRotator(proxy_list, rotation_type='request', rotation_interval=5)

for url in urls:
    proxies = rotator.get_proxy()
    response = requests.get(url, proxies=proxies, timeout=10)

プロキシの種類の選択

ソーシャルメディアや厳重な保護があるプラットフォームでの大量操作には、住宅用プロキシを使用してください。これらは通常の家庭用接続のように見え、ブラックリストに載ることは稀です。データセンターは、速度が重要なあまり保護されていないリソースに適しています。

ブラウザフィンガープリンティングとTLSフィンガープリンティング

IPのローテーションを行っても、ブラウザとTLS接続の技術的フィンガープリンティングによって特定される可能性があります。これらのパラメータは各クライアントに固有で、偽装が難しいです。

TLSフィンガープリンティング

HTTPS接続を確立する際、クライアントはサポートされている暗号と拡張機能のセットを含むClientHelloを送信します。この組み合わせは各ライブラリに固有です。例えば、PythonのrequestsはOpenSSLを使用しており、そのフィンガープリンティングはChromeとは簡単に区別できます。

解決策: ブラウザのTLSフィンガープリンティングを模倣するライブラリを使用してください。Pythonの場合、curl_cffi、tls_client、または完全なブラウザエミュレーションのためのplaywright/puppeteerが適しています。

# インストール: pip install curl-cffi
from curl_cffi import requests

# Chrome 110を模倣
response = requests.get(
    'https://example.com',
    impersonate="chrome110",
    proxies={'https': 'http://proxy:port'}
)

# 代替: tls_client
import tls_client

session = tls_client.Session(
    client_identifier="chrome_108",
    random_tls_extension_order=True
)

response = session.get('https://example.com')

HTTP/2フィンガープリンティング

TLSに加えて、アンチボットシステムはHTTP/2のパラメータを分析します: ヘッダーの順序、SETTINGSフレームの設定、ストリームの優先順位。標準ライブラリは、ChromeやFirefoxの正確なヘッダーの順序を遵守していません。

# Chrome用の正しいヘッダーの順序
headers = {
    ':method': 'GET',
    ':authority': 'example.com',
    ':scheme': 'https',
    ':path': '/',
    'sec-ch-ua': '"Not_A Brand";v="8", "Chromium";v="120"',
    'sec-ch-ua-mobile': '?0',
    'sec-ch-ua-platform': '"Windows"',
    'upgrade-insecure-requests': '1',
    'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64)...',
    'accept': 'text/html,application/xhtml+xml...',
    'sec-fetch-site': 'none',
    'sec-fetch-mode': 'navigate',
    'sec-fetch-user': '?1',
    'sec-fetch-dest': 'document',
    'accept-encoding': 'gzip, deflate, br',
    'accept-language': 'en-US,en;q=0.9',
}

CanvasとWebGLフィンガープリンティング

ブラウザは、GPU、ドライバ、OSに応じて異なる方法でグラフィックを描画します。サイトはこれを使用してデバイスのユニークなフィンガープリンティングを作成します。ヘッドレスブラウザ（Selenium、Puppeteer）を使用する際は、自動化の兆候を隠すことが重要です:

// Puppeteer: ヘッドレスモードの隠蔽
const puppeteer = require('puppeteer-extra');
const StealthPlugin = require('puppeteer-extra-plugin-stealth');

puppeteer.use(StealthPlugin());

const browser = await puppeteer.launch({
    headless: true,
    args: [
        '--disable-blink-features=AutomationControlled',
        '--no-sandbox',
        '--disable-setuid-sandbox',
        `--proxy-server=${proxyUrl}`
    ]
});

const page = await browser.newPage();

// navigator.webdriverのオーバーライド
await page.evaluateOnNewDocument(() => {
    Object.defineProperty(navigator, 'webdriver', {
        get: () => false,
    });
});

ヘッダー、クッキー、セッションの管理

HTTPヘッダーとクッキーの正しい取り扱いは、実際のユーザーを模倣するために重要です。これらのパラメータのエラーは、ブロックの一般的な原因です。

必須ヘッダー

Chromeブラウザを模倣するための最小限のヘッダーセット:

import requests

headers = {
    'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
    'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,*/*;q=0.8',
    'Accept-Language': 'en-US,en;q=0.9',
    'Accept-Encoding': 'gzip, deflate, br',
    'DNT': '1',
    'Connection': 'keep-alive',
    'Upgrade-Insecure-Requests': '1',
    'Sec-Fetch-Dest': 'document',
    'Sec-Fetch-Mode': 'navigate',
    'Sec-Fetch-Site': 'none',
    'Sec-Fetch-User': '?1',
    'Cache-Control': 'max-age=0',
}

session = requests.Session()
session.headers.update(headers)

クッキーの管理

多くのサイトは、最初の訪問時にトラッキングクッキーを設定し、その後のリクエストでそれらの存在を確認します。クッキーが存在しない、または不一致の場合はボットの兆候です。

import requests
import pickle

class SessionManager:
    def __init__(self, session_file='session.pkl'):
        self.session_file = session_file
        self.session = requests.Session()
        self.load_session()
        
    def load_session(self):
        """保存されたセッションの読み込み"""
        try:
            with open(self.session_file, 'rb') as f:
                cookies = pickle.load(f)
                self.session.cookies.update(cookies)
        except FileNotFoundError:
            pass
            
    def save_session(self):
        """再利用のためにクッキーを保存"""
        with open(self.session_file, 'wb') as f:
            pickle.dump(self.session.cookies, f)
            
    def request(self, url, **kwargs):
        response = self.session.get(url, **kwargs)
        self.save_session()
        return response

# 使用例
manager = SessionManager('instagram_session.pkl')
response = manager.request('https://www.instagram.com/explore/')

RefererとOrigin

ヘッダー Referer と Origin は、ユーザーがどこから来たのかを示します。これらが欠如しているか、無効な値がある場合は赤信号です。

# 正しいシーケンス: ホーム → カテゴリ → 商品
session = requests.Session()

# ステップ 1: ホームにアクセス
response = session.get('https://example.com/')

# ステップ 2: カテゴリに移動
response = session.get(
    'https://example.com/category/electronics',
    headers={'Referer': 'https://example.com/'}
)

# ステップ 3: 商品を閲覧
response = session.get(
    'https://example.com/product/12345',
    headers={'Referer': 'https://example.com/category/electronics'}
)

人間の行動の模倣

技術的なパラメータは、全体の半分に過ぎません。現代のアンチボットシステムは、ユーザーがページとどのように相互作用するか、どれくらいの時間を費やすか、マウスがどのように動くかといった行動パターンを分析します。

スクロールとマウスの動き

SeleniumやPuppeteerを使用する際は、ランダムなマウスの動きやページのスクロールを追加してください:

from selenium import webdriver
from selenium.webdriver.common.action_chains import ActionChains
import random
import time

def human_like_mouse_move(driver):
    """ページ上でのランダムなマウスの動き"""
    action = ActionChains(driver)
    
    for _ in range(random.randint(3, 7)):
        x = random.randint(0, 1000)
        y = random.randint(0, 800)
        action.move_by_offset(x, y)
        action.pause(random.uniform(0.1, 0.3))
    
    action.perform()

def human_like_scroll(driver):
    """自然なスクロールの模倣"""
    total_height = driver.execute_script("return document.body.scrollHeight")
    current_position = 0
    
    while current_position < total_height:
        # ランダムなスクロールステップ
        scroll_step = random.randint(100, 400)
        current_position += scroll_step
        
        driver.execute_script(f"window.scrollTo(0, {current_position});")
        
        # 変動を持たせた待機
        time.sleep(random.uniform(0.5, 1.5))
        
        # 時々少し戻る（人間がするように）
        if random.random() < 0.2:
            back_scroll = random.randint(50, 150)
            current_position -= back_scroll
            driver.execute_script(f"window.scrollTo(0, {current_position});")
            time.sleep(random.uniform(0.3, 0.8))

# 使用例
driver = webdriver.Chrome()
driver.get('https://example.com')

human_like_mouse_move(driver)
time.sleep(random.uniform(2, 4))
human_like_scroll(driver)

ページ上の時間

実際のユーザーはページ上で時間を過ごします: コンテンツを読み、画像を見ています。リンクを瞬時に移動するボットは簡単に認識されます。

def realistic_page_view(driver, url, min_time=5, max_time=15):
    """
    アクティビティを伴うリアルなページビュー
    """
    driver.get(url)
    
    # 初期遅延（読み込みと「読書」）
    time.sleep(random.uniform(2, 4))
    
    # スクロール
    human_like_scroll(driver)
    
    # 追加のアクティビティ
    total_time = random.uniform(min_time, max_time)
    elapsed = 0
    
    while elapsed < total_time:
        action_choice = random.choice(['scroll', 'mouse_move', 'pause'])
        
        if action_choice == 'scroll':
            # 少し上または下にスクロール
            scroll_amount = random.randint(-200, 300)
            driver.execute_script(f"window.scrollBy(0, {scroll_amount});")
            pause = random.uniform(1, 3)
            
        elif action_choice == 'mouse_move':
            human_like_mouse_move(driver)
            pause = random.uniform(0.5, 2)
            
        else:  # pause
            pause = random.uniform(2, 5)
        
        time.sleep(pause)
        elapsed += pause

ナビゲーションパターン

疑わしいパターンを避けてください: 深いページへの直接移動、ホームページの無視、すべての要素をスキップせずに順番に訪問すること。

Cloudflare、DataDome、その他の保護の回避

専門のアンチボットシステムは、包括的なアプローチを必要とします。これらは、JavaScriptチャレンジ、CAPTCHA、リアルタイムの行動分析を使用します。

Cloudflare

Cloudflareは、Browser Integrity Check、JavaScript Challenge、CAPTCHAなど、複数の保護レベルを使用しています。基本的な保護を回避するには、正しいTLSフィンガープリンティングとJavaScriptの実行が必要です:

# オプション 1: cloudscraper (JSチャレンジの自動解決)
import cloudscraper

scraper = cloudscraper.create_scraper(
    browser={
        'browser': 'chrome',
        'platform': 'windows',
        'desktop': True
    }
)

response = scraper.get('https://protected-site.com')

# オプション 2: undetected-chromedriver (複雑なケース用)
import undetected_chromedriver as uc

options = uc.ChromeOptions()
options.add_argument('--proxy-server=http://proxy:port')

driver = uc.Chrome(options=options)
driver.get('https://protected-site.com')

# チャレンジを通過するまで待機
time.sleep(5)

# requests用のクッキーを取得
cookies = driver.get_cookies()
session = requests.Session()
for cookie in cookies:
    session.cookies.set(cookie['name'], cookie['value'])

DataDome

DataDomeは、ユーザーの行動をリアルタイムで分析します: マウスの動き、キーボードの癖、タイミング。回避するには、アクティビティを模倣する完全なブラウザが必要です:

from playwright.sync_api import sync_playwright
import random

def bypass_datadome(url, proxy=None):
    with sync_playwright() as p:
        browser = p.chromium.launch(
            headless=False,  # DataDomeはヘッドレスを検出します
            proxy={'server': proxy} if proxy else None
        )
        
        context = browser.new_context(
            viewport={'width': 1920, 'height': 1080},
            user_agent='Mozilla/5.0 (Windows NT 10.0; Win64; x64)...'
        )
        
        page = context.new_page()
        
        # 自動化を隠すためのスクリプトのインジェクション
        page.add_init_script("""
            Object.defineProperty(navigator, 'webdriver', {get: () => false});
            window.chrome = {runtime: {}};
        """)
        
        page.goto(url)
        
        # 人間の行動を模倣
        time.sleep(random.uniform(2, 4))
        
        # ランダムなマウスの動き
        for _ in range(random.randint(5, 10)):
            page.mouse.move(
                random.randint(100, 1800),
                random.randint(100, 1000)
            )
            time.sleep(random.uniform(0.1, 0.3))
        
        # スクロール
        page.evaluate(f"window.scrollTo(0, {random.randint(300, 800)})")
        time.sleep(random.uniform(1, 2))
        
        content = page.content()
        browser.close()
        
        return content

CAPTCHA

CAPTCHAを自動的に解決するには、認識サービス（2captcha、Anti-Captcha）を使用するか、回避戦略を採用してください:

• CAPTCHAを引き起こさないレベルまでリクエストの頻度を下げる
• 良好な評判を持つクリーンな住宅用IPを使用する
• 認証されたアカウントを介して作業する（これらはCAPTCHAの閾値が高い）
• 時間を分散させて負荷を分散する（ピーク時間を避ける）

モニタリングとブロックの処理

最良の実践を行っても、ブロックは避けられません。迅速にそれを検出し、適切に処理することが重要です。

ブロックのインジケーター

リトライシステム

import requests
from requests.adapters import HTTPAdapter
from requests.packages.urllib3.util.retry import Retry

def create_session_with_retries():
    """
    自動リトライとエラー処理を備えたセッション
    """
    session = requests.Session()
    
    retry_strategy = Retry(
        total=5,
        backoff_factor=2,  # 2, 4, 8, 16, 32秒
        status_forcelist=[429, 500, 502, 503, 504],
        method_whitelist=["GET", "POST"]
    )
    
    adapter = HTTPAdapter(max_retries=retry_strategy)
    session.mount("http://", adapter)
    session.mount("https://", adapter)
    
    return session

def safe_request(url, session, max_attempts=3):
    """
    ブロック処理を伴うリクエスト
    """
    for attempt in range(max_attempts):
        try:
            response = session.get(url, timeout=15)
            
            # ブロックの確認
            if response.status_code == 403:
                print(f"IPがブロックされました。プロキシをローテーションします...")
                # プロキシ変更のロジック
                continue
                
            elif response.status_code == 429:
                wait_time = int(response.headers.get('Retry-After', 60))
                print(f"レート制限に達しました。{wait_time}s待機します...")
                time.sleep(wait_time)
                continue
                
            elif 'captcha' in response.text.lower():
                print("CAPTCHAが検出されました")
                # CAPTCHA解決のロジックまたはスキップ
                return None
                
            return response
            
        except requests.exceptions.Timeout:
            print(f"試行 {attempt + 1} でタイムアウト")
            time.sleep(5 * (attempt + 1))
            
        except requests.exceptions.ProxyError:
            print("プロキシエラー。ローテーションします...")
            # プロキシ変更
            continue
            
    return None

ロギングと分析

戦略を最適化するためにメトリクスを追跡してください:

import logging
from collections import defaultdict
from datetime import datetime

class ScraperMetrics:
    def __init__(self):
        self.stats = {
            'total_requests': 0,
            'successful': 0,
            'rate_limited': 0,
            'blocked': 0,
            'captcha': 0,
            'errors': 0,
            'proxy_failures': defaultdict(int)
        }
        
    def log_request(self, status, proxy=None):
        self.stats['total_requests'] += 1
        
        if status == 200:
            self.stats['successful'] += 1
        elif status == 429:
            self.stats['rate_limited'] += 1
        elif status == 403:
            self.stats['blocked'] += 1
            if proxy:
                self.stats['proxy_failures'][proxy] += 1
                
    def get_success_rate(self):
        if self.stats['total_requests'] == 0:
            return 0
        return (self.stats['successful'] / self.stats['total_requests']) * 100
        
    def print_report(self):
        print(f"\n=== スクレイピングレポート ===")
        print(f"総リクエスト数: {self.stats['total_requests']}")
        print(f"成功率: {self.get_success_rate():.2f}%")
        print(f"レート制限: {self.stats['rate_limited']}")
        print(f"ブロック: {self.stats['blocked']}")
        print(f"CAPTCHA: {self.stats['captcha']}")
        
        if self.stats['proxy_failures']:
            print(f"\n問題のあるプロキシ:")
            for proxy, count in sorted(
                self.stats['proxy_failures'].items(), 
                key=lambda x: x[1], 
                reverse=True
            )[:5]:
                print(f"  {proxy}: {count} 回の失敗")

# 使用例
metrics = ScraperMetrics()

for url in urls:
    response = safe_request(url, session)
    if response:
        metrics.log_request(response.status_code, current_proxy)
    
metrics.print_report()

結論

保護を回避するための戦略は、継続的な調整と改善が必要です。技術的なパラメータと行動パターンの両方を考慮し、効果的なアプローチを構築しましょう。成功率を高めるためには、常に最新の情報を追い、実践を重ねることが重要です。