Spider 框架拆解：從並發爬取到暫停恢復、自動 Proxy 輪換與即時統計

在 Scrapling 系列的前幾篇文章裡，我們已經掌握了 Scrapling 的基礎 API 使用技巧，從如何選取元素、提取資料，到如何用不同類型的 Fetcher Session 應對各種網站場景。然而，當我們需要大規模、有組織地爬取數百甚至數千個頁面時，單純依靠 Session 搭配迴圈的方式很快就會遇到瓶頸：如何管理並發請求？如何在爬蟲中途斷線後優雅恢復？如何在同一套爬蟲中靈活切換不同的 Session 策略？又如何即時掌握爬蟲的運行健康狀態？本文以撰稿時 PyPI 上的最新版本 Scrapling 0.4.15 為準，程式範例都對照官方文件調整過。

▲ D4Vinci/Scrapling 的 GitHub 專案首頁，README 說明安裝方式、支援的抓取引擎與基本用法。

這正是 Scrapling Spider 框架要解決的核心問題。Spider 是 Scrapling 提供的一套類 Scrapy 的爬蟲框架，它將爬蟲從一次性腳本提升為可維護、可監控、可恢復的工程化系統。本文將從頭拆解 Spider 的每個核心機制，並透過實戰範例帶你全面掌握這個強大的框架。

一、Scrapy-like API：熟悉的設計，極低的學習門檻

如果你是 Python 爬蟲開發者，幾乎不可能沒有聽過 Scrapy 這個誕生於 2008 年的爬蟲框架，它至今仍是 Python 生態系中最受歡迎的爬蟲工具之一。Scrapling Spider 在設計哲學上向 Scrapy 致敬，採用了非常相似的 API 風格，讓曾有 Scrapy 使用經驗的開發者幾乎可以無痛轉移。

▲ D4Vinci/Scrapling 的 Releases 分頁，列出正式版本與更新說明，方便核對文中提到的版號。

1.1 Spider 類別的基本結構

一個典型的 Scrapling Spider 繼承自 Spider 基底類別，並且一定要指定 name 作為識別。方法只需要兩個：負責生成初始 Request 的 start_requests（也可以直接設定 start_urls 讓框架代勞），以及負責處理伺服器回傳 Response、從中提取資料或產生新 Request 的 async def parse。整體結構與 Scrapy 的設計如出一轍，差別在於 parse 是原生非同步函式。

讓我們來看一個最簡單的範例：

from scrapling.spiders import Spider, Request

class MySpider(Spider):
    name = "my_spider"

    async def start_requests(self):
        urls = [
            "https://example.com/page1",
            "https://example.com/page2",
            "https://example.com/page3",
        ]
        for url in urls:
            yield Request(url=url, callback=self.parse_page)

    async def parse_page(self, response):
        title = response.css("h1::text").get("")
        content = response.css("article p::text").getall()
        yield {"title": title, "content": " ".join(content), "url": response.url}

# 正式執行，start() 會回傳 CrawlResult
result = MySpider().start()
print(result.stats.items_scraped)

如果你熟悉 Scrapy，會發現這段程式碼的骨架幾乎一樣。最大的差異有兩點：第一，Scrapling 的 parse callback 是 async 函式，因為它從底層就是非同步架構；第二，兩邊雖然都叫 response.css()，但背後的解析引擎不同，Scrapling 以 lxml 為基礎，另外加上自適應元素追蹤，遇到網站改版時的容錯能力比較高。

1.2 Request / Response 物件模型

Scrapling Spider 的核心資料模型是 Request 與 Response 物件。Request 封裝了目標 URL、HTTP 方法、標頭、Cookie 以及最重要的 callback 函式指標；Response 則封裝了伺服器回傳的內容、狀態碼、原始 URL，以及經過 Scrapling 引擎解析後的 DOM 樹。

Request 物件支援的關鍵參數包括：

url：目標頁面的 URL
callback：處理該 Request 回傳 Response 的非同步函式
method：HTTP 方法，預設為 GET，可指定 POST、PUT 等
headers：自訂 HTTP 標頭
cookies：自訂 Cookie
data：POST 請求的資料內容
meta：傳遞自訂資料的字典，可在不同 callback 之間共享
priority：請求優先級，用於控制請求處理順序
dont_filter：是否跳過去重過濾

