爬虫系列文章总目录：【2022 年】Python3 爬虫学习教程，本教程内容多数来自于《Python3网络爬虫开发实战（第二版）》一书，目前截止 2022 年，可以将爬虫基本技术进行系统讲解，同时将最新前沿爬虫技术如异步、JavaScript 逆向、AST、安卓逆向、Hook、智能解析、群控技术、WebAssembly、大规模分布式、Docker、Kubernetes 等，市面上目前就仅有《Python3 网络爬虫开发实战（第二版）》一书了，点击了解详情。

很多情况下，一些网站的页面或资源我们通常需要登录才能看到。比如说访问 GitHub 的个人设置页面，如果不登录是无法查看的；比如说 12306 买票提交订单的页面，如果不登录是无法提交订单的；比如说要发一条微博，如果不登录是无法发送的。

我们之前学习的案例都是爬取的无需登录即可访问的站点，但是诸如上面例子的情况非常非常多，那假如我们想要用爬虫来访问这些页面，比如用爬虫修改 GitHub 的个人设置，用爬虫提交购票订单，用爬虫发微博，能做到吗？

答案是可以，这里就需要用到一些模拟登录相关的技术了。

那么本节我们就先来了解一下模拟登录的一些基本原理和实现吧。

1. 网站登录验证的实现

我们要实现模拟登录，那就得首先了解网站登录验证的实现。

登录一般是需要两个内容，用户名和密码，有的网站可能是手机号和验证码，有的是微信扫码，有的是 OAuth 验证等等，但根本上来说，都是把一些可供认证的信息提交给了服务器。

比如这里我们就拿用户名和密码来说吧。用户在一个网页表单里面输入了这些内容，然后点击登录按钮的一瞬间，浏览器客户端就会向服务器发送一个登录请求，这个请求里面肯定就包含了用户名和密码信息，这时候，服务器需要处理一下这些信息，然后返回给客户端一个类似「凭证」的东西，有了这个「凭证」以后呢，客户端拿着这个「凭证」再去访问某些需要登录才能查看的页面，服务器自然就能”放行“了，返回对应的内容或执行对应的操作就好了。

形象点说呢，我们拿登录发微博和买票坐火车这两件事来类比。发微博就好像要坐火车，没票是没法坐火车的吧，要坐火车怎么办呢？当然是先买票了，我们拿钱去火车站买个票，有了票之后，进站口查验一下，没问题就自然能去坐火车了，这个票就是坐火车的「凭证」。那发微博也一样，我们有用户名和密码，请求下服务器，获得一个「凭证」，这就相当于买到了火车票，然后在发微博的时候拿着这个「凭证」去请求服务器，服务器校验没问题，自然就把微博发出去了。

那么问题来了，这个「凭证」到底是怎么生成和验证的呢？目前比较流行的实现方式有两种，一种是基于 Session + Cookie 的验证，一种是基于 JWT（JSON Web Token）的验证，下面我们来介绍下。

2. Session 和 Cookie

我们在第一章了解了 Session 和 Cookie 的基本概念。简而言之呢，Session 就是存在服务端的，里面保存了用户此次访问的会话信息，Cookie 则是保存在用户本地浏览器的，它会在每次用户访问网站的时候发送给服务器，Cookie 会作为 Request Headers 的一部分发送给服务器，服务器根据 Cookie 里面包含的信息判断找出其 Session 对象并做一些校验，不同的 Session 对象里面维持了不同访问用户的状态，服务器可以根据这些信息决定返回 Response 的内容。

我们以用户登录的情形来说吧，其实不同的网站对于用户的登录状态的实现是可能不同的，但是 Session 和 Cookie 一定是相互配合工作的。

下面梳理如下：

• 比如说，Cookie 里面可能只存了 Session ID 相关信息，服务器能根据 Cookie 找到对应的 Session，用户登录之后，服务器会把对应的 Session 里面标记一个字段，代表已登录状态或者其他信息（如角色、登录时间）等等，这样用户每次访问网站的时候都带着 Cookie 来访问，服务器就能找到对应的 Session，然后看一下 Session 里面的状态是登录状态，那就可以返回对应的结果或执行某些操作。
• 当然 Cookie 里面也可能直接存了某些凭证信息。比如说用户在发起登录请求之后，服务器校验通过，返回给客户端的 Response Headers 里面可能带有 Set-Cookie 字段，里面可能就包含了类似凭证的信息，这样客户端会执行设置 Cookie 的操作，将这些信息保存到 Cookie 里面，以后再访问网页时携带这些 Cookie 信息，服务器拿着这里面的信息校验，自然也能实现登录状态检测了。

以上两种情况几乎能涵盖大部分的 Session 和 Cookie 登录验证的实现，具体的实现逻辑因服务器而异，但 Session 和 Cookie 一定是需要相互配合才能实现的。

3. JWT

Web 开发技术是一直在发展的，近几年前后端分离的趋势越来越火，很多 Web 网站都采取了前后端分离的技术来实现。而且传统的基于 Session 和 Cookie 的校验也存在一定问题，比如服务器需要维护登录用户的 Session 信息，而且分布式部署不方便，也不太适合前后端分离的项目。

所以，JWT 技术应运而生。

JWT，英文全称为 JSON Web Token，是为了在网络应用环境间传递声明而执行的一种基于 JSON 的开放标准。实际上就是在每次登录的时候通过一个 Token 字符串来校验登录状态。JWT 的声明一般被用来在身份提供者和服务提供者之间传递被认证的用户身份信息，以便于从资源服务器获取资源，也可以增加一些额外的业务逻辑所必须的声明信息，所以这个 Token 也可直接被用于认证，也可传递一些额外信息。

有了 JWT，一些认证就不需要借助于 Session 和 Cookie 了，服务器也无须维护 Session 信息，减少了服务器的开销。服务器只需要有一个校验 JWT 的功能就好了，同时也可以做到分布式部署和跨语言的支持。

JWT 通常就是一个加密的字符串，它也有自己的标准，类似下面的这种格式：

1

eyJ0eXAxIjoiMTIzNCIsImFsZzIiOiJhZG1pbiIsInR5cCI6IkpXVCIsImFsZyI6IkhTMjU2In0.eyJVc2VySWQiOjEyMywiVXNlck5hbWUiOiJhZG1pbiIsImV4cCI6MTU1MjI4Njc0Ni44Nzc0MDE4fQ.pEgdmFAy73walFonEm2zbxg46Oth3dlT02HR9iVzXa8

我们可以发现中间有两个用来分割的 . ，因此可以把它看成是一个三段式的加密字符串。

它由三部分构成，分别是 Header、Payload、Signature。

• Header，声明了 JWT 的签名算法，如 RSA、SHA256 等，也可能包含 JWT 编号或类型等数据，然后对整个信息进行 Base64 编码即可。
• Payload，通常用来存放一些业务需要但不敏感的信息，如 UserID 等，另外它也有很多默认是字段，如 JWT 签发者、JWT 接受者、JWT 过期时间等，Base64 编码即可。
• Signature，就是一个签名，是把 Header、Payload 的信息用秘钥 secret 加密后形成的，这个 secret 是保存在服务器端的，不能被轻易泄露。如此一来，即使一些 Payload 的信息被篡改，服务器也能通过 Signature 判断出非法请求，拒绝服务。

这三部分通过 . 组合起来就形成了 JWT 的字符串，就是用户的访问凭证。

所以这个登录认证流程也很简单了，用户拿着用户名密码登录，然后服务器生成 JWT 字符串返回给客户端。客户端每次请求都带着这个 JWT 就行了，服务器会自动判断其有效情况，如果有效，自然就返回对应的数据。JWT 的传输就多种多样了，可以将其放在 Request Headers 中，也可以放在 URL 里，甚至也有的网站把它放在 Cookie 里面，但总而言之，能传给服务器进行校验就好了。

好，到此为止呢，我们就已经了解了网站登录验证的实现了。

4. 模拟登录

好，那了解了网站登录验证的实现后，模拟登录自然就有思路了。

下面我们同样分两种认证方式来说明。

Session 和 Cookie

基于 Session 和 Cookie 的模拟登录，如果我们要用爬虫实现的话，其实最主要的就是把 Cookie 的信息维护好就行了，因为爬虫就相当于客户端浏览器，我们模拟好浏览器做的事情就好了。

一般怎么实现模拟登录呢？接下来我们结合之前所讲的技术总结一下。

• 第一，如果我们已经在浏览器中登录了自己的账号，要想用爬虫模拟，那么可以直接把 Cookie 复制过来交给爬虫。这是最省时省力的方式，相当于我们用浏览器手动操作登录了。我们把 Cookie 放到代码里，爬虫每次请求的时候再将其放到 Request Headers 中，完全模拟了浏览器的操作。之后服务器会通过 Cookie 校验登录状态，如果没问题，自然就可以执行某些操作或返回某些内容了。
• 第二，如果我们不想有任何手工操作，那么可以直接使用爬虫模拟登录过程。其实登录的过程多数也是一个 POST 请求。我们用爬虫提交了用户名、密码等信息给服务器，服务器返回的 Response Headers 里面可能会带有 Set-Cookie 的字段，我们只需要把这些 Cookie 保存下来就行了。所以，最主要的就是把这个过程中的 Cookie 维持好。当然这里可能会遇到一些困难，比如登录过程中伴随着各种校验参数，不好直接模拟请求；网站设置 Cookie 的过程是通过 JavaScript 实现的，所以可能还得仔细分析下其中的逻辑，尤其是我们用 requests 这样的请求库进行模拟登录的时候，遇到的问题经常比较多。
• 第三，我们也可以用一些简单的方式来实现模拟登录，即实现登录过程的自动化。比如我们用 Selenium、Pyppeteer 或 Playwright 来驱动浏览器模拟执行一些操作，如填写用户名和密码、提交表单等。登录成功后，通过 Selenium 或 Pyppeteer 获取当前浏览器的 Cookie 并保存即可。这样后续就可以拿着 Cookie 的内容发起请求，同样也能实现模拟登录。

以上介绍的就是一些常用的爬虫模拟登录的方案，其目的是维护好客户端的 Cookie 信息。总之，每次请求都携带好 Cookie 信息就能实现模拟登录了。

JWT

基于 JWT 的模拟登录思路也比较清晰了，由于 JWT 的字符串就是用户访问的凭证，所以模拟登录只需要做到下面几步。

• 第一步，模拟网站登录操作的请求。比如拿着用户名和密码信息请求登录接口，获取服务器返回的结果，这个结果中通常包含 JWT 字符串的信息，将它保存即可。
• 第二步，后续的请求携带 JWT 进行访问。在 JWT 不过期的情况下，通常能正常访问和执行对应的操作。携带方式多种多样，因网站而异。
• 第三步，如果 JWT 过期了，可能需要再次进行第一步，重新获取 JWT。

当然，模拟登录的过程肯定会带有一些其他的加密参数，需要根据实际情况具体分析。

4. 账号池

如果爬虫要求爬取的数据量比较大或爬取速度比较快，而网站又有单账号并发限制或者访问状态检测等反爬虫手段，那么我们的账号可能就会无法访问或者面临封号的风险了。

这时候一般怎么办呢？

我们可以使用分流的方案来实现。假设某个网站设置一分钟之内检测到同一个账号访问 3 次或 3 次以上则封号，我们就可以建立一个账号池，用多个账号来随机访问或爬取数据，这样就能大幅提高爬虫的并发量，降低被封号的风险了。比如我们可以准备 100 个账号，然后 100 个账号都模拟登录，把对应的 Cookie 或 JWT 存下来，每次访问的时候随机取一个来，由于账号多，所以每个账号被取用的概率也就降下来了，这样就能避免单账号并发过大的问题，也降低封号风险。

5. 总结

本节我们首先了解了 Session + Cookie 和 JWT 模拟登录的原理，接着初步了解了两种模拟登录方式的实现思路，最后初步介绍了一下账号池的作用。

后文我们会通过几个实战案例来实现上述两种方案的模拟登录，为了更好地理解后文的实战内容，建议好好理解本节所介绍的内容。

系列文章总目录：【2022 年】Python3 爬虫学习教程，本教程内容多数来自于《Python3网络爬虫开发实战（第二版）》一书，目前截止 2022 年，可以将爬虫基本技术进行系统讲解，同时将最新前沿爬虫技术如异步、JavaScript 逆向、AST、安卓逆向、Hook、智能解析、群控技术、WebAssembly、大规模分布式、Docker、Kubernetes 等，市面上目前就仅有《Python3 网络爬虫开发实战（第二版）》一书了，点击了解详情。

在上一节中，我们介绍了异步爬虫的基本原理和 asyncio 的基本用法，并且在最后简单提及了使用 aiohttp 来实现网页爬取的过程。在本节中，我们来介绍一下 aiohttp 的常见用法。

1. 基本介绍

前面介绍的 asyncio 模块内部实现了对 TCP、UDP、SSL 协议的异步操作，但是对于 HTTP 请求来说，我们就需要用到 aiohttp 来实现了。

aiohttp 是一个基于 asyncio 的异步 HTTP 网络模块，它既提供了服务端，又提供了客户端。其中我们用服务端可以搭建一个支持异步处理的服务器，就是用来处理请求并返回响应的，类似于 Django、Flask、Tornado 等一些 Web 服务器。而客户端可以用来发起请求，类似于使用 requests 发起一个 HTTP 请求然后获得响应，但 requests 发起的是同步的网络请求，aiohttp 则是异步的。

本节中，我们主要了解一下 aiohttp 客户端部分的用法。

2. 基本实例

首先，我们来看一个基本的 aiohttp 请求案例，代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

import aiohttp
import asyncio

async def fetch(session, url):
    async with session.get(url) as response:
        return await response.text(), response.status

async def main():
    async with aiohttp.ClientSession() as session:
        html, status = await fetch(session, 'https://cuiqingcai.com')
        print(f'html: {html[:100]}...')
        print(f'status: {status}')

if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())

这里我们使用 aiohttp 来爬取我的个人博客，获得了源码和响应状态码并输出出来，运行结果如下：

1
2
3
4
5
6

html: <!DOCTYPE HTML>
<html>
<head>
<meta charset="UTF-8">
<meta name="baidu-tc-verification" content=...
status: 200

这里网页源码过长，只截取输出了一部分。可以看到，这里我们成功获取了网页的源代码及响应状态码 200，也就完成了一次基本的 HTTP 请求，即我们成功使用 aiohttp 通过异步的方式来进行了网页爬取。当然，这个操作用之前讲的 requests 也可以做到。

可以看到，其请求方法的定义和之前有了明显的区别，主要有如下几点：

• 首先在导入库的时候，我们除了必须要引入 aiohttp 这个库之外，还必须要引入 asyncio 这个库。因为要实现异步爬取，需要启动协程，而协程则需要借助于 asyncio 里面的事件循环来执行。除了事件循环，asyncio 里面也提供了很多基础的异步操作。
• 异步爬取方法的定义和之前有所不同，在每个异步方法前面统一要加 async 来修饰。
• with as 语句前面同样需要加 async 来修饰。在 Python 中，with as 语句用于声明一个上下文管理器，能够帮我们自动分配和释放资源。而在异步方法中，with as 前面加上 async 代表声明一个支持异步的上下文管理器。
• 对于一些返回 coroutine 的操作，前面需要加 await 来修饰。比如 response 调用 text 方法，查询 API 可以发现，其返回的是 coroutine 对象，那么前面就要加 await；而对于状态码来说，其返回值就是一个数值类型，那么前面就不需要加 await。所以，这里可以按照实际情况处理，参考官方文档说明，看看其对应的返回值是怎样的类型，然后决定加不加 await 就可以了。
• 最后，定义完爬取方法之后，实际上是 main 方法调用了 fetch 方法。要运行的话，必须要启用事件循环，而事件循环就需要使用 asyncio 库，然后使用 run_until_complete 方法来运行。

