爬蟲平臺Crawlab核心原理--自動提取字段算法

時間 2019-11-06

原文原文鏈接

⚠注意: 可配置爬蟲如今僅在Python版本（v0.2.1-v0.2.4）可用，在最新版本Golang版本（v0.3.0）還暫時不可用，後續會加上，請關注近期更新javascript

背景

實際的大型爬蟲開發項目中，爬蟲工程師會被要求抓取監控幾十上百個網站。通常來講這些網站的結構大同小異，不一樣的主要是被抓取項的提取規則。傳統方式是讓爬蟲工程師寫一個通用框架，而後將各網站的提取規則作成可配置的，而後將配置工做交給更初級的工程師或外包出去。這樣作將爬蟲開發流水線化，提升了部分生產效率。可是，配置的工做仍是一個苦力活兒，仍是很是消耗人力。所以，自動提取字段應運而生。css

自動提取字段是Crawlab在版本v0.2.2中在可配置爬蟲基礎上開發的新功能。它讓用戶不用作任何繁瑣的提取規則配置，就能夠自動提取出可能的要抓取的列表項，作到真正的「一鍵抓取」，順利的話，開發一個網站的爬蟲能夠半分鐘內完成。市面上有利用機器學習的方法來實現自動抓取要提取的抓取規則，有一些能夠作到精準提取，但遺憾的是平臺要收取高額的費用，我的開發者或小型公司通常承擔不起。html

Crawlab的自動提取字段是根據人爲抓取的模式來模擬的，所以不用通過任何訓練就可使用。並且，Crawlab的自動提取字段功能不會向用戶收取費用，由於Crawlab自己就是免費的。java

算法介紹

算法的核心來自於人的行爲自己，經過查找網頁中看起來像列表的元素來定位列表及抓取項。通常咱們查找列表項是怎樣的一個過程呢？有人說：這還不容易嗎，一看就知道那個是各列表呀！兄弟，拜託... 我們是在程序的角度談這個的，它只理解HTML、CSS、JS這些代碼，並不像你那樣智能。node

咱們識別一個列表，首先要看它是否是有不少相似的子項；其次，這些列表一般來講看起來比較「複雜」，含有不少看得見的元素；最後，咱們還要關注分頁，分頁按鈕通常叫作「下一頁」、「下頁」、「Next」、「Next Page」等等。python

用程序能夠理解的語言，咱們把以上規則總結以下：git

列表項github

從根節點自上而下遍歷標籤；
對於每個標籤，若是包含多個一樣的子標籤，判斷爲列表標籤候選；
取子標籤（遞歸）個數最多的列表標籤候選爲列表標籤；

列表子項算法

對以上規則提取的列表標籤，對每一個子標籤（遞歸）進行遍歷
將有href的a標籤爲加入目標字段；
將有text的標籤爲加入目標字段。

分頁微信

對於每個標籤，若是標籤文本爲特定文本（「下一頁」、「下頁」、「next page」、「next」），選取該標籤爲目標標籤。

這樣，咱們就設計好了自動提取列表項、列表子項、分頁的規則。剩下的就是寫代碼了。我知道這樣的設計過於簡單，也過於理想，沒有考慮到一些特殊狀況。後面咱們將經過在一些知名網站上測試看看咱們的算法表現如何。

算法實現

算法實現很簡單。爲了更好的操做HTML標籤，咱們選擇了lxml庫做爲HTML的操做庫。lxml是python的一個解析庫，支持HTML和XML的解析，支持XPath、CSS解析方式，並且解析效率很是高。

自上而下的遍歷語法是sel.iter()。sel是etree.Element，而iter會從根節點自上而下遍歷各個元素，直到遍歷完全部元素。它是一個generator。

構造解析樹

在獲取到頁面的HTML以後，咱們須要調用lxml中的etree.HTML方法構造解析樹。代碼很簡單以下，其中r爲requests.get的Response

# get html parse tree
sel = etree.HTML(r.content)
複製代碼

這段帶代碼在SpiderApi._get_html方法裏。源碼請見這裏。

輔助函數

在開始構建算法以前，咱們須要實現一些輔助函數。全部函數是封裝在SpiderApi類中的，因此寫法與類方法同樣。

@staticmethod
def _get_children(sel):
    # 獲取全部不包含comments的子節點
    return [tag for tag in sel.getchildren() if type(tag) != etree._Comment]
複製代碼