這樣熟悉的設計讓 Scrapy 使用者幾乎不需要學習成本就能直接上手。但 Scrapling 並非僅僅模仿 Scrapy，它在核心架構上做了許多現代化的改進，其中最顯著的就是非同步原生的支援與靈活的 Session 管理。

二、並發機制：concurrent_requests 與單一網域並發上限

在實際的爬蟲專案中，我們很少需要逐頁依序請求 — 那樣的速度太慢了。並發（concurrency）是提高爬取效率的關鍵，但同時也必須注意不要對目標伺服器造成過大的負擔，否則很容易被封鎖 IP 或觸發 rate limit。

▲ D4Vinci/Scrapling 的 Issues 分頁，可見使用者回報與討論，評估維護狀態與常見踩坑時可對照閱讀。

Scrapling Spider 提供了兩層並發控制機制：全域的 concurrent_requests，以及針對單一網域的 concurrent_requests_per_domain。

2.1 concurrent_requests：全域並發控制

concurrent_requests 參數控制 Spider 同時進行的最大請求數量，預設值是 4。舉例來說，當你設定 concurrent_requests=10 時，Spider 最多會同時發出 10 個 HTTP 請求，當其中一個請求完成並進入 parse callback 處理階段後，才會從佇列中取出下一個 Request 來發送。

這個參數的設定需要根據目標伺服器的承載能力以及你的網路頻寬來決定。一般來說：

小型網站或部落格：建議設定 3–5，避免對小型伺服器造成壓力
中型網站：建議設定 10–20，在速度與禮貌之間取得平衡
大型 API 或 CDN 加速站點：建議設定 20–50，但需注意不要觸發 rate limit
內部系統或 API：可設定到 100 以上，前提是確認伺服器能負荷

設定方式非常直覺：

class MySpider(Spider):
    concurrent_requests = 10
    # 其餘設定...

2.2 concurrent_requests_per_domain：單站流量控制

很多時候，一個爬蟲需要同時爬取多個不同的網站。在這種情況下，全域的 concurrent_requests 控制雖然能限制整體的並發量，卻無法防止某個特定網站被過度頻繁地請求。例如，你可能同時爬取十個網站，但其中一個網站的回應特別慢，這時全域的並發數量可能全部被這個慢速網站佔用，導致其他網站的請求也被阻塞。

concurrent_requests_per_domain 就是用來解決這個問題的。它限制「同一個網域」最多能同時跑幾個請求，確保任何單一站點都不會佔用全部的並發額度：

class MySpider(Spider):
    name = "my_spider"
    concurrent_requests = 20             # 全域同時最多 20 個請求
    concurrent_requests_per_domain = 5   # 每個網域各自最多 5 個，0 代表不限制

這樣的設計非常適合需要從多個資料來源聚合資料的爬蟲，例如比價網站、新聞聚合器或資料庫同步工具。

2.3 download_delay：避免被封鎖的緩衝

除了並發數量之外，download_delay 參數可以在每個請求之間插入固定的延遲時間，進一步降低對目標伺服器的請求頻率。設定方式同樣直覺：

class MySpider(Spider):
    download_delay = 1.5  # 每個請求之間等待 1.5 秒

需要注意的是，download_delay 會對每一個請求都加上固定等待，不分網域，它和全域並發上限是兩套獨立的禮貌機制。如果目標網站有 robots.txt，直接把 robots_txt_obey = True 開起來最省事，框架會遵守 Disallow 規則與 Crawl-delay；而當 AutoThrottle 開啟時，download_delay 與 robots.txt 的 Crawl-delay 會變成延遲的下限值，動態降速不會把它們蓋掉。

三、Checkpoint 機制：Ctrl+C 優雅關閉與斷點續爬

在大型爬蟲專案中，最讓人沮喪的莫過於爬蟲跑了幾個小時、已經抓取了上萬筆資料後，因為網路波動、伺服器重啟或意外的 Ctrl+C 而被迫中斷，所有進度化為烏有。Scrapling Spider 的 checkpoint 機制正是為了解決這個痛點而設計的。

3.1 運作原理

