Python 异步网络爬虫 II

上一部分整理了如何利用 aiohttp 和 asyncio 执行异步网络请求，接下来我们将在此基础上实现一个简洁、普适的爬虫框架。

一般网站抓取的流程是这样的：

从入口页面开始提取一组下一级页面的链接，然后递归地执行下去，直到最后一层页面为止。唯一不同的是对每一级页面所要抓取的信息，也就是需要的正则表达式不同，除此之外，请求页面、分析内容、正则匹配的步骤是重复的，因此可以将上面的过程简化为：

其中虚线框中的步骤可以抽象出来，即模拟浏览器行为的 Client，Cleaning 方法用于对正则匹配的结果进行清理，并将下一级所需的入口地址返回给 Client，在这一过程中也可能涉及到数据输出到数据库的过程。
这里我参考了 Flask（或 Sanic）框架的设计，即利用 Python 装饰器的语法特性，将不同页面的 Cleaning 方法注册到 Client 中：

          
import asyncio  
import aiohttp  
import async_timeout

import re

class AvGot(object):  
  def __init__(self, loop=None):
    self.loop = loop

    self.headers = {
      "User-Agent": ("Mozilla/5.0 (Macintosh; Intel Mac OS X 10_11_1)"
      " AppleWebKit/537.36 (KHTML, like Gecko)"
      " Chrome/54.0.2840.87 Safari/537.36"),
    }
    self.conn = aiohttp.TCPConnector(verify_ssl=False)
    self.session = aiohttp.ClientSession(loop=loop,
                    connector=self.conn,
                    headers=self.headers)

    self.pipe = []

  async def fetch(self, url=""):
    with async_timeout(10):
      async with self.session.get(url) as resp:
        return await resp.text()
  async def extract(self, url="", regexp=r''):
    html = await self.fetch(url)
    matches = re.findall(regexp, html)
    return matches

  def entry(self, url='', regexp=r''):
    def wrapper(callback):
      self.pipe.append(callback)
    return wrapper

  def close(self):
    self.session()
    self.loop.close()
  def run(self):
    pass

loop = asyncio.get_event_loop()  
av = AvGot(loop)

reTag = re.compile('<a href="(\/tag\/.*?)">(.*?)<\/a>'))  
ROOT  = "https://movie.douban.com/tag/"

@av.entry(ROOT, reTag)
async def entry_callback(result):  
  # await db.save(result) 保存到数据库
  def clean(row):
    return ("https://movie.douban.com{}".format(row[0]), row[1])
  return list(map(clean, result))[:2]

av.run()  
av.close()

这时我们遇到一个比较棘手的问题：由于从当前页面提取数据（extract()）的过程是异步的， 而在上一个页面执行完成之前是无法进入下一个页面的，也就是说同一级页面之间是异步的，不同层级页面之间是同步的，那么如何在 asyncio 中安排这种任务？
这其实是一个带有递归属性的生产者/消费者模型，上级页面作为生产者只有在经过网络请求之后才能生产出下级所有入口链接，而下一级的消费者将成为下下一级的生产者……我们可以将事件循环看作是一个“传送带”，一些可能造成阻塞的任务（如extract()）会被挂起，等阻塞任务完成后重新进入队列等待被执行：

在上面的问题中，不同层级页面的提取过程可以被封装成 Task 并丢进任务队列，只是不同任务携带不同的页面地址、正则表达式、Cleaning 回调函数等属性，至于这些任务在具体执行时如何调度，就丢给事件循环去操心好了（这也是使用 asyncio 的一条基本原则）：

          
# 下面只列举更改后的代码
from collections import namedtuple

# Task 所需要携带的属性
Node = namedtuple("Node", ["url", "re", "callback"])

