資訊專欄INFORMATION COLUMN
摘要：這篇文章的題目有點(diǎn)大，但這并不是說我自覺對(duì)爬蟲這塊有多大見解，我只不過是想將自己的一些經(jīng)驗(yàn)付諸于筆，對(duì)于如何寫一個(gè)爬蟲框架，我想一步一步地結(jié)合具體代碼來講述如何從零開始編寫一個(gè)自己的爬蟲框架年到如今，我花精力比較多的一個(gè)開源項(xiàng)目算是了，這是
這篇文章的題目有點(diǎn)大，但這并不是說我自覺對(duì)Python爬蟲這塊有多大見解，我只不過是想將自己的一些經(jīng)驗(yàn)付諸于筆，對(duì)于如何寫一個(gè)爬蟲框架，我想一步一步地結(jié)合具體代碼來講述如何從零開始編寫一個(gè)自己的爬蟲框架
2018年到如今，我花精力比較多的一個(gè)開源項(xiàng)目算是Ruia了，這是一個(gè)基于Python3.6+的異步爬蟲框架，當(dāng)時(shí)也獲得一些推薦，比如Github Trending Python語(yǔ)言榜單第二，目前Ruia還在開發(fā)中，Star數(shù)目不過700+，如果各位有興趣，歡迎一起開發(fā)，來波star我也不會(huì)拒絕哈~
什么是爬蟲框架
說這個(gè)之前，得先說說什么是框架：
是實(shí)現(xiàn)業(yè)界標(biāo)準(zhǔn)的組件規(guī)范：比如眾所周知的MVC開發(fā)規(guī)范
提供規(guī)范所要求之基礎(chǔ)功能的軟件產(chǎn)品：比如Django框架就是MVC的開發(fā)框架，但它還提供了其他基礎(chǔ)功能幫助我們快速開發(fā)，比如中間件、認(rèn)證系統(tǒng)等
框架的關(guān)注點(diǎn)在于規(guī)范二字，好，我們要寫的Python爬蟲框架規(guī)范是什么？
很簡(jiǎn)單，爬蟲框架就是對(duì)爬蟲流程規(guī)范的實(shí)現(xiàn)，不清楚的朋友可以看上一篇文章談?wù)剬?duì)Python爬蟲的理解，下面總結(jié)一下爬蟲流程：
請(qǐng)求&響應(yīng)
解析
持久化
這三個(gè)流程有沒有可能以一種優(yōu)雅的形式串聯(lián)起來，Ruia目前是這樣實(shí)現(xiàn)的，請(qǐng)看代碼示例：
可以看到，Item & Field類結(jié)合一起實(shí)現(xiàn)了字段的解析提取，Spider類結(jié)合Request * Response類實(shí)現(xiàn)了對(duì)爬蟲程序整體的控制，從而可以如同流水線一般編寫爬蟲，最后返回的item可以根據(jù)使用者自身的需求進(jìn)行持久化，這幾行代碼，我們就實(shí)現(xiàn)了獲取目標(biāo)網(wǎng)頁(yè)請(qǐng)求、字段解析提取、持久化這三個(gè)流程
實(shí)現(xiàn)了基本流程規(guī)范之后，我們繼而就可以考慮一些基礎(chǔ)功能，讓使用者編寫爬蟲可以更加輕松，比如：中間件(Ruia里面的Middleware)、提供一些hook讓用戶編寫爬蟲更方便(比如ruia-motor)
這些想明白之后，接下來就可以愉快地編寫自己心目中的爬蟲框架了
如何踏出第一步
首先，我對(duì)Ruia爬蟲框架的定位很清楚，基于asyncio & aiohttp的一個(gè)輕量的、異步爬蟲框架，怎么實(shí)現(xiàn)呢，我覺得以下幾點(diǎn)需要遵守：
輕量級(jí)，專注于抓取、解析和良好的API接口
插件化，各個(gè)模塊耦合程度盡量低，目的是容易編寫自定義插件
速度，異步無阻塞框架，需要對(duì)速度有一定追求
什么是爬蟲框架如今我們已經(jīng)很清楚了，現(xiàn)在急需要做的就是將流程規(guī)范利用Python語(yǔ)言實(shí)現(xiàn)出來，怎么實(shí)現(xiàn)，分為哪幾個(gè)模塊，可以看如下圖示：
?
同時(shí)讓我們結(jié)合上面一節(jié)的Ruia代碼來從業(yè)務(wù)邏輯角度看看這幾個(gè)模塊到底是什么意思：
Request：請(qǐng)求
Response：響應(yīng)
Item & Field：解析提取
Spider：爬蟲程序的控制中心，將請(qǐng)求、響應(yīng)、解析、存儲(chǔ)結(jié)合起來
這四個(gè)部分我們可以簡(jiǎn)單地使用五個(gè)類來實(shí)現(xiàn)，在開始講解之前，請(qǐng)先克隆Ruia框架到本地：
# 請(qǐng)確保本地Python環(huán)境是3.6+ git clone https://github.com/howie6879/ruia.git # 安裝pipenv pip install pipenv # 安裝依賴包 pipenv install --dev
然后用PyCharm打開Ruia項(xiàng)目：
選擇剛剛pipenv配置好的python解釋器：
此時(shí)可以完整地看到項(xiàng)目代碼：
好，環(huán)境以及源碼準(zhǔn)備完畢，接下來將結(jié)合代碼講述一個(gè)爬蟲框架的編寫流程
Request & Response
Request類的目的是對(duì)aiohttp加一層封裝進(jìn)行模擬請(qǐng)求，功能如下：
封裝GET、POST兩種請(qǐng)求方式
增加回調(diào)機(jī)制
自定義重試次數(shù)、休眠時(shí)間、超時(shí)、重試解決方案、請(qǐng)求是否成功驗(yàn)證等功能
將返回的一系列數(shù)據(jù)封裝成Response類返回
接下來就簡(jiǎn)單了，不過就是實(shí)現(xiàn)上述需求，首先，需要實(shí)現(xiàn)一個(gè)函數(shù)來抓取目標(biāo)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)
實(shí)際運(yùn)行一下，會(huì)輸出請(qǐng)求狀態(tài)200，就這樣簡(jiǎn)單封裝一下，我們已經(jīng)有了自己的請(qǐng)求類Request，接下來只需要再完善一下重試機(jī)制以及將返回的屬性封裝一下就基本完成了：
# 重試函數(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來實(shí)際請(qǐng)求一個(gè)目標(biāo)網(wǎng)頁(yè)，如下：
這段代碼請(qǐng)求了目標(biāo)網(wǎng)頁(yè)https://docs.python-ruia.org/并返回了Response對(duì)象，其中Response提供屬性介紹如下：
Field & Item
實(shí)現(xiàn)了對(duì)目標(biāo)網(wǎng)頁(yè)的請(qǐng)求，接下來就是對(duì)目標(biāo)網(wǎng)頁(yè)進(jìn)行字段提取，我覺得ORM的思想很適合用在這里，我們只需要定義一個(gè)Item類，類里面每個(gè)屬性都可以用Field類來定義，然后只需要傳入url或者html，執(zhí)行過后Item類里面 定義的屬性會(huì)自動(dòng)被提取出來變成目標(biāo)字段值
可能說起來比較拗口，下面直接演示一下可能你就明白這樣寫的好，假設(shè)你的需求是獲取HackerNews網(wǎng)頁(yè)的title和url，可以這樣實(shí)現(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)被賦與實(shí)際的目標(biāo)值，這樣寫起來是不是很簡(jiǎn)潔清晰也很明了呢？
來看看怎么實(shí)現(xiàn)，Field類的目的是提供多種方式讓開發(fā)者提取網(wǎng)頁(yè)字段，比如：
XPath
CSS Selector
RE
所以我們只需要根據(jù)需求，定義父類然后再利用不同的提取方式實(shí)現(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
核心類就是上面的代碼，具體實(shí)現(xiàn)請(qǐng)看ruia/field.py
接下來繼續(xù)說Item部分，這部分實(shí)際上是對(duì)ORM那塊的實(shí)現(xiàn)，用到的知識(shí)點(diǎn)是元類，因?yà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 = {}
這一層弄明白接下來就很簡(jiǎn)單了，還記得上一篇文章《談?wù)剬?duì)Python爬蟲的理解》里面說的四個(gè)類型的目標(biāo)網(wǎng)頁(yè)么：
單頁(yè)面單目標(biāo)
單頁(yè)面多目標(biāo)
多頁(yè)面單目標(biāo)
多頁(yè)面多目標(biāo)
本質(zhì)來說就是要獲取網(wǎng)頁(yè)的單目標(biāo)以及多目標(biāo)（多頁(yè)面可以放在Spider那塊實(shí)現(xiàn)），Item類只需要定義兩個(gè)方法就能實(shí)現(xiàn)：
get_item()：?jiǎn)文繕?biāo)
get_items()：多目標(biāo)，需要定義好target_item
具體實(shí)現(xiàn)見：ruia/item.py
Spider
在Ruia框架中，為什么要有Spider，有以下原因：
真實(shí)世界爬蟲是多個(gè)頁(yè)面的（或深度或廣度），利用Spider可以對(duì)這些進(jìn)行 有效的管理
制定一套爬蟲程序的編寫標(biāo)準(zhǔn)，可以讓開發(fā)者容易理解、交流，能迅速產(chǎn)出高質(zhì)量爬蟲程序
自由地定制插件
接下來說說代碼實(shí)現(xiàn)，Ruia框架的API寫法我有參考Scrapy，各個(gè)函數(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()
使用起來還是挺簡(jiǎn)潔的，輸出如下：