Checkpoint 機制的核心概念是：Spider 會定期把「還沒送出的請求佇列」與「已經看過的請求指紋（fingerprint）」寫進指定的工作目錄，並在收到中斷訊號時再存一次。當爬蟲重新啟動，Spider 會自動載入最近的 checkpoint，從中斷的地方繼續執行，不會重複爬取已經完成的頁面；如果整趟順利跑完，這個目錄會被自動清掉。

啟用方式很簡單，只要在建立 Spider 實例時傳入 crawldir 工作目錄，也可以順便指定存檔的間隔秒數：

from scrapling.spiders import Spider

class MySpider(Spider):
    name = "my_spider"
    concurrent_requests = 5

    async def parse(self, response):
        # ... 解析邏輯
        pass

# 第一次執行，每 120 秒存一次檔
spider = MySpider(crawldir="crawl_data/my_spider", interval=120.0)
result = spider.start()

if result.paused:
    print("爬取暫停，重新執行同一段程式就會自動接續")
else:
    print("爬取完成，checkpoint 目錄已清空")

3.2 優雅關閉的內部流程

當使用者按下 Ctrl+C 或系統發送 SIGINT 訊號時，Scrapling Spider 不會粗暴地終止程序，而是執行以下優雅關閉流程：

停止接收新的 Request 進入佇列
允許正在執行中的 Request 完成其 HTTP 請求與 parse callback
將所有 pending Request 序列化到 checkpoint 檔案
將已看過的請求指紋寫入 checkpoint，避免恢復後重複爬取同樣的頁面
以原子寫入的方式存檔（先寫暫存檔再更名），避免中斷時把狀態檔寫壞
關閉所有網路連接與 Session
輸出最終的爬蟲統計數據

這樣的設計確保了即使在爬蟲被中斷的情況下，也不會遺失資料或需要從頭開始。

3.3 Checkpoint 的實際應用場景

Checkpoint 機制在以下場景中特別有價值：

大規模資料爬取：需要爬取數十萬甚至上百萬頁面時，單次 session 無法完成所有工作
定時爬蟲：配合 cron job 或排程工具，每次執行時從上次中斷處繼續
不穩定的網路環境：在 VPN 切換、代理不穩或目標網站偶爾回應失敗的情況下，能夠自動恢復
資源受限的環境：在記憶體或時間有限的容器或雲端函式中，分段完成爬取任務

要提醒的是，checkpoint 是框架內部的序列化狀態檔，不是設計給人直接編輯的格式，想重來一次最乾脆的做法是把 crawldir 整個目錄刪掉再跑。反過來說，這個工作目錄也是斷點續爬的唯一依據，如果部署時把它掛在容器內的暫存區，容器一重建就會失去續爬能力。

四、多 Session 管理：HTTP 快速通道與 Stealth 隱身通道

Scrapling 最大的特色之一就是對 Session 的靈活管理，而這個特色在 Spider 框架中被發揮到了極致。Spider 允許在同一個爬蟲實例中混用多種 Session 類型，根據不同的頁面需求選擇最合適的通道。

4.1 兩種 Session 類型

Scrapling 提供幾種不同定位的 Session：

FetcherSession：基於純 HTTP 請求的快速通道。它不需要瀏覽器引擎，直接發送 HTTP 請求並解析回應。速度極快、資源消耗低，適用於一般的 REST API、靜態網頁或不需要 JavaScript 渲染的頁面。
AsyncStealthySession：基於 Playwright 的隱身瀏覽器通道。它會啟動一個真正的瀏覽器實例（Chromium），能夠執行 JavaScript、處理動態渲染的內容，並內建了多種反偵測機制來繞過 Cloudflare、Akamai 等 WAF 保護。
DynamicSession：同樣走瀏覽器引擎，但定位是「需要完整瀏覽器行為、不需要隱身對抗」的頁面，比 stealth 通道輕、啟動更快，適合有 JavaScript 渲染但沒有嚴格反爬的網站。

4.2 在同一個 Spider 中混用 Session

這是最令人興奮的部分：你不需要為不同類型的頁面撰寫兩套爬蟲。在 Scrapling Spider 中，你可以設定多個 Session 並在 Request 層級指定使用哪一個：

from scrapling.spiders import Spider, SessionManager, Request, Response
from scrapling.fetchers import FetcherSession, AsyncStealthySession