注意：在 Python 3.7 及以后的版本中，我们可以使用 asyncio.run(main()) 来代替最后的启动操作，不需要显示声明事件循环，run 方法内部会自动启动一个事件循环。但这里为了兼容更多的 Python 版本，依然还是显式声明了事件循环。

3. URL 参数设置

对于 URL 参数的设置，我们可以借助于 params 参数，传入一个字典即可，示例如下：

1
2
3
4
5
6
7
8
9
10
11

import aiohttp
import asyncio

async def main():
    params = {'name': 'germey', 'age': 25}
    async with aiohttp.ClientSession() as session:
        async with session.get('https://httpbin.org/get', params=params) as response:
            print(await response.text())

if __name__ == '__main__':
    asyncio.get_event_loop().run_until_complete(main())

运行结果如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

{
  "args": {
    "age": "25",
    "name": "germey"
  },
  "headers": {
    "Accept": "*/*",
    "Accept-Encoding": "gzip, deflate",
    "Host": "httpbin.org",
    "User-Agent": "Python/3.7 aiohttp/3.6.2",
    "X-Amzn-Trace-Id": "Root=1-5e85eed2-d240ac90f4dddf40b4723ef0"
  },
  "origin": "17.20.255.122",
  "url": "https://httpbin.org/get?name=germey&age=25"
}

这里可以看到，其实际请求的 URL 为 https://httpbin.org/get?name=germey&age=25，其 URL 请求参数就对应了 params 的内容。

4. 其他请求类型

另外，aiohttp 还支持其他请求类型，如 POST、PUT、DELETE 等，这和 requests 的使用方式有点类似，示例如下：

1
2
3
4
5
6

session.post('http://httpbin.org/post', data=b'data')
session.put('http://httpbin.org/put', data=b'data')
session.delete('http://httpbin.org/delete')
session.head('http://httpbin.org/get')
session.options('http://httpbin.org/get')
session.patch('http://httpbin.org/patch', data=b'data')

要使用这些方法，只需要把对应的方法和参数替换一下即可。

5. POST 请求

对于 POST 表单提交，其对应的请求头的 Content-Type 为 application/x-www-form-urlencoded，我们可以用如下方式来实现，代码示例如下：

1
2
3
4
5
6
7
8
9
10
11

import aiohttp
import asyncio

async def main():
    data = {'name': 'germey', 'age': 25}
    async with aiohttp.ClientSession() as session:
        async with session.post('https://httpbin.org/post', data=data) as response:
            print(await response.text())

if __name__ == '__main__':
    asyncio.get_event_loop().run_until_complete(main())

运行结果如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

{
  "args": {},
  "data": "",
  "files": {},
  "form": {
    "age": "25",
    "name": "germey"
  },
  "headers": {
    "Accept": "*/*",
    "Accept-Encoding": "gzip, deflate",
    "Content-Length": "18",
    "Content-Type": "application/x-www-form-urlencoded",
    "Host": "httpbin.org",
    "User-Agent": "Python/3.7 aiohttp/3.6.2",
    "X-Amzn-Trace-Id": "Root=1-5e85f0b2-9017ea603a68dc285e0552d0"
  },
  "json": null,
  "origin": "17.20.255.58",
  "url": "https://httpbin.org/post"
}

对于 POST JSON 数据提交，其对应的请求头的 Content-Type 为 application/json，我们只需要将 post 方法的 data 参数改成 json 即可，代码示例如下：

1
2
3
4
5

async def main():
    data = {'name': 'germey', 'age': 25}
    async with aiohttp.ClientSession() as session:
        async with session.post('https://httpbin.org/post', json=data) as response:
            print(await response.text())

运行结果如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

{
  "args": {},
  "data": "{\"name\": \"germey\", \"age\": 25}",
  "files": {},
  "form": {},
  "headers": {
    "Accept": "*/*",
    "Accept-Encoding": "gzip, deflate",
    "Content-Length": "29",
    "Content-Type": "application/json",
    "Host": "httpbin.org",
    "User-Agent": "Python/3.7 aiohttp/3.6.2",
    "X-Amzn-Trace-Id": "Root=1-5e85f03e-c91c9a20c79b9780dbed7540"
  },
  "json": {
    "age": 25,
    "name": "germey"
  },
  "origin": "17.20.255.58",
  "url": "https://httpbin.org/post"
}

可以发现，其实现也和 requests 非常像，不同的参数支持不同类型的请求内容。

6. 响应

对于响应来说，我们可以用如下方法分别获取响应的状态码、响应头、响应体、响应体二进制内容、响应体 JSON 结果，示例如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

import aiohttp
import asyncio

async def main():
    data = {'name': 'germey', 'age': 25}
    async with aiohttp.ClientSession() as session:
        async with session.post('https://httpbin.org/post', data=data) as response:
            print('status:', response.status)
            print('headers:', response.headers)
            print('body:', await response.text())
            print('bytes:', await response.read())
            print('json:', await response.json())

if __name__ == '__main__':
    asyncio.get_event_loop().run_until_complete(main())

运行结果如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

status: 200
headers: <CIMultiDictProxy('Date': 'Thu, 02 Apr 2020 14:13:05 GMT', 'Content-Type': 'application/json', 'Content-Length': '503', 'Connection': 'keep-alive', 'Server': 'gunicorn/19.9.0', 'Access-Control-Allow-Origin': '*', 'Access-Control-Allow-Credentials': 'true')>
body: {
  "args": {},
  "data": "",
  "files": {},
  "form": {
    "age": "25",
    "name": "germey"
  },
  "headers": {
    "Accept": "*/*",
    "Accept-Encoding": "gzip, deflate",
    "Content-Length": "18",
    "Content-Type": "application/x-www-form-urlencoded",
    "Host": "httpbin.org",
    "User-Agent": "Python/3.7 aiohttp/3.6.2",
    "X-Amzn-Trace-Id": "Root=1-5e85f2f1-f55326ff5800b15886c8e029"
  },
  "json": null,
  "origin": "17.20.255.58",
  "url": "https://httpbin.org/post"
}

bytes: b'{\n  "args": {}, \n  "data": "", \n  "files": {}, \n  "form": {\n    "age": "25", \n    "name": "germey"\n  }, \n  "headers": {\n    "Accept": "*/*", \n    "Accept-Encoding": "gzip, deflate", \n    "Content-Length": "18", \n    "Content-Type": "application/x-www-form-urlencoded", \n    "Host": "httpbin.org", \n    "User-Agent": "Python/3.7 aiohttp/3.6.2", \n    "X-Amzn-Trace-Id": "Root=1-5e85f2f1-f55326ff5800b15886c8e029"\n  }, \n  "json": null, \n  "origin": "17.20.255.58", \n  "url": "https://httpbin.org/post"\n}\n'
json: {'args': {}, 'data': '', 'files': {}, 'form': {'age': '25', 'name': 'germey'}, 'headers': {'Accept': '*/*', 'Accept-Encoding': 'gzip, deflate', 'Content-Length': '18', 'Content-Type': 'application/x-www-form-urlencoded', 'Host': 'httpbin.org', 'User-Agent': 'Python/3.7 aiohttp/3.6.2', 'X-Amzn-Trace-Id': 'Root=1-5e85f2f1-f55326ff5800b15886c8e029'}, 'json': None, 'origin': '17.20.255.58', 'url': 'https://httpbin.org/post'}

这里我们可以看到有些字段前面需要加 await，有的则不需要。其原则是，如果它返回的是一个 coroutine 对象（如 async 修饰的方法），那么前面就要加 await，具体可以看 aiohttp 的 API，其链接为：https://docs.aiohttp.org/en/stable/client_reference.html。

7. 超时设置

对于超时设置，我们可以借助 ClientTimeout 对象，比如这里要设置 1 秒的超时，可以这么实现：

1
2
3
4
5
6
7
8
9
10
11

import aiohttp
import asyncio

async def main():
    timeout = aiohttp.ClientTimeout(total=1)
    async with aiohttp.ClientSession(timeout=timeout) as session:
        async with session.get('https://httpbin.org/get') as response:
            print('status:', response.status)

if __name__ == '__main__':
    asyncio.get_event_loop().run_until_complete(main())

如果在 1 秒之内成功获取响应的话，运行结果如下：

1

200

如果超时的话，会抛出 TimeoutError 异常，其类型为 asyncio.TimeoutError，我们再进行异常捕获即可。

另外，声明 ClientTimeout 对象时还有其他参数，如 connect、socket_connect 等，详细可以参考官方文档：https://docs.aiohttp.org/en/stable/client_quickstart.html#timeouts。

8. 并发限制

由于 aiohttp 可以支持非常大的并发，比如上万、十万、百万都是能做到的，但对于这么大的并发量，目标网站很可能在短时间内无法响应，而且很可能瞬时间将目标网站爬挂掉，所以我们需要控制一下爬取的并发量。

一般情况下，我们可以借助于 asyncio 的 Semaphore 来控制并发量，示例如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

import asyncio
import aiohttp

CONCURRENCY = 5
URL = 'https://www.baidu.com'

semaphore = asyncio.Semaphore(CONCURRENCY)
session = None

async def scrape_api():
    async with semaphore:
        print('scraping', URL)
        async with session.get(URL) as response:
            await asyncio.sleep(1)
            return await response.text()

async def main():
    global session
    session = aiohttp.ClientSession()
    scrape_index_tasks = [asyncio.ensure_future(scrape_api()) for _ in range(10000)]
    await asyncio.gather(*scrape_index_tasks)


if __name__ == '__main__':
    asyncio.get_event_loop().run_until_complete(main())

这里我们声明了 CONCURRENCY（代表爬取的最大并发量）为 5，同时声明爬取的目标 URL 为百度。接着，我们借助于 Semaphore 创建了一个信号量对象，将其赋值为 semaphore，这样我们就可以用它来控制最大并发量了。怎么使用呢？这里我们把它直接放置在对应的爬取方法里面，使用 async with 语句将 semaphore 作为上下文对象即可。这样的话，信号量可以控制进入爬取的最大协程数量，即我们声明的 CONCURRENCY 的值。

在 main 方法里面，我们声明了 10000 个 task，将其传递给 gather 方法运行。倘若不加以限制，这 10000 个 task 会被同时执行，并发数量太大。但有了信号量的控制之后，同时运行的 task 的数量最大会被控制在 5 个，这样就能给 aiohttp 限制速度了。

9. 总结

本节我们了解了 aiohttp 的基本使用方法，更详细的内容还是推荐大家到官方文档查阅，详见 https://docs.aiohttp.org/。

本节代码：https://github.com/Python3WebSpider/AsyncTest。

爬虫系列文章总目录：【2022 年】Python3 爬虫学习教程，本教程内容多数来自于《Python3 网络爬虫开发实战（第二版）》一书，目前截止 2022 年，可以将爬虫基本技术进行系统讲解，同时将最新前沿爬虫技术如异步、JavaScript 逆向、AST、安卓逆向、Hook、智能解析、群控技术、WebAssembly、大规模分布式、Docker、Kubernetes 等，市面上目前就仅有《Python3 网络爬虫开发实战（第二版）》一书了，点击了解详情。

前面我们讲解了 Ajax 的分析方法，利用 Ajax 接口我们可以非常方便地完成数据爬取。只要我们能找到 Ajax 接口的规律，就可以通过某些参数构造出对应的请求，数据自然就能轻松爬取到。

但是在很多情况下，一些 Ajax 请求的接口通常会包含加密参数，如token、sign 等，如：https://spa2.scrape.center/，它的 Ajax 接口是包含一个 token 参数的，如图所示。

由于请求接口时必须加上 token 参数，所以我们如果不深入分析找到 token 的构造逻辑，是难以直接模拟这些 Ajax 请求的。

此时解决方法通常有两种：一种就是深挖其中的逻辑，把其中 token 的构造逻辑完全找出来，再用 Python 复现，构造 Ajax 请求；另外一种方法就是直接通过模拟浏览器的方式来绕过这个过程，因为在浏览器里我们可以看到这个数据，如果能把看到的数据直接爬取下来，当然也就能获取对应的信息了。

由于第一种方法难度较高，这里我们就先介绍第二种方法：模拟浏览器爬取。

这里使用的工具为 Selenium，这里就来先了解一下 Selenium 的基本使用方法。

Selenium 是一个自动化测试工具，利用它可以驱动浏览器执行特定的动作，如点击、下拉等操作，同时还可以获取浏览器当前呈现的页面的源代码，做到可见即可爬。对于一些 JavaScript 动态渲染的页面来说，此种抓取方式非常有效。本节中，就让我们来感受一下它的强大之处吧。

1. 准备工作

本节以 Chrome 为例来讲解 Selenium 的用法。在开始之前，请确保已经正确安装好了 Chrome 浏览器并配置好了 ChromeDriver。另外，还需要正确安装好 Python 的 Selenium 库。

安装方法可以参考：https://setup.scrape.center/selenium，全部配置完成之后，我们便可以开始本节的学习了。

2. 基本用法

准备工作做好之后，首先来大体看一下 Selenium 的功能。示例如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.common.keys import Keys
from selenium.webdriver.support import expected_conditions as EC
from selenium.webdriver.support.wait import WebDriverWait

browser = webdriver.Chrome()
try:
    browser.get('https://www.baidu.com')
    input = browser.find_element_by_id('kw')
    input.send_keys('Python')
    input.send_keys(Keys.ENTER)
    wait = WebDriverWait(browser, 10)
    wait.until(EC.presence_of_element_located((By.ID, 'content_left')))
    print(browser.current_url)
    print(browser.get_cookies())
    print(browser.page_source)
finally:
    browser.close()

运行代码后发现，会自动弹出一个 Chrome 浏览器。浏览器首先会跳转到百度，然后在搜索框中输入 Python，接着跳转到搜索结果页，如图所示。

此时在控制台的输出结果如下：

1
2
3

https://www.baidu.com/s?ie=utf-8&f=8&rsv_bp=0&rsv_idx=1&tn=baidu&wd=Python&rsv_pq=c94d0df9000a72d0&rsv_t=07099xvun1ZmC0bf6eQvygJ43IUTTUOl5FCJVPgwG2YREs70GplJjH2F%2BCQ&rqlang=cn&rsv_enter=1&rsv_sug3=6&rsv_sug2=0&inputT=87&rsv_sug4=87
[{'secure': False, 'value': 'B490B5EBF6F3CD402E515D22BCDA1598', 'domain': '.baidu.com', 'path': '/', 'httpOnly': False, 'name': 'BDORZ', 'expiry': 1491688071.707553}, {'secure': False, 'value': '22473_1441_21084_17001', 'domain': '.baidu.com', 'path': '/', 'httpOnly': False, 'name': 'H_PS_PSSID'}, {'secure': False, 'value': '12883875381399993259_00_0_I_R_2_0303_C02F_N_I_I_0', 'domain': '.www.baidu.com', 'path': '/', 'httpOnly': False, 'name': '__bsi', 'expiry': 1491601676.69722}]
<!DOCTYPE html><!--STATUS OK-->...</html>

源代码过长，在此省略。可以看到，我们得到的当前 URL、Cookies 和源代码都是浏览器中的真实内容。

所以说，如果用 Selenium 来驱动浏览器加载网页的话，就可以直接拿到 JavaScript 渲染的结果了，不用担心使用的是什么加密系统。

