Scrapyウェブスクレイピング: JavaScriptページを扱うスパイダーを構築する

python3 -m venv .venv
source .venv/bin/activate
pip install "scrapy==2.17.0" "twisted==26.4.0" "pyopenssl==25.3.0"

python3 -c "import importlib.metadata as m; print(m.version('scrapy'), m.version('twisted'), m.version('pyopenssl'))"

import scrapy


class QuotesSpider(scrapy.Spider):
    name = "quotes"
    start_urls = ["https://quotes.toscrape.com/"]

    def parse(self, response):
        for quote in response.css("div.quote"):
            yield {
                "text": quote.css("span.text::text").get(),
                "author": quote.css("small.author::text").get(),
                "tags": quote.css("div.tags a.tag::text").getall(),
            }

        next_page = response.css("li.next a::attr(href)").get()
        if next_page:
            yield response.follow(next_page, callback=self.parse)

scrapy runspider quotes_spider.py -O quotes.json \
  -s LOG_LEVEL=ERROR \
  -s CLOSESPIDER_PAGECOUNT=3

    start_urls = ["https://quotes.toscrape.com/js/"]

scrapy runspider quotes_spider.py -O js.json -s LOG_LEVEL=ERROR -s CLOSESPIDER_PAGECOUNT=1

import json
import os

from scrapy.http import HtmlResponse

UNLOCKER = "https://api.scrapeless.com/api/v2/unlocker/request"


class ScrapelessMiddleware:
    """各リクエストを上流でレンダリングし、その後Scrapyに通常のHtmlResponseを渡します。"""

    def __init__(self, token, country):
        self.token = token
        self.country = country

    @classmethod
    def from_crawler(cls, crawler):
        return cls(
            os.environ["SCRAPELESS_API_KEY"],
            crawler.settings.get("SCRAPELESS_PROXY_COUNTRY", "US"),
        )

    def process_request(self, request, spider):
        if request.meta.get("scrapeless"):
            return None
        payload = {
            "actor": "unlocker.webunlocker",
            "input": {
                "url": request.url,
                "proxy_country": self.country,
                "js_render": True,
            },
        }
        return request.replace(
            url=UNLOCKER,
            method="POST",
            body=json.dumps(payload),
            headers={"Content-Type": "application/json", "x-api-token": self.token},
            meta={**request.meta, "scrapeless": True, "origin_url": request.url},
            dont_filter=True,
        )

def process_response(self, request, response, spider):
        if not request.meta.get("scrapeless"):
            return response
        rendered = json.loads(response.text)["data"]
        return HtmlResponse(
            url=request.meta["origin_url"],
            body=rendered,
            encoding="utf-8",
            request=request,
        )

export SCRAPELESS_API_KEY="your_api_key"

PYTHONPATH=. scrapy runspider quotes_spider.py -O js_unlocked.json \
  -s LOG_LEVEL=ERROR \
  -s CLOSESPIDER_PAGECOUNT=1 \
  -s 'DOWNLOADER_MIDDLEWARES={"scrapeless_middleware.ScrapelessMiddleware": 543}'

import json
import os

import scrapy
from scrapy import signals
from scrapy.crawler import CrawlerProcess
from scrapy.http import HtmlResponse

UNLOCKER = "https://api.scrapeless.com/api/v2/unlocker/request"


class ScrapelessMiddleware:
    """各リクエストを上流でレンダリングし、Scrapyに通常のHtmlResponseを渡します。"""

    def __init__(self, token, country):
        self.token = token
        self.country = country

    @classmethod
    def from_crawler(cls, crawler):
        return cls(
            os.environ["SCRAPELESS_API_KEY"],
            crawler.settings.get("SCRAPELESS_PROXY_COUNTRY", "US"),
        )

    def process_request(self, request, spider):
        if request.meta.get("scrapeless"):
            return None
        payload = {
            "actor": "unlocker.webunlocker",
            "input": {"url": request.url, "proxy_country": self.country, "js_render": True},
        }
        return request.replace(
            url=UNLOCKER,
            method="POST",
            body=json.dumps(payload),
            headers={"Content-Type": "application/json", "x-api-token": self.token},
            meta={**request.meta, "scrapeless": True, "origin_url": request.url},
            dont_filter=True,
        )

    def process_response(self, request, response, spider):
        if not request.meta.get("scrapeless"):
            return response
        rendered = json.loads(response.text)["data"]
        return HtmlResponse(
            url=request.meta["origin_url"],
            body=rendered,
            encoding="utf-8",
            request=request,
        )


class QuotesSpider(scrapy.Spider):
    name = "quotes"

    def __init__(self, url, **kwargs):
        super().__init__(**kwargs)

self.start_urls = [url]

    def parse(self, response):
        for quote in response.css("div.quote"):
            yield {
                "text": quote.css("span.text::text").get(),
                "author": quote.css("small.author::text").get(),
                "tags": quote.css("div.tags a.tag::text").getall(),
            }


class ScrapelessQuotesSpider(QuotesSpider):
    name = "quotes-scrapeless"
    custom_settings = {"DOWNLOADER_MIDDLEWARES": {ScrapelessMiddleware: 543}}


def main():
    import importlib.metadata as md

    print(
        "scrapy", md.version("scrapy"),
        "| twisted", md.version("twisted"),
        "| pyopenssl", md.version("pyopenssl"),
    )

    jobs = [
        ("静的ページ、プレーンなScrapy", QuotesSpider, "https://quotes.toscrape.com/"),
        ("JavaScriptページ、プレーンなScrapy", QuotesSpider, "https://quotes.toscrape.com/js/"),
        ("JavaScriptページ、Scrapelessミドルウェア", ScrapelessQuotesSpider, "https://quotes.toscrape.com/js/"),
    ]
    collected = {label: [] for label, _, _ in jobs}

    # Scrapyは信号ハンドラーを弱い参照で保持するため、各ハンドラーへの強い参照を保持します。
    handlers = []

    def collector(label):
        def on_item(item, response, spider):
            collected[label].append(item)
        handlers.append(on_item)
        return on_item

    process = CrawlerProcess({
        "LOG_LEVEL": "ERROR",
        "SCRAPELESS_PROXY_COUNTRY": "US",
    })
    for label, spider_cls, url in jobs:
        crawler = process.create_crawler(spider_cls)
        crawler.signals.connect(collector(label), signal=signals.item_scraped)
        process.crawl(crawler, url=url)
    process.start()

    for label, _, _ in jobs:
        items = collected[label]
        print(f"{label}: {len(items)} アイテム")
        if items:
            print(f"  最初の著者: {items[0]['author']}")
            print(f"  最初のタグ: {items[0]['tags']}")

    print("両方のスパイダーに共有されるparseメソッド:", ScrapelessQuotesSpider.parse is QuotesSpider.parse)


if __name__ == "__main__":
    main()

scrapy 2.17.0 | twisted 26.4.0 | pyopenssl 25.3.0
静的ページ、プレーンなScrapy: 10 アイテム
  最初の著者: アルバート・アインシュタイン
  最初のタグ: ['change', 'deep-thoughts', 'thinking', 'world']
JavaScriptページ、プレーンなScrapy: 0 アイテム
JavaScriptページ、Scrapelessミドルウェア: 10 アイテム
  最初の著者: アルバート・アインシュタイン
  最初のタグ: ['change', 'deep-thoughts', 'thinking', 'world']
両方のスパイダーに共有されるparseメソッド: True