@staticmethod
def _get_text_child_tags(sel):
    # 遞歸獲取全部文本子節點（根節點）
    tags = []
    for tag in sel.iter():
        if type(tag) != etree._Comment and tag.text is not None and tag.text.strip() != '':
            tags.append(tag)
    return tags
複製代碼

@staticmethod
def _get_a_child_tags(sel):
    # 遞歸獲取全部超連接子節點（根節點）
    tags = []
    for tag in sel.iter():
        if tag.tag == 'a':
            if tag.get('href') is not None and not tag.get('href').startswith('#') and not tag.get(
                    'href').startswith('javascript'):
                tags.append(tag)
    return tags
複製代碼

獲取列表項

下面是核心中的核心！同窗們請集中注意力。

咱們來編寫獲取列表項的代碼。如下是得到列表標籤候選列表list_tag_list的代碼。看起來稍稍有些複雜，但其實邏輯很簡單：對於每個節點，咱們得到全部子節點（一級），過濾出高於閾值（默認10）的節點，而後過濾出節點的子標籤類別惟一的節點。這樣候選列表就獲得了。

list_tag_list = []
threshold = spider.get('item_threshold') or 10
# iterate all child nodes in a top-down direction
for tag in sel.iter():
    # get child tags
    child_tags = self._get_children(tag)

    if len(child_tags) < threshold:
        # if number of child tags is below threshold, skip
        continue
    else:
        # have one or more child tags
        child_tags_set = set(map(lambda x: x.tag, child_tags))

        # if there are more than 1 tag names, skip
        if len(child_tags_set) > 1:
            continue

        # add as list tag
        list_tag_list.append(tag)
複製代碼

接下來咱們將從候選列表中篩選出包含最多文本子節點的節點。聽起來有些拗口，打個比方：一個電商網站的列表子項，也就是產品項，必定是有許多例如價格、產品名、賣家等信息的，所以會包含不少文本節點。咱們就是經過這種方式過濾掉文本信息很少的列表（例如菜單列表、類別列表等等），獲得最終的列表。在代碼裏咱們存爲max_tag。

# find the list tag with the most child text tags
max_tag = None
max_num = 0
for tag in list_tag_list:
    _child_text_tags = self._get_text_child_tags(self._get_children(tag)[0])
    if len(_child_text_tags) > max_num:
        max_tag = tag
        max_num = len(_child_text_tags)
複製代碼

下面，咱們將生成列表項的CSS選擇器。如下代碼實現的邏輯主要就是根據上面獲得的目標標籤根據其id或class屬性來生成CSS選擇器。

# get list item selector
item_selector = None
if max_tag.get('id') is not None:
    item_selector = f'#{max_tag.get("id")} > {self._get_children(max_tag)[0].tag}'
elif max_tag.get('class') is not None:
    cls_str = '.'.join([x for x in max_tag.get("class").split(' ') if x != ''])
    if len(sel.cssselect(f'.{cls_str}')) == 1:
        item_selector = f'.{cls_str} > {self._get_children(max_tag)[0].tag}'
複製代碼

找到目標列表項以後，咱們須要作的就是將它下面的文本標籤和超連接標籤提取出來。代碼以下，就不細講了。感興趣的讀者能夠看源碼來理解。

# get list fields
fields = []
if item_selector is not None:
    first_tag = self._get_children(max_tag)[0]
    for i, tag in enumerate(self._get_text_child_tags(first_tag)):
        if len(first_tag.cssselect(f'{tag.tag}')) == 1:
            fields.append({
                'name': f'field{i + 1}',
                'type': 'css',
                'extract_type': 'text',
                'query': f'{tag.tag}',
            })
        elif tag.get('class') is not None:
            cls_str = '.'.join([x for x in tag.get("class").split(' ') if x != ''])
            if len(tag.cssselect(f'{tag.tag}.{cls_str}')) == 1:
                fields.append({
                    'name': f'field{i + 1}',
                    'type': 'css',
                    'extract_type': 'text',
                    'query': f'{tag.tag}.{cls_str}',
                })

    for i, tag in enumerate(self._get_a_child_tags(self._get_children(max_tag)[0])):
        # if the tag is <a...></a>, extract its href
        if tag.get('class') is not None:
            cls_str = '.'.join([x for x in tag.get("class").split(' ') if x != ''])
            fields.append({
                'name': f'field{i + 1}_url',
                'type': 'css',
                'extract_type': 'attribute',
                'attribute': 'href',
                'query': f'{tag.tag}.{cls_str}',
            })
複製代碼