每天15分钟玩转Scrapy - 动态页面

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

import scrapy

class QuotesJsStaticSpider(scrapy.Spider):
    """路线一：普通请求。页面里的名言是 JS 拼出来的，静态 HTML 里空空如也。"""

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

    def parse(self, response):
        self.logger.info("路线一 普通请求，div.quote 数量 = %d", len(response.css("div.quote")))

1

2026-09-12 23:44:30 [quotesjsstatic] INFO: 路线一 普通请求，div.quote 数量 = 0

 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

import json

import scrapy

class QuotesJsDataSpider(scrapy.Spider):
    """路线二：页面里既然有现成数据，就直接把 JSON 抠出来。"""

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

    def parse(self, response):
        script = response.xpath('//script[contains(text(), "var data")]/text()').get()
        if not script:
            self.logger.error("路线二 没找到内嵌数据源")
            return
        start = script.find("var data = [") + len("var data = ")
        data, _ = json.JSONDecoder().raw_decode(script[start:])
        self.logger.info("路线二 内嵌数据源，拿到 %d 条", len(data))
        for row in data:
            yield {
                "text": row["text"],
                "author": row["author"]["name"],
                "tags": row["tags"],
            }

1

2026-09-12 23:44:35 [quotesjsdata] INFO: 路线二 内嵌数据源，拿到 10 条

1
2
3

1. parsel .re() 跨行        -> None
2. 贪婪 .*  + json.loads    -> JSONDecodeError: Extra data: line 132 column 2 (char 4045)
3. 非贪婪 .*?               -> 成功 10 条

1
2

start = script.find("var data = [") + len("var data = ")
data, end = json.JSONDecoder().raw_decode(script[start:])

1
2

pip install scrapy-playwright
playwright install chromium

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

# 把 http/https 交给 playwright 的下载器去处理
DOWNLOAD_HANDLERS = {
    "http": "scrapy_playwright.handler.ScrapyPlaywrightDownloadHandler",
    "https": "scrapy_playwright.handler.ScrapyPlaywrightDownloadHandler",
}

PLAYWRIGHT_BROWSER_TYPE = "chromium"
PLAYWRIGHT_LAUNCH_OPTIONS = {"headless": True}
PLAYWRIGHT_DEFAULT_NAVIGATION_TIMEOUT = 30_000
PLAYWRIGHT_MAX_PAGES_PER_CONTEXT = 4
PLAYWRIGHT_MAX_CONTEXTS = 1

 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

import scrapy
from scrapy_playwright.page import PageMethod

class QuotesJsRenderSpider(scrapy.Spider):
    """路线三：交给浏览器把页面渲染好，再按普通选择器抓。"""

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

    async def start(self):
        for url in self.start_urls:
            yield scrapy.Request(
                url,
                meta={
                    "playwright": True,
                    "playwright_page_methods": [
                        # 等名言节点真的出现，别让框架拿到半成品页面
                        PageMethod("wait_for_selector", "div.quote"),
                    ],
                },
            )

    def parse(self, response):
        quotes = response.css("div.quote")
        self.logger.info("路线三 浏览器渲染后，div.quote 数量 = %d", len(quotes))
        for q in quotes:
            yield {
                "text": q.css("span.text::text").get(),
                "author": q.css("small.author::text").get(),
                "tags": q.css("div.tags a.tag::text").getall(),
            }

1
2
3
4
5
6

2026-09-12 23:44:24 [quotesjsrender] INFO: 路线三 浏览器渲染后，div.quote 数量 = 10
{'downloader/request_count': 2,
 'elapsed_time_seconds': 4.9159651000009035,
 'item_scraped_count': 10,
 'playwright/request_count': 7,
 'playwright/request_count/navigation': 1}

 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

import scrapy
from scrapy_playwright.page import PageMethod

