python爬蟲詳細解析附案例

什么是爬蟲框架

說這個之前，得先說說什么是框架：

• 是實現業界標准的組件規范：比如眾所周知的MVC開發規范
• 提供規范所要求之基礎功能的軟件產品：比如Django框架就是MVC的開發框架，但它還提供了其他基礎功能幫助我們快速開發，比如中間件、認證系統等

框架的關注點在於規范二字，好，我們要寫的Python爬蟲框架規范是什么？

很簡單，爬蟲框架就是對爬蟲流程規范的實現，不清楚的朋友可以看上一篇文章談談對Python爬蟲的理解，下面總結一下爬蟲流程：

• 請求&響應
• 解析
• 持久化

這三個流程有沒有可能以一種優雅的形式串聯起來，Ruia目前是這樣實現的，請看代碼示例：

 

 

可以看到，Item & Field類結合一起實現了字段的解析提取，Spider類結合Request * Response類實現了對爬蟲程序整體的控制，從而可以如同流水線一般編寫爬蟲，最后返回的item可以根據使用者自身的需求進行持久化，這幾行代碼，我們就實現了獲取目標網頁請求、字段解析提取、持久化這三個流程

實現了基本流程規范之后，我們繼而就可以考慮一些基礎功能，讓使用者編寫爬蟲可以更加輕松，比如：中間件(Ruia里面的Middleware)、提供一些hook讓用戶編寫爬蟲更方便(比如ruia-motor)

這些想明白之后，接下來就可以愉快地編寫自己心目中的爬蟲框架了

如何踏出第一步

首先，我對Ruia爬蟲框架的定位很清楚，基於asyncio & aiohttp的一個輕量的、異步爬蟲框架，怎么實現呢，我覺得以下幾點需要遵守：

• 輕量級，專注於抓取、解析和良好的API接口
• 插件化，各個模塊耦合程度盡量低，目的是容易編寫自定義插件
• 速度，異步無阻塞框架，需要對速度有一定追求

什么是爬蟲框架如今我們已經很清楚了，現在急需要做的就是將流程規范利用Python語言實現出來，怎么實現，分為哪幾個模塊，可以看如下圖示：



 

 

同時讓我們結合上面一節的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項目：

 

 

選擇剛剛pipenv配置好的python解釋器：

 

 

此時可以完整地看到項目代碼：

 

 

好，環境以及源碼准備完畢，接下來將結合代碼講述一個爬蟲框架的編寫流程

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來實際請求一個目標網頁，如下：

 

 

這段代碼請求了目標網頁https://docs.python-ruia.org/並返回了Response對象，其中Response提供屬性介紹如下：

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())

 

 

從輸出結果可以看到，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 <GET: https://news.ycombinator.com/news?p=1>
[2019:03:14 10:29:04] INFO  Request <GET: https://news.ycombinator.com/news?p=2>
[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的請求控制，目前采用的是生產消費者模式來處理，具體函數如下：

 

 

詳細代碼，見ruia/spider.py

更多

至此，爬蟲框架的核心部分已經實現完畢，基礎功能同樣一個不落地實現了，接下來要做的就是：

• 實現更多優雅地功能
• 實現更多的插件，讓生態豐富起來
• 修BU

更多推薦：

https://www.cnblogs.com/leetcodetijie/p/13175504.html