class MultiSessionSpider(Spider):
    name = "multi_session"
    start_urls = ["https://example.com/list"]

    def configure_sessions(self, manager: SessionManager):
        manager.add("fast", FetcherSession(impersonate="chrome"))
        # lazy=True：第一次真的用到才啟動，免得白開一個瀏覽器
        manager.add("stealth", AsyncStealthySession(block_webrtc=True), lazy=True)

    async def parse(self, response: Response):
        for item_url in response.css(".item a::attr(href)").getall():
            request = Request(url=item_url, callback=self.parse_detail)
            request.sid = "stealth" if "protected" in item_url else "fast"
            yield request

    async def parse_detail(self, response: Response):
        # 兩個通道的回應物件操作方式完全相同
        yield {"url": response.url, "title": response.css("h1::text").get("")}

這樣的設計讓爬蟲能夠靈活應對混合型網站。舉例來說，一個電商網站的首頁和商品列表頁可能沒有特別的防護，可以走快速的 FetcherSession；但商品詳情頁或結帳頁面可能受到 Cloudflare 保護，這時可以自動切換到 AsyncStealthySession。同一個爬蟲、同一套 parse 邏輯，無需撰寫兩套程式碼。

4.3 Session 的生命週期管理

Spider 會自動管理每個 Session 的生命週期，包括：

啟用：Spider 啟動時準備已註冊的 Session；標記 lazy=True 的通道（例如瀏覽器）會等到第一次用到才啟動
連線池：維護每個 Session 的連線池，避免重複建立 TCP 連線
Cookie 管理：每個 Session 維護獨立的 Cookie jar
Proxy 綁定：每個 Session 可以綁定不同的 Proxy 配置
關閉：在 Spider 優雅關閉時，自動關閉所有 Session 並釋放資源

這意味著你不需要手動管理 Session 的開啟與關閉，Spider 框架會幫你處理好所有細節。

五、Proxy Rotator：自動代理輪換機制

在商業爬蟲或大規模資料採集中，代理輪換是必不可少的環節。Scrapling Spider 內建了 ProxyRotator 模組，支援自訂的代理輪換策略。

5.1 基本配置

ProxyRotator 可以接受多種格式的代理列表：

from scrapling.spiders import Spider, SessionManager
from scrapling.fetchers import FetcherSession, AsyncStealthySession, ProxyRotator

cheap_proxies = ProxyRotator([
    "http://proxy1.example.com:8080",
    "http://proxy2.example.com:8080",
])

# 瀏覽器通道用的代理可以帶帳密，格式是 dict
expensive_proxies = ProxyRotator([
    {"server": "http://residential1.example.com:8080", "username": "user", "password": "pass"},
    {"server": "http://mobile1.example.com:8080", "username": "user", "password": "pass"},
])


class ProxiedSpider(Spider):
    name = "proxied_spider"
    start_urls = ["https://example.com"]

    def configure_sessions(self, manager: SessionManager):
        manager.add("fast", FetcherSession(proxy_rotator=cheap_proxies))
        manager.add("stealth", AsyncStealthySession(proxy_rotator=expensive_proxies), lazy=True)

5.2 輪換策略

ProxyRotator 預設就是循環輪換，依序把請求分配到清單中的每一個代理，用完一輪再從頭開始。如果你需要別種分配方式，就傳入自訂的選擇函式：

循環輪換（預設）：依序使用清單中的代理，跑完一輪再從頭，適合代理品質平均、只需要分散來源 IP 的場景
自訂策略函式：由你決定下一個要用哪個代理，例如按照你自己記錄的成功率或地區排序

自訂策略的實作方式如下：

from scrapling.core._types import ProxyType

def pick_next(proxies: list, current_index: int) -> tuple[ProxyType, int]:
    """自訂策略：回傳下一個要用的代理，以及新的索引位置"""
    next_index = (current_index + 1) % len(proxies)
    return proxies[next_index], next_index

proxy_rotator = ProxyRotator(
    ["http://proxy1.example.com:8080", "http://proxy2.example.com:8080"],
    strategy=pick_next,
)

5.3 被判定封鎖時的重試機制

代理輪換真正的價值，是在某條線路被擋掉時還能自動換路。Scrapling Spider 內建了 blocked 判斷與重試：回應狀態碼落在 401、403、407、429、444、500、502、503、504 時，框架會先呼叫 is_blocked()；判定為封鎖後，複製一份請求交給 retry_blocked_request()，讓你有機會換 Session、換代理或補標頭，再重新排進佇列，最多重試 max_blocked_retries 次（預設 3）。