下面来详细了解一下 Selenium 的用法。

3. 声明浏览器对象

Selenium 支持非常多的浏览器，如 Chrome、Firefox、Edge 等，还有 Android、BlackBerry 等手机端的浏览器。我们可以用如下方式初始化：

1
2
3
4
5
6

from selenium import webdriver

browser = webdriver.Chrome()
browser = webdriver.Firefox()
browser = webdriver.Edge()
browser = webdriver.Safari()

这样就完成了浏览器对象的初始化并将其赋值为 browser 对象。接下来，我们要做的就是调用 browser 对象，让其执行各个动作以模拟浏览器操作。

4. 访问页面

我们可以用 get 方法来请求网页，其参数传入链接 URL 即可。比如，这里用 get 方法访问淘宝，然后打印出源代码，代码如下：

1
2
3
4
5
6

from selenium import webdriver

browser = webdriver.Chrome()
browser.get('https://www.taobao.com')
print(browser.page_source)
browser.close()

运行后发现，此时弹出了 Chrome 浏览器并且自动访问了淘宝，然后控制台输出了淘宝页面的源代码，随后浏览器关闭。

通过这几行简单的代码，我们可以实现浏览器的驱动并获取网页源码，非常便捷。

5. 查找节点

Selenium 可以驱动浏览器完成各种操作，比如填充表单、模拟点击等。比如，我们想要完成向某个输入框输入文字的操作，总需要知道这个输入框在哪里吧？而 Selenium 提供了一系列查找节点的方法，我们可以用这些方法来获取想要的节点，以便下一步执行一些动作或者提取信息。

单个节点

比如，想要从淘宝页面中提取搜索框这个节点，首先要观察它的源代码，如图所示。

可以发现，它的 id 是 q，name 也是 q。此外，还有许多其他属性，此时我们就可以用多种方式获取它了。比如，find_element_by_name 是根据 name 值获取，find_element_by_id 是根据 id 获取。另外，还有根据 XPath、CSS 选择器等获取的方式。

下面我们用代码实现一下：

1
2
3
4
5
6
7
8
9

from selenium import webdriver

browser = webdriver.Chrome()
browser.get('https://www.taobao.com')
input_first = browser.find_element_by_id('q')
input_second = browser.find_element_by_css_selector('#q')
input_third = browser.find_element_by_xpath('//*[@id="q"]')
print(input_first, input_second, input_third)
browser.close()

这里我们使用 3 种方式获取输入框，分别是根据 ID、CSS 选择器和 XPath 获取，它们返回的结果完全一致。运行结果如下：

1
2
3

<selenium.webdriver.remote.webelement.WebElement (session="5e53d9e1c8646e44c14c1c2880d424af", element="0.5649563096161541-1")>
<selenium.webdriver.remote.webelement.WebElement (session="5e53d9e1c8646e44c14c1c2880d424af", element="0.5649563096161541-1")>
<selenium.webdriver.remote.webelement.WebElement (session="5e53d9e1c8646e44c14c1c2880d424af", element="0.5649563096161541-1")>

可以看到，这 3 个节点都是 WebElement 类型，是完全一致的。

下面列出所有获取单个节点的方法：

1
2
3
4
5
6
7
8

find_element_by_id
find_element_by_name
find_element_by_xpath
find_element_by_link_text
find_element_by_partial_link_text
find_element_by_tag_name
find_element_by_class_name
find_element_by_css_selector

另外，Selenium 还提供了通用方法 find_element，它需要传入两个参数：查找方式 By 和值。实际上，它就是 find_element_by_id 这种方法的通用函数版本，比如 find_element_by_id(id) 就等价于 find_element(By.ID, id)，二者得到的结果完全一致。我们用代码实现一下：

1
2
3
4
5
6
7
8

from selenium import webdriver
from selenium.webdriver.common.by import By

browser = webdriver.Chrome()
browser.get('https://www.taobao.com')
input_first = browser.find_element(By.ID, 'q')
print(input_first)
browser.close()

实际上，这种查找方式的功能和上面列举的查找函数完全一致，不过参数更加灵活。

多个节点

如果查找的目标在网页中只有一个，那么完全可以用 find_element 方法。但如果有多个节点，再用 find_element 方法查找，就只能得到第一个节点了。如果要查找所有满足条件的节点，需要用 find_elements 这样的方法。注意，在这个方法的名称中，element 多了一个 s，注意区分。

比如，要查找淘宝左侧导航条的所有条目，就可以这样来实现：

1
2
3
4
5
6
7

from selenium import webdriver

browser = webdriver.Chrome()
browser.get('https://www.taobao.com')
lis = browser.find_elements_by_css_selector('.service-bd li')
print(lis)
browser.close()

运行结果如下：

1

[<selenium.webdriver.remote.webelement.WebElement (session="c26290835d4457ebf7d96bfab3740d19", element="0.09221044033125603-1")>, <selenium.webdriver.remote.webelement.WebElement (session="c26290835d4457ebf7d96bfab3740d19", element="0.09221044033125603-2")>, <selenium.webdriver.remote.webelement.WebElement (session="c26290835d4457ebf7d96bfab3740d19", element="0.09221044033125603-3")>...<selenium.webdriver.remote.webelement.WebElement (session="c26290835d4457ebf7d96bfab3740d19", element="0.09221044033125603-16")>]

这里简化了输出结果，中间部分省略。

可以看到，得到的内容变成了列表类型，列表中的每个节点都是 WebElement 类型。

也就是说，如果我们用 find_element 方法，只能获取匹配的第一个节点，结果是 WebElement 类型。如果用 find_elements 方法，则结果是列表类型，列表中的每个节点都是 WebElement 类型。

这里列出所有获取多个节点的方法：

1
2
3
4
5
6
7
8

find_elements_by_id
find_elements_by_name
find_elements_by_xpath
find_elements_by_link_text
find_elements_by_partial_link_text
find_elements_by_tag_name
find_elements_by_class_name
find_elements_by_css_selector

当然，我们也可以直接用 find_elements 方法来选择，这时可以这样写：

1

lis = browser.find_elements(By.CSS_SELECTOR, '.service-bd li')

结果是完全一致的。

6. 节点交互

Selenium 可以驱动浏览器来执行一些操作，也就是说可以让浏览器模拟执行一些动作。比较常见的用法有：输入文字时用 send_keys 方法，清空文字时用 clear 方法，点击按钮时用 click 方法。示例如下：

1
2
3
4
5
6
7
8
9
10
11
12

from selenium import webdriver
import time

browser = webdriver.Chrome()
browser.get('https://www.taobao.com')
input = browser.find_element_by_id('q')
input.send_keys('iPhone')
time.sleep(1)
input.clear()
input.send_keys('iPad')
button = browser.find_element_by_class_name('btn-search')
button.click()

这里首先驱动浏览器打开淘宝，然后用 find_element_by_id 方法获取输入框，然后用 send_keys 方法输入 iPhone 文字，等待一秒后用 clear 方法清空输入框，再次调用 send_keys 方法输入 iPad 文字，之后再用 find_element_by_class_name 方法获取搜索按钮，最后调用 click 方法完成搜索动作。

通过上面的方法，我们完成了一些常见节点的操作，更多的操作可以参见官方文档的交互动作介绍 ：http://selenium-python.readthedocs.io/api.html#module-selenium.webdriver.remote.webelement。

7. 动作链

在上面的实例中，一些交互动作都是针对某个节点执行的。比如，对于输入框，我们就调用它的输入文字和清空文字方法；对于按钮，就调用它的点击方法。其实，还有另外一些操作，它们没有特定的执行对象，比如鼠标拖曳、键盘按键等，这些动作用另一种方式来执行，那就是动作链。

比如，现在实现一个节点的拖曳操作，将某个节点从一处拖曳到另外一处，可以这样实现：

1
2
3
4
5
6
7
8
9
10
11
12

from selenium import webdriver
from selenium.webdriver import ActionChains

browser = webdriver.Chrome()
url = 'http://www.runoob.com/try/try.php?filename=jqueryui-api-droppable'
browser.get(url)
browser.switch_to.frame('iframeResult')
source = browser.find_element_by_css_selector('#draggable')
target = browser.find_element_by_css_selector('#droppable')
actions = ActionChains(browser)
actions.drag_and_drop(source, target)
actions.perform()

首先，打开网页中的一个拖曳实例，然后依次选中要拖曳的节点和拖曳到的目标节点，接着声明 ActionChains 对象并将其赋值为 actions 变量，然后通过调用 actions 变量的 drag_and_drop 方法，再调用 perform 方法执行动作，此时就完成了拖曳操作，如图所示。

更多的动作链操作可以参考官方文档的动作链介绍：http://selenium-python.readthedocs.io/api.html#module-selenium.webdriver.common.action_chains。

8. 执行 JavaScript

对于某些操作，Selenium API 并没有提供。比如，下拉进度条，它可以直接模拟运行 JavaScript，此时使用 execute_script 方法即可实现，代码如下：

1
2
3
4
5
6

from selenium import webdriver

browser = webdriver.Chrome()
browser.get('https://www.zhihu.com/explore')
browser.execute_script('window.scrollTo(0, document.body.scrollHeight)')
browser.execute_script('alert("To Bottom")')

这里就利用 execute_script 方法将进度条下拉到最底部，然后弹出 alert 提示框。

所以说有了这个方法，基本上 API 没有提供的所有功能都可以用执行 JavaScript 的方式来实现了。

9. 获取节点信息

前面说过，通过 page_source 属性可以获取网页的源代码，接着就可以使用解析库（如正则表达式、Beautiful Soup、pyquery 等）来提取信息了。

不过，既然 Selenium 已经提供了选择节点的方法，返回的是 WebElement 类型，那么它也有相关的方法和属性来直接提取节点信息，如属性、文本等。这样的话，我们就可以不用通过解析源代码来提取信息了，非常方便。

接下来，我们就来看看怎样获取节点信息吧。

获取属性

我们可以使用 get_attribute 方法来获取节点的属性，但是其前提是先选中这个节点，示例如下：

1
2
3
4
5
6
7
8

from selenium import webdriver

browser = webdriver.Chrome()
url = 'https://spa2.scrape.center/'
browser.get(url)
logo = browser.find_element_by_class_name('logo-image')
print(logo)
print(logo.get_attribute('src'))

运行之后，程序便会驱动浏览器打开该页面，然后获取 class 为 logo-image 的节点，最后打印出它的 src。

控制台的输出结果如下：

1
2

<selenium.webdriver.remote.webelement.WebElement (session="7f4745d35a104759239b53f68a6f27d0", element="cd7c72b4-4920-47ed-91c5-ea06601dc509")>
https://spa2.scrape.center/img/logo.a508a8f0.png

通过 get_attribute 方法，然后传入想要获取的属性名，就可以得到它的值了。

获取文本值

每个 WebElement 节点都有 text 属性，直接调用这个属性就可以得到节点内部的文本信息，这相当于 pyquery 的 text 方法，示例如下：

1
2
3
4
5
6
7

from selenium import webdriver

browser = webdriver.Chrome()
url = 'https://spa2.scrape.center/'
browser.get(url)
input = browser.find_element_by_class_name('logo-title')
print(input.text)

这里依然先打开页面，然后获取 class 为 logo-title 这个节点，再将其文本值打印出来。

控制台的输出结果如下：

1

Scrape

获取 ID、位置、标签名和大小

另外，WebElement 节点还有一些其他属性，比如 id 属性可以获取节点 ID，location 属性可以获取该节点在页面中的相对位置，tag_name 属性可以获取标签名称，size 属性可以获取节点的大小，也就是宽高，这些属性有时候还是很有用的。示例如下：

1
2
3
4
5
6
7
8
9
10

from selenium import webdriver

browser = webdriver.Chrome()
url = 'https://spa2.scrape.center/'
browser.get(url)
input = browser.find_element_by_class_name('logo-title')
print(input.id)
print(input.location)
print(input.tag_name)
print(input.size)

这里首先获得 class 为 logo-title 这个节点，然后调用其 id、location、tag_name、size 属性来获取对应的属性值。

10. 切换 Frame

我们知道网页中有一种节点叫作 iframe，也就是子 Frame，相当于页面的子页面，它的结构和外部网页的结构完全一致。Selenium 打开页面后，它默认是在父级 Frame 里面操作，而此时如果页面中还有子 Frame，它是不能获取到子 Frame 里面的节点的。这时就需要使用 switch_to.frame 方法来切换 Frame。示例如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

import time
from selenium import webdriver
from selenium.common.exceptions import NoSuchElementException

browser = webdriver.Chrome()
url = 'http://www.runoob.com/try/try.php?filename=jqueryui-api-droppable'
browser.get(url)
browser.switch_to.frame('iframeResult')
try:
    logo = browser.find_element_by_class_name('logo')
except NoSuchElementException:
    print('NO LOGO')
browser.switch_to.parent_frame()
logo = browser.find_element_by_class_name('logo')
print(logo)
print(logo.text)

控制台输出结果如下：

1
2
3

NO LOGO
<selenium.webdriver.remote.webelement.WebElement (session="4bb8ac03ced4ecbdefef03ffdc0e4ccd", element="0.13792611320464965-2")>
RUNOOB.COM

这里还是以前面演示动作链操作的网页为实例，首先通过 switch_to.frame 方法切换到子 Frame 里面，然后尝试获取子 Frame 里的 logo 节点（这是找不到的），如果找不到的话，就会抛出 NoSuchElementException 异常，异常被捕捉之后，就会输出 NO LOGO。接下来，重新切换回父级 Frame，然后再次重新获取节点，发现此时可以成功获取了。

所以，当页面中包含子 Frame 时，如果想获取子 Frame 中的节点，需要先调用 switch_to.frame 方法切换到对应的 Frame，然后再进行操作。

11. 延时等待

在 Selenium 中，get 方法会在网页框架加载结束后结束执行，此时如果获取 page_source，可能并不是浏览器完全加载完成的页面，如果某些页面有额外的 Ajax 请求，我们在网页源代码中也不一定能成功获取到。所以，这里需要延时等待一定时间，确保节点已经加载出来。

这里等待方式有两种：一种是隐式等待，一种是显式等待。

隐式等待

当使用隐式等待执行测试的时候，如果 Selenium 没有在 DOM 中找到节点，将继续等待，超出设定时间后，则抛出找不到节点的异常。换句话说，当查找节点而节点并没有立即出现的时候，隐式等待将等待一段时间再查找 DOM，默认的时间是 0。示例如下：

1
2
3
4
5
6
7

from selenium import webdriver

browser = webdriver.Chrome()
browser.implicitly_wait(10)
browser.get('https://spa2.scrape.center/')
input = browser.find_element_by_class_name('logo-image')
print(input)

这里我们用 implicitly_wait 方法实现了隐式等待。

显式等待

隐式等待的效果其实并没有那么好，因为我们只规定了一个固定时间，而页面的加载时间会受到网络条件的影响。

这里还有一种更合适的显式等待方法，它指定要查找的节点，然后指定一个最长等待时间。如果在规定时间内加载出来了这个节点，就返回查找的节点；如果到了规定时间依然没有加载出该节点，则抛出超时异常。示例如下：

1
2
3
4
5
6
7
8
9
10
11

from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC

browser = webdriver.Chrome()
browser.get('https://www.taobao.com/')
wait = WebDriverWait(browser, 10)
input = wait.until(EC.presence_of_element_located((By.ID, 'q')))
button = wait.until(EC.element_to_be_clickable((By.CSS_SELECTOR, '.btn-search')))
print(input, button)

