大規模データ収集のためのスケーラブルな非同期Webクローラーの構築

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

import aiohttp
import asyncio
from bs4 import BeautifulSoup
from urllib.parse import urljoin, urlparse

class AsyncCrawler:
    def __init__(self, max_concurrent=100, delay=1.0):
        self.max_concurrent = max_concurrent
        self.delay = delay
        self.session = None
        self.visited = set()
        self.queue = asyncio.Queue()

    async def fetch(self, url):
        """単一URLを非同期でフェッチ"""
        try:
            async with self.session.get(url, timeout=10) as response:
                if response.status == 200:
                    return await response.text()
                else:
                    print(f"Error {response.status}: {url}")
                    return None
        except asyncio.TimeoutError:
            print(f"Timeout: {url}")
            return None
        except Exception as e:
            print(f"Error fetching {url}: {e}")
            return None

    async def parse(self, html, base_url):
        """HTMLからすべてのリンクを抽出"""
        soup = BeautifulSoup(html, 'html.parser')
        links = []

        for link in soup.find_all('a', href=True):
            href = link['href']
            absolute_url = urljoin(base_url, href)
            links.append(absolute_url)

        return links

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22

from urllib.robotparser import RobotFileParser

class RobotsChecker:
    def __init__(self):
        self.parsers = {}

    async def can_fetch(self, url, user_agent='*'):
        """robots.txtに基づきURLのクロール可否を確認"""
        parsed = urlparse(url)
        robots_url = f"{parsed.scheme}://{parsed.netloc}/robots.txt"

        if robots_url not in self.parsers:
            parser = RobotFileParser()
            parser.set_url(robots_url)
            try:
                parser.read()
                self.parsers[robots_url] = parser
            except:
                # robots.txtが存在しない場合、クロールを許可
                return True

        return self.parsers[robots_url].can_fetch(user_agent, url)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

import time

class RateLimiter:
    def __init__(self, requests_per_second=10):
        self.delay = 1.0 / requests_per_second
        self.last_request = {}

    async def wait(self, domain):
        """ドメインごとにレート制限を適用"""
        now = time.time()

        if domain in self.last_request:
            elapsed = now - self.last_request[domain]
            if elapsed < self.delay:
                await asyncio.sleep(self.delay - elapsed)

        self.last_request[domain] = time.time()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

from playwright.async_api import async_playwright

async def fetch_with_js(url):
    """JavaScriptレンダリング付きでURLをフェッチ"""
    async with async_playwright() as p:
        browser = await p.chromium.launch()
        page = await browser.new_page()

        try:
            await page.goto(url, wait_until='networkidle')
            content = await page.content()
            return content
        finally:
            await browser.close()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41

import sqlite3

class CrawlDatabase:
    def __init__(self, db_path='crawler.db'):
        self.conn = sqlite3.connect(db_path)
        self.create_tables()

    def create_tables(self):
        """データベーススキーマの作成"""
        self.conn.execute('''
            CREATE TABLE IF NOT EXISTS urls (
                id INTEGER PRIMARY KEY,
                url TEXT UNIQUE,
                status INTEGER,
                content_type TEXT,
                crawled_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        ''')

        self.conn.execute('''
            CREATE TABLE IF NOT EXISTS links (
                source_url TEXT,
                target_url TEXT,
                anchor_text TEXT,
                FOREIGN KEY (source_url) REFERENCES urls(url)
            )
        ''')

        self.conn.commit()

    def add_url(self, url, status, content_type):
        """クロールしたURLを保存"""
        try:
            self.conn.execute(
                'INSERT INTO urls (url, status, content_type) VALUES (?, ?, ?)',
                (url, status, content_type)
            )
            self.conn.commit()
        except sqlite3.IntegrityError:
            # URLは既に存在
            pass

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48

async def crawl_website(start_url, max_pages=1000):
    """メインのクローリングパイプライン"""
    crawler = AsyncCrawler(max_concurrent=50)
    robots = RobotsChecker()
    limiter = RateLimiter(requests_per_second=5)
    db = CrawlDatabase()

    # セッションの初期化
    async with aiohttp.ClientSession() as session:
        crawler.session = session
        await crawler.queue.put(start_url)

        pages_crawled = 0

        while not crawler.queue.empty() and pages_crawled < max_pages:
            url = await crawler.queue.get()

            # 訪問済みかチェック
            if url in crawler.visited:
                continue

            # robots.txtのチェック
            if not await robots.can_fetch(url):
                print(f"Blocked by robots.txt: {url}")
                continue

            # レート制限
            domain = urlparse(url).netloc
            await limiter.wait(domain)

            # フェッチとパース
            html = await crawler.fetch(url)
            if html:
                crawler.visited.add(url)
                pages_crawled += 1

                # リンクの抽出
                links = await crawler.parse(html, url)

                # 新しいリンクをキューに追加
                for link in links:
                    if link not in crawler.visited:
                        await crawler.queue.put(link)

                # データベースに保存
                db.add_url(url, 200, 'text/html')

                print(f"Crawled {pages_crawled}/{max_pages}: {url}")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

class DiskQueue:
    def __init__(self, db_path='queue.db'):
        self.conn = sqlite3.connect(db_path)
        self.conn.execute('''
            CREATE TABLE IF NOT EXISTS queue (
                url TEXT PRIMARY KEY,
                priority INTEGER DEFAULT 0,
                added_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        ''')

    async def put(self, url, priority=0):
        self.conn.execute(
            'INSERT OR IGNORE INTO queue (url, priority) VALUES (?, ?)',
            (url, priority)
        )
        self.conn.commit()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

from urllib.parse import urlparse, urlunparse

def normalize_url(url):
    """重複を防ぐためのURL正規化"""
    parsed = urlparse(url)

    # HTTPSを強制
    scheme = 'https'

    # wwwプレフィックスを除去
    netloc = parsed.netloc.replace('www.', '')

    # 末尾スラッシュを除去
    path = parsed.path.rstrip('/')

    # デフォルトポートを除去
    netloc = netloc.replace(':80', '').replace(':443', '')

    # 再構築
    return urlunparse((scheme, netloc, path, '', parsed.query, ''))