class ToughSpider(Spider):
    name = "tough_spider"
    start_urls = ["https://example.com"]
    max_blocked_retries = 5

    def configure_sessions(self, manager: SessionManager):
        manager.add("fast", FetcherSession(impersonate=["chrome", "firefox", "safari"]))
        manager.add("stealth", AsyncStealthySession(block_webrtc=True), lazy=True)

    async def is_blocked(self, response: Response) -> bool:
        # 覆寫預設的狀態碼判斷，改看回應內容
        body = response.body.decode("utf-8", errors="ignore")
        return "access denied" in body.lower()

    async def retry_blocked_request(self, request: Request, response: Response) -> Request:
        # 被擋掉了，改用隱身通道再試一次
        request.sid = "stealth"
        self.logger.info(f"Retrying blocked request: {request.url}")
        return request

重試的請求會跳過去重、排在較後面的優先序，所以不會卡住整個佇列；如果原本帶著代理資訊，重試前會被清掉，讓輪換器重新分配一條線路。另外要留意，官方的 ProxyRotator 只負責把代理輪流發出去，並沒有健康檢查或自動剔除機制，代理品質的監控要自己接。

六、Streaming 模式：即時監控爬蟲健康狀態

當爬蟲開始大規模運作時，你會需要知道它目前的狀態：已經爬取多少頁面？花了多少時間？有多少請求失敗或被擋？Scrapling Spider 的 streaming 模式提供了即時的監控能力。

6.1 啟用 Streaming 模式

Streaming 不是靠 run() 的參數，而是改用非同步的 stream()，在資料產出的當下就拿到手：

import anyio
from scrapling.spiders import Spider

async def main():
    spider = MySpider(crawldir="crawl_data/my_spider")
    async for item in spider.stream():
        print(f"抓到一筆：{item}")
        print(f"累計 {spider.stats.items_scraped} 筆，"
              f"已發出 {spider.stats.requests_count} 個請求")

anyio.run(main)

迭代的同時，你可以直接讀 spider.stats（CrawlStats 物件）拿到即時統計，常用欄位包括：

items_scraped：已成功產出的資料筆數
items_dropped：被 hook 或過濾條件丟掉的筆數
requests_count：已發出的請求總數
failed_requests_count：失敗的請求數
blocked_requests_count：被判定為封鎖的請求數
robots_disallowed_count：因為 robots.txt 而跳過的請求數
response_bytes：已接收的位元組數
elapsed_seconds 與 requests_per_second：總耗時與平均速度
cache_hits 與 cache_misses：開啟 development mode 時的快取命中狀況

6.2 把統計接進你的監控系統

官方沒有 monitor_callback 這種參數。要接監控系統，就在迭代過程自己判斷，或覆寫生命週期 hook，把數字送到 Prometheus、Datadog 或自家儀表板：

import anyio

async def main():
    spider = MySpider()
    async for item in spider.stream():
        stats = spider.stats
        if stats.failed_requests_count > 100:
            await send_alert(f"失敗請求已達 {stats.failed_requests_count} 次")
        await log_to_database({
            "items": stats.items_scraped,
            "requests": stats.requests_count,
            "blocked": stats.blocked_requests_count,
            "bytes": stats.response_bytes,
        })

anyio.run(main)

如果不想在迭代裡寫判斷，也可以覆寫 on_start()、on_error()、on_scraped_item()、on_close() 這幾個生命週期 hook，把上報、清理與收尾邏輯集中管理。

6.3 自動調速

長時間任務更關鍵的是 AutoThrottle。它會觀察每個網域的實際回應速度，自動調整「送往下一個請求前的延遲」：伺服器快就加速，慢或被刁難就退讓，這對不穩定的目標網站特別有用。開啟方式是把幾個類別屬性打開：

class AdaptiveSpider(Spider):
    name = "adaptive_spider"
    start_urls = ["https://example.com"]
    autothrottle_enabled = True
    autothrottle_start_delay = 5.0        # 對一個網域的第一次請求先等 5 秒
    autothrottle_max_delay = 60.0         # 延遲上限
    autothrottle_target_concurrency = 8   # 想維持的並發水位