这里首先引入 WebDriverWait 这个对象，指定最长等待时间，然后调用它的 until 方法，传入等待条件 expected_conditions。比如，这里传入了 presence_of_element_located 这个条件，代表节点出现的意思，其参数是节点的定位元组，也就是 ID 为 q 的节点搜索框。

这样可以做到的效果就是，在 10 秒内如果 ID 为 q 的节点（即搜索框）成功加载出来，就返回该节点；如果超过 10 秒还没有加载出来，就抛出异常。

对于按钮，可以更改一下等待条件，比如改为 element_to_be_clickable，也就是可点击，所以查找按钮时查找 CSS 选择器为 .btn-search 的按钮，如果 10 秒内它是可点击的，也就是成功加载出来了，就返回这个按钮节点；如果超过 10 秒还不可点击，也就是没有加载出来，就抛出异常。

运行代码，在网速较佳的情况下是可以成功加载出来的。

控制台的输出如下：

1
2

<selenium.webdriver.remote.webelement.WebElement (session="07dd2fbc2d5b1ce40e82b9754aba8fa8", element="0.5642646294074107-1")>
<selenium.webdriver.remote.webelement.WebElement (session="07dd2fbc2d5b1ce40e82b9754aba8fa8", element="0.5642646294074107-2")>

可以看到，控制台成功输出了两个节点，它们都是 WebElement 类型。

如果网络有问题，10 秒内没有成功加载，那就抛出 TimeoutException 异常，此时控制台的输出如下：

1
2
3
4
5

TimeoutException Traceback (most recent call last)
<ipython-input-4-f3d73973b223> in <module>()
      7 browser.get('https://www.taobao.com/')
      8 wait = WebDriverWait(browser, 10)
----> 9 input = wait.until(EC.presence_of_element_located((By.ID, 'q')))

关于等待条件，其实还有很多，比如判断标题内容，判断某个节点内是否出现了某文字等。下表列出了所有的等待条件。

更多等待条件的参数及用法介绍可以参考官方文档：http://selenium-python.readthedocs.io/api.html#module-selenium.webdriver.support.expected_conditions。

12. 前进后退

平常使用浏览器时，都有前进和后退功能，Selenium 也可以完成这个操作，它使用 back 方法后退，使用 forward 方法前进。示例如下：

1
2
3
4
5
6
7
8
9
10
11

import time
from selenium import webdriver

browser = webdriver.Chrome()
browser.get('https://www.baidu.com/')
browser.get('https://www.taobao.com/')
browser.get('https://www.python.org/')
browser.back()
time.sleep(1)
browser.forward()
browser.close()

这里我们连续访问 3 个页面，然后调用 back 方法回到第二个页面，接下来再调用 forward 方法又可以前进到第三个页面。

13. Cookies

使用 Selenium，还可以方便地对 Cookies 进行操作，例如获取、添加、删除 Cookies 等。示例如下：

1
2
3
4
5
6
7
8
9

from selenium import webdriver

browser = webdriver.Chrome()
browser.get('https://www.zhihu.com/explore')
print(browser.get_cookies())
browser.add_cookie({'name': 'name', 'domain': 'www.zhihu.com', 'value': 'germey'})
print(browser.get_cookies())
browser.delete_all_cookies()
print(browser.get_cookies())

首先，我们访问了知乎。加载完成后，浏览器实际上已经生成 Cookies 了。接着，调用 get_cookies 方法获取所有的 Cookies。然后，我们添加一个 Cookie，这里传入一个字典，有 name、domain 和 value 等内容。接下来，再次获取所有的 Cookies。可以发现，结果就多了这一项新加的 Cookie。最后，调用 delete_all_cookies 方法删除所有的 Cookies。再重新获取，发现结果就为空了。

控制台的输出如下：

1
2
3

[{'secure': False, 'value': '"NGM0ZTM5NDAwMWEyNDQwNDk5ODlkZWY3OTkxY2I0NDY=|1491604091|236e34290a6f407bfbb517888849ea509ac366d0"', 'domain': '.zhihu.com', 'path': '/', 'httpOnly': False, 'name': 'l_cap_id', 'expiry': 1494196091.403418}, ...]
[{'secure': False, 'value': 'germey', 'domain': '.www.zhihu.com', 'path': '/', 'httpOnly': False, 'name': 'name'}, {'secure': False, 'value': '"NGM0ZTM5NDAwMWEyNDQwNDk5ODlkZWY3OTkxY2I0NDY=|1491604091|236e34290a6f407bfbb517888849ea509ac366d0"', 'domain': '.zhihu.com', 'path': '/', 'httpOnly': False, 'name': 'l_cap_id', 'expiry': 1494196091.403418}, ...]
[]

通过以上方法来操作 Cookies 还是非常方便的。

14. 选项卡管理

在访问网页的时候，会开启一个个选项卡。在 Selenium 中，我们也可以对选项卡进行操作。示例如下：

1
2
3
4
5
6
7
8
9
10
11
12

import time
from selenium import webdriver

browser = webdriver.Chrome()
browser.get('https://www.baidu.com')
browser.execute_script('window.open()')
print(browser.window_handles)
browser.switch_to.window(browser.window_handles[1])
browser.get('https://www.taobao.com')
time.sleep(1)
browser.switch_to.window(browser.window_handles[0])
browser.get('https://python.org')

控制台的输出如下：

1

['CDwindow-4f58e3a7-7167-4587-bedf-9cd8c867f435', 'CDwindow-6e05f076-6d77-453a-a36c-32baacc447df']

这里首先访问了百度，然后调用了 execute_script 方法，这里传入 window.open 这个 JavaScript 语句新开启一个选项卡。接下来，我们想切换到该选项卡。这里调用 window_handles 属性获取当前开启的所有选项卡，返回的是选项卡的代号列表。要想切换选项卡，只需要调用 switch_to.window 方法即可，其中参数是选项卡的代号。这里我们将第二个选项卡代号传入，即跳转到第二个选项卡，接下来在第二个选项卡下打开一个新页面，然后切换回第一个选项卡重新调用 switch_to.window 方法，再执行其他操作即可。

15. 异常处理

在使用 Selenium 的过程中，难免会遇到一些异常，例如超时、节点未找到等错误，一旦出现此类错误，程序便不会继续运行了。这里我们可以使用 try except 语句来捕获各种异常。

首先，演示一下节点未找到的异常，示例如下：

1
2
3
4
5

from selenium import webdriver

browser = webdriver.Chrome()
browser.get('https://www.baidu.com')
browser.find_element_by_id('hello')

这里首先打开百度页面，然后尝试选择一个并不存在的节点，此时就会遇到异常。

运行之后控制台的输出如下：

1
2
3
4
5

NoSuchElementException Traceback (most recent call last)
<ipython-input-23-978945848a1b> in <module>()
      3 browser = webdriver.Chrome()
      4 browser.get('https://www.baidu.com')
----> 5 browser.find_element_by_id('hello')

可以看到，这里抛出了 NoSuchElementException 异常，这通常是节点未找到的异常。为了防止程序遇到异常而中断，我们需要捕获这些异常，示例如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14

from selenium import webdriver
from selenium.common.exceptions import TimeoutException, NoSuchElementException

browser = webdriver.Chrome()
try:
    browser.get('https://www.baidu.com')
except TimeoutException:
    print('Time Out')
try:
    browser.find_element_by_id('hello')
except NoSuchElementException:
    print('No Element')
finally:
    browser.close()

这里我们使用 try except 来捕获各类异常。比如，我们对 find_element_by_id 查找节点的方法捕获 NoSuchElementException 异常，这样一旦出现这样的错误，就进行异常处理，程序也不会中断了。

控制台的输出如下：

1

No Element

关于更多的异常类，可以参考官方文档：：http://selenium-python.readthedocs.io/api.html#module-selenium.common.exceptions。

16. 反屏蔽

现在很多网站都加上了对 Selenium 的检测，来防止一些爬虫的恶意爬取。即如果检测到有人在使用 Selenium 打开浏览器，那就直接屏蔽。

在大多数情况下，检测的基本原理是检测当前浏览器窗口下的 window.navigator 对象是否包含 webdriver 这个属性。因为在正常使用浏览器的情况下，这个属性是 undefined，然而一旦我们使用了 Selenium，Selenium 会给 window.navigator 设置 webdriver 属性。很多网站就通过 JavaScript 判断如果 webdriver 属性存在，那就直接屏蔽。

这边有一个典型的案例网站：https://antispider1.scrape.center/，这个网站就使用了上述原理实现了 WebDriver 的检测，如果使用 Selenium 直接爬取的话，那就会返回如图所示的页面。

这时候我们可能想到直接使用 JavaScript 语句把这个 webdriver 属性置空，比如通过调用 execute_script 方法来执行如下代码：

1

Object.defineProperty(navigator, "webdriver", { get: () => undefined });

这行 JavaScript 语句的确可以把 webdriver 属性置空，但是 execute_script 调用这行 JavaScript 语句实际上是在页面加载完毕之后才执行的，执行太晚了，网站早在最初页面渲染之前就已经对 webdriver 属性进行了检测，所以用上述方法并不能达到效果。

在 Selenium 中，我们可以使用 CDP（即 Chrome Devtools-Protocol，Chrome 开发工具协议）来解决这个问题，通过它我们可以实现在每个页面刚加载的时候执行 JavaScript 代码，执行的 CDP 方法叫作 Page.addScriptToEvaluateOnNewDocument，然后传入上文的 JavaScript 代码即可，这样我们就可以在每次页面加载之前将 webdriver 属性置空了。另外，我们还可以加入几个选项来隐藏 WebDriver 提示条和自动化扩展信息，代码实现如下：

1
2
3
4
5
6
7
8
9
10
11

from selenium import webdriver
from selenium.webdriver import ChromeOptions

option = ChromeOptions()
option.add_experimental_option('excludeSwitches', ['enable-automation'])
option.add_experimental_option('useAutomationExtension', False)
browser = webdriver.Chrome(options=option)
browser.execute_cdp_cmd('Page.addScriptToEvaluateOnNewDocument', {
    'source': 'Object.defineProperty(navigator, "webdriver", {get: () => undefined})'
})
browser.get('https://antispider1.scrape.center/')

这样整个页面就能被加载出来了，如图所示。

对于大多数情况，以上方法均可以实现 Selenium 反屏蔽。但对于一些特殊网站，如果它有更多的 WebDriver 特征检测，可能需要具体排查。

17. 无头模式

我们可以观察到，上面的案例在运行的时候，总会弹出一个浏览器窗口，虽然有助于观察页面爬取状况，但在有时候窗口弹来弹去也会形成一些干扰。

Chrome 浏览器从 60 版本已经支持了无头模式，即 Headless。无头模式在运行的时候不会再弹出浏览器窗口，减少了干扰，而且它减少了一些资源的加载，如图片等，所以也在一定程度上节省了资源加载时间和网络带宽。

我们可以借助于 ChromeOptions 来开启 Chrome Headless 模式，代码实现如下：

1
2
3
4
5
6
7
8
9

from selenium import webdriver
from selenium.webdriver import ChromeOptions

option = ChromeOptions()
option.add_argument('--headless')
browser = webdriver.Chrome(options=option)
browser.set_window_size(1366, 768)
browser.get('https://www.baidu.com')
browser.get_screenshot_as_file('preview.png')

这里我们通过 ChromeOptions 的 add_argument 方法添加了一个参数 --headless，开启了无头模式。在无头模式下，我们最好设置一下窗口的大小，接着打开页面，最后我们调用 get_screenshot_as_file 方法输出了页面的截图。

运行代码之后，我们发现 Chrome 窗口就不会再弹出来了，代码依然正常运行，最后输出的页面如图所示。

这样我们就在无头模式下完成了页面的抓取和截图操作。

18. 总结

现在，我们基本上对 Selenium 的常规用法有了大体的了解。使用 Selenium，处理 JavaScript 渲染的页面不再是难事，后面我们会用一个实例来演示 Selenium 爬取网站的流程。

本节代码：https://github.com/Python3WebSpider/SeleniumTest。

爬虫系列文章总目录：【2022 年】Python3 爬虫学习教程，本教程内容多数来自于《Python3网络爬虫开发实战（第二版）》一书，目前截止 2022 年，可以将爬虫基本技术进行系统讲解，同时将最新前沿爬虫技术如异步、JavaScript 逆向、AST、安卓逆向、Hook、智能解析、群控技术、WebAssembly、大规模分布式、Docker、Kubernetes 等，市面上目前就仅有《Python3 网络爬虫开发实战（第二版）》一书了，点击了解详情。

想查数据，就免不了搜索，而搜索离不开搜索引擎。百度、谷歌都是非常庞大、复杂的搜索引擎，它们几乎索引了互联网上开放的所有网页和数据。然而对于我们自己的业务数据来说，没必要用这么复杂的技术。如果我们想实现自己的搜索引擎，为了便于存储和检索，Elasticsearch 就是不二选择。它是一个全文搜索引擎，可以快速存储、搜索和分析海量数据。

所以，如果我们我们将爬取到的数据存储到 Elasticsearch 里面，那将会非常方便检索。

1. Elasticsearch 介绍

Elasticsearch 是一个开源的搜索引擎，建立在一个全文搜索引擎库 Apache Lucene™ 基础之上。

那 Lucene 又是什么呢？Lucene 可能是目前存在的（不论开源还是私有的）拥有最先进、高性能和全功能搜索引擎功能的库，但也仅仅只是一个库。要想用 Lucene，我们需要编写 Java 并引用 Lucene 包才可以，而且我们需要对信息检索有一定程度的理解。

为了解决这个问题，Elasticsearch 就诞生了。Elasticsearch 也是使用 Java 编写的，它的内部使用 Lucene 做索引与搜索，但是它的目标是使全文检索变得简单，相当于 Lucene 的一层封装，它提供了一套简单一致的 RESTful API 来帮助我们实现存储和检索。

所以 Elasticsearch 仅仅就是一个简易版的 Lucene 封装吗？那就大错特错了，Elasticsearch 不仅仅是 Lucene，并且也不仅仅只是一个全文搜索引擎。它可以这样准确形容：

• 一个分布式的实时文档存储，每个字段可以被索引与搜索；
• 一个分布式实时分析搜索引擎；
• 能胜任上百个服务节点的扩展，并支持 PB 级别的结构化或者非结构化数据。

总之，它是一个非常强大的搜索引擎，维基百科、Stack Overflow、GitHub 都纷纷采用它来做搜索，不仅仅提供强大的检索能力，也提供强大的存储能力。

2. Elasticsearch 相关概念

在 Elasticsearch 中有几个基本概念，如节点、索引、文档等，下面分别说明一下。理解了这些概念，对熟悉 Elasticsearch 是非常有帮助的。

节点和集群

Elasticsearch 本质上是一个分布式数据库，允许多台服务器协同工作，每台服务器可以运行多个 Elasticsearch 实例。

单个 Elasticsearch 实例称为一个节点（Node），一组节点构成一个集群（Cluster）。

索引

索引，即 Index，Elasticsearch 会索引所有字段，经过处理后写入一个反向索引（Inverted Index）。查找数据的时候，直接查找该索引。

所以，Elasticsearch 数据管理的顶层单位就叫作索引，其实就相当于 MySQL、MongoDB 等中数据库的概念。另外，值得注意的是，每个索引 （即数据库）的名字必须小写。

文档

文档，即 Document。索引里面单条记录称为文档，许多条文档构成了一个索引。

同一个索引里面的文档，不要求有相同的结构（Schema），但是最好保持一致，因为这样有利于提高搜索效率。