class CatalogSpider(scrapy.Spider):
    """把几个常用的 PageMethod 摆在一起，看它们各自返回什么。"""

    name = "catalog"
    allowed_domains = ["quotes.toscrape.com"]

    async def start(self):
        # 这几个 PageMethod 对象建出来之后自己留着引用，
        # 跑完请求就能从 .result 上把返回值读回来
        title = PageMethod("title")
        count = PageMethod("evaluate", "document.querySelectorAll('div.quote').length")
        shot = PageMethod("screenshot", path="out/js_page.png", full_page=True)

        yield scrapy.Request(
            "https://quotes.toscrape.com/js/",
            meta={
                "playwright": True,
                "playwright_page_methods": [
                    # 页面加载完成 != 渲染完成，等节点真的出现再往下走
                    PageMethod("wait_for_selector", "div.quote"),
                    PageMethod("wait_for_load_state", "networkidle"),
                    title,
                    count,
                    shot,
                ],
            },
            callback=self.parse,
            cb_kwargs={"pms": {"title": title, "count": count, "shot": shot}},
        )

    def parse(self, response, pms):
        self.logger.info("page.title() 返回 -> %s", pms["title"].result)
        self.logger.info("page.evaluate() 返回 -> %s", pms["count"].result)
        self.logger.info(
            "page.screenshot() 返回 -> %s 类型，长度 %s",
            type(pms["shot"].result).__name__,
            len(pms["shot"].result or b""),
        )
        self.logger.info("同一个响应，选择器数到 %d 条", len(response.css("div.quote")))

1
2
3
4
5
6
7

2026-09-13 06:06:37 [catalog] INFO: page.title() 返回 -> Quotes to Scrape
2026-09-13 06:06:37 [catalog] INFO: page.evaluate() 返回 -> 10
2026-09-13 06:06:37 [catalog] INFO: page.screenshot() 返回 -> bytes 类型，长度 100833
2026-09-13 06:06:37 [catalog] INFO: 同一个响应，选择器数到 10 条
 'elapsed_time_seconds': 2.781685699999798,
 'item_scraped_count': 10,
 'playwright/request_count': 7,

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

async def scroll_to_bottom(page, rounds=6, pause_ms=700):
    """一屏一屏往下滚，直到名言条数不再增加为止，返回最终条数。"""
    total = 0
    for i in range(rounds):
        await page.evaluate("window.scrollTo(0, document.body.scrollHeight)")
        await page.wait_for_timeout(pause_ms)
        n = await page.evaluate("document.querySelectorAll('div.quote').length")
        print("   第 %d 次滚动后，页面上的名言数 = %d" % (i + 1, n))
        if n == total:
            # 这一轮没有新东西进来，说明到底了
            return n
        total = n
    return total

1
2
3
4
5
6
7

"""按资源类型拦截请求，给渲染提速。"""

BLOCKED_RESOURCE_TYPES = {"image", "font", "media"}

def should_abort(request):
    """返回 True 表示这次请求不发了。request 上带 resource_type。"""
    return request.resource_type in BLOCKED_RESOURCE_TYPES

1
2
3
4
5

class RenderLeanSpider(scrapy.Spider):
    name = "renderlean"
    custom_settings = {
        "PLAYWRIGHT_ABORT_REQUEST": "jsrender.abort.should_abort",
    }

1
2
3

2026-09-13 06:09:10 [uaprobe] INFO: 浏览器里 navigator.userAgent -> Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) HeadlessChrome/151.0.7922.34 Safari/537.36
2026-09-13 06:09:10 [uaprobe] INFO: 浏览器里 navigator.webdriver -> True
2026-09-13 06:09:10 [uaprobe] INFO: Scrapy 这一层的默认 UA -> b'Scrapy/2.19.0 (+https://scrapy.org)'

1
2
3
4
5
6
7
8

async def parse(self, response):
    page = response.meta["playwright_page"]
    try:
        await page.wait_for_selector("div.quote")
        # ... 解析
    finally:
        # 这一行不能漏，漏了页面不会回收，跑久了浏览器就卡住了
        await page.close()

1
2
3
4

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

1
2
3
4

[asyncio] ERROR: Task was destroyed but it is pending!
task: <Task pending name='Task-3' coro=<_ThreadedLoopAdapter._process_queue() running at ...scrapy_playwright\_loop.py:58> ...>
Exception ignored while closing generator <coroutine object _ThreadedLoopAdapter._process_queue ...>:
RuntimeError: Event loop is closed