python如何编写爬虫_如何实现一个Python爬虫框架
image
這篇文章的題目有點大，但這并不是說我自覺對Python爬蟲這塊有多大見解，我只不過是想將自己的一些經驗付諸于筆，對于如何寫一個爬蟲框架，我想一步一步地結合具體代碼來講述如何從零開始編寫一個自己的爬蟲框架
2018年到如今，我花精力比較多的一個開源項目算是Ruia了，這是一個基于Python3.6+的異步爬蟲框架，當時也獲得一些推薦，比如Github Trending Python語言榜單第二，目前Ruia還在開發中，Star數目不過700+，如果各位有興趣，歡迎一起開發，來波star我也不會拒絕哈~
什么是爬蟲框架
說這個之前，得先說說什么是框架：
是實現業界標準的組件規范：比如眾所周知的MVC開發規范
提供規范所要求之基礎功能的軟件產品：比如Django框架就是MVC的開發框架，但它還提供了其他基礎功能幫助我們快速開發，比如中間件、認證系統等
框架的關注點在于規范二字，好，我們要寫的Python爬蟲框架規范是什么？
很簡單，爬蟲框架就是對爬蟲流程規范的實現，不清楚的朋友可以看上一篇文章談談對Python爬蟲的理解，下面總結一下爬蟲流程：
請求&響應
解析
持久化
這三個流程有沒有可能以一種優雅的形式串聯起來，Ruia目前是這樣實現的，請看代碼示例：
image
可以看到，Item & Field類結合一起實現了字段的解析提取，Spider類結合Request * Response類實現了對爬蟲程序整體的控制，從而可以如同流水線一般編寫爬蟲，最后返回的item可以根據使用者自身的需求進行持久化，這幾行代碼，我們就實現了獲取目標網頁請求、字段解析提取、持久化這三個流程
實現了基本流程規范之后，我們繼而就可以考慮一些基礎功能，讓使用者編寫爬蟲可以更加輕松，比如：中間件(Ruia里面的Middleware)、提供一些hook讓用戶編寫爬蟲更方便(比如ruia-motor)
這些想明白之后，接下來就可以愉快地編寫自己心目中的爬蟲框架了
如何踏出第一步
首先，我對Ruia爬蟲框架的定位很清楚，基于asyncio & aiohttp的一個輕量的、異步爬蟲框架，怎么實現呢，我覺得以下幾點需要遵守：
輕量級，專注于抓取、解析和良好的API接口
插件化，各個模塊耦合程度盡量低，目的是容易編寫自定義插件
速度，異步無阻塞框架，需要對速度有一定追求
什么是爬蟲框架如今我們已經很清楚了，現在急需要做的就是將流程規范利用Python語言實現出來，怎么實現，分為哪幾個模塊，可以看如下圖示：
image
同時讓我們結合上面一節的Ruia代碼來從業務邏輯角度看看這幾個模塊到底是什么意思：
Request：請求
Response：響應
Item & Field：解析提取
Spider：爬蟲程序的控制中心，將請求、響應、解析、存儲結合起來
這四個部分我們可以簡單地使用五個類來實現，在開始講解之前，請先克隆Ruia框架到本地：
# 請確保本地Python環境是3.6+
git clone https://github.com/howie6879/ruia.git
# 安裝pipenv
pip install pipenv
# 安裝依賴包
pipenv install --dev
然后用PyCharm打開Ruia項目：
[站外圖片上傳中...(image-87e8ff-1552612815023)]
選擇剛剛pipenv配置好的python解釋器：
[圖片上傳失敗...(image-6ca198-1552612815023)]
此時可以完整地看到項目代碼：
image
好，環境以及源碼準備完畢，接下來將結合代碼講述一個爬蟲框架的編寫流程
Request & Response
Request類的目的是對aiohttp加一層封裝進行模擬請求，功能如下：
封裝GET、POST兩種請求方式
增加回調機制
自定義重試次數、休眠時間、超時、重試解決方案、請求是否成功驗證等功能
將返回的一系列數據封裝成Response類返回
接下來就簡單了，不過就是實現上述需求，首先，需要實現一個函數來抓取目標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)
實際運行一下，會輸出請求狀態200，就這樣簡單封裝一下，我們已經有了自己的請求類Request，接下來只需要再完善一下重試機制以及將返回的屬性封裝一下就基本完成了：
# 重試函數
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來實際請求一個目標網頁，如下：
image
這段代碼請求了目標網頁https://docs.python-ruia.org/并返回了Response對象，其中Response提供屬性介紹如下：
image
Field & Item
實現了對目標網頁的請求，接下來就是對目標網頁進行字段提取，我覺得ORM的思想很適合用在這里，我們只需要定義一個Item類，類里面每個屬性都可以用Field類來定義，然后只需要傳入url或者html，執行過后Item類里面 定義的屬性會自動被提取出來變成目標字段值
可能說起來比較拗口，下面直接演示一下可能你就明白這樣寫的好，假設你的需求是獲取HackerNews網頁的title和url，可以這樣實現：
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())
[站外圖片上傳中...(image-19d70a-1552612815023)]
從輸出結果可以看到，title和url屬性已經被賦與實際的目標值，這樣寫起來是不是很簡潔清晰也很明了呢？
來看看怎么實現，Field類的目的是提供多種方式讓開發者提取網頁字段，比如：
XPath
CSS Selector
RE
所以我們只需要根據需求，定義父類然后再利用不同的提取方式實現子類即可，代碼如下：
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
核心類就是上面的代碼，具體實現請看ruia/field.py
接下來繼續說Item部分，這部分實際上是對ORM那塊的實現，用到的知識點是元類，因為我們需要控制類的創建行為：
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 = {}
這一層弄明白接下來就很簡單了，還記得上一篇文章《談談對Python爬蟲的理解》里面說的四個類型的目標網頁么：
單頁面單目標
單頁面多目標
多頁面單目標
多頁面多目標
本質來說就是要獲取網頁的單目標以及多目標（多頁面可以放在Spider那塊實現），Item類只需要定義兩個方法就能實現：
get_item()：單目標
get_items()：多目標，需要定義好target_item
具體實現見：ruia/item.py
Spider
在Ruia框架中，為什么要有Spider，有以下原因：
真實世界爬蟲是多個頁面的（或深度或廣度），利用Spider可以對這些進行 有效的管理
制定一套爬蟲程序的編寫標準，可以讓開發者容易理解、交流，能迅速產出高質量爬蟲程序
自由地定制插件
接下來說說代碼實現，Ruia框架的API寫法我有參考Scrapy，各個函數之間的聯結也是使用回調，但是你也可以直接使用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()
使用起來還是挺簡潔的，輸出如下：
[2019:03:14 10:29:04] INFO Spider Spider started!
[2019:03:14 10:29:04] INFO Spider Worker started: 4380434912
[2019:03:14 10:29:04] INFO Spider Worker started: 4380435048
[2019:03:14 10:29:04] INFO Request
[2019:03:14 10:29:04] INFO Request
[2019:03:14 10:29:08] INFO Spider Stopping spider: Ruia
[2019:03:14 10:29:08] INFO Spider Total requests: 2
[2019:03:14 10:29:08] INFO Spider Time usage: 0:00:03.426335
[2019:03:14 10:29:08] INFO Spider Spider finished!
Spider的核心部分在于對請求URL的請求控制，目前采用的是生產消費者模式來處理，具體函數如下：
image
詳細代碼，見ruia/spider.py
更多
至此，爬蟲框架的核心部分已經實現完畢，基礎功能同樣一個不落地實現了，接下來要做的就是：
實現更多優雅地功能
實現更多的插件，讓生態豐富起來
修BUG
項目地址點擊閱讀原文或者在github搜索ruia，如果你有興趣，請參與進來吧！
如果覺得寫得不錯，點個好看來個star唄~
image
總結
以上是生活随笔為你收集整理的python如何编写爬虫_如何实现一个Python爬虫框架的全部內容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： pythonmessage用法_djan
- 下一篇： python多列排序_Python pr