类型

文档可以分组，比如 weather 这个索引里面，既可以按城市分组（北京和上海），也可以按气候分组（晴天和雨天）。这种分组就叫作类型（Type），它是虚拟的逻辑分组，用来过滤文档，类似 MySQL 中的数据表、MongoDB 中的 Collection。

不同的类型应该有相似的结构。举例来说，id 字段不能在这个组中是字符串，在另一个组中是数值。这是与关系型数据库的表的一个区别。性质完全不同的数据（比如 products 和 logs）应该存成两个索引，而不是一个索引里面的两个类型（虽然可以做到）。

根据规划，Elastic 6.x 版只允许每个索引包含一个类型，Elastic 7.x 开始将会将其彻底移除。

字段

每个文档都类似一个 JSON 结构，它包含了许多字段，每个字段都有其对应的值，多个字段组成了一个文档，其实就可以类比 MySQL 数据表中的字段。

在 Elasticsearch 中，文档归属于一种类型（Type），而这些类型存在于索引中，我们可以画一些简单的对比图来类比传统关系型数据库：

1
2

Relational DB -> Databases -> Tables -> Rows -> Columns
Elasticsearch -> Indices   -> Types  -> Documents -> Fields

以上就是 Elasticsearch 里面的一些基本概念，通过和关系型数据库的对比更加有助于理解。

3. 准备工作

在开始本节实际操作之前，请确保已经正确安装好了 Elasticsearch，安装方式可以参考：https://setup.scrape.center/elasticsearch，安装完成之后确保其在本地 9200 端口上正常运行即可。

Elasticsearch 实际上提供了一系列 Restful API 来进行存取和查询操作，我们可以使用 curl 等命令或者直接调用 API 来进行数据存储和修改操作，但总归来说并不是很方便。所以这里我们就直接介绍一个专门用来对接 Elasticsearch 操作的 Python 库，名称也叫做 Elasticsearch，使用 pip3 安装即可：

1

pip3 install elasticsearch

更详细的安装方式可以参考：https://setup.scrape.center/elasticsearch-py。

安装好了之后我们就可以开始本节的学习了。

4. 创建索引

我们先来看下怎样创建一个索引，这里我们创建一个名为 news 的索引：

1
2
3
4
5

from elasticsearch import Elasticsearch

es = Elasticsearch()
result = es.indices.create(index='news', ignore=400)
print(result)

这里我们首先创建了一个 Elasticsearch 对象，并且没有设置任何参数，默认情况下它会连接本地 9200 端口运行的 Elasticsearch 服务，我们也可以设置特定的连接信息，如：

1
2
3
4

es = Elasticsearch(
    ['https://[username:password@]hostname:port'],
    verify_certs=True, # 是否验证 SSL 证书
)

第一个参数我们可以构造特定格式的链接字符串并传入，hostname 和 port 即 Elasticsearch 运行的地址和端口，username 和 password 是可选的，代表连接 Elasticsearch 需要的用户名和密码，另外而且还有其他的参数设置，比如 verify_certs 代表是否验证证书有效性。更多参数的设置可以参考：https://elasticsearch-py.readthedocs.io/en/latest/api.html#elasticsearch。

声明 Elasticsearch 对象之后，我们调用了 es 的 indices 对象的 create 方法传入了 index 的名称，如果创建成功，会返回如下结果：

1

{'acknowledged': True, 'shards_acknowledged': True, 'index': 'news'}

可以看到，其返回结果是 JSON 格式，其中的 acknowledged 字段表示创建操作执行成功。

但这时如果我们再把代码执行一次的话，就会返回如下结果：

1

{'error': {'root_cause': [{'type': 'resource_already_exists_exception', 'reason': 'index [news/hHEYozoqTzK_qRvV4j4a3w] already exists', 'index_uuid': 'hHEYozoqTzK_qRvV4j4a3w', 'index': 'news'}], 'type': 'resource_already_exists_exception', 'reason': 'index [news/hHEYozoqTzK_qRvV4j4a3w] already exists', 'index_uuid': 'hHEYozoqTzK_qRvV4j4a3w', 'index': 'news'}, 'status': 400}

它提示创建失败，status 状态码是 400，错误原因是索引已经存在了。

注意在这里的代码中，我们使用的 ignore 参数为 400，这说明如果返回结果是 400 的话，就忽略这个错误，不会报错，程序不会抛出异常。

假如我们不加 ignore 这个参数的话：

1
2
3

es = Elasticsearch()
result = es.indices.create(index='news')
print(result)

再次执行就会报错了：

1
2

raise HTTP_EXCEPTIONS.get(status_code, TransportError)(status_code, error_message, additional_info)
elasticsearch.exceptions.RequestError: TransportError(400, 'resource_already_exists_exception', 'index [news/QM6yz2W8QE-bflKhc5oThw] already exists')

这样程序的执行就会出现问题。因此，我们需要善用 ignore 参数，把一些意外情况排除，这样可以保证程序正常执行而不会中断。

创建完之后，我们还可以设置下索引的字段映射定义，可以参考：https://elasticsearch-py.readthedocs.io/en/latest/api.html?#elasticsearch.client.IndicesClient.put_mapping。

5. 删除索引

删除索引也是类似的，代码如下：

1
2
3
4
5

from elasticsearch import Elasticsearch

es = Elasticsearch()
result = es.indices.delete(index='news', ignore=[400, 404])
print(result)

这里也使用了 ignore 参数来忽略索引不存在而删除失败导致程序中断的问题。

如果删除成功，会输出如下结果：

1

{'acknowledged': True}

如果索引已经被删除，再执行删除，则会输出如下结果：

1

{'error': {'root_cause': [{'type': 'index_not_found_exception', 'reason': 'no such index [news]', 'resource.type': 'index_or_alias', 'resource.id': 'news', 'index_uuid': '_na_', 'index': 'news'}], 'type': 'index_not_found_exception', 'reason': 'no such index [news]', 'resource.type': 'index_or_alias', 'resource.id': 'news', 'index_uuid': '_na_', 'index': 'news'}, 'status': 404}

这个结果表明当前索引不存在，删除失败。返回的结果同样是 JSON，状态码是 404，但是由于我们添加了 ignore 参数，忽略了 404 状态码，因此程序正常执行，输出 JSON 结果，而不是抛出异常。

6. 插入数据

Elasticsearch 就像 MongoDB 一样，在插入数据的时候可以直接插入结构化字典数据，插入数据可以调用 create 方法。例如，这里我们插入一条新闻数据：

1
2
3
4
5
6
7
8
9
10
11

from elasticsearch import Elasticsearch

es = Elasticsearch()
es.indices.create(index='news', ignore=400)

data = {
  'title': '乘风破浪不负韶华，奋斗青春圆梦高考',
  'url': 'http://view.inews.qq.com/a/EDU2021041600732200'
}
result = es.create(index='news', id=1, body=data)
print(result)

这里我们首先声明了一条新闻数据，包括标题和链接，然后通过调用 create 方法插入了这条数据。在调用 create 方法时，我们传入了 4 个参数，index 参数代表了索引名称，id 则是数据的唯一标识 ID，body 则代表了文档的具体内容。

运行结果如下：

1

{'_index': 'news', '_type': '_doc', '_id': '1', '_version': 1, 'result': 'created', '_shards': {'total': 2, 'successful': 1, 'failed': 0}, '_seq_no': 0, '_primary_term': 1}

结果中 result 字段为 created，代表该数据插入成功。

另外，其实我们也可以使用 index 方法来插入数据。但与 create 不同的是，create 方法需要我们指定 id 字段来唯一标识该条数据，而 index 方法则不需要，如果不指定 id，会自动生成一个 id。调用 index 方法的写法如下：

1

es.index(index='news', body=data)

create 方法内部其实也是调用了 index 方法，是对 index 方法的封装。

7. 更新数据

更新数据也非常简单，我们同样需要指定数据的 id 和内容，调用 update 方法即可，代码如下：

1
2
3
4
5
6
7
8
9
10

from elasticsearch import Elasticsearch

es = Elasticsearch()
data = {
    'title': '乘风破浪不负韶华，奋斗青春圆梦高考',
    'url': 'http://view.inews.qq.com/a/EDU2021041600732200',
    'date': '2021-07-05'
}
result = es.update(index='news', body=data, id=1)
print(result)

这里我们为数据增加了一个日期字段，然后调用了 update 方法，结果如下：

1

{'_index': 'news', '_type': '_doc', '_id': '1', '_version': 2, 'result': 'updated', '_shards': {'total': 2, 'successful': 1, 'failed': 0}, '_seq_no': 1, '_primary_term': 1}

可以看到，返回结果中 result 字段为 updated，即表示更新成功。另外，我们还注意到一个字段 _version，这代表更新后的版本号数，2 代表这是第二个版本。因为之前已经插入过一次数据，所以第一次插入的数据是版本 1，可以参见上例的运行结果，这次更新之后版本号就变成了 2，以后每更新一次，版本号都会加 1。

另外，更新操作利用 index 方法同样可以做到，其写法如下：

1

es.index(index='news', body=data, id=1)

可以看到，index 方法可以代替我们完成插入和更新数据这两个操作。如果数据不存在，就执行插入操作，如果已经存在，就执行更新操作，非常方便。

8. 删除数据

如果想删除一条数据，可以调用 delete 方法并指定需要删除的数据 id 即可。其写法如下：

1
2
3
4
5

from elasticsearch import Elasticsearch

es = Elasticsearch()
result = es.delete(index='news', id=1)
print(result)

运行结果如下：

1

{'_index': 'news', '_type': '_doc', '_id': '1', '_version': 2, 'result': 'deleted', '_shards': {'total': 2, 'successful': 1, 'failed': 0}, '_seq_no': 3, '_primary_term': 1}

可以看到，运行结果中 result 字段为 deleted，代表删除成功；_version 变成了 3，又增加了 1。

9. 查询数据

上面的几个操作都是非常简单的操作，普通的数据库如 MongoDB 都可以完成，看起来并没有什么了不起的，Elasticsearch 更特殊的地方在于其异常强大的检索功能。

对于中文来说，我们需要安装一个分词插件，这里使用的是 elasticsearch-analysis-ik，其 GitHub 链接为https://github.com/medcl/elasticsearch-analysis-ik。这里我们使用 Elasticsearch 的另一个命令行工具 elasticsearch-plugin 来安装，这里安装的版本是 7.13.2，请确保和 Elasticsearch 的版本对应起来，命令如下：

1

elasticsearch-plugin install https://github.com/medcl/elasticsearch-analysis-ik/releases/download/v7.13.2/elasticsearch-analysis-ik-7.13.2.zip

这里的版本号请替换成你的 Elasticsearch 版本号。

安装之后，我们需要重新启动 Elasticsearch，启动之后它会自动加载安装好的插件。

首先，我们重新新建一个索引并指定需要分词的字段，相应代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

from elasticsearch import Elasticsearch

es = Elasticsearch()
mapping = {
    'properties': {
        'title': {
            'type': 'text',
            'analyzer': 'ik_max_word',
            'search_analyzer': 'ik_max_word'
        }
    }
}
es.indices.delete(index='news', ignore=[400, 404])
es.indices.create(index='news', ignore=400)
result = es.indices.put_mapping(index='news', body=mapping)
print(result)

这里我们先将之前的索引删除了，然后新建了一个索引，接着更新了它的 mapping 信息。mapping 信息中指定了分词的字段，指定了字段的类型 type 为 text，分词器 analyzer 和搜索分词器 search_analyzer 为 ik_max_word，即使用我们刚才安装的中文分词插件。如果不指定的话，则使用默认的英文分词器。

接下来，我们插入几条新数据：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

from elasticsearch import Elasticsearch

es = Elasticsearch()

datas = [
    {
        'title': '高考结局大不同',
        'url': 'https://k.sina.com.cn/article_7571064628_1c3454734001011lz9.html',
    },
    {
        'title': '进入职业大洗牌时代，“吃香”职业还吃香吗？',
        'url': 'https://new.qq.com/omn/20210828/20210828A025LK00.html',
    },
    {
        'title': '乘风破浪不负韶华，奋斗青春圆梦高考',
        'url': 'http://view.inews.qq.com/a/EDU2021041600732200',
    },
    {
        'title': '他，活出了我们理想的样子',
        'url': 'https://new.qq.com/omn/20210821/20210821A020ID00.html',
    }
]

for data in datas:
    es.index(index='news', body=data)

这里我们指定了 4 条数据，它们都带有 title 和url 字段，然后通过 index 方法将其插入 Elasticsearch 中，索引名称为 news。

接下来，我们根据关键词查询一下相关内容：

1
2

result = es.search(index='news')
print(result)

可以看到，这里查询出了插入的 4 条数据：

1

{'took': 11, 'timed_out': False, '_shards': {'total': 1, 'successful': 1, 'skipped': 0, 'failed': 0}, 'hits': {'total': {'value': 4, 'relation': 'eq'}, 'max_score': 1.0, 'hits': [{'_index': 'news', '_type': '_doc', '_id': 'jebpkHsBm-BAny-7hOYp', '_score': 1.0, '_source': {'title': '高考结局大不同', 'url': 'https://k.sina.com.cn/article_7571064628_1c3454734001011lz9.html'}}, {'_index': 'news', '_type': '_doc', '_id': 'jubpkHsBm-BAny-7hObz', '_score': 1.0, '_source': {'title': '进入职业大洗牌时代，“吃香”职业还吃香吗？', 'url': 'https://new.qq.com/omn/20210828/20210828A025LK00.html'}}, {'_index': 'news', '_type': '_doc', '_id': 'j-bpkHsBm-BAny-7heZN', '_score': 1.0, '_source': {'title': '乘风破浪不负韶华，奋斗青春圆梦高考', 'url': 'http://view.inews.qq.com/a/EDU2021041600732200'}}, {'_index': 'news', '_type': '_doc', '_id': 'kObpkHsBm-BAny-7hean', '_score': 1.0, '_source': {'title': '他，活出了我们理想的样子', 'url': 'https://new.qq.com/omn/20210821/20210821A020ID00.html'}}]}}

可以看到，返回结果会出现在 hits 字段里面，其中 total 字段标明了查询的结果条目数，max_score 代表了最大匹配分数。

另外，我们还可以进行全文检索，这才是体现 Elasticsearch 搜索引擎特性的地方：

1
2
3
4
5
6
7
8
9
10
11
12
13
14

from elasticsearch import Elasticsearch
import json

dsl = {
    'query': {
        'match': {
            'title': '高考 圆梦'
        }
    }
}

es = Elasticsearch()
result = es.search(index='news', body=dsl)
print(result)

这里我们使用 Elasticsearch 支持的 DSL 语句来进行查询，使用 match 指定全文检索，检索的字段是 title，内容是“中国 领事馆”，搜索结果如下：

1

{'took': 6, 'timed_out': False, '_shards': {'total': 1, 'successful': 1, 'skipped': 0, 'failed': 0}, 'hits': {'total': {'value': 2, 'relation': 'eq'}, 'max_score': 1.7796917, 'hits': [{'_index': 'news', '_type': '_doc', '_id': 'j-bpkHsBm-BAny-7heZN', '_score': 1.7796917, '_source': {'title': '乘风破浪不负韶华，奋斗青春圆梦高考', 'url': 'http://view.inews.qq.com/a/EDU2021041600732200'}}, {'_index': 'news', '_type': '_doc', '_id': 'jebpkHsBm-BAny-7hOYp', '_score': 0.81085134, '_source': {'title': '高考结局大不同', 'url': 'https://k.sina.com.cn/article_7571064628_1c3454734001011lz9.html'}}]}}

