資訊專欄INFORMATION COLUMN
摘要：這篇文章的題目有點大，但這并不是說我自覺對爬蟲這塊有多大見解，我只不過是想將自己的一些經(jīng)驗付諸于筆，對于如何寫一個爬蟲框架，我想一步一步地結(jié)合具體代碼來講述如何從零開始編寫一個自己的爬蟲框架年到如今，我花精力比較多的一個開源項目算是了，這是
這篇文章的題目有點大，但這并不是說我自覺對Python爬蟲這塊有多大見解，我只不過是想將自己的一些經(jīng)驗付諸于筆，對于如何寫一個爬蟲框架，我想一步一步地結(jié)合具體代碼來講述如何從零開始編寫一個自己的爬蟲框架
2018年到如今，我花精力比較多的一個開源項目算是Ruia了，這是一個基于Python3.6+的異步爬蟲框架，當時也獲得一些推薦，比如Github Trending Python語言榜單第二，目前Ruia還在開發(fā)中，Star數(shù)目不過700+，如果各位有興趣，歡迎一起開發(fā)，來波star我也不會拒絕哈~
什么是爬蟲框架
說這個之前，得先說說什么是框架：
是實現(xiàn)業(yè)界標準的組件規(guī)范：比如眾所周知的MVC開發(fā)規(guī)范
提供規(guī)范所要求之基礎(chǔ)功能的軟件產(chǎn)品：比如Django框架就是MVC的開發(fā)框架，但它還提供了其他基礎(chǔ)功能幫助我們快速開發(fā)，比如中間件、認證系統(tǒng)等
框架的關(guān)注點在于規(guī)范二字，好，我們要寫的Python爬蟲框架規(guī)范是什么？
很簡單，爬蟲框架就是對爬蟲流程規(guī)范的實現(xiàn)，不清楚的朋友可以看上一篇文章談?wù)剬ython爬蟲的理解，下面總結(jié)一下爬蟲流程：
請求&響應(yīng)
解析
持久化
這三個流程有沒有可能以一種優(yōu)雅的形式串聯(lián)起來，Ruia目前是這樣實現(xiàn)的，請看代碼示例：
可以看到，Item & Field類結(jié)合一起實現(xiàn)了字段的解析提取，Spider類結(jié)合Request * Response類實現(xiàn)了對爬蟲程序整體的控制，從而可以如同流水線一般編寫爬蟲，最后返回的item可以根據(jù)使用者自身的需求進行持久化，這幾行代碼，我們就實現(xiàn)了獲取目標網(wǎng)頁請求、字段解析提取、持久化這三個流程
實現(xiàn)了基本流程規(guī)范之后，我們繼而就可以考慮一些基礎(chǔ)功能，讓使用者編寫爬蟲可以更加輕松，比如：中間件(Ruia里面的Middleware)、提供一些hook讓用戶編寫爬蟲更方便(比如ruia-motor)
這些想明白之后，接下來就可以愉快地編寫自己心目中的爬蟲框架了
如何踏出第一步
首先，我對Ruia爬蟲框架的定位很清楚，基于asyncio & aiohttp的一個輕量的、異步爬蟲框架，怎么實現(xiàn)呢，我覺得以下幾點需要遵守：
輕量級，專注于抓取、解析和良好的API接口
插件化，各個模塊耦合程度盡量低，目的是容易編寫自定義插件
速度，異步無阻塞框架，需要對速度有一定追求
什么是爬蟲框架如今我們已經(jīng)很清楚了，現(xiàn)在急需要做的就是將流程規(guī)范利用Python語言實現(xiàn)出來，怎么實現(xiàn)，分為哪幾個模塊，可以看如下圖示：
?
同時讓我們結(jié)合上面一節(jié)的Ruia代碼來從業(yè)務(wù)邏輯角度看看這幾個模塊到底是什么意思：
Request：請求
Response：響應(yīng)
Item & Field：解析提取
Spider：爬蟲程序的控制中心，將請求、響應(yīng)、解析、存儲結(jié)合起來
這四個部分我們可以簡單地使用五個類來實現(xiàn)，在開始講解之前，請先克隆Ruia框架到本地：
# 請確保本地Python環(huán)境是3.6+ git clone https://github.com/howie6879/ruia.git # 安裝pipenv pip install pipenv # 安裝依賴包 pipenv install --dev
然后用PyCharm打開Ruia項目：
選擇剛剛pipenv配置好的python解釋器：
此時可以完整地看到項目代碼：
好，環(huán)境以及源碼準備完畢，接下來將結(jié)合代碼講述一個爬蟲框架的編寫流程
Request & Response
Request類的目的是對aiohttp加一層封裝進行模擬請求，功能如下：
封裝GET、POST兩種請求方式
增加回調(diào)機制
自定義重試次數(shù)、休眠時間、超時、重試解決方案、請求是否成功驗證等功能
將返回的一系列數(shù)據(jù)封裝成Response類返回
接下來就簡單了，不過就是實現(xiàn)上述需求，首先，需要實現(xiàn)一個函數(shù)來抓取目標url，比如命名為fetch:
import asyncio
import aiohttp
import async_timeout
from typing import Coroutine
class Request:
# Default config
REQUEST_CONFIG = {
"RETRIES": 3,
"DELAY": 0,
"TIMEOUT": 10,
"RETRY_FUNC": Coroutine,
"VALID": Coroutine
}
METHOD = ["GET", "POST"]
def __init__(self, url, method="GET", request_config=None, request_session=None):
self.url = url
self.method = method.upper()
self.request_config = request_config or self.REQUEST_CONFIG
self.request_session = request_session
@property
def current_request_session(self):
if self.request_session is None:
self.request_session = aiohttp.ClientSession()
self.close_request_session = True
return self.request_session
async def fetch(self):
"""Fetch all the information by using aiohttp"""
if self.request_config.get("DELAY", 0) > 0:
await asyncio.sleep(self.request_config["DELAY"])
timeout = self.request_config.get("TIMEOUT", 10)
async with async_timeout.timeout(timeout):
resp = await self._make_request()
try:
resp_data = await resp.text()
except UnicodeDecodeError:
resp_data = await resp.read()
resp_dict = dict(
rl=self.url,
method=self.method,
encoding=resp.get_encoding(),
html=resp_data,
cookies=resp.cookies,
headers=resp.headers,
status=resp.status,
history=resp.history
)
await self.request_session.close()
return type("Response", (), resp_dict)
async def _make_request(self):
if self.method == "GET":
request_func = self.current_request_session.get(self.url)
else:
request_func = self.current_request_session.post(self.url)
resp = await request_func
return resp
if __name__ == "__main__":
loop = asyncio.get_event_loop()
resp = loop.run_until_complete(Request("https://docs.python-ruia.org/").fetch())
print(resp.status)
實際運行一下，會輸出請求狀態(tài)200，就這樣簡單封裝一下，我們已經(jīng)有了自己的請求類Request，接下來只需要再完善一下重試機制以及將返回的屬性封裝一下就基本完成了：
# 重試函數(shù)
async def _retry(self):
if self.retry_times > 0:
retry_times = self.request_config.get("RETRIES", 3) - self.retry_times + 1
self.retry_times -= 1
retry_func = self.request_config.get("RETRY_FUNC")
if retry_func and iscoroutinefunction(retry_func):
request_ins = await retry_func(weakref.proxy(self))
if isinstance(request_ins, Request):
return await request_ins.fetch()
return await self.fetch()
最終代碼見ruia/request.py即可，接下來就可以利用Request來實際請求一個目標網(wǎng)頁，如下：
這段代碼請求了目標網(wǎng)頁https://docs.python-ruia.org/并返回了Response對象，其中Response提供屬性介紹如下：
Field & Item
實現(xiàn)了對目標網(wǎng)頁的請求，接下來就是對目標網(wǎng)頁進行字段提取，我覺得ORM的思想很適合用在這里，我們只需要定義一個Item類，類里面每個屬性都可以用Field類來定義，然后只需要傳入url或者html，執(zhí)行過后Item類里面 定義的屬性會自動被提取出來變成目標字段值
可能說起來比較拗口，下面直接演示一下可能你就明白這樣寫的好，假設(shè)你的需求是獲取HackerNews網(wǎng)頁的title和url，可以這樣實現(xiàn)：
import asyncio
from ruia import AttrField, TextField, Item
class HackerNewsItem(Item):
target_item = TextField(css_select="tr.athing")
title = TextField(css_select="a.storylink")
url = AttrField(css_select="a.storylink", attr="href")
async def main():
async for item in HackerNewsItem.get_items(url="https://news.ycombinator.com/"):
print(item.title, item.url)
if __name__ == "__main__":
items = asyncio.run(main())
從輸出結(jié)果可以看到，title和url屬性已經(jīng)被賦與實際的目標值，這樣寫起來是不是很簡潔清晰也很明了呢？
來看看怎么實現(xiàn)，Field類的目的是提供多種方式讓開發(fā)者提取網(wǎng)頁字段，比如：
XPath
CSS Selector
RE
所以我們只需要根據(jù)需求，定義父類然后再利用不同的提取方式實現(xiàn)子類即可，代碼如下：
class BaseField(object):
"""
BaseField class
"""
def __init__(self, default: str = "", many: bool = False):
"""
Init BaseField class
url: http://lxml.de/index.html
:param default: default value
:param many: if there are many fields in one page
"""
self.default = default
self.many = many
def extract(self, *args, **kwargs):
raise NotImplementedError("extract is not implemented.")
class _LxmlElementField(BaseField):
pass
class AttrField(_LxmlElementField):
"""
This field is used to get attribute.
"""
pass
class HtmlField(_LxmlElementField):
"""
This field is used to get raw html data.
"""
pass
class TextField(_LxmlElementField):
"""
This field is used to get text.
"""
pass
class RegexField(BaseField):
"""
This field is used to get raw html code by regular expression.
RegexField uses standard library `re` inner, that is to say it has a better performance than _LxmlElementField.
"""
pass
核心類就是上面的代碼，具體實現(xiàn)請看ruia/field.py
接下來繼續(xù)說Item部分，這部分實際上是對ORM那塊的實現(xiàn)，用到的知識點是元類，因為我們需要控制類的創(chuàng)建行為：
class ItemMeta(type):
"""
Metaclass for an item
"""
def __new__(cls, name, bases, attrs):
__fields = dict({(field_name, attrs.pop(field_name))
for field_name, object in list(attrs.items())
if isinstance(object, BaseField)})
attrs["__fields"] = __fields
new_class = type.__new__(cls, name, bases, attrs)
return new_class
class Item(metaclass=ItemMeta):
"""
Item class for each item
"""
def __init__(self):
self.ignore_item = False
self.results = {}
這一層弄明白接下來就很簡單了，還記得上一篇文章《談?wù)剬ython爬蟲的理解》里面說的四個類型的目標網(wǎng)頁么：
單頁面單目標
單頁面多目標
多頁面單目標
多頁面多目標
本質(zhì)來說就是要獲取網(wǎng)頁的單目標以及多目標（多頁面可以放在Spider那塊實現(xiàn)），Item類只需要定義兩個方法就能實現(xiàn)：
get_item()：單目標
get_items()：多目標，需要定義好target_item
具體實現(xiàn)見：ruia/item.py
Spider
在Ruia框架中，為什么要有Spider，有以下原因：
真實世界爬蟲是多個頁面的（或深度或廣度），利用Spider可以對這些進行 有效的管理
制定一套爬蟲程序的編寫標準，可以讓開發(fā)者容易理解、交流，能迅速產(chǎn)出高質(zhì)量爬蟲程序
自由地定制插件
接下來說說代碼實現(xiàn)，Ruia框架的API寫法我有參考Scrapy，各個函數(shù)之間的聯(lián)結(jié)也是使用回調(diào)，但是你也可以直接使用await，可以直接看代碼示例：
from ruia import AttrField, TextField, Item, Spider
class HackerNewsItem(Item):
target_item = TextField(css_select="tr.athing")
title = TextField(css_select="a.storylink")
url = AttrField(css_select="a.storylink", attr="href")
class HackerNewsSpider(Spider):
start_urls = [f"https://news.ycombinator.com/news?p={index}" for index in range(1, 3)]
async def parse(self, response):
async for item in HackerNewsItem.get_items(html=response.html):
yield item
if __name__ == "__main__":
HackerNewsSpider.start()
使用起來還是挺簡潔的，輸出如下：
