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

所以我们只需要根据需求，定义父类然后再利用不同的提取方式实现子类即可，代码如下：

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: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爬虫框架相关推荐

python文件运行哪一个_如何使一个python文件运行另一个？
本问题已经有最佳答案,请猛点这里访问. 如何创建一个python文件来运行另一个? 例如,我有两个.py文件.我想运行一个文件,然后让它运行另一个.py文件. docs.python.org / / ...
python大数据免费_安利大家一个Python大数据分析神器
对于Pandas运行速度的提升方法,之前已经介绍过很多回了,里面经常提及Dask,很多朋友没接触过可能不太了解,今天就推荐一下这个神器.StB免费资源网 1.什么是Dask?StB免费资源网 Pand ...
python判断对象类型_如何查看一个Python对象的类型
怎么查看一个对象的类型在Python中有两种类型判断函数,type()和isinstance(). 使用type() 首先,我们来判断对象类型,使用type()函数: 基本类型都可以用type()判 ...
c4d python使用方法_Cinema 4D Python脚本编写必须知道的5行Python代码教程
Cinema 4D Python脚本编写必须知道的5行Python代码教程在这期Cinema 4D python教程中的第一篇.Alpha Pixel将介绍如何使用python脚本进行设置,并查看一 ...
pythonscrapy爬虫_零基础写python爬虫之使用Scrapy框架编写爬虫
网络爬虫,是在网上进行数据抓取的程序,使用它能够抓取特定网页的HTML数据.虽然我们利用一些库开发一个爬虫程序,但是使用框架可以大大提高效率,缩短开发时间.Scrapy是一个使用Python编写的,轻 ...
python分布式爬虫系统_如何构建一个分布式爬虫：理论篇
前言本系列文章计划分三个章节进行讲述,分别是理论篇.基础篇和实战篇.理论篇主要为构建分布式爬虫而储备的理论知识,基础篇会基于理论篇的知识写一个简易的分布式爬虫,实战篇则会以微博为例,教大家做一个比较 ...
python爬虫分布式怎么构造_如何构建一个分布式爬虫：基础篇
继上篇(如何构建一个分布式爬虫:理论篇)我们谈论了Celery的基本知识后,本篇继续讲解如何一步步使用Celery构建分布式爬虫.这次我们抓取的对象定为celery官方文档(Celery - Dist ...
python如何编写数据库_如何在几分钟内用Python编写一个简单的玩具数据库
python如何编写数据库 MySQL, PostgreSQL, Oracle, Redis, and many more, you just name it - databases are a re ...
python爬虫分布式怎么构造_如何构建一个分布式爬虫：实战篇
本篇文章将是『如何构建一个分布式爬虫』系列文章的最后一篇,拟**从实战角度**来介绍如何构建一个*稳健的分布式微博爬虫*.这里我*没敢谈高效*,抓过微博数据的同学应该都知道微博的反爬虫能力,也知道微博 ...

python如何编写爬虫_如何实现一个Python爬虫框架

python如何编写爬虫_如何实现一个Python爬虫框架相关推荐

最新文章

热门文章