这里我们看到匹配的结果有两条，第一条的分数为 1.7796917，第二条的分数为 0.81085134，这是因为第一条匹配的数据中含有“高考”和“圆梦”两个词，第二条匹配的数据中不包含“圆梦”，但是包含了“高考”这个词，所以也被检索出来了，但是分数比较低。

因此，可以看出，检索时会对对应的字段进行全文检索，结果还会按照检索关键词的相关性进行排序，这就是一个基本的搜索引擎雏形。

另外，Elasticsearch 还支持非常多的查询方式，这里就不再一一展开描述了，总之其功能非常强大，详情可以参考官方文档：https://www.elastic.co/guide/en/elasticsearch/reference/master/query-dsl.html。

10. 总结

以上便是对 Elasticsearch 的基本介绍以及使用 Python 操作 Elasticsearch 的基本用法，但这仅仅是 Elasticsearch 的基本功能，它还有更多强大的功能等待着我们去探索。

本节代码地址：https://github.com/Python3WebSpider/ElasticSearchTest。

爬虫系列文章总目录：【2022 年】Python3 爬虫学习教程，本教程内容多数来自于《Python3网络爬虫开发实战（第二版）》一书，目前截止 2022 年，可以将爬虫基本技术进行系统讲解，同时将最新前沿爬虫技术如异步、JavaScript 逆向、AST、安卓逆向、Hook、智能解析、群控技术、WebAssembly、大规模分布式、Docker、Kubernetes 等，市面上目前就仅有《Python3 网络爬虫开发实战（第二版）》一书了，点击了解详情。

我们知道爬虫是 IO 密集型任务，比如如果我们使用 requests 库来爬取某个站点的话，发出一个请求之后，程序必须要等待网站返回响应之后才能接着运行，而在等待响应的过程中，整个爬虫程序是一直在等待的，实际上没有做任何事情。对于这种情况，我们有没有优化方案呢？

当然有，下面我们就来了解一下异步爬虫的基本概念和实现。

要实现异步机制的爬虫，那自然和协程脱不了关系。

1. 案例引入

在介绍协程之前，我们先来看一个案例网站，链接地址为：https://httpbin.org/delay/5，如果我们访问这个链接，需要等待五秒之后才能得到结果，这是因为服务器强制等待了 5 秒的时间才返回响应。

平时我们浏览网页的时候，绝大部分网页响应速度还是很快的，如果我们写爬虫来爬取的话，发出 Request 到收到 Response 的时间不会很长，因此我们需要等待的时间并不多。

然而像上面这个网站，一次 Request 就需要 5 秒才能得到 Response，如果我们用 requests 写爬虫来爬取的话，那每次 requests 都要等待 5 秒才能拿到结果了。

我们来测试下，下面我们来用 requests 写一个遍历程序，直接遍历 100 次试试看，实现代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

import requests
import logging
import time

logging.basicConfig(level=logging.INFO,
                    format='%(asctime)s - %(levelname)s: %(message)s')

TOTAL_NUMBER = 100
URL = 'https://httpbin.org/delay/5'

start_time = time.time()
for _ in range(1, TOTAL_NUMBER + 1):
    logging.info('scraping %s', URL)
    response = requests.get(URL)
end_time = time.time()
logging.info('total time %s seconds', end_time - start_time)

这里我们直接用循环的方式构造了 100 个 Request，使用的是 requests 单线程，在爬取之前和爬取之后记录了时间，最后输出爬取了 100 个页面消耗的时间。

运行结果如下：

1
2
3
4
5
6
7
8
9
10
11
12
13

2020-08-03 01:01:36,781 - INFO: scraping https://httpbin.org/delay/5
2020-08-03 01:01:43,410 - INFO: scraping https://httpbin.org/delay/5
2020-08-03 01:01:50,029 - INFO: scraping https://httpbin.org/delay/5
2020-08-03 01:01:56,702 - INFO: scraping https://httpbin.org/delay/5
2020-08-03 01:02:03,345 - INFO: scraping https://httpbin.org/delay/5
2020-08-03 01:02:09,958 - INFO: scraping https://httpbin.org/delay/5
2020-08-03 01:02:16,500 - INFO: scraping https://httpbin.org/delay/5
2020-08-03 01:02:23,143 - INFO: scraping https://httpbin.org/delay/5
...
2020-08-03 01:12:19,867 - INFO: scraping https://httpbin.org/delay/5
2020-08-03 01:12:26,479 - INFO: scraping https://httpbin.org/delay/5
2020-08-03 01:12:33,083 - INFO: scraping https://httpbin.org/delay/5
2020-08-03 01:12:39,758 - INFO: total time 662.9764430522919 seconds

由于每个页面至少要等待 5 秒才能加载出来，因此 100 个页面至少要花费 500 秒的时间，加上网站本身负载的问题，总的爬取时间最终为 663 秒，大约 11 分钟。

这在实际情况下是很常见的，有些网站本身加载速度就比较慢，稍慢的可能 1~3 秒，更慢的说不定 10 秒以上。如果我们用 requests 单线程这么爬取的话，总的耗时是非常多的。此时如果我们开了多线程或多进程来爬取的话，其爬取速度确实会成倍提升，那是否有更好的解决方案呢？

本节就来了解一下使用协程来加速的方法，此种方法对于 IO 密集型任务非常有效。如将其应用到网络爬虫中，爬取效率甚至可以成百倍地提升。

2. 基础知识

在了解协程之前，我们首先了解一些基础概念，如阻塞和非阻塞、同步和异步、多进程和协程。

阻塞

阻塞状态指程序未得到所需计算资源时被挂起的状态。程序在等待某个操作完成期间，自身无法继续干别的事情，则称该程序在该操作上是阻塞的。

常见的阻塞形式有：网络 I/O 阻塞、磁盘 I/O 阻塞、用户输入阻塞等。阻塞是无处不在的，包括 CPU 切换上下文时，所有的进程都无法真正干事情，它们也会被阻塞。如果是多核 CPU，则正在执行上下文切换操作的核不可被利用。

非阻塞

程序在等待某操作的过程中，自身不被阻塞，可以继续运行干别的事情，则称该程序在该操作上是非阻塞的。

非阻塞并不是在任何程序级别、任何情况下都存在的。仅当程序封装的级别可以囊括独立的子程序单元时，它才可能存在非阻塞状态。

非阻塞的存在是因为阻塞存在，正因为某个操作阻塞导致的耗时与效率低下，我们才要把它变成非阻塞的。

同步

不同程序单元为了完成某个任务，在执行过程中需靠某种通信方式以协调一致，此时这些程序单元是同步执行的。

例如在购物系统中更新商品库存时，需要用“行锁”作为通信信号，让不同的更新请求强制排队顺序执行，那更新库存的操作是同步的。

简言之，同步意味着有序。

异步

为了完成某个任务，有时不同程序单元之间无须通信协调也能完成任务，此时不相关的程序单元之间可以是异步的。

例如，爬取下载网页。调度程序调用下载程序后，即可调度其他任务，而无须与该下载任务保持通信以协调行为。不同网页的下载、保存等操作都是无关的，也无须相互通知协调。这些异步操作的完成时刻并不确定。

简言之，异步意味着无序。

多进程

多进程就是利用 CPU 的多核优势，在同一时间并行执行多个任务，可以大大提高执行效率。

协程

协程，英文叫作 coroutine，又称微线程、纤程，它是一种用户态的轻量级线程。

协程拥有自己的寄存器上下文和栈。协程调度切换时，将寄存器上下文和栈保存到其他地方，在切回来的时候，恢复先前保存的寄存器上下文和栈。因此，协程能保留上一次调用时的状态，即所有局部状态的一个特定组合，每次过程重入时，就相当于进入上一次调用的状态。

协程本质上是个单进程，它相对于多进程来说，无须线程上下文切换的开销，无须原子操作锁定及同步的开销，编程模型也非常简单。

我们可以使用协程来实现异步操作，比如在网络爬虫场景下，我们发出一个请求之后，需要等待一定时间才能得到响应，但其实在这个等待过程中，程序可以干许多其他事情，等到响应得到之后才切换回来继续处理，这样可以充分利用 CPU 和其他资源，这就是协程的优势。

3. 协程的用法

接下来，让我们来了解一下协程的实现。从 Python 3.4 开始，Python 中加入了协程的概念，但这个版本的协程还是以生成器对象为基础，Python 3.5 则增加了 async/await，使得协程的实现更加方便。

Python 中使用协程最常用的库莫过于 asyncio，所以本节会以 asyncio 为基础来介绍协程的用法。

首先，我们需要了解下面几个概念：

• event_loop：事件循环，相当于一个无限循环，我们可以把一些函数注册到这个事件循环上，当满足条件发生的时候，就会调用对应的处理方法。
• coroutine：中文翻译叫协程，在 Python 中常指代协程对象类型，我们可以将协程对象注册到时间循环中，它会被事件循环调用。我们可以使用 async 关键字来定义一个方法，这个方法在调用时不会立即被执行，而是返回一个协程对象。
• task：任务，它是对协程对象的进一步封装，包含了任务的各个状态。
• future：代表将来执行或没有执行的任务的结果，实际上和 task 没有本质区别。

另外，我们还需要了解 async/await 关键字，它是从 Python 3.5 才出现的，专门用于定义协程。其中，async 定义一个协程，await 用来挂起阻塞方法的执行。

4. 准备工作

在本节开始之前，请确保安装的 Python 版本为 3.5 及以上，如果版本是 3.4 及以下，则下方的案例是不能运行的。

具体的安装方法可以参考：https://setup.scrape.center/python。

安装好合适的 Python 版本之后我们就可以开始本节的学习了。

5. 定义协程

首先，我们来定义一个协程，体验一下它和普通进程在实现上的不同之处，代码如下：

1
2
3
4
5
6
7
8
9
10
11
12

import asyncio

async def execute(x):
    print('Number:', x)

coroutine = execute(1)
print('Coroutine:', coroutine)
print('After calling execute')

loop = asyncio.get_event_loop()
loop.run_until_complete(coroutine)
print('After calling loop')

运行结果如下：

1
2
3
4

Coroutine: <coroutine object execute at 0x1034cf830>
After calling execute
Number: 1
After calling loop

首先，我们引入了 asyncio 这个包，这样我们才可以使用 async 和 await，然后使用 async 定义了一个 execute 方法，该方法接收一个数字参数，执行之后会打印这个数字。

随后我们直接调用了这个方法，然而这个方法并没有执行，而是返回了一个 coroutine 协程对象。随后我们使用 get_event_loop 方法创建了一个事件循环 loop，并调用了 loop 对象的 run_until_complete 方法将协程注册到事件循环 loop 中，然后启动。最后，我们才看到 execute 方法打印了输出结果。

可见，async 定义的方法就会变成一个无法直接执行的 coroutine 对象，必须将其注册到事件循环中才可以执行。

前面我们还提到了 task，它是对 coroutine 对象的进一步封装，比 coroutine 对象多了运行状态，比如 running、finished 等，我们可以用这些状态来获取协程对象的执行情况。

在上面的例子中，当我们将 coroutine 对象传递给 run_until_complete 方法的时候，实际上它进行了一个操作，就是将 coroutine 封装成了 task 对象。我们也可以显式地进行声明，如下所示：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

import asyncio

async def execute(x):
    print('Number:', x)
    return x

coroutine = execute(1)
print('Coroutine:', coroutine)
print('After calling execute')

loop = asyncio.get_event_loop()
task = loop.create_task(coroutine)
print('Task:', task)
loop.run_until_complete(task)
print('Task:', task)
print('After calling loop')

运行结果如下：

1
2
3
4
5
6

Coroutine: <coroutine object execute at 0x10e0f7830>
After calling execute
Task: <Task pending coro=<execute() running at demo.py:4>>
Number: 1
Task: <Task finished coro=<execute() done, defined at demo.py:4> result=1>
After calling loop

这里我们定义了 loop 对象之后，接着调用了它的 create_task 方法将 coroutine 对象转化为 task 对象，随后我们打印输出一下，发现它是 pending 状态。接着，我们将 task 对象添加到事件循环中执行，随后打印输出 task 对象，发现它的状态变成了 finished，同时还可以看到其 result 变成了 1，也就是我们定义的 execute 方法的返回结果。

另外，定义 task 对象还有一种方式，就是直接通过 asyncio 的 ensure_future 方法，返回结果也是 task 对象，这样的话我们就可以不借助 loop 来定义。即使我们还没有声明 loop，也可以提前定义好 task 对象，写法如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

import asyncio

async def execute(x):
    print('Number:', x)
    return x

coroutine = execute(1)
print('Coroutine:', coroutine)
print('After calling execute')

task = asyncio.ensure_future(coroutine)
print('Task:', task)
loop = asyncio.get_event_loop()
loop.run_until_complete(task)
print('Task:', task)
print('After calling loop')

运行结果如下：

1
2
3
4
5
6

Coroutine: <coroutine object execute at 0x10aa33830>
After calling execute
Task: <Task pending coro=<execute() running at demo.py:4>>
Number: 1
Task: <Task finished coro=<execute() done, defined at demo.py:4> result=1>
After calling loop

可以发现，其运行效果都是一样的。

6. 绑定回调

另外，我们也可以为某个 task 绑定一个回调方法。比如，我们来看下面的例子：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

import asyncio
import requests

async def request():
    url = 'https://www.baidu.com'
    status = requests.get(url)
    return status

def callback(task):
    print('Status:', task.result())

coroutine = request()
task = asyncio.ensure_future(coroutine)
task.add_done_callback(callback)
print('Task:', task)

loop = asyncio.get_event_loop()
loop.run_until_complete(task)
print('Task:', task)

这里我们定义了一个 request 方法，请求了百度，获取其状态码，但是这个方法里面我们没有任何 print 语句。随后我们定义了一个 callback 方法，这个方法接收一个参数，是 task 对象，然后调用 print 方法打印了 task 对象的结果。这样我们就定义好了一个 coroutine 对象和一个回调方法。我们现在希望的效果是，当 coroutine 对象执行完毕之后，就去执行声明的 callback 方法。

那么它们两者怎样关联起来呢？很简单，只需要调用 add_done_callback 方法即可。我们将 callback 方法传递给封装好的 task 对象，这样当 task 执行完毕之后，就可以调用 callback 方法了。同时 task 对象还会作为参数传递给 callback 方法，调用 task 对象的 result 方法就可以获取返回结果了。

运行结果如下：

1
2
3

Task: <Task pending coro=<request() running at demo.py:5> cb=[callback() at demo.py:11]>
Status: <Response [200]>
Task: <Task finished coro=<request() done, defined at demo.py:5> result=<Response [200]>>

实际上不用回调方法，直接在 task 运行完毕之后，也可以直接调用 result 方法获取结果，如下所示：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

import asyncio
import requests

async def request():
    url = 'https://www.baidu.com'
    status = requests.get(url)
    return status

coroutine = request()
task = asyncio.ensure_future(coroutine)
print('Task:', task)

loop = asyncio.get_event_loop()
loop.run_until_complete(task)
print('Task:', task)
print('Task Result:', task.result())

运行结果是一样的：

1
2
3

Task: <Task pending coro=<request() running at demo.py:4>>
Task: <Task finished coro=<request() done, defined at demo.py:4> result=<Response [200]>>
Task Result: <Response [200]>

7. 多任务协程

上面的例子我们只执行了一次请求，如果想执行多次请求，应该怎么办呢？我们可以定义一个 task 列表，然后使用 asyncio 的 wait 方法即可执行。看下面的例子：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

