Scrapyとは？Pythonクローリングフレームワークの全体像と実践的な使い方

pip install scrapy

python -m venv scrapy-env
source scrapy-env/bin/activate   # macOS / Linux
scrapy-env\Scripts\activate      # Windows
pip install scrapy

conda install -c conda-forge scrapy

scrapy version
# Scrapy 2.14.1

scrapy startproject myproject
cd myproject

myproject/
    scrapy.cfg
    myproject/
        __init__.py
        items.py        # データ構造の定義
        middlewares.py   # ミドルウェア
        pipelines.py     # データ保存処理
        settings.py      # 各種設定
        spiders/         # Spider格納ディレクトリ
            __init__.py

scrapy genspider sample_spider example.com

import scrapy


class BooksSpider(scrapy.Spider):
    name = "books"
    start_urls = ["https://books.toscrape.com/catalogue/category/books/fantasy_19/index.html"]

    def parse(self, response):
        for book in response.css("article.product_pod"):
            yield {
                "title": book.css("h3 a::attr(title)").get(),
                "price": book.css(".price_color::text").get(),
                "availability": book.css(".availability::text").getall()[-1].strip(),
            }

        # 次ページへのリンクをたどる
        next_page = response.css("li.next a::attr(href)").get()
        if next_page:
            yield response.follow(next_page, callback=self.parse)

scrapy crawl books -o books.json

scrapy crawl books -o books.csv      # CSV出力
scrapy crawl books -o books.jsonl    # JSON Lines出力

# リクエスト間隔（秒）
DOWNLOAD_DELAY = 1

# 同時並行リクエスト数
CONCURRENT_REQUESTS = 8

# 同一ドメインへの同時リクエスト上限
CONCURRENT_REQUESTS_PER_DOMAIN = 4

# robots.txtの遵守
ROBOTSTXT_OBEY = True

# User-Agentの設定
USER_AGENT = "myproject (+http://www.example.com)"

# レスポンスキャッシュ（開発時に便利）
HTTPCACHE_ENABLED = True
HTTPCACHE_EXPIRATION_SECS = 3600

# pipelines.py
class PriceValidationPipeline:
    def process_item(self, item, spider):
        price_str = item.get("price", "")
        if price_str.startswith("£"):
            item["price"] = float(price_str.replace("£", ""))
        return item


class DuplicateFilterPipeline:
    def __init__(self):
        self.seen_titles = set()

    def process_item(self, item, spider):
        title = item.get("title")
        if title in self.seen_titles:
            raise scrapy.exceptions.DropItem(f"Duplicate: {title}")
        self.seen_titles.add(title)
        return item

ITEM_PIPELINES = {
    "myproject.pipelines.DuplicateFilterPipeline": 100,
    "myproject.pipelines.PriceValidationPipeline": 200,
}

scrapy shell "https://books.toscrape.com"
>>> response.css("article.product_pod h3 a::attr(title)").getall()

pip install scrapy-playwright
playwright install chromium

DOWNLOAD_HANDLERS = {
    "http": "scrapy_playwright.handler.ScrapyPlaywrightDownloadHandler",
    "https": "scrapy_playwright.handler.ScrapyPlaywrightDownloadHandler",
}

TWISTED_REACTOR = "twisted.internet.asyncioreactor.AsyncioSelectorReactor"

import scrapy


class DynamicSpider(scrapy.Spider):
    name = "dynamic"
    start_urls = ["https://example.com/spa-page"]

    def start_requests(self):
        for url in self.start_urls:
            yield scrapy.Request(
                url,
                meta={"playwright": True, "playwright_include_page": True},
            )

    async def parse(self, response):
        page = response.meta["playwright_page"]
        await page.wait_for_selector(".loaded-content")
        # ページ内容が読み込まれた状態でresponseを取得
        content = await page.content()
        await page.close()

        sel = scrapy.Selector(text=content)
        for item in sel.css(".loaded-content .item"):
            yield {
                "name": item.css("::text").get(),
            }

scrapy shell "https://books.toscrape.com"

>>> response.css("title::text").get()
'All products | Books to Scrape - Sandbox'

>>> len(response.css("article.product_pod"))
20

>>> response.xpath('//article[@class="product_pod"]//h3/a/@title').getall()[:3]
['A Light in the Attic', 'Tipping the Velvet', 'Soumission']

pip install shub
shub deploy

def parse(self, response):
    # データ抽出処理
    for item in response.css(".item"):
        yield {"name": item.css("::text").get()}

    # 次ページへ遷移
    next_url = response.css("a.next::attr(href)").get()
    if next_url:
        yield response.follow(next_url, callback=self.parse)

scrapy crawl books -a category=fantasy -a max_pages=5

class BooksSpider(scrapy.Spider):
    name = "books"

    def __init__(self, category=None, max_pages=10, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.category = category
        self.max_pages = int(max_pages)

scrapy crawl books --loglevel WARNING

def start_requests(self):
    yield scrapy.FormRequest(
        "https://example.com/login",
        formdata={"user": "myuser", "pass": "mypass"},
        callback=self.after_login,
    )