當對方回 429 並附帶 Retry-After 時，框架會跟著把延遲拉到對方要求的秒數，不用自己寫退避邏輯，最後每個網域實際採用的延遲也會記在 stats.autothrottle_delays 裡。要留意的是，它調整的是延遲，不是並發上限。

七、Development Mode：快取頁面加速開發迭代

在開發爬蟲的過程中，最耗時的部分往往不是撰寫選擇器或解析邏輯，而是等待 HTTP 請求完成。每次修改一段 parse 邏輯、想測試新的 CSS 選擇器或 XPath 表達式，都需要重新發送一次 HTTP 請求，等待網路往返。如果目標網站速度較慢或有反爬限制，這個過程會變得極為痛苦。

Scrapling Spider 的 development mode 正是為了解決這個開發體驗問題而設計的。

7.1 啟用 Development Mode

class MySpider(Spider):
    name = "my_spider"
    start_urls = ["https://example.com"]
    development_mode = True
    development_cache_dir = "/tmp/my_spider_cache"   # 不指定就用預設路徑

7.2 Development Mode 的運作方式

當 development mode 啟用時，Spider 的行為會變成這樣：

第一次執行：照正常流程發送 HTTP 請求，並把每個回應（狀態碼、標頭、內容）存到快取目錄，預設路徑是 .scrapling_cache/{spider.name}
第二次以後：同一個請求指紋直接讀快取，完全不碰網路，也不受 download_delay、rate limit 與 blocked 重試的影響
快取鍵：預設用請求指紋，只要動到 fp_include_kwargs、fp_include_headers、fp_keep_fragments 這些會影響指紋的設定，就會重新抓一次
快取有效期：沒有自動過期機制，每個回應存成一個 {fingerprint}.json，想重新抓就手動刪掉快取目錄

這表示開發過程中只有第一次執行需要等網路往返，之後每一次改選擇器、改解析邏輯都是瞬間完成。官方也在文件裡提醒：development mode 只適合開發階段，正式上線前記得關掉。

7.3 實際開發流程

一個典型的 Dev Mode 開發流程如下：

撰寫 Spider 的基本架構，包括 start_requests 和初步的 parse 邏輯
以 dev mode 執行 Spider，確認所有 Request 都能正常發送與接收
調整 parse 邏輯中的選擇器或資料提取方式
再次以 dev mode 執行 Spider — 這次不需要等待網路請求，Response 直接從快取讀取
重複步驟 3–4 直到資料提取邏輯完全正確
把 development_mode 拿掉，正式執行爬蟲

這樣做最省時間的地方，是選擇器除錯與資料清洗這兩段反覆試錯的過程：改一行、重跑一次，不用再等一輪網路往返。實際省下多少時間會因專案而異，但迭代節奏的差別非常有感。

八、結果匯出：內建 JSON／JSONL／CSV／XML 與自訂處理

爬蟲最終的目的是產生有價值的資料。Scrapling Spider 的 start() 會回傳一個 CrawlResult，裡面同時帶著這次抓到的資料與統計，要匯出只要一行。

8.1 內建 Export 格式

Spider 支援多種輸出格式：

JSON：標準 JSON 陣列格式，每個 item 為一個 JSON 物件
JSON Lines (JSONL)：每行一個 JSON 物件，適合大規模資料串流處理
CSV：逗號分隔值，適合 Excel 或 Google Sheets 匯入
XML：每個 item 包成 <item>，外層是 <items>，標籤名稱可以自己換

設定方式極為簡單：

result = MySpider().start()

# 標準 JSON，加 indent 會排版
result.items.to_json("output/scraped_data.json", indent=True)

# 每行一筆，適合串流處理
result.items.to_jsonl("output/scraped_data.jsonl")

# CSV 或 XML（CSV 可用 fields 指定欄位順序、delimiter 換成 Tab）
result.items.to_csv("output/scraped_data.csv")
result.items.to_xml("output/scraped_data.xml")

8.2 自訂處理：on_scraped_item

如果你的爬蟲需要把資料直接寫進資料庫、送到 API 或做即時清洗，就覆寫 on_scraped_item()：每產出一筆資料就會被呼叫一次，回傳 None 代表丟棄這筆。