import asyncio
import requests

async def request():
    url = 'https://www.baidu.com'
    status = requests.get(url)
    return status

tasks = [asyncio.ensure_future(request()) for _ in range(5)]
print('Tasks:', tasks)

loop = asyncio.get_event_loop()
loop.run_until_complete(asyncio.wait(tasks))

for task in tasks:
    print('Task Result:', task.result())

这里我们使用一个 for 循环创建了 5 个 task，组成了一个列表，然后把这个列表首先传递给了 asyncio 的 wait 方法，再将其注册到时间循环中，就可以发起 5 个任务了。最后，我们再将任务的运行结果输出出来，具体如下：

1
2
3
4
5
6

Tasks: [<Task pending coro=<request() running at demo.py:5>>, <Task pending coro=<request() running at demo.py:5>>, <Task pending coro=<request() running at demo.py:5>>, <Task pending coro=<request() running at demo.py:5>>, <Task pending coro=<request() running at demo.py:5>>]
Task Result: <Response [200]>
Task Result: <Response [200]>
Task Result: <Response [200]>
Task Result: <Response [200]>
Task Result: <Response [200]>

可以看到，5 个任务被顺次执行了，并得到了运行结果。

8. 协程实现

前面说了这么一通，又是 async，又是 coroutine，又是 task，又是 callback，但似乎并没有看出协程的优势？反而写法上更加奇怪和麻烦了。别急，上面的案例只是为后面的使用作铺垫。接下来，我们正式来看下协程在解决 IO 密集型任务上有怎样的优势。

在上面的代码中，我们用一个网络请求作为示例，这就是一个耗时等待操作，因为我们请求网页之后需要等待页面响应并返回结果。耗时等待操作一般都是 IO 操作，比如文件读取、网络请求等。协程对于处理这种操作是有很大优势的，当遇到需要等待的情况时，程序可以暂时挂起，转而去执行其他操作，从而避免一直等待一个程序而耗费过多的时间，充分利用资源。

为了表现出协程的优势，我们还是以本节开头介绍的网站 https://httpbin.org/delay/5 为例，因为该网站响应比较慢，所以我们可以通过爬取时间来直观感受到爬取速度的提升。

为了让大家更好地理解协程的正确使用方法，这里我们先来看看大家使用协程时常犯的错误，后面再给出正确的例子来对比一下。

首先，我们还是拿之前的 requests 库来进行网页请求，接下来再重新使用上面的方法请求一遍：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

import asyncio
import requests
import time

start = time.time()

async def request():
    url = 'https://httpbin.org/delay/5'
    print('Waiting for', url)
    response = requests.get(url)
    print('Get response from', url, 'response', response)

tasks = [asyncio.ensure_future(request()) for _ in range(10)]
loop = asyncio.get_event_loop()
loop.run_until_complete(asyncio.wait(tasks))

end = time.time()
print('Cost time:', end - start)

这里我们还是创建了 10 个 task，然后将 task 列表传给 wait 方法并注册到时间循环中执行。

运行结果如下：

1
2
3
4
5
6
7
8
9
10

Waiting for https://httpbin.org/delay/5
Get response from https://httpbin.org/delay/5 response <Response [200]>
Waiting for https://httpbin.org/delay/5
...
Get response from https://httpbin.org/delay/5 response <Response [200]>
Waiting for https://httpbin.org/delay/5
Get response from https://httpbin.org/delay/5 response <Response [200]>
Waiting for https://httpbin.org/delay/5
Get response from https://httpbin.org/delay/5 response <Response [200]>
Cost time: 66.64284420013428

可以发现，这和正常的请求并没有什么区别，依然还是顺次执行的，耗时 66 秒，平均一个请求耗时 6.6 秒，说好的异步处理呢？

其实，要实现异步处理，我们得先要有挂起的操作，当一个任务需要等待 IO 结果的时候，可以挂起当前任务，转而去执行其他任务，这样我们才能充分利用好资源。上面的方法都是一本正经地串行走下来，连个挂起都没有，怎么可能实现异步？想太多了。

要实现异步，接下来我们再了解一下 await 的用法，它可以将耗时等待的操作挂起，让出控制权。当协程执行的时候遇到 await，时间循环就会将本协程挂起，转而去执行别的协程，直到其他协程挂起或执行完毕。

所以，我们可能会将代码中的 request 方法改成如下的样子：

1
2
3
4
5

async def request():
    url = 'https://httpbin.org/delay/5'
    print('Waiting for', url)
    response = await requests.get(url)
    print('Get response from', url, 'response', response)

仅仅是在 requests 前面加了一个关键字 await，然而此时执行代码，会得到如下报错：

1
2
3
4
5
6
7
8
9
10
11

Waiting for https://httpbin.org/delay/5
Waiting for https://httpbin.org/delay/5
Waiting for https://httpbin.org/delay/5
Waiting for https://httpbin.org/delay/5
...
Task exception was never retrieved
future: <Task finished coro=<request() done, defined at demo.py:8> exception=TypeError("object Response can't be used in 'await' expression")>
Traceback (most recent call last):
  File "demo.py", line 11, in request
    response = await requests.get(url)
TypeError: object Response can't be used in 'await' expression

这次它遇到 await 方法确实挂起了，也等待了，但是最后却报了这个错误。这个错误的意思是 requests 返回的 Response 对象不能和 await 一起使用，为什么呢？因为根据官方文档说明，await 后面的对象必须是如下格式之一（具体可以参见 https://www.python.org/dev/peps/pep-0492/#await-expression）：

• 一个原生 coroutine 对象；
• 一个由 types.coroutine 修饰的生成器，这个生成器可以返回 coroutine 对象；
• 一个包含 __await__ 方法的对象返回的一个迭代器。

reqeusts 返回的 Response 对象不符合上面任一条件，因此就会报上面的错误了。

有的读者可能已经发现了，既然 await 后面可以跟一个 coroutine 对象，那么我用 async 把请求的方法改成 coroutine 对象不就可以了吗？所以就改写成如下的样子：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

import asyncio
import requests
import time

start = time.time()

async def get(url):
    return requests.get(url)

async def request():
    url = 'https://httpbin.org/delay/5'
    print('Waiting for', url)
    response = await get(url)
    print('Get response from', url, 'response', response)

tasks = [asyncio.ensure_future(request()) for _ in range(10)]
loop = asyncio.get_event_loop()
loop.run_until_complete(asyncio.wait(tasks))

end = time.time()
print('Cost time:', end - start)

这里我们将请求页面的方法独立出来，并用 async 修饰，这样就得到了一个 coroutine 对象。运行一下看看：

1
2
3
4
5
6
7
8
9
10
11
12

Waiting for https://httpbin.org/delay/5
Get response fromhttps://httpbin.org/delay/5 response <Response [200]>
Waiting for https://httpbin.org/delay/5
Get response from https://httpbin.org/delay/5 response <Response [200]>
Waiting for https://httpbin.org/delay/5
...
Get response from https://httpbin.org/delay/5 response <Response [200]>
Waiting for https://httpbin.org/delay/5
Get response from https://httpbin.org/delay/5 response <Response [200]>
Waiting for https://httpbin.org/delay/5
Get response from https://httpbin.org/delay/5 response <Response [200]>
Cost time: 65.394437756259273

还是不行，它还不是异步执行的，也就是说我们仅仅将涉及 IO 操作的代码封装到 async 修饰的方法里面是不可行的。我们必须要使用支持异步操作的请求方式才可以实现真正的异步，所以这里就需要 aiohttp 派上用场了。

9. 使用 aiohttp

aiohttp 是一个支持异步请求的库，配合使用它和 asyncio，我们可以非常方便地实现异步请求操作。我们使用 pip3 安装即可：

1

pip3 install aiohttp

具体的安装方法可以参考：https://setup.scrape.center/aiohttp。

aiohttp 的官方文档链接为 https://aiohttp.readthedocs.io/，它分为两部分，一部分是 Client，一部分是 Server，详细的内容可以参考官方文档。

下面我们将 aiohttp 用上来，将代码改成如下样子：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

import asyncio
import aiohttp
import time

start = time.time()

async def get(url):
    session = aiohttp.ClientSession()
    response = await session.get(url)
    await response.text()
    await session.close()
    return response

async def request():
    url = 'https://httpbin.org/delay/5'
    print('Waiting for', url)
    response = await get(url)
    print('Get response from', url, 'response', response)

tasks = [asyncio.ensure_future(request()) for _ in range(10)]
loop = asyncio.get_event_loop()
loop.run_until_complete(asyncio.wait(tasks))

end = time.time()
print('Cost time:', end - start)

这里我们将请求库由 requests 改成了 aiohttp，通过 aiohttp 的 ClientSession 类的 get 方法进行请求，结果如下：

1
2
3
4
5
6
7
8
9
10
11
12

Waiting for https://httpbin.org/delay/5
Waiting for https://httpbin.org/delay/5
Waiting for https://httpbin.org/delay/5
Waiting for https://httpbin.org/delay/5
...
Get response from https://httpbin.org/delay/5 response <ClientResponse(https://httpbin.org/delay/5) [200 OK]>
<CIMultiDictProxy('Date': 'Sun, 09 Aug 2020 14:30:22 GMT', 'Content-Type': 'application/json', 'Content-Length': '360', 'Connection': 'keep-alive', 'Server': 'gunicorn/19.9.0', 'Access-Control-Allow-Origin': '*', 'Access-Control-Allow-Credentials': 'true')>

...
Get response from https://httpbin.org/delay/5 response <ClientResponse(https://httpbin.org/delay/5) [200 OK]>
<CIMultiDictProxy('Date': 'Sun, 09 Aug 2020 14:30:22 GMT', 'Content-Type': 'application/json', 'Content-Length': '360', 'Connection': 'keep-alive', 'Server': 'gunicorn/19.9.0', 'Access-Control-Allow-Origin': '*', 'Access-Control-Allow-Credentials': 'true')>
Cost time: 6.033240079879761

成功了！我们发现这次请求的耗时由 51 秒直接变成了 6 秒，耗费时间减少了非常多。

在代码里面，我们使用了 await，后面跟了 get 方法。在执行这 10 个协程的时候，如果遇到了 await，就会将当前协程挂起，转而去执行其他协程，直到其他协程也挂起或执行完毕，再执行下一个协程。

开始运行时，时间循环会运行第一个 task。针对第一个 task 来说，当执行到第一个 await 跟着的 get 方法时，它被挂起，但这个 get 方法第一步的执行是非阻塞的，挂起之后立马被唤醒，所以立即又进入执行，创建了 ClientSession 对象，接着遇到了第二个 await，调用了 session.get 请求方法，然后就被挂起了。由于请求需要耗时很久，所以一直没有被唤醒，好在第一个 task 被挂起了，那么接下来该怎么办呢？事件循环会寻找当前未被挂起的协程继续执行，于是就转而执行第二个 task 了，也是一样的流程操作，直到执行了第十个 task 的 session.get 方法之后，全部的 task 都被挂起了。所有 task 都已经处于挂起状态，那咋办？只好等待了。5 秒之后，几个请求几乎同时都有了响应，然后几个 task 也被唤醒接着执行，输出请求结果，最后总耗时 6 秒！

怎么样？这就是异步操作的便捷之处，当遇到阻塞式操作时，任务被挂起，程序接着去执行其他任务，而不是傻傻地等着，这样可以充分利用 CPU 时间，而不必把时间浪费在等待 IO 上。

有人会说，既然这样的话，在上面的例子中，在发出网络请求后，既然接下来的 5 秒都是在等待的，在 5 秒之内，CPU 可以处理的 task 数量远不止这些，那么岂不是我们放 10 个、20 个、50 个、100 个、1000 个 task 一起执行，最后得到所有结果的耗时不都是差不多的吗？因为这几个任务被挂起后都是一起等待的。

理论来说，确实是这样的，不过有个前提，那就是服务器在同一时刻接受无限次请求都能保证正常返回结果，也就是服务器无限抗压。另外，还要忽略 IO 传输时延，确实可以做到无限 task 一起执行且在预想时间内得到结果。但由于不同服务器处理的实现机制不同，可能某些服务器并不能承受这么高的并发，因此响应速度也会减慢。

这里我们以百度为例，测试一下并发数量为 1、3、5、10…500 的情况下的耗时情况，代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

import asyncio
import aiohttp
import time


def test(number):
    start = time.time()

    async def get(url):
        session = aiohttp.ClientSession()
        response = await session.get(url)
        await response.text()
        await session.close()
        return response

    async def request():
        url = 'https://www.baidu.com/'
        await get(url)

    tasks = [asyncio.ensure_future(request()) for _ in range(number)]
    loop = asyncio.get_event_loop()
    loop.run_until_complete(asyncio.wait(tasks))

    end = time.time()
    print('Number:', number, 'Cost time:', end - start)

for number in [1, 3, 5, 10, 15, 30, 50, 75, 100, 200, 500]:
    test(number)

运行结果如下：

1
2
3
4
5
6
7
8
9
10
11

Number: 1 Cost time: 0.05885505676269531
Number: 3 Cost time: 0.05773782730102539
Number: 5 Cost time: 0.05768704414367676
Number: 10 Cost time: 0.15174412727355957
Number: 15 Cost time: 0.09603095054626465
Number: 30 Cost time: 0.17843103408813477
Number: 50 Cost time: 0.3741800785064697
Number: 75 Cost time: 0.2894289493560791
Number: 100 Cost time: 0.6185381412506104
Number: 200 Cost time: 1.0894129276275635
Number: 500 Cost time: 1.8213098049163818

可以看到，即使我们增加了并发数量，但在服务器能承受高并发的前提下，其爬取速度几乎不太受影响。

综上所述，使用了异步请求之后，我们几乎可以在相同的时间内实现成百上千倍次的网络请求，把这个运用在爬虫中，速度提升可谓是非常可观了。

10. 总结

以上便是 Python 中协程的基本原理和用法，在后面一节中我们会详细介绍 aiohttp 的用法和爬取实战，实现快速高并发的爬取。

本节代码：https://github.com/Python3WebSpider/AsyncTest

爬虫系列文章总目录：【2022 年】Python3 爬虫学习教程，本教程内容多数来自于《Python3 网络爬虫开发实战（第二版）》一书，目前截止 2022 年，可以将爬虫基本技术进行系统讲解，同时将最新前沿爬虫技术如异步、JavaScript 逆向、AST、安卓逆向、Hook、智能解析、群控技术、WebAssembly、大规模分布式、Docker、Kubernetes 等，市面上目前就仅有《Python3 网络爬虫开发实战（第二版）》一书了，点击了解详情。

Playwright 是微软在 2020 年初开源的新一代自动化测试工具，它的功能类似于 Selenium、Pyppeteer 等，都可以驱动浏览器进行各种自动化操作。它的功能也非常强大，对市面上的主流浏览器都提供了支持，API 功能简洁又强大。虽然诞生比较晚，但是现在发展得非常火热。

1. Playwright 的特点

• Playwright 支持当前所有主流浏览器，包括 Chrome 和 Edge（基于 Chromium）、Firefox、Safari（基于 WebKit） ，提供完善的自动化控制的 API。
• Playwright 支持移动端页面测试，使用设备模拟技术可以使我们在移动 Web 浏览器中测试响应式 Web 应用程序。
• Playwright 支持所有浏览器的 Headless 模式和非 Headless 模式的测试。
• Playwright 的安装和配置非常简单，安装过程中会自动安装对应的浏览器和驱动，不需要额外配置 WebDriver 等。
• Playwright 提供了自动等待相关的 API，当页面加载的时候会自动等待对应的节点加载，大大简化了 API 编写复杂度。