class AvGot(object):  
  _ENTRY = "ENTRY_NODE"
  def __init__(self, loop=None):
    self._prev_node = None
    self.pipe = {}

    # 异步任务队列
    self.queue = asyncio.Queue()
  def entry(self, url="", regexp=""):
    def wrapper(callback):
      node = Node(url, regexp, callback)
      if self.pipe.get(self._ENTRY) is None:
        self.pipe[self._ENTRY] = node
      else:
        # 以 Cleaning 函数而不是 node 作为 Key
        # 因为任务队列中需要构造新的 node
        self.pipe[self.prev_node.callback] = node
      self._prev_node = node
    return wrapper
  def register(self, regexp=r''):
    """
     除入口页面
     其他页面地址 url 依赖上级页面提取结果
    """
    return self.entry("", regexp)

  def run(self):
    # 将入口页面放入队列
    self.queue.put_nowait(self.pipe.get(self._ENTRY))

    async def _runner():
      producer = asyncio.ensure_future(self._worker())
      await self.queue.join()
      producer.cancel()

    self.loop.run_until_complete(_runner())


  async def _worker():
    while True:
      node = self.queue.get()

      # Cleaning 函数在这里回调，并产生下一级页面入口
      results = await node.callback(await self.extract(node.url, node.re))

      if results is not None:
        for page in results:
          # 从 pipe 链表中取出下一级的 node
          p = self.pipe.get(node.callback)
          if p is not None:
            # 根据结果中的 url 构造新的任务并放回到队列里
            next_node = Node(page[0], p.re, p.callback)
            self.queue.put_nowait(next_node)
      self.queue.task_done()

以上就是异步爬虫的基本结构，有一点需要约定好的是所有的 Cleaning 方法必须以列表形式返回清洗之后的结果，且下一级页面入口必须在第一位（最后一页除外）。接下来做一个简单的测试，以豆瓣电影分类页面为入口，进入该类别列表，最后进入电影详情页面，并提取电影时长和评分：

        
ROOT = "https://movie.douban.com/tag/"

# 正则：类别地址与类别名称
reTag = re.compile('<a href="(\/tag\/.*?)">(.*?)<\/a>')  
# 正则：详情页面链接及电影标题
reLinkTitle = re.compile('<a href="(https:\/\/movie\.douban\.com\/subject\/\d+/)".*?>([\s\S]*?)<\/a>')  
# 正则：电影时长及评分
reRuntimeRate = re.compile('<span property="v:runtime" content="(\d+)">[\s\S]*?<strong class="ll rating_num" property="v:average">(.*?)<\/strong>')

@av.entry(ROOT, reTag)
async def entry_callback(results):  
  # 构造列表页地址
  return list(map(lambda row: ("https://movie.douban.com{}".format(row[0]), row[1]), result))
@av.register(reLinkTitle)
async def list_page(result):  
  # 显示未清理前结果
  print(result)
  def clean(row):
    return (row[0], re.sub(r'<.*?>|\s', "", row[1]))
  return list(map(clean, result))

@av.register(reRuntimeRate)
async def detail_page(result):  
  print(result)

av.run()  
av.close()

从上面的执行的结果可以看出，正则表达式有时候不能（或不便）直接精确过滤我们所需内容，因此可以在 Cleaning 函数中进行清理（如去掉多余 Tag 或空位符等），另外：

• 不像 Flask，这里通过 register 注册方法的顺序必须与页面处理顺序保持一致；
• 与 Sanic 一样，注册方法必须也是 Coroutine （async def），同时可以在其中异步执行数据库存储操作；
• 上级页面信息实际上可以通过扩展 Node 直接传递给下级页面，这在某些相关页面中甚至是必须的；

总结

抽象这一框架的目的主要有以下几点：

1. 学习使用 asyncio 库及基于协程的异步；
2. 将网络爬虫的编写过程聚焦到页面关系分析、精确正则表达式及少量数据清理上；
3. 简化使用，降低学习成本。

仍有以下内容需要完成：

1. 错误捕捉与 logging，让调试过程更简单；
2. 寻找不适应该框架的情况，进行 upgrade；
3. 性能测试；
4. 完善浏览器模拟：Headers、proxy、Referer等……

未完待续。

参考

1. Sanic
2. Asyncio Doc::Producer/consumer

原文链接：http://blog.rainy.im/2016/11/09/python-async-webscraper-ii/

收藏

喜欢 0