class MySpider(Spider):
    name = "my_spider"
    start_urls = ["https://example.com"]

    async def on_scraped_item(self, item: dict) -> dict | None:
        if not item.get("title"):
            return None   # 標題空的就不要
        await db.execute(
            "INSERT INTO scraped_data (title, content, url, scraped_at) VALUES ($1, $2, $3, $4)",
            item["title"], item["content"], item["url"], datetime.now()
        )
        return item

這個 hook 是 async 的，所以在裡面等資料庫或打 API 不會擋住其他協程；但它是在爬蟲流程內被等待的，如果你在裡面做很慢的事，整體速度還是會被拖下來，需要真正的背景處理就自己丟進佇列。

8.3 資料管線整合

要同時寫多個目的地，就在同一個 hook 裡依序處理，再用內建的 exporter 留一份檔案當備份：

class MySpider(Spider):
    name = "my_spider"
    start_urls = ["https://example.com"]

    async def on_scraped_item(self, item: dict) -> dict | None:
        await write_to_postgres(item)
        await send_to_elasticsearch(item)
        await notify_webhook(item)
        return item

result = MySpider().start()
result.items.to_jsonl("output/backup.jsonl")

這樣的組合讓 Spider 不只是把資料抓下來，也能直接接進既有的資料管線。

九、實戰範例：完整的 Spider 應用

讓我們將以上所有概念整合到一個完整的實戰範例中。以下是一個多 Session、支援斷點續爬、具備代理輪換與即時監控的完整 Spider：

import anyio
from scrapling.spiders import Spider, SessionManager, Request, Response
from scrapling.fetchers import FetcherSession, AsyncStealthySession, ProxyRotator

cheap_proxies = ProxyRotator([
    "http://proxy1.pool:8080",
    "http://proxy2.pool:8080",
])


class NewsAggregatorSpider(Spider):
    """新聞聚合爬蟲：整合並發控制、多 Session、斷點續爬與代理輪換"""

    name = "news_aggregator"

    # 並發與禮貌設定
    concurrent_requests = 15
    concurrent_requests_per_domain = 5
    download_delay = 0.5
    autothrottle_enabled = True
    max_blocked_retries = 4

    def configure_sessions(self, manager: SessionManager):
        manager.add("fast", FetcherSession(impersonate="chrome", proxy_rotator=cheap_proxies))
        manager.add("stealth", AsyncStealthySession(block_webrtc=True, proxy_rotator=cheap_proxies), lazy=True)

    async def start_requests(self):
        # 一般新聞列表頁走快速通道
        for category in ["tech", "finance", "science"]:
            request = Request(
                url=f"https://news.example.com/{category}",
                callback=self.parse_category,
            )
            request.sid = "fast"
            yield request

        # 受保護的 API 走隱身通道
        protected = Request(
            url="https://protected-api.source.com/latest",
            callback=self.parse_protected_api,
        )
        protected.sid = "stealth"
        yield protected

    async def parse_category(self, response: Response):
        for article in response.css(".article-card"):
            yield {
                "title": article.css("h2::text").get(""),
                "summary": article.css(".summary::text").get(""),
                "url": article.css("a::attr(href)").get(""),
                "category": response.url.split("/")[-1],
                "source": "news.example.com",
            }

        # 分頁處理
        next_page = response.css(".pagination .next::attr(href)").get("")
        if next_page:
            yield response.follow(next_page, callback=self.parse_category)

    async def parse_protected_api(self, response: Response):
        for item in response.json()["articles"]:
            yield {
                "title": item["title"],
                "content": item["body"],
                "url": item["url"],
                "category": item["category"],
                "source": "protected-api.source.com",
            }

    async def retry_blocked_request(self, request: Request, response: Response) -> Request:
        # 被擋掉就換隱身通道
        request.sid = "stealth"
        self.logger.info(f"Retrying blocked request: {request.url}")
        return request


# 執行：把 crawldir 一起帶上，中斷後重跑就會自動接續
async def main():
    spider = NewsAggregatorSpider(crawldir="crawl_data/news_aggregator")
    async for item in spider.stream():
        await push_to_queue(item)

    stats = spider.stats
    print(f"共 {stats.items_scraped} 筆，失敗 {stats.failed_requests_count} 筆，"
          f"平均 {stats.requests_per_second:.1f} req/s")