本节我们就来了解下 Playwright 的使用方法。

2. 安装

要使用 Playwright，需要 Python 3.7 版本及以上，请确保 Python 的版本符合要求。

要安装 Playwright，可以直接使用 pip3，命令如下：

1

pip3 install playwright

安装完成之后需要进行一些初始化操作：

1

playwright install

这时候 Playwrigth 会安装 Chromium, Firefox and WebKit 浏览器并配置一些驱动，我们不必关心中间配置的过程，Playwright 会为我们配置好。

具体的安装说明可以参考：https://setup.scrape.center/playwright。

安装完成之后，我们便可以使用 Playwright 启动 Chromium 或 Firefox 或 WebKit 浏览器来进行自动化操作了。

3. 基本使用

Playwright 支持两种编写模式，一种是类似 Pyppetter 一样的异步模式，另一种是像 Selenium 一样的同步模式，我们可以根据实际需要选择使用不同的模式。

我们先来看一个基本同步模式的例子：

1
2
3
4
5
6
7
8
9
10

from playwright.sync_api import sync_playwright

with sync_playwright() as p:
    for browser_type in [p.chromium, p.firefox, p.webkit]:
        browser = browser_type.launch(headless=False)
        page = browser.new_page()
        page.goto('https://www.baidu.com')
        page.screenshot(path=f'screenshot-{browser_type.name}.png')
        print(page.title())
        browser.close()

首先我们导入了 sync_playwright 方法，然后直接调用了这个方法，该方法返回的是一个 PlaywrightContextManager 对象，可以理解是一个浏览器上下文管理器，我们将其赋值为变量 p。

接着我们调用了 PlaywrightContextManager 对象的 chromium、firefox、webkit 属性依次创建了一个 Chromium、Firefox 以及 Webkit 浏览器实例，接着用一个 for 循环依次执行了它们的 launch 方法，同时设置了 headless 参数为 False。

注意：如果不设置为 False，默认是无头模式启动浏览器，我们看不到任何窗口。

launch 方法返回的是一个 Browser 对象，我们将其赋值为 browser 变量。然后调用 browser 的 new_page 方法，相当于新建了一个选项卡，返回的是一个 Page 对象，将其赋值为 page，这整个过程其实和 Pyppeteer 非常类似。接着我们就可以调用 page 的一系列 API 来进行各种自动化操作了，比如调用 goto，就是加载某个页面，这里我们访问的是百度的首页。接着我们调用了 page 的 screenshot 方法，参数传一个文件名称，这样截图就会自动保存为该图片名称，这里名称中我们加入了 browser_type 的 name 属性，代表浏览器的类型，结果分别就是 chromium, firefox, webkit。另外我们还调用了 title 方法，该方法会返回页面的标题，即 HTML 中 title 节点中的文字，也就是选项卡上的文字，我们将该结果打印输出到控制台。最后操作完毕，调用 browser 的 close 方法关闭整个浏览器，运行结束。

运行一下，这时候我们可以看到有三个浏览器依次启动并加载了百度这个页面，分别是 Chromium、Firefox 和 Webkit 三个浏览器，页面加载完成之后，生成截图、控制台打印结果就退出了。

这时候当前目录便会生成三个截图文件，都是百度的首页，文件名中都带有了浏览器的名称，如图所示：

控制台运行结果如下：

1
2
3

百度一下，你就知道
百度一下，你就知道
百度一下，你就知道

通过运行结果我们可以发现，我们非常方便地启动了三种浏览器并完成了自动化操作，并通过几个 API 就完成了截图和数据的获取，整个运行速度是非常快的，者就是 Playwright 最最基本的用法。

当然除了同步模式，Playwright 还提供异步的 API，如果我们项目里面使用了 asyncio，那就应该使用异步模式，写法如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14

import asyncio
from playwright.async_api import async_playwright

async def main():
    async with async_playwright() as p:
        for browser_type in [p.chromium, p.firefox, p.webkit]:
            browser = await browser_type.launch()
            page = await browser.new_page()
            await page.goto('https://www.baidu.com')
            await page.screenshot(path=f'screenshot-{browser_type.name}.png')
            print(await page.title())
            await browser.close()

asyncio.run(main())

可以看到整个写法和同步模式基本类似，导入的时候使用的是 async_playwright 方法，而不再是 sync_playwright 方法。写法上添加了 async/await 关键字的使用，最后的运行效果是一样的。

另外我们注意到，这例子中使用了 with as 语句，with 用于上下文对象的管理，它可以返回一个上下文管理器，也就对应一个 PlaywrightContextManager 对象，无论运行期间是否抛出异常，它能够帮助我们自动分配并且释放 Playwright 的资源。

4. 代码生成

Playwright 还有一个强大的功能，那就是可以录制我们在浏览器中的操作并将代码自动生成出来，有了这个功能，我们甚至都不用写任何一行代码，这个功能可以通过 playwright 命令行调用 codegen 来实现，我们先来看看 codegen 命令都有什么参数，输入如下命令：

1

playwright codegen --help

结果类似如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

Usage: npx playwright codegen [options] [url]

open page and generate code for user actions

Options:
  -o, --output <file name>     saves the generated script to a file
  --target <language>          language to use, one of javascript, python, python-async, csharp (default: "python")
  -b, --browser <browserType>  browser to use, one of cr, chromium, ff, firefox, wk, webkit (default: "chromium")
  --channel <channel>          Chromium distribution channel, "chrome", "chrome-beta", "msedge-dev", etc
  --color-scheme <scheme>      emulate preferred color scheme, "light" or "dark"
  --device <deviceName>        emulate device, for example  "iPhone 11"
  --geolocation <coordinates>  specify geolocation coordinates, for example "37.819722,-122.478611"
  --load-storage <filename>    load context storage state from the file, previously saved with --save-storage
  --lang <language>            specify language / locale, for example "en-GB"
  --proxy-server <proxy>       specify proxy server, for example "http://myproxy:3128" or "socks5://myproxy:8080"
  --save-storage <filename>    save context storage state at the end, for later use with --load-storage
  --timezone <time zone>       time zone to emulate, for example "Europe/Rome"
  --timeout <timeout>          timeout for Playwright actions in milliseconds (default: "10000")
  --user-agent <ua string>     specify user agent string
  --viewport-size <size>       specify browser viewport size in pixels, for example "1280, 720"
  -h, --help                   display help for command

Examples:

  $ codegen
  $ codegen --target=python
  $ codegen -b webkit https://example.com

可以看到这里有几个选项，比如 -o 代表输出的代码文件的名称；—target 代表使用的语言，默认是 python，即会生成同步模式的操作代码，如果传入 python-async 就会生成异步模式的代码；-b 代表的是使用的浏览器，默认是 Chromium，其他还有很多设置，比如 —device 可以模拟使用手机浏览器，比如 iPhone 11，—lang 代表设置浏览器的语言，—timeout 可以设置页面加载超时时间。

好，了解了这些用法，那我们就来尝试启动一个 Firefox 浏览器，然后将操作结果输出到 script.py 文件，命令如下：

1

playwright codegen -o script.py -b firefox

这时候就弹出了一个 Firefox 浏览器，同时右侧会输出一个脚本窗口，实时显示当前操作对应的代码。

我们可以在浏览器中做任何操作，比如打开百度，然后点击输入框并输入 nba，然后再点击搜索按钮，浏览器窗口如下：

可以看见浏览器中还会高亮显示我们正在操作的页面节点，同时还显示了对应的选择器字符串 input[name="wd"]，右侧的窗口如图所示：

在操作过程中，该窗口中的代码就实时变化，可以看到这里生成了我们一系列操作的对应代码，比如在搜索框中输入 nba，就对应如下代码：

1

page.fill("input[name=\"wd\"]", "nba")

操作完毕之后，关闭浏览器，Playwright 会生成一个 script.py 文件，内容如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

from playwright.sync_api import sync_playwright

def run(playwright):
    browser = playwright.firefox.launch(headless=False)
    context = browser.new_context()

    # Open new page
    page = context.new_page()

    # Go to https://www.baidu.com/
    page.goto("https://www.baidu.com/")

    # Click input[name="wd"]
    page.click("input[name=\"wd\"]")

    # Fill input[name="wd"]
    page.fill("input[name=\"wd\"]", "nba")

    # Click text=百度一下
    with page.expect_navigation():
        page.click("text=百度一下")

    context.close()
    browser.close()

with sync_playwright() as playwright:
    run(playwright)

可以看到这里生成的代码和我们之前写的示例代码几乎差不多，而且也是完全可以运行的，运行之后就可以看到它又可以复现我们刚才所做的操作了。

所以，有了这个功能，我们甚至都不用编写任何代码，只通过简单的可视化点击就能把代码生成出来，可谓是非常方便了！

另外这里有一个值得注意的点，仔细观察下生成的代码，和前面的例子不同的是，这里 new_page 方法并不是直接通过 browser 调用的，而是通过 context 变量调用的，这个 context 又是由 browser 通过调用 new_context 方法生成的。有读者可能就会问了，这个 context 究竟是做什么的呢？

其实这个 context 变量对应的是一个 BrowserContext 对象，BrowserContext 是一个类似隐身模式的独立上下文环境，其运行资源是单独隔离的，在做一些自动化测试过程中，每个测试用例我们都可以单独创建一个 BrowserContext 对象，这样可以保证每个测试用例之间互不干扰，具体的 API 可以参考 https://playwright.dev/python/docs/api/class-browsercontext。

5. 移动端浏览器支持

Playwright 另外一个特色功能就是可以支持移动端浏览器的模拟，比如模拟打开 iPhone 12 Pro Max 上的 Safari 浏览器，然后手动设置定位，并打开百度地图并截图。首先我们可以选定一个经纬度，比如故宫的经纬度是 39.913904, 116.39014，我们可以通过 geolocation 参数传递给 Webkit 浏览器并初始化。

示例代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

from playwright.sync_api import sync_playwright

with sync_playwright() as p:
    iphone_12_pro_max = p.devices['iPhone 12 Pro Max']
    browser = p.webkit.launch(headless=False)
    context = browser.new_context(
        **iphone_12_pro_max,
        locale='zh-CN',
        geolocation={'longitude': 116.39014, 'latitude': 39.913904},
        permissions=['geolocation']
    )
    page = context.new_page()
    page.goto('https://amap.com')
    page.wait_for_load_state(state='networkidle')
    page.screenshot(path='location-iphone.png')
    browser.close()

这里我们先用 PlaywrightContextManager 对象的 devices 属性指定了一台移动设备，这里传入的是手机的型号，比如 iPhone 12 Pro Max，当然也可以传其他名称，比如 iPhone 8，Pixel 2 等。

前面我们已经了解了 BrowserContext 对象，BrowserContext 对象也可以用来模拟移动端浏览器，初始化一些移动设备信息、语言、权限、位置等信息，这里我们就用它来创建了一个移动端 BrowserContext 对象，通过 geolocation 参数传入了经纬度信息，通过 permissions 参数传入了赋予的权限信息，最后将得到的 BrowserContext 对象赋值为 context 变量。

接着我们就可以用 BrowserContext 对象来新建一个页面，还是调用 new_page 方法创建一个新的选项卡，然后跳转到高德地图，并调用了 wait_for_load_state 方法等待页面某个状态完成，这里我们传入的 state 是 networkidle，也就是网络空闲状态。因为在页面初始化和加载过程中，肯定是伴随有网络请求的，所以加载过程中肯定不算 networkidle 状态，所以这里我们传入 networkidle 就可以标识当前页面和数据加载完成的状态。加载完成之后，我们再调用 screenshot 方法获取当前页面截图，最后关闭浏览器。

运行下代码，可以发现这里就弹出了一个移动版浏览器，然后加载了高德地图，并定位到了故宫的位置，如图所示：

输出的截图也是浏览器中显示的结果。

所以这样我们就成功实现了移动端浏览器的模拟和一些设置，其操作 API 和 PC 版浏览器是完全一样的。

6. 选择器

前面我们注意到 click 和 fill 等方法都传入了一个字符串，这些字符串有的符合 CSS 选择器的语法，有的又是 text= 开头的，感觉似乎没太有规律的样子，它到底支持怎样的匹配规则呢？下面我们来了解下。

传入的这个字符串，我们可以称之为 Element Selector，它不仅仅支持 CSS 选择器、XPath，Playwright 还扩展了一些方便好用的规则，比如直接根据文本内容筛选，根据节点层级结构筛选等等。

文本选择

文本选择支持直接使用 text= 这样的语法进行筛选，示例如下：

1

page.click("text=Log in")

这就代表选择文本是 Log in 的节点，并点击。

CSS 选择器

CSS 选择器之前也介绍过了，比如根据 id 或者 class 筛选：

1
2

page.click("button")
page.click("#nav-bar .contact-us-item")

根据特定的节点属性筛选：

1
2

page.click("[data-test=login-button]")
page.click("[aria-label='Sign in']")

CSS 选择器 + 文本

我们还可以使用 CSS 选择器结合文本值进行海选，比较常用的就是 has-text 和 text，前者代表包含指定的字符串，后者代表字符串完全匹配，示例如下：

1
2

page.click("article:has-text('Playwright')")
page.click("#nav-bar :text('Contact us')")

第一个就是选择文本中包含 Playwright 的 article 节点，第二个就是选择 id 为 nav-bar 节点中文本值等于 Contact us 的节点。

CSS 选择器 + 节点关系

还可以结合节点关系来筛选节点，比如使用 has 来指定另外一个选择器，示例如下：

1

page.click(".item-description:has(.item-promo-banner)")

比如这里选择的就是选择 class 为 item-description 的节点，且该节点还要包含 class 为 item-promo-banner 的子节点。

另外还有一些相对位置关系，比如 right-of 可以指定位于某个节点右侧的节点，示例如下：

1

page.click("input:right-of(:text('Username'))")

这里选择的就是一个 input 节点，并且该 input 节点要位于文本值为 Username 的节点的右侧。

XPath

当然 XPath 也是支持的，不过 xpath 这个关键字需要我们自行制定，示例如下：

1

page.click("xpath=//button")

这里需要在开头指定 xpath= 字符串，代表后面是一个 XPath 表达式。

关于更多选择器的用法和最佳实践，可以参考官方文档：https://playwright.dev/python/docs/selectors。

7. 常用操作方法

上面我们了解了浏览器的一些初始化设置和基本的操作实例，下面我们再对一些常用的操作 API 进行说明。

常见的一些 API 如点击 click，输入 fill 等操作，这些方法都是属于 Page 对象的，所以所有的方法都从 Page 对象的 API 文档查找就好了，文档地址：https://playwright.dev/python/docs/api/class-page。

下面介绍几个常见的 API 用法。

事件监听

Page 对象提供了一个 on 方法，它可以用来监听页面中发生的各个事件，比如 close、console、load、request、response 等等。

比如这里我们可以监听 response 事件，response 事件可以在每次网络请求得到响应的时候触发，我们可以设置对应的回调方法获取到对应 Response 的全部信息，示例如下：

1
2
3
4
5
6
7
8
9
10
11
12

from playwright.sync_api import sync_playwright

def on_response(response):
    print(f'Statue {response.status}: {response.url}')

with sync_playwright() as p:
    browser = p.chromium.launch(headless=False)
    page = browser.new_page()
    page.on('response', on_response)
    page.goto('https://spa6.scrape.center/')
    page.wait_for_load_state('networkidle')
    browser.close()