if __name__ == "__main__":
    anyio.run(main)

這個範例爬蟲展示了以下所有功能的整合應用：

全域並發上限與單網域並發限制
多 Session 混用（FetcherSession + AsyncStealthySession）
Checkpoint 斷點續爬
Proxy 自動輪換
用 stream() 迭代時同步讀取即時統計
on_scraped_item 客製處理與 to_jsonl() 匯出

十、Spider 的工程化價值總結

經過以上深入的分析與實戰範例，我們可以看到 Scrapling Spider 不僅僅是一個爬蟲框架，它更是一個完整的工程化解決方案。以下是 Spider 帶給開發者的核心價值：

10.1 從腳本到系統的升級

大多數爬蟲專案始於一個簡單的 Python 腳本：發送請求、解析回應、儲存資料。但當爬蟲規模增長到需要處理數萬個頁面、需要團隊協作維護、需要整合到 CI/CD 管線中時，腳本的模式就顯得力不從心了。Spider 框架透過以下設計讓爬蟲升級為真正的工程系統：

模組化架構：start_requests、parse callback、on_scraped_item 各司其職，程式碼組織清晰
可配置性：所有行為參數化，透過類別屬性或執行參數即可調整行為，無需修改核心邏輯
可測試性：可以單獨測試每個 parse callback 的邏輯
可擴展性：透過自訂 Session、自訂 Proxy 輪換策略、自訂生命週期 hook 等方式擴展

10.2 生產環境就緒

Scrapling Spider 從設計之初就考慮了生產環境的實際需求：

優雅退出：不會因為網路波動或手動中斷而遺失資料
自動恢復：Checkpoint 機制確保任務可以分段完成
即時監控：Streaming 模式讓你可以隨時掌握爬蟲狀態
資源控制：全域與單網域並發上限、download_delay 與 AutoThrottle，確保爬蟲不會失控
錯誤處理：blocked 偵測與自動重試，加上可覆寫的 on_error 統一收斂錯誤

10.3 開發效率的提升

Dev Mode 的設計充分體現了 Scrapling 對開發體驗的重視。開發者不再需要忍受漫長的等待來測試每一個細微的修改，這不僅提升了開發速度，更重要的是保持了開發者的心流狀態。配合 on_scraped_item 與內建的匯出方法，從開發到上線的流程被大幅簡化。

10.4 適用場景一覽

Scrapling Spider 特別適合以下場景：

大規模電商監控：同時監控數百個電商平台的價格與庫存
新聞與內容聚合：從多個新聞來源定期抓取內容
社交媒體數據採集：爬取社群平台的公開資料
研究與學術資料收集：從學術資料庫、政府開放資料平台收集資料
SEO 與網站分析：大規模分析網站結構與內容
資料庫遷移與同步：從舊系統大量遷移資料到新系統

結語：Scrapling Spider 為爬蟲開發樹立了新的標竿

回顧 Scrapling Spider 的核心設計，它巧妙地平衡了三個看似矛盾的需求：強大與簡單、彈性與規範、效率與穩定性。它借鑒了 Scrapy 經過時間驗證的優秀設計，但又在非同步支援、Session 管理、Dev Mode 等方面做出了現代化的創新。

對於有 Scrapy 經驗的開發者來說，Scrapling Spider 幾乎無需學習成本；對於爬蟲新手來說，它清晰的文件與直覺的 API 設計也讓入門門檻降到最低。更重要的是，它將爬蟲開發從「撰寫一次性腳本」的思維模式提升到了「設計可維護工程系統」的高度。

系列下一篇是實測驗證：解析速度如何做到 BeautifulSoup 的 784 倍？效能優化實戰技巧，會把鏡頭轉向效能調校。如果你想補齊動態渲染與反爬機制的處理，可以搭配三種 Fetcher 實戰對決：純 HTTP、隱身瀏覽器、完整 Playwright 如何繞過 Cloudflare Turnstile，以及整合 LLM 做智能提取的AI 協作與自適應爬蟲的未來一起讀。

如果你對 Scrapling 有任何問題，歡迎在下方留言討論。如果這篇文章對你有幫助，也請分享給更多需要的朋友！

Spider 框架拆解：從並發爬取到暫停恢復、自動 Proxy 輪換與即時統計