本文预计阅读需3min

你好，我是你老朋友Payne，大家都或许过我之前写的水文-JS解密入门，没看过的童鞋开源回头看看啊。里面主要讲述了Hash MD5的例子，以及加密与解密，相关的。那么今天我们去搞一下MD5的“父亲”， Hash。主要阐述了什么是哈希，哈希运用方向以及hash碰撞及解决方向，请查阅

Hash算法

哈希

（hash）也叫散列。Hash算法，是将一个不定长的输入，通过散列函数变换成一个定长的输出，即散列值。

应用

Hash主要应用在数据结构以及密码学领域。

数据结构：

使用Hash算法的数据结构叫做哈希表，也叫散列表，主要是为了提高查询的效率。它通过把关键码值映射到表中一个位置来访问记录，以加快查找的速度。这个映射函数就是hash function，存放记录的数组叫做哈希表。 在数据结构中应用时，有时需要较高的运算速度而弱化考虑抗碰撞性，可以使用自己构建的哈希函数。

密码学：

这种hash散列变换是一种单向运算，具有不可逆性即不能根据散列值还原出输入信息，因此严格意义上讲Hash算法是一种消息摘要算法，不是一种加密算法。常见的hash算法有：SM3、MD5、SHA-1等 。 在不同的应用场景下，hash函数的选择也会有所侧重。比如在管理数据结构时，主要要考虑运算的快速性，并且要保证hash均匀分布；而应用在密码学中就要优先考虑 抗碰撞性 ，避免出现两段不同明文hash值相同的情况发生。

哈希碰撞:

不同的值经过Hash function得到同一值则产生哈希碰撞， 防止哈希碰撞的最有效方法，就是扩大哈希值的取值空间

1
2
3

16个二进制位的哈希值，产生碰撞的可能性是 65536 分之一。如果有65537个用户，就一定会产生碰撞。哈希值的长度扩大到32个二进制位，碰撞的可能性就会下降到 4,294,967,296 分之一。

更长的哈希值意味着更大的存储空间、更多的计算，将影响性能和成本。开发者s必须做出抉择，在安全与成本之间找到平衡。

hash碰撞解决办法：

1. 开放地址法

1. 线性探测再散列
2. 二次探测再散列

3. 伪随机探测再散列

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

   所谓的开放定址法就是一旦发生了冲突，就去寻找下一个空的散列地址，只要散列表足够大，空的散列地址总能找到，并将记录存入 。 开放寻址法：Hi=(H(key) + di) MOD m,i=1,2,…，k(k<=m-1)，其中H(key)为散列函数，m为散列表长，di为增量序列，可有下列三种取法： 1)． di=1,2,3,…，m-1，称线性探测再散列； 2)． di=1^2,(-1)^2,2^2,(-2)^2,(3)^2,…，±(k)^2,(k<=m/2)称二次探测再散列； 3)． di=伪随机数序列，称伪随机探测再散列。 用开放定址法解决冲突的做法是：当冲突发生时，使用某种探测技术（线性探测法、二次探测法（解决线性探测的堆积问题）、随机探测法（和二次探测原理一致，不一样的是：二次探测以定值跳跃，而随机探测的散列地址跳跃长度是不定值））在散列表中形成一个探测序列。 沿此序列逐个单元地查找，直到找到给定的关键字，或者碰到一个开放的地址（即该地址单元为空）为止插入即可。

4. 再哈希法

再哈希法又叫双哈希法，有多个不同的Hash函数，当发生冲突时，使用第二个，第三个，….，等哈希函数去计算地址，直到无冲突。 虽然不易发生聚集，但是增加了计算时间。

1. 链地址法(Java hashmap)

链地址法的基本思想是：每个哈希表节点都有一个next指针，多个哈希表节点可以用next指针构成一个单向链表，将所有关键字为同义词的结点链接在同一个单链表中，

1. 建立公共溢出区

建立公共溢出区的基本思想是：假设哈希函数的值域是 [1,m-1]，则设向量 HashTable[0…m-1] 为基本表， 每个分量存放一个记录，另外设向量 OverTable[0…v] 为溢出表，所有关键字和基本表中关键字为同义词的记录，不管它们由哈希函数得到的哈希地址是什么，一旦发生冲突，都填入溢出表。

你好，我是悦创。 本篇将开启我自己啃代理池的心得，将逐步放送，因为代理池搭建较为复杂，这里我就尽可能把代理池分成几篇来讲，同时也保证，在我其他篇放出来之前，每一篇都是你们的新知识。 学习就像看小说一样，一次一篇就会显得额外的轻松！ 当你把学习当作某个娱乐的事情来做，你会发现不一样的世界！ 我们无法延长生命长度，但是我们延长生命宽度，学习编程就是扩展生命最有力的武器！

1. 看完之后你会得到什么

• 返回 yield；
• eval 的使用；
• 多个代理网站同时抓取；
• 使用异步测试代理是否可用；
• Python 的元类编程简单介绍；
• 正则表达式、PyQuery 提取数据；
• 模块化编程；

废话不多说，马上步入正题！

2. 你需要的准备

在学习本篇文章时，希望你已经具备如下技能或者知识点：

1. Python 环境（推荐 Python 3.7+）；
2. Python 爬虫常用库；
3. Python 基本语法；
4. 面向对象编程；
5. yield、eval 的使用；
6. 模块化编程；

3. 课前预习知识点

对于代理池的搭建呢，虽然我已经尽可能地照顾到绝大多数地小白，把代理地的搭建呢，进行了拆分，不过对于培训机构或者自学不是特别精进的小伙伴来说还是有些难度的，对于这些呢？我这里也给大家准备了知识点的扫盲。以下内容节选自我个人博客文章，这里为了扫盲就提供必要部分，想看完整的文章可以点击此链接：https://www.aiyc.top/archives/605.html

3.1 Python 的元类编程

你好，我是悦创。 好久不见，最近在啃数学、Java、英语没有来更新公众号，那么今天就来更新一下啦！ 又到了每日一啃，啃代码的小悦。今天我遇到了 Python 原类，然而我啥也不懂。只能靠百度和谷歌了，主要还是用谷歌来查啦，百度前几条永远是广告准确度也不行（个人观点）,也顺便参考了几个博客：廖雪峰网站，添加了一点自己的观点和理解。

3.1.1 type()

动态语言和静态语言最大的不同，就是函数和类的定义，不是编译时定义的，而是运行时动态创建的。 比方说我们要定义一个 Hello 的 class，就写一个 hello.py 模块（这里我实际创建的是：the_test_code_project.py 模块）：

1
2
3

class Hello(object):
def hello(self, name='world'):
print('Hello, %s.' % name)

当 Python 解释器载入 hello 模块时，就会依次执行该模块的所有语句，执行结果就是动态创建出一个 Hello 的 class 对象，测试如下：

1
2
3
4
5

if __name__ == '__main__':
h = Hello()
h.hello()
print(type(Hello))
print(type(h))

运行结果如下：

1
2
3

Hello, world.
<class 'type'>
<class '__main__.Hello'>

其中，上面的输出结果：__main__.Hello 等价于 <class 'the_test_code_project.Hello'> 运行的方式不同显示的方式也不同，但含义是一样的。 type() 函数可以查看一个类型或变量的类型，为了让小白更轻松，我也写了个例子：

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

number = 12
string = 'Hello AIYC!'
float_number = 12.1
list_data = [1, '2', 3.5, 'AIYC'] # 可变
tuples = (1, '2', 3.5, 'AIYC') # 不可变

if __name__ == '__main__':
print(type(number))
print(type(string))
print(type(float_number))
print(type(list_data))
print(type(tuples))

运行结果：

1
2
3
4
5

<class 'int'>
<class 'str'>
<class 'float'>
<class 'list'>
<class 'tuple'>

Hello 是一个 class，它的类型就是 type ，而 h 是一个实例，它的类型就是class Hello。 我们说 class 的定义是运行时动态创建的，而创建 class 的方法就是使用 type() 函数。 type() 函数既可以返回一个对象的类型，又可以创建出新的类型，比如，我们可以通过 type() 函数创建出Hello 类，而无需通过 class Hello(object)... 的定义：

1
2
3
4
5
6

def fn(self, name='world'): # 先定义函数
print('Hello, %s.' % name)

Hello = type('Hello', (object,), dict(hello=fn)) # 创建 Hello class
# Hello = type('Class_Name', (object,), dict(hello=fn)) # 创建 Hello class
# type(类名, 父类的元组（针对继承的情况，可以为空），包含属性的字典（名称和值）)

我们接下来来调用一下代码，看输出的结果如何：

1
2
3
4
5

if __name__ == '__main__':
h = Hello()
h.hello()
print(type(Hello))
print(type(h))

这里推荐写成：if __name__ == '__main__': 使代码更加的规范。 运行结果：

1
2
3

Hello, world.
<class 'type'>
<class '__main__.Hello'>

要创建一个 class 对象，type() 函数依次传入 3 个参数：

1. class 的名称；
2. 继承的父类集合，注意 Python 支持多重继承，如果只有一个父类，别忘了 tuple 的单元素写法；（这个个 tuple 单元素写法起初本人不太理解，然后一查并认真观察了一下上面的代码就想起来 tuple 单元素写法需要加逗号（，），就是你必须这么写：tuple_1 = (1,) 而不能这么写：tuple_2 = (1) ，tuple_2 = (1) 的写法，Python 会自动认为是一个整数而不是一个元组）
3. class 的方法名称与函数绑定，这里我们把函数 fn 绑定到方法名 hello 上。

通过 type() 函数创建的类和直接写 class 是完全一样的，因为 Python 解释器遇到 class 定义时，仅仅是扫描一下 class 定义的语法，然后调用 type() 函数创建出 class。（直接 Class 创建也是） 正常情况下，我们都用 class Xxx... 来定义类，但是，type() 函数也允许我们动态创建出类来，也就是说，动态语言本身支持运行期动态创建类，这和静态语言有非常大的不同，要在静态语言运行期创建类，必须构造源代码字符串再调用编译器，或者借助一些工具生成字节码实现，本质上都是动态编译，会非常复杂。

3.1.2 metaclass

除了使用 type() 动态创建类以外，要控制类的 创建行为 ，还可以使用 metaclass。 metaclass，直译为元类，简单的解释就是：

• 当我们定义了类以后，就可以根据这个类创建出实例，所以：先定义类，然后创建实例。

• 但是如果我们想创建出类呢？

那就必须根据 metaclass 创建出类，所以：先定义 metaclass ，然后创建类。连接起来就是：先定义 metaclass ，就可以创建类，最后创建实例。 所以，metaclass 允许你创建类或者修改类 。换句话说，你可以把类看成是 metaclass 创建出来的“实例”。

metaclass 是 Python 面向对象里最难理解，也是最难使用的魔术代码。正常情况下，你不会碰到需要使用metaclass的情况，所以，以下内容看不懂也没关系，因为基本上你不会用到。（然而还是被我遇见了，而且还是看不懂，但经过大佬的指点就只是知道如何使用，但并不了解其中的原理，所以才有了此篇。)

我们先看一个简单的例子，这个 metaclass 可以给我们自定义的 MyList 增加一个 add 方法：

1
2
3
4

class ListMetaclass(type):
def __new__(cls, name, bases, attrs):
attrs['add'] = lambda self, value: self.append(value) # 加上新的方法
return type.__new__(cls, name, bases, attrs) # 返回修改后的定义

定义 ListMetaclass ，按照默认习惯，metaclass 的类名总是以 Metaclass 结尾，以便清楚地表示这是一个metaclass 。

有了 ListMetaclass ，我们在定义类的时候还要指示使用 ListMetaclass 来定制类，传入关键字参数 metaclass ：

1
2

class MyList(list, metaclass=ListMetaclass):
pass

当我们传入关键字参数 metaclass 时，魔术就生效了，它指示 Python 解释器在创建 MyList 时，要通过ListMetaclass.__new__() 来创建，在此，我们可以修改类的定义，比如，加上新的方法，然后，返回修改后的定义。 __new__() 方法接收到的参数依次是：

1. 当前准备创建的类的对象；
2. 类的名字；
3. 类继承的父类集合；
4. 类的方法集合。

测试一下 MyList 是否可以调用 add() 方法：

1
2
3
4
5
6

L = MyList()
L.add(1)
print(L)

# 输出
[1]

而普通的 list 没有 add() 方法：

1
2
3
4
5

\>>> L2 = list()
>>> L2.add(1)
Traceback (most recent call last):
File "<stdin>", line 1, in <module>
AttributeError: 'list' object has no attribute 'add'

这时候，我想你应该和我会有同样的问题，动态修改有什么意义？ 直接在 MyList 定义中写上 add() 方法不是更简单吗？ 正常情况下，确实应该直接写，我觉得通过 metaclass 修改纯属变态。

3.2 Python eval() 函数

对于 Python 的 eval 呢，大部分人是这么定义的：eval() 函数用来执行一个字符串表达式，并返回表达式的值。 这句话，有可能对于新手来说并不是非常好理解，我们还是用例子来顿悟一下吧。 以下是 eval() 方法的语法：

1

eval(expression[, globals[, locals]])

参数

• expression — 表达式。
• globals — 变量作用域，全局命名空间，如果被提供，则必须是一个字典对象。
• locals — 变量作用域，局部命名空间，如果被提供，可以是任何映射对象。

返回值 返回表达式计算结果。 实际操作

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

In [3]: x = 7

In [4]: eval( '3 * x' )
Out[4]: 21

In [5]: eval('pow(2,2)')
Out[5]: 4

In [6]: eval('2 + 2')
Out[6]: 4

In [7]: n=81

In [8]: eval("n + 4")
Out[8]: 85

再来个函数的操作：

1
2
3
4

In [1]: str1 = "print('Hello World')"

In [2]: eval(str1)
Hello World

ok，到此零基础小白关怀文章就完成了，我就不继续赘述啦！ 看到这里，你们的身体还行吗？ 我正在的干货要开始了！ 进入我们的 show time 环节，起飞了、起飞了，做好准备了！

4. 抓取免费代理

这里呢，我就不再带带大家手摸手的教学如何抓取代理了，因为当你能看这篇文章时，相信你已经对爬虫是有所入门了，如果还没入门的小伙伴可以关注本公众号，往期文章也有零基础教学和基础的课程资源，可以公众号后台回复：Python爬虫。即可获取资源，如果失效也别担心加小悦好友即可！（资源多多噢！）

小白不建议报名培训机构，毕竟现在培训结构收智商税比较多，还是需要多多鉴别噢！

4.1 目标的代理网站

1. 快代理：https://www.kuaidaili.com/
2. 西刺代理：https://www.xicidaili.com
3. 66代理：http://www.66ip.cn/
4. 无忧代理：http://www.data5u.com
5. 开心代理-高匿：http://www.kxdaili.com/dailiip/
6. 云代理：http://www.ip3366.net/free/

对于每个网站的代码呢，具体实现也是比较简单的，这里我就不做过多的赘述啦！ 接下来我都直接上代码，不过在上代理的爬虫代码前，我们需要来写个元类，代码如下：

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

class ProxyMetaclass(type):
"""
定义 ProxyMetaclass ，按照默认习惯，metaclass 的类名总是以 Metaclass 结尾，以便清楚地表示这是一个metaclass ：
元类，在 FreeProxyGetter 类中加入
CrawlFunc_list 和 CrawlFuncCount
两个参数，分别表示爬虫函数，和爬虫函数的数量。
"""
def __new__(cls, name, bases, attrs):
count = 0
attrs['CrawlFunc_List'] = [] # 添加：CrawlFunc_List 列表方法
for k, v in attrs.items():
if 'crawl_' in k:
# 判断这个函数里面是否携带 crawl_ 也就是利用 value in xxx
attrs['CrawlFunc_List'].append(k)
count += 1
attrs['CrawlFuncCount'] = count # 检测添加的函数的总数量
return type.__new__(cls, name, bases, attrs) # 返回所修改的

详细的代码含义，已经写在上面了，具体的这里 就不赘述了，如果泥有任何不理解的可以去点击阅读原文在我的博客网站下留言，和后台回复数字“3”，加小编好友，拉你入交流群。

4.2 代码编写

4.2.1 请求函数编写

单独保存成一个文件：utils.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
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70

"""
project = 'Code', file_name = 'utils.py', author = 'AI悦创'
time = '2020/6/1 12:34', product_name = PyCharm, 公众号：AI悦创
code is far away from bugs with the god animal protecting
I love animals. They taste delicious.
"""
import requests
import asyncio
import aiohttp
from requests.exceptions import ConnectionError
from fake_useragent import UserAgent,FakeUserAgentError
import random

def get_page(url, options={}):
"""
构造随机请求头，如果不理解的可以阅读此文章：
两行代码设置 Scrapy UserAgent：https://www.aiyc.top/archives/533.html
"""
try:
ua = UserAgent()
except FakeUserAgentError:
pass
# 生成随机的请求头，加 try...except... 使代码更加健壮
base_headers = {
'User-Agent': ua.random,
'Accept-Encoding': 'gzip, deflate, sdch',
'Accept-Language': 'zh-CN,zh;q=0.8'
}
# 如果使用者有传入请求头，则将此请求头和随机生成的合成在一起
headers = dict(base_headers, **options)
# 当前请求的 url
print('Getting', url)
try:
r = requests.get(url, headers=headers)
print('Getting result', url, r.status_code)
if r.status_code == 200:
return r.text
return None
except ConnectionError:
print('Crawling Failed', url)
return None

class Downloader(object):
"""
一个异步下载器，可以对代理源异步抓取，但是容易被 BAN。
self._htmls: 把请求的 html 放入列表当中
升级版的请求类
"""

def __init__(self, urls):
self.urls = urls
self._htmls = []

async def download_single_page(self, url):
"""
下载单页
"""
async with aiohttp.ClientSession() as session:
async with session.get(url) as response:
self._htmls.append(await response.text())

def download(self):
loop = asyncio.get_event_loop()
tasks = [self.download_single_page(url) for url in self.urls]
loop.run_until_complete(asyncio.wait(tasks))

@property
def htmls(self):
self.download()
return self._htmls

上面请求函数编写完成之后，我们就需要在抓取代理的代码文件中，进行导包，代码如下：

1
2

# files:Spider.py
from utils import get_page

4.2.2 代理抓取代码

导入所需要的库：

1
2
3
4

import requests
from utils import get_page
from pyquery import PyQuery as pq
import re

接下来我们需要创建一个类 FreeProxyGetter 并使用元类创建。

1
2

class FreeProxyGetter(object, metaclass=ProxyMetaclass):
pass

1. 快代理

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

def crawl_kuaidaili(self):
"""
快代理
:return:
"""
for page in range(1, 4):
# 国内高匿代理
start_url = 'https://www.kuaidaili.com/free/inha/{}/'.format(page)
html = get_page(start_url)
# print(html)
pattern = re.compile(
'<td data-title="IP">(.*)</td>s*<td data-title="PORT">(w+)</td>'
)
# s * 匹配空格，起到换行作用
# ip_addres = re.findall(pattern, html) # 写法一
ip_addres = pattern.findall(str(html))
for adress, port in ip_addres:
# print(adress, port)
result = f"{adress}:{port}".strip()
yield result

2. 西刺代理

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

def crawl_xicidaili(self):
"""
西刺代理
:return:
"""
for page in range(1, 4):
start_url = 'https://www.xicidaili.com/wt/{}'.format(page)
html = get_page(start_url)
# print(html)
ip_adress = re.compile(
'<td class="country"><img src="//fs.xicidaili.com/images/flag/cn.png" alt="Cn" /></td>s*<td>(.*?)</td>s*<td>(.*?)</td>'
)
# s* 匹配空格，起到换行作用
re_ip_adress = ip_adress.findall(str(html))
for adress, port in re_ip_adress:
result = f"{adress}:{port}".strip()
yield result

3. 66代理

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

def crawl_daili66(self, page_count=4):
"""
66代理
:param page_count:
:return:
"""
start_url = 'http://www.66ip.cn/{}.html'
urls = [start_url.format(page) for page in range(1, page_count + 1)]
for url in urls:
print('Crawling', url)
html = get_page(url)
if html:
doc = pq(html)
trs = doc('.containerbox table tr:gt(0)').items()
for tr in trs:
ip = tr.find('td:nth-child(1)').text()
port = tr.find('td:nth-child(2)').text()
yield ':'.join([ip, port])

4. 无忧代理

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

def crawl_data5u(self):
"""
无忧代理
:return:
"""
start_url = 'http://www.data5u.com'
html = get_page(start_url)
# print(html)
ip_adress = re.compile(
'<ul class="l2">s*<span><li>(.*?)</li></span>s*<span style="width: 100px;"><li class=".*">(.*?)</li></span>'
)
# s * 匹配空格，起到换行作用
re_ip_adress = ip_adress.findall(str(html))
for adress, port in re_ip_adress:
result = f"{adress}:{port}"
yield result.strip()

5. 开心代理-高匿

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

def crawl_kxdaili(self):
"""
开心代理-高匿
:return:
"""
for i in range(1, 4):
start_url = 'http://www.kxdaili.com/dailiip/1/{}.html'.format(i)
try:
html = requests.get(start_url)
if html.status_code == 200:
html.encoding = 'utf-8'
# print(html.text)
ip_adress = re.compile('<tr.*?>s*<td>(.*?)</td>s*<td>(.*?)</td>')
# s* 匹配空格，起到换行作用
re_ip_adress = ip_adress.findall(str(html.text))
for adress, port in re_ip_adress:
result = f"{adress}:{port}"
yield result.strip()
return None
except:
pass

6. 云代理

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

def IP3366Crawler(self):
"""
云代理
parse html file to get proxies
:return:
"""
start_url = 'http://www.ip3366.net/free/?stype=1&page={page}'
urls = [start_url.format(page=i) for i in range(1, 8)]
# s * 匹配空格，起到换行作用
ip_address = re.compile('<tr>s*<td>(.*?)</td>s*<td>(.*?)</td>')
for url in urls:
html = get_page(url)
re_ip_address = ip_address.findall(str(html))
for adress, port in re_ip_address:
result = f"{adress}:{port}"
yield result.strip()

至此，代理网站的代码已经全部编写完成，接下来我们需要了解的是然后调用此类，直接调用吗？ 显然不是，我们还需要编写一个运行调用的函数，代码如下：

1
2
3
4
5
6
7
8
9
10

def run(self):
# print(self.__CrawlFunc__)
proxies = []
callback = self.CrawlFunc_List
for i in callback:
print('Callback', i)
for proxy in eval("self.{}()".format(i)):
print('Getting', proxy, 'from', i)
proxies.append(proxy)
return proxies

以上代理的代码中，我并没有抓取每个网站全部页数，如果有需要可以自行调节。 这里我们可以直接写一个调用代码，来测试代码是否正常，写在：Spider.py 也就是代理的代码中，代码如下：

1
2
3

if __name__ == '__main__':
Tester = FreeProxyGetter()
Tester.run()

运行结果如下： 结果较多，省略大部分结果。

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

Getting 14.115.70.177:4216 from crawl_xicidaili
Getting 116.22.48.220:4216 from crawl_xicidaili
Getting 223.241.3.120:4216 from crawl_xicidaili
......
Getting 60.188.1.27:8232 from crawl_data5u
Callback crawl_kxdaili
Getting 117.71.165.208:3000 from crawl_kxdaili
Getting 223.99.197.253:63000 from crawl_kxdaili
Getting 60.255.186.169:8888 from crawl_kxdaili
Getting 180.168.13.26:8000 from crawl_kxdaili
Getting 125.124.51.226:808 from crawl_kxdaili
Getting 47.99.145.67:808 from crawl_kxdaili
Getting 222.186.55.41:8080 from crawl_kxdaili
Getting 36.6.224.30:3000 from crawl_kxdaili
Getting 110.43.42.200:8081 from crawl_kxdaili
Getting 39.137.69.8:80 from crawl_kxdaili

ok，至此代理成功抓取，接下来就是我们的异步检测了。

4.2.3 异步检测

这里，我们需要编写一个 error.py 来自定义包：

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

class ResourceDepletionError(Exception):

def __init__(self):
Exception.__init__(self)

def __str__(self):
return repr('The proxy source is exhausted')

class PoolEmptyError(Exception):

def __init__(self):
Exception.__init__(self)

def __str__(self):
return repr('The proxy pool is empty')

异步检测代码，这里我就不做任何数据存储，有需求的可以自行添加，下一篇将添加一个存入 redis 数据库的，敬请关注！

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
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79

# files:schedule.py
import asyncio
import time
from multiprocessing.context import Process

import aiohttp
from db import RedisClient
from Spider import FreeProxyGetter
from error import ResourceDepletionError
from aiohttp import ServerDisconnectedError, ClientResponseError, ClientConnectorError
from socks import ProxyConnectionError

get_proxy_timeout = 9

class ValidityTester(object):
"""
代理的有效性测试
"""
def __init__(self):
self.raw_proxies = None
self.usable_proxies = [] # 可用代理
self.test_api = 'http://www.baidu.com'

def set_raw_proxies(self, proxies):
self.raw_proxies = proxies # 临时存储一些代理

async def test_single_proxy(self, proxy):
"""
python3.5 之后出现的新特性
text one proxy, if valid, put them to usable_proxies.
"""
try:
async with aiohttp.ClientSession() as session:
try:
if isinstance(proxy, bytes):
proxy = proxy.decode('utf-8')
real_proxy = 'http://' + proxy
print('Testing', proxy)
async with session.get(self.test_api, proxy=real_proxy, timeout=get_proxy_timeout) as response:
if response.status == 200:
print('Valid proxy', proxy)
except (ProxyConnectionError, TimeoutError, ValueError):
print('Invalid proxy', proxy)
except (ServerDisconnectedError, ClientResponseError, ClientConnectorError) as s:
print(s)
pass

def test(self):
"""
aio test all proxies.
"""
print('ValidityTester is working')
try:
loop = asyncio.get_event_loop()
tasks = [self.test_single_proxy(proxy) for proxy in self.raw_proxies]
loop.run_until_complete(asyncio.wait(tasks))
except ValueError:
print('Async Error')

class Schedule(object):
@staticmethod
def valid_proxy():
"""
Get half of proxies which in Spider
"""
tester = ValidityTester()
free_proxy_getter = FreeProxyGetter()
# 获取 Spider 数据库数据
# 调用测试类
tester.set_raw_proxies(free_proxy_getter.run())
# 开始测试
tester.test()

def run(self):
print('Ip processing running')
valid_process = Process(target=Schedule.valid_proxy) # 获取代理并筛选
valid_process.start()
if __name__ == '__main__':
schedule = Schedule().run()

对于上面的部分代理代码，我为了输出结果更加直观，把 print() 输出的，包括 utils.py 的 print() 全部注释了， 运行结果（省略部分输出）

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
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66

Testing 182.46.252.84:9999
Testing 183.63.188.250:8808
Testing 39.137.69.10:8080
Testing 119.190.149.226:8060
Testing 219.159.38.201:56210
Testing 58.246.143.32:8118
Testing 114.99.117.13:3000
Testing 114.229.229.248:8118
Testing 119.57.108.53:53281
Testing 223.247.94.164:4216
Testing 219.159.38.208:56210
Testing 59.44.78.30:42335
Testing 112.74.87.172:3128
Testing 182.92.235.109:3128
Testing 58.62.115.3:4216
Testing 183.162.171.37:4216
Testing 223.241.5.46:4216
Testing 223.243.4.231:4216
Testing 223.241.5.228:4216
Testing 183.162.171.56:4216
Testing 223.241.4.146:4216
Testing 223.241.4.200:4216
Testing 223.241.4.182:4216
Testing 118.187.50.114:8080
Testing 223.214.176.170:3000
Testing 125.126.123.208:60004
Testing 163.125.248.39:8088
Testing 119.57.108.89:53281
Testing 222.249.238.138:8080
Testing 125.126.97.77:60004
Testing 222.182.54.210:8118
Valid proxy 58.220.95.90:9401
Valid proxy 60.191.45.92:8080
Valid proxy 222.186.55.41:8080
Valid proxy 60.205.132.71:80
Valid proxy 124.90.49.146:8888

Valid proxy 218.75.109.86:3128
Valid proxy 27.188.64.70:8060

Valid proxy 27.188.62.3:8060
Valid proxy 221.180.170.104:8080
Valid proxy 39.137.69.9:8080

Valid proxy 39.137.69.6:80
Valid proxy 39.137.69.10:8080
Valid proxy 114.229.6.215:8118
Valid proxy 221.180.170.104:8080
Valid proxy 218.89.14.142:8060
Valid proxy 116.196.85.150:3128

Valid proxy 122.224.65.197:3128
Valid proxy 112.253.11.103:8000
Valid proxy 112.253.11.103:8000
Valid proxy 112.253.11.113:8000
Valid proxy 124.90.52.200:8888

Cannot connect to host 113.103.226.99:4216 ssl:default [Connect call failed ('113.103.226.99', 4216)]
Cannot connect to host 36.249.109.59:8221 ssl:default [Connect call failed ('36.249.109.59', 8221)]
Cannot connect to host 125.126.121.66:60004 ssl:default [Connect call failed ('125.126.121.66', 60004)]
Cannot connect to host 27.42.168.46:48919 ssl:default [Connect call failed ('27.42.168.46', 48919)]
Cannot connect to host 163.125.113.220:8118 ssl:default [Connect call failed ('163.125.113.220', 8118)]
Cannot connect to host 114.55.63.34:808 ssl:default [Connect call failed ('114.55.63.34', 808)]
Cannot connect to host 119.133.207.10:4216 ssl:default [Connect call failed ('119.133.207.10', 4216)]
Cannot connect to host 58.62.115.3:4216 ssl:default [Connect call failed ('58.62.115.3', 4216)]
Cannot connect to host 117.71.164.232:3000 ssl:default [Connect call failed ('117.71.164.232', 3000)]

到此，多站抓取异步检测，就已经完全实现，也是实现了模块化编程。我也 顺便把项目结构分享一下：

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

C:.
│ error.py
│ info.txt
│ schedule.py
│ Spider.py
│ the_test_code_crawler.py
│ utils.py
│
└─__pycache__
db.cpython-37.pyc
error.cpython-37.pyc
Spider.cpython-37.pyc
utils.cpython-37.pyc

本篇我想你主要学习到的目标也就是开头所说的，这里就不赘述了，下一篇我将带大家把抓取到可用的代理进行存储到 redis 数据库中，尽请关注！ 源代码文件获取：公众号后台回放：ProxyPool-1 也欢迎加入我的交流群，一起交流！

之前开发了一个工具包 GerapyPyppeteer，GitHub 地址为 https://github.com/Gerapy/GerapyPyppeteer，这个包实现了 Scrapy 和 Pyppeteer 的对接，利用它我们就可以方便地实现 Scrapy 使用 Pyppeteer 爬取动态渲染的页面了。 另外，很多朋友在运行爬虫的时候可能会使用到 Docker，想把 Scrapy 和 Pyppeteer 打包成 Docker 运行，但是这个打包和测试过程中大家可能会遇到一些问题，在这里对 Pyppeteer 打包 Docker 的坑简单做一下总结。

概述

Pyppeteer 打包 Docker 主要是有这么几个坑点：

• 依赖没有安装，导致无法正确安装和启动 Pyppeteer。
• 没有关闭沙盒模式，导致可能出现 Browser closed unexpectedly 错误
• 没有提前安装好 Pyppeteer，导致每次启动时都要重新安装

下面我们分别对三个问题做下简单的 Troubleshooting。

安装依赖

首先说第一个，安装依赖。 因为 Docker 大部分都是基于 Linux 系统的，比如我常用的基础镜像就是 python:3.7，剩余 Debian 系列，当然还有很多其他的版本，具体可以查看 https://hub.docker.com/_/python 了解下。 但是对于 Pyppeteer 来说，python:3.7 内置的依赖库并不够，我们还需要额外进行安装，安装完毕之后还需要清空下 apt list，一句 Dockerfile 命令如下：

1
2
3
4
5
6
7

RUN apt-get update && 
  apt-get -y install libnss3 xvfb gconf-service libasound2 libatk1.0-0 libc6 libcairo2 libcups2 
  libdbus-1-3 libexpat1 libfontconfig1 libgbm1 libgcc1 libgconf-2-4 libgdk-pixbuf2.0-0 libglib2.0-0 
  libgtk-3-0 libnspr4 libpango-1.0-0 libpangocairo-1.0-0 libstdc++6 libx11-6 libx11-xcb1 libxcb1 
  libxcomposite1 libxcursor1 libxdamage1 libxext6 libxfixes3 libxi6 libxrandr2 libxrender1 libxss1 
  libxtst6 ca-certificates fonts-liberation libappindicator1 libnss3 lsb-release xdg-utils wget && 
  rm -rf /var/lib/apt/lists/*

关闭沙盒模式

在 Docker 中如果直接启动 Pyppeteer，我们还需要关闭沙盒模式，否则可能会遇到如下错误：

1
2

pyppeteer.errors.BrowserError: Browser closed unexpectedly:
[0924/153706.301300:ERROR:zygote_host_impl_linux.cc(89)] Running as root without --no-sandbox is not supported. See https://crbug.com/638180

这里提示我们要关闭沙盒模式，这里只需要在启动 Pyppeteer 的时候，给 launch 方法的 args 参数多加一个 \--no-sandbox 即可，写法如下：

1

browser = await pyppeteer.launch(options={'args': ['--no-sandbox']})

这样就不会再遇到上面的错误了。

提前安装

另外建议在打包 Docker 的时候就提前把 Pyppeteer 提前安装好，可以单独使用一句 RUN 命令安装即可。

1

RUN pip install -U pip && pip install pyppeteer && pyppeteer-install

这里是提前安装了一下 Pyppteer 这个 Python 库，然后利用 Python 库提供的 pyppeteer-install 命令提前下载了 Chromium 浏览器。 这样后面启动的时候就可以直接唤起 Chromium 浏览器进行爬取了。

总结

最后看下完整 Dockerfile，内容如下：

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

FROM python:3.7

RUN apt-get update && 
  apt-get -y install libnss3 xvfb gconf-service libasound2 libatk1.0-0 libc6 libcairo2 libcups2 
  libdbus-1-3 libexpat1 libfontconfig1 libgbm1 libgcc1 libgconf-2-4 libgdk-pixbuf2.0-0 libglib2.0-0 
  libgtk-3-0 libnspr4 libpango-1.0-0 libpangocairo-1.0-0 libstdc++6 libx11-6 libx11-xcb1 libxcb1 
  libxcomposite1 libxcursor1 libxdamage1 libxext6 libxfixes3 libxi6 libxrandr2 libxrender1 libxss1 
  libxtst6 ca-certificates fonts-liberation libappindicator1 libnss3 lsb-release xdg-utils wget && 
  rm -rf /var/lib/apt/lists/*

RUN pip install -U pip && pip install pyppeteer && pyppeteer-install

WORKDIR /code
COPY requirements.txt .
RUN pip install -r requirements.txt
COPY . .
CMD python3 run.py

这里首先就是安装了必须的依赖库，然后安装了 Pyppeteer 并下载了 Chromium 浏览器，最后拷贝项目运行即可。 当然最后的一句 CMD 大家可以随意指定入口。 最后大家可以体验一个实例来感受下 Scrapy 和 Pyppeteer 对接后在 Docker 中的运行效果：

1

docker run germey/gerapy-pyppeteer-example

如果大家对 Scrapy 和 Pyppeteer 感兴趣也可以看下我写的这个库 GerapyPyppeteer，GitHub 地址为 https://github.com/Gerapy/GerapyPyppeteer，感谢支持！

解析页面是做爬虫的过程中的重要环节，而且如果站点多了，解析也会变得非常复杂，所以智能化解析就可能是一个不错的解决方案。如果我们能够容忍一定的错误率，那么我们可以利用智能化解析算法帮我们提取一些内容，简单高效。 那有没有办法做到一个网站的全自动化解析呢？ 比如来了一个博客网站，我能首先识别出来这是一个列表页还是文章（详情）页，然后提取列表页的每篇文章的链接，然后跳转到每篇文章（详情）页再提取文章相关信息。 那么这里面可能就有四个关键部分：

• 判断当前所在的页面是列表页还是文章（详情）页
• 识别出列表页下一页的链接
• 识别出列表页所有列表链接
• 识别出文章（详情）页的文章内容和其他信息

如果我们能把这四步都用算法实现出来，那么我们只需要一个网站的主站链接就能轻松地把内容规整地爬取下来了。 那么这篇文章我们就来简单说下第一步，如何判断当前所在的页面的列表页还是文章（详情）页。

注：后文中文章页统一称之为详情页。

示例

列表页和详情页不知道大家有没有基本的概念了，列表页就是导航页，里面带有好多文章或新闻或详情链接，我们选一个链接点进去就是详情页。 比如说这里拿新浪体育来说，首页如图所示： 看到这里面有很多链接，就是一些页面导航集合，这个页面就是列表页。 然后我们随便点开一篇新闻报道，如图所示： 这里就是一篇新闻报道，带有醒目的标题、发布时间、正文等内容，这就是详情页。 现在我们要做的就是用一个算法来凭借 HTML 代码区分出来哪个是列表页，哪个是详情页。 最后的输入输出如下：

• 输入：一个页面的 HTML 代码
• 输出：这个页面是列表页还是详情页，并输出二者的判定概率。

模型选用

首先我们确认下这个问题是个什么问题。 很明显，结果要么是列表页，要么是详情页，就是个二分类问题。 那二分类问题怎么解决呢？实现一个基本的分类模型就好了。大范围就是传统机器学习和现在比较流行的深度学习。总体上来说，深度学习的精度和处理能力会强一点，但是想想我们现在的应用场景，后者要追求精度的话可能需要更多的标注数据，而前者也有比较不错的易用的模型了，比如 SVM。 所以，我们不妨先选用 SVM 模型来实现一个基本的二分类模型来试试看，效果如果已经很好了或者提升空间不大了，那就直接用就好了，如果效果比较差，那我们再选用其他模型来优化。 好，那就定下来了，我们用 SVM 模型来实现一下试试。

数据标注

既然要做分类模型，那么最重要的当然就是数据标注了，我们分两组就好了，一组是列表页，一组是详情页，我们先用手工配合爬虫找一些列表页和详情页的 HTML 代码，然后将其保存下来。 结果类似如下： 每个文件夹几百个就行了，数量不用太多，五花八门的页面混起来更好。

特征提取

既然要做 SVM，那么我们得想清楚要分清两个类别需要哪些特征。既然是特征，那我们就要选出二者不同的特征，这样更加有区分度。 比如这里我大体总结了有这么几个特征：

• 文本密度：正文页通常会包含密集的文字，比如一个 p 节点内部就包含几十上百个文字，如果用单个节点内的文字数目来表示文本密度的话，那么详情页的部分内容文本密度会很高。
• 超链接节点的数量和比例：一般来说列表页通常会包含多个超链接，而且很大比例都是超链接文本，而详情页却有很多的文字并不是超链接，比如正文。
• 符号密度：一般来说列表页通常会是一些标题导航，一般可能都不会包含句号，而详情页的正文内容通常就会包含句号等内容，如果按照单位文字所包含的标点符号数量来表示符号密度的话，后者的符号密度也会高一些。
• 列表簇的数目：一般来说，列表页通常会包含多个具有公共父节点的条目，多个条目构成一个列表簇，虽然说详情页侧栏也会包含一些列表，但至少这个数量也可以成为一个特征来判别。
• meta 信息：有一些特殊的 meta 信息是列表页独有的，比如只有详情页才会有发布时间，而列表页通常是没有的。
• 正文标题和 title 标题相似度：一般来说，详情页的正文标题和 title 标题很可能是相同的内容，而列表页通常则是站点的名称。

以上便是我简单列的几个特征，还有很多其他的特征也可以来挖掘，比如视觉特征等等。

模型训练

真正代码实现的过程就是将现有的 HTML 文本进行预处理，把上面的一些特征提取出来，然后直接声明一个 SVM 分类模型即可。 这里声明了一个 feature 名字和对应的处理方法：

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

self.feature_funcs = {
    'number_of_a_char': number_of_a_char,
    'number_of_a_char_log10': self._number_of_a_char_log10,
    'number_of_char': number_of_char,
    'number_of_char_log10': self._number_of_char_log10,
    'rate_of_a_char': self._rate_of_a_char,
    'number_of_p_descendants': number_of_p_descendants,
    'number_of_a_descendants': number_of_a_descendants,
    'number_of_punctuation': number_of_punctuation,
    'density_of_punctuation': density_of_punctuation,
    'number_of_clusters': self._number_of_clusters,
    'density_of_text': density_of_text,
    'max_density_of_text': self._max_density_of_text,
    'max_number_of_p_children': self._max_number_of_p_children,
    'has_datetime_meta': self._has_datetime_mata,
    'similarity_of_title': self._similarity_of_title,
}
self.feature_names = self.feature_funcs.keys()

以上方法就是特征和对应的获取方法，具体根据实际情况实现即可。 然后关键的部分就是对数据的处理和模型的训练了，关键代码如下：

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
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46

list_file_paths = list(glob(f'{DATASETS_LIST_DIR}/*.html'))
detail_file_paths = list(glob(f'{DATASETS_DETAIL_DIR}/*.html'))

x_data, y_data = [], []

for index, list_file_path in enumerate(list_file_paths):
    logger.log('inspect', f'list_file_path {list_file_path}')
    element = file2element(list_file_path)
    if element is None:
        continue
    preprocess4list_classifier(element)
    x = self.features_to_list(self.features(element))
    x_data.append(x)
    y_data.append(1)

for index, detail_file_path in enumerate(detail_file_paths):
    logger.log('inspect', f'detail_file_path {detail_file_path}')
    element = file2element(detail_file_path)
    if element is None:
        continue
    preprocess4list_classifier(element)
    x = self.features_to_list(self.features(element))
    x_data.append(x)
    y_data.append(0)

# preprocess data
ss = StandardScaler()
x_data = ss.fit_transform(x_data)
joblib.dump(ss, self.scaler_path)
x_train, x_test, y_train, y_test = train_test_split(x_data, y_data, test_size=0.2, random_state=5)

# set up grid search
c_range = np.logspace(-5, 20, 5, base=2)
gamma_range = np.logspace(-9, 10, 5, base=2)
param_grid = [
    {'kernel': ['rbf'], 'C': c_range, 'gamma': gamma_range},
    {'kernel': ['linear'], 'C': c_range},
]
grid = GridSearchCV(SVC(probability=True), param_grid, cv=5, verbose=10, n_jobs=-1)
clf = grid.fit(x_train, y_train)
y_true, y_pred = y_test, clf.predict(x_test)
logger.log('inspect', f'n{classification_report(y_true, y_pred)}')
score = grid.score(x_test, y_test)
logger.log('inspect', f'test accuracy {score}')
# save model
joblib.dump(grid.best_estimator_, self.model_path)

这里首先对数据进行预处理，然后将每个 feature 存 存到 x_data 中，标注结果存到 y_data 中。接着我们使用 StandardScaler 对数据进行标准化处理，然后进行随机切分。最后使用 GridSearch 训练了一个 SVM 模型然后保存了下来。 以上便是基本的模型训练过程，具体的代码可以再完善一下。

使用

以上的流程我已经实现了，并且发布了一个开源 Python 包，名字叫做 Gerapy AutoExtractor，GitHub 地址为 https://github.com/Gerapy/GerapyAutoExtractor。 大家如需使用可以使用 pip 安装：

1

pip3 install gerapy-auto-extractor

这个库针对于以上算法提供了四个方法：

• is_detail：判断是否是详情页
• is_list：判断是否是列表页
• probability_of_detail：是详情页的概率，结果是 0-1
• probability_of_list：是列表页的概率，结果是 0-1

例如，我们随便可以找几个网址存下来，比如把上文的列表页和详情页的 HTML 代码存下来分别保存为 list.html 和 detail.html。 测试代码如下：

1
2
3
4
5
6
7
8
9
10

from gerapy_auto_extractor import is_detail, is_list, probability_of_detail, probability_of_list
from gerapy_auto_extractor.helpers import content, jsonify

html = content('detail.html')
print(probability_of_detail(html), probability_of_list(html))
print(is_detail(html), is_list(html))

html = content('list.html')
print(probability_of_detail(html), probability_of_list(html))
print(is_detail(html), is_list(html))

这里我们就调用了以上四个方法来实现了页面类型和置信度的判断。 类似的输出结果如下：

1
2
3
4

0.9990605314033392 0.0009394685966607814
True False
0.033477426883441685 0.9665225731165583
False True

这样我们就成功得到了两个页面的类别和置信度了。 以上便是判断列表页和详情页的原理和实现，如需了解更多请关注项目 Gerapy Auto Extractor，GitHub 链接为 https://github.com/Gerapy/GerapyAutoExtractor，多谢支持。

Airtest

首先需要安装 Airtest，使用 pip3 即可：

1

pip3 install airtest

初始化 device

如果设备没有被初始化的话会进行初始化，并把初始化的设备作为当前设备。 用法如下：

1
2
3
4
5
6
7
8
9

def init_device(platform="Android", uuid=None, **kwargs):
    """
    Initialize device if not yet, and set as current device.

    :param platform: Android, IOS or Windows
    :param uuid: uuid for target device, e.g. serialno for Android, handle for Windows, uuid for iOS
    :param kwargs: Optional platform specific keyword args, e.g. `cap_method=JAVACAP` for Android
    :return: device instance
    """

示例如下：

1
2

device = init_device('Android')
print(device)

运行结果如下：

1

<airtest.core.android.android.Android object at 0x1018f3a58>

可以发现它返回的是一个 Android 对象。 这个 Android 对象实际上属于 airtest.core.android 这个包，继承自 airtest.core.device.Device 这个类，与之并列的还有 airtest.core.ios.ios.IOS 、airtest.core.linux.linux.Linux、airtest.core.win.win.Windows 等。这些都有一些针对 Device 操作的 API，下面我们以 airtest.core.android.android.Android 为例来总结一下。

• get_default_device：获取默认 device
• uuid：获取当前 Device 的 UUID
• list_app：列举所有 App
• path_app：打印输出某个 App 的完整路径
• check_app：检查某个 App 是否在当前设备上
• start_app：启动某个 App
• start_app_timing：启动某个 App，然后计算时间
• stop_app：停止某个 App
• clear_app：清空某个 App 的全部数据
• install_app：安装某个 App
• install_multiple_app：安装多个 App
• uninstall_app：卸载某个 App
• snapshot：屏幕截图
• shell：获取 Adb Shell 执行的结果
• keyevent：执行键盘操作
• wake：唤醒当前设备
• home：点击 HOME 键
• text：向设备输入内容
• touch：点击屏幕某处的位置
• double_click：双击屏幕某处的位置
• swipe：滑动屏幕，由一点到另外一点
• pinch：手指捏和操作
• logcat：日志记录操作
• getprop：获取某个特定属性的值
• get_ip_address：获取 IP 地址
• get_top_activity：获取当前 Activity
• get_top_activity_name_and_pid：获取当前 Activity 的名称和进程号
• get_top_activity_name：获取当前 Activity 的名称
• is_keyboard_shown：判断当前键盘是否出现了
• is_locked：设备是否锁定了
• unlock：解锁设备
• display_info：获取当前显示信息，如屏幕宽高等
• get_display_info：同 display_info
• get_current_resolution：获取当前设备分辨率
• get_render_resolution：获取当前渲染分辨率
• start_recording：开始录制
• stop_recording：结束录制
• adjust_all_screen：调整屏幕适配分辨率

了解了上面的方法之后，我们可以用一个实例来感受下它们的用法：

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
29
30
31
32
33
34
35

from airtest.core.android import Android
from airtest.core.api import *
import logging

logging.getLogger("airtest").setLevel(logging.WARNING)

device: Android = init_device('Android')
is_locked = device.is_locked()
print(f'is_locked: {is_locked}')

if is_locked:
    device.unlock()

device.wake()

app_list = device.list_app()
print(f'app list {app_list}')

uuid = device.uuid
print(f'uuid {uuid}')

display_info = device.get_display_info()
print(f'display info {display_info}')

resolution = device.get_render_resolution()
print(f'resolution {resolution}')

ip_address = device.get_ip_address()
print(f'ip address {ip_address}')

top_activity = device.get_top_activity()
print(f'top activity {top_activity}')

is_keyboard_shown = device.is_keyboard_shown()
print(f'is keyboard shown {is_keyboard_shown}')

这里我们调用了设备的一些操作方法，获取了一些基本状态，运行结果如下：

1
2
3
4
5
6
7
8

is_locked: False
app list ['com.kimcy929.screenrecorder', 'com.android.providers.telephony', 'io.appium.settings', 'com.android.providers.calendar', 'com.android.providers.media', 'com.goldze.mvvmhabit', 'com.android.wallpapercropper', 'com.android.documentsui', 'com.android.galaxy4', 'com.android.externalstorage', 'com.android.htmlviewer', 'com.android.quicksearchbox', 'com.android.mms.service', 'com.android.providers.downloads', 'mark.qrcode', ..., 'com.google.android.play.games', 'io.kkzs', 'tv.danmaku.bili', 'com.android.captiveportallogin']
uuid emulator-5554
display info {'id': 0, 'width': 1080, 'height': 1920, 'xdpi': 320.0, 'ydpi': 320.0, 'size': 6.88, 'density': 2.0, 'fps': 60.0, 'secure': True, 'rotation': 0, 'orientation': 0.0, 'physical_width': 1080, 'physical_height': 1920}
resolution (0.0, 0.0, 1080.0, 1920.0)
ip address 10.0.2.15
top activity ('com.microsoft.launcher.dev', 'com.microsoft.launcher.Launcher', '16040')
is keyboard shown False

连接 device

连接 device 需要传入设备的 uri，格式类似 android://adbhost:adbport/serialno?param=value&param2=value2，使用方法如下：

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

def connect_device(uri):
    """
    Initialize device with uri, and set as current device.

    :param uri: an URI where to connect to device, e.g. `android://adbhost:adbport/serialno?param=value&param2=value2`
    :return: device instance
    :Example:
        * ``android:///`` # local adb device using default params
        * ``android://adbhost:adbport/1234566?cap_method=javacap&touch_method=adb``  # remote device using custom params
        * ``windows:///`` # local Windows application
        * ``ios:///`` # iOS device
    """

示例如下：

1
2
3
4
5
6

from airtest.core.android import Android
from airtest.core.api import *

uri = 'Android://127.0.0.1:5037/emulator-5554'
device: Android = connect_device(uri)
print(device)

运行结果如下：

1

<airtest.core.android.android.Android object at 0x110246940>

其实返回结果和 init_device 是一样的，最后 connect_device 方法就是调用了 init_device 方法。

获取当前 device

就是直接调用 device 方法，定义如下：

1
2
3
4
5
6
7

def device():
    """
    Return the current active device.

    :return: current device instance
    """
    return G.DEVICE

获取所有 device

在 airtest 中有一个全局变量 G，获取所有 device 的方法如下：

1
2
3
4
5
6
7

from airtest.core.android import Android
from airtest.core.api import *

print(G.DEVICE_LIST)
uri = 'Android://127.0.0.1:5037/emulator-5554'
device: Android = connect_device(uri)
print(G.DEVICE_LIST)

运行结果如下：

1
2

[]
[<airtest.core.android.android.Android object at 0x10ba03978>]

这里需要注意的是，在最开始没有调用 connect_device 方法之前，DEVICE_LIST 是空的，在调用之后 DEVICE_LIST 会自动添加已经连接的 device，DEVICE_LIST 就是已经连接的 device 列表

切换 device

我们可以使用 set_current 方法切换当前连接的 device，传入的是 index，定义如下：

1
2
3
4
5
6
7
8
9

def set_current(idx):
    """
    Set current active device.

    :param idx: uuid or index of initialized device instance
    :raise IndexError: raised when device idx is not found
    :return: None
    :platforms: Android, iOS, Windows
    """

这个方法没有返回值，调用 set_current 方法切换 device 之后，再调用 device 方法就可以获取当前 device 对象了。

执行命令行

可以使用 shell 方法传入 cmd 来执行命令行，定义如下：

1
2
3
4
5
6
7
8
9
10

@logwrap
def shell(cmd):
    """
    Start remote shell in the target device and execute the command

    :param cmd: command to be run on device, e.g. "ls /data/local/tmp"
    :return: the output of the shell cmd
    :platforms: Android
    """
    return G.DEVICE.shell(cmd)

直接调用 adb 命令就好了，例如获取内存信息就可以使用如下命令：

1
2
3
4
5
6
7

from airtest.core.api import *

uri = 'Android://127.0.0.1:5037/emulator-5554'
connect_device(uri)

result = shell('cat /proc/meminfo')
print(result)

运行结果如下：

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
29
30
31
32
33
34
35
36
37
38
39
40

MemTotal:        3627908 kB
MemFree:         2655560 kB
MemAvailable:    2725928 kB
Buffers:            3496 kB
Cached:           147472 kB
SwapCached:            0 kB
Active:           744592 kB
Inactive:         126332 kB
Active(anon):     723292 kB
Inactive(anon):    16344 kB
Active(file):      21300 kB
Inactive(file):   109988 kB
Unevictable:           0 kB
Mlocked:               0 kB
HighTotal:       2760648 kB
HighFree:        2073440 kB
LowTotal:         867260 kB
LowFree:          582120 kB
SwapTotal:             0 kB
SwapFree:              0 kB
Dirty:                 0 kB
Writeback:             0 kB
AnonPages:        720100 kB
Mapped:           127720 kB
Shmem:             19428 kB
Slab:              76196 kB
SReclaimable:       7392 kB
SUnreclaim:        68804 kB
KernelStack:        7896 kB
PageTables:         8544 kB
NFS_Unstable:          0 kB
Bounce:                0 kB
WritebackTmp:          0 kB
CommitLimit:     1813952 kB
Committed_AS:   21521776 kB
VmallocTotal:     122880 kB
VmallocUsed:       38876 kB
VmallocChunk:      15068 kB
DirectMap4k:       16376 kB
DirectMap4M:      892928 kB

启动和停止 App

启动和停止 App 直接传入包名即可，其实它们就是调用的 device 的 start_app 和 stop_app 方法，定义如下：

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

@logwrap
def start_app(package, activity=None):
    """
    Start the target application on device

    :param package: name of the package to be started, e.g. "com.netease.my"
    :param activity: the activity to start, default is None which means the main activity
    :return: None
    :platforms: Android, iOS
    """
    G.DEVICE.start_app(package, activity)

@logwrap
def stop_app(package):
    """
    Stop the target application on device

    :param package: name of the package to stop, see also `start_app`
    :return: None
    :platforms: Android, iOS
    """
    G.DEVICE.stop_app(package)

用法示例如下：

1
2
3
4
5
6
7
8
9

from airtest.core.api import *

uri = 'Android://127.0.0.1:5037/emulator-5554'
connect_device(uri)

package = 'com.tencent.mm'
start_app(package)
sleep(10)
stop_app(package)

这里我指定了微信的包名，然后调用 start_app 启动了微信，然后等待了 10 秒，然后调用了 stop_app 停止了微信。

安装和卸载

安装和卸载也是一样，也是调用了 device 的 install 和 uninstall 方法，定义如下：

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

@logwrap
def install(filepath, **kwargs):
    """
    Install application on device

    :param filepath: the path to file to be installed on target device
    :param kwargs: platform specific `kwargs`, please refer to corresponding docs
    :return: None
    :platforms: Android
    """
    return G.DEVICE.install_app(filepath, **kwargs)

@logwrap
def uninstall(package):
    """
    Uninstall application on device

    :param package: name of the package, see also `start_app`
    :return: None
    :platforms: Android
    """
    return G.DEVICE.uninstall_app(package)

截图

截图使用 snapshot 即可完成，可以设定存储的文件名称，图片质量等。 定义如下：

1
2
3
4
5
6
7
8
9
10
11

def snapshot(filename=None, msg="", quality=ST.SNAPSHOT_QUALITY):
    """
    Take the screenshot of the target device and save it to the file.

    :param filename: name of the file where to save the screenshot. If the relative path is provided, the default
                     location is ``ST.LOG_DIR``
    :param msg: short description for screenshot, it will be recorded in the report
    :param quality: The image quality, integer in range [1, 99]
    :return: absolute path of the screenshot
    :platforms: Android, iOS, Windows
    """

示例如下：

1
2
3
4
5
6
7
8
9

from airtest.core.api import *

uri = 'Android://127.0.0.1:5037/emulator-5554'
connect_device(uri)

package = 'com.tencent.mm'
start_app(package)
sleep(3)
snapshot('weixin.png', quality=30)

运行之后在当前目录会生成一个 weixin.png 的图片，如图所示：

唤醒和首页

唤醒和回到首页分别也是调用了 device 的 wake 和 home 方法，定义如下：

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

@logwrap
def wake():
    """
    Wake up and unlock the target device

    :return: None
    :platforms: Android

    .. note:: Might not work on some models
    """
    G.DEVICE.wake()

@logwrap
def home():
    """
    Return to the home screen of the target device.

    :return: None
    :platforms: Android, iOS
    """
    G.DEVICE.home()

直接调用即可。

点击屏幕

点击屏幕是 touch 方法，可以传入一张图或者绝对位置，同时可以指定点击次数，定义如下：

1
2
3
4
5
6
7
8
9
10
11

@logwrap
def touch(v, times=1, **kwargs):
    """
    Perform the touch action on the device screen

    :param v: target to touch, either a Template instance or absolute coordinates (x, y)
    :param times: how many touches to be performed
    :param kwargs: platform specific `kwargs`, please refer to corresponding docs
    :return: finial position to be clicked
    :platforms: Android, Windows, iOS
    """

例如我现在的手机屏幕是这样子： 这里我截图下来一张图片，如图所示： 然后我们把这个图片声明成一个 Template 传入，示例如下：

1
2
3
4
5

from airtest.core.api import *

uri = 'Android://127.0.0.1:5037/emulator-5554'
connect_device(uri)
touch(Template('tpl.png'))

启动之后它就会识别出这张图片的位置，然后点击。 或者我们可以指定点击的绝对位置，示例如下：

1
2
3
4
5
6

from airtest.core.api import *

uri = 'Android://127.0.0.1:5037/emulator-5554'
connect_device(uri)
home()
touch((400, 600), times=2)

另外上述的 touch 方法还可以完全等同于 click 方法。 如果要双击的话，还可以使用调用 double_click 方法，传入参数也可以是 Template 或者绝对位置。

滑动

滑动可以使用 swipe 方法，可以传入起始和终止位置，两个位置都可以传入绝对位置或者 Template，定义如下：

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

@logwrap
def swipe(v1, v2=None, vector=None, **kwargs):
    """
    Perform the swipe action on the device screen.

    There are two ways of assigning the parameters
        * ``swipe(v1, v2=Template(...))``   # swipe from v1 to v2
        * ``swipe(v1, vector=(x, y))``      # swipe starts at v1 and moves along the vector.

    :param v1: the start point of swipe,
               either a Template instance or absolute coordinates (x, y)
    :param v2: the end point of swipe,
               either a Template instance or absolute coordinates (x, y)
    :param vector: a vector coordinates of swipe action, either absolute coordinates (x, y) or percentage of
                   screen e.g.(0.5, 0.5)
    :param **kwargs: platform specific `kwargs`, please refer to corresponding docs
    :raise Exception: general exception when not enough parameters to perform swap action have been provided
    :return: Origin position and target position
    :platforms: Android, Windows, iOS
    """

比如这里我们可以定义手指向右滑动：

1
2
3
4
5
6

from airtest.core.api import *

uri = 'Android://127.0.0.1:5037/emulator-5554'
connect_device(uri)
home()
swipe((200, 300), (900, 300))

放大缩小

放大缩小是使用的 pinch 方法，可以指定放大还是缩小，同时还可以指定中心位置点和放大缩小的比率。 定义如下：

1
2
3
4
5
6
7
8
9
10
11

@logwrap
def pinch(in_or_out='in', center=None, percent=0.5):
    """
    Perform the pinch action on the device screen

    :param in_or_out: pinch in or pinch out, enum in ["in", "out"]
    :param center: center of pinch action, default as None which is the center of the screen
    :param percent: percentage of the screen of pinch action, default is 0.5
    :return: None
    :platforms: Android
    """

示例如下：

1
2
3
4
5
6

from airtest.core.api import *

uri = 'Android://127.0.0.1:5037/emulator-5554'
connect_device(uri)
home()
pinch(in_or_out='out', center=(300, 300), percent=0.4)

键盘事件

可以使用 keyevent 来输入某个键，例如 home、back 等等，keyevent 的定义如下：

1
2
3
4
5
6
7
8
9

def keyevent(keyname, **kwargs):
    """
    Perform key event on the device

    :param keyname: platform specific key name
    :param **kwargs: platform specific `kwargs`, please refer to corresponding docs
    :return: None
    :platforms: Android, Windows, iOS
    """

示例如下：

1

keyevent("HOME")

这样就表示按了 HOME 键。

输入内容

输入内容需要使用 text 方法，当然前提是这个 widget 需要是 active 状态，text 的定义如下：

1
2
3
4
5
6
7
8
9
10

@logwrap
def text(text, enter=True, **kwargs):
    """
    Input text on the target device. Text input widget must be active first.

    :param text: text to input, unicode is supported
    :param enter: input `Enter` keyevent after text input, default is True
    :return: None
    :platforms: Android, Windows, iOS
    """

等待和判断

可以使用 wait 方法等待某个内容加载出来，需要传入的是 Template，定义如下：

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

@logwrap
def wait(v, timeout=None, interval=0.5, intervalfunc=None):
    """
    Wait to match the Template on the device screen

    :param v: target object to wait for, Template instance
    :param timeout: time interval to wait for the match, default is None which is ``ST.FIND_TIMEOUT``
    :param interval: time interval in seconds to attempt to find a match
    :param intervalfunc: called after each unsuccessful attempt to find the corresponding match
    :raise TargetNotFoundError: raised if target is not found after the time limit expired
    :return: coordinates of the matched target
    :platforms: Android, Windows, iOS
    """

同时也使用 exists 方法判断某个内容是否存在，定义如下：

1
2
3
4
5
6
7
8
9

@logwrap
def exists(v):
    """
    Check whether given target exists on device screen

    :param v: target to be checked
    :return: False if target is not found, otherwise returns the coordinates of the target
    :platforms: Android, Windows, iOS
    """

断言

另外 Airtest 还提供了几个断言语句来判断结果是否存在或者相同，定义如下：

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
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63

@logwrap
def assert_exists(v, msg=""):
    """
    Assert target exists on device screen

    :param v: target to be checked
    :param msg: short description of assertion, it will be recorded in the report
    :raise AssertionError: if assertion fails
    :return: coordinates of the target
    :platforms: Android, Windows, iOS
    """
    try:
        pos = loop_find(v, timeout=ST.FIND_TIMEOUT, threshold=ST.THRESHOLD_STRICT)
        return pos
    except TargetNotFoundError:
        raise AssertionError("%s does not exist in screen, message: %s" % (v, msg))

@logwrap
def assert_not_exists(v, msg=""):
    """
    Assert target does not exist on device screen

    :param v: target to be checked
    :param msg: short description of assertion, it will be recorded in the report
    :raise AssertionError: if assertion fails
    :return: None.
    :platforms: Android, Windows, iOS
    """
    try:
        pos = loop_find(v, timeout=ST.FIND_TIMEOUT_TMP)
        raise AssertionError("%s exists unexpectedly at pos: %s, message: %s" % (v, pos, msg))
    except TargetNotFoundError:
        pass

@logwrap
def assert_equal(first, second, msg=""):
    """
    Assert two values are equal

    :param first: first value
    :param second: second value
    :param msg: short description of assertion, it will be recorded in the report
    :raise AssertionError: if assertion fails
    :return: None
    :platforms: Android, Windows, iOS
    """
    if first != second:
        raise AssertionError("%s and %s are not equal, message: %s" % (first, second, msg))

@logwrap
def assert_not_equal(first, second, msg=""):
    """
    Assert two values are not equal

    :param first: first value
    :param second: second value
    :param msg: short description of assertion, it will be recorded in the report
    :raise AssertionError: if assertion
    :return: None
    :platforms: Android, Windows, iOS
    """
    if first == second:
        raise AssertionError("%s and %s are equal, message: %s" % (first, second, msg))

这几个断言比较常用的就是 assert_exists 和 assert_not_exists 判断某个目标是否存在于屏幕上，同时还可以传入 msg，它可以被记录到 report 里面。 以上就是 Airtest 的 API 的用法，它提供了一些便捷的方法封装，同时还对接了图像识别等技术。 但 Airtest 也有一定的局限性，比如不能根据 DOM 树来选择对应的节点，依靠图像识别也有一定的不精确之处，所以还需要另外一个库 —— Poco。

Poco

利用 Poco 我们可以支持 DOM 选择，例如编写 XPath 等来定位某一个节点。 首先需要安装 Poco，使用 pip3 即可：

1

pip3 install pocoui

安装好了之后我们便可以使用它来选择一些节点了，示例如下：

1
2
3
4
5
6
7
8
9

from airtest.core.api import *
from poco.drivers.android.uiautomation import AndroidUiautomationPoco

poco = AndroidUiautomationPoco()

uri = 'Android://127.0.0.1:5037/emulator-5554'
connect_device(uri)
home()
poco(text='Weather').click()

比如这里我们就声明了 AndroidUiautomationPoco 这个 Poco 对象，然后调用了 poco 传入一些选择器，选中之后执行 click 方法。 这个选择器非常强大，可以传入 name 和各个属性值，具体的使用方法见：https://poco.readthedocs.io/en/latest/source/poco.pocofw.html

1
2
3
4
5
6
7
8
9
10
11

from airtest.core.api import *
from poco.drivers.android.uiautomation import AndroidUiautomationPoco
from poco.proxy import UIObjectProxy

poco = AndroidUiautomationPoco()

uri = 'Android://127.0.0.1:5037/emulator-5554'
connect_device(uri)
home()
object: UIObjectProxy = poco("com.microsoft.launcher.dev:id/workspace")
print(object)

poco 返回的是 UIObjectProxy 对象，它提供了其他的操作 API，例如选取子节点，兄弟节点，父节点等等，同时可以调用各个操作方法，如 click、pinch、scroll 等等。 具体的操作文档可以参见：https://poco.readthedocs.io/en/latest/source/poco.proxy.html 下面简单总结：

• attr：获取节点属性
• child：获取子节点
• children：获取所有子节点
• click：点击
• double_click：双击
• drag_to：将某个节点拖拽到另一个节点
• exists：某个节点是否存在
• focus：获得焦点
• get_bounds：获取边界
• get_name：获取节点名
• get_position：获取节点位置
• get_size：获取节点大小
• get_text：获取文本内容
• long_click：长按
• offspring：选出包含直接子代的后代
• parent：父节点
• pinch：缩放
• scroll：滑动
• set_text：设置文字
• sibling：兄弟节点
• swipe：滑动
• wait：等待
• wait_for_appearance：等待某个节点出现
• wait_for_disappearance：等待某个节点消失

以上的这些方法混用的话就可以执行各种节点的选择和相应的操作。

最近遇到一个问题，那就是需要给别人共享一下 Kubernetes 的某个资源的使用和访问权限，这个仅仅存在于某个 namespace 下，但是我又不能把管理员权限全都给它，我想只给他授予这一个 Namespace 下的权限，那应该怎么办呢？ 比如我这边是需要只想授予 postgresql 这个 Namespace 的权限，这里我就需要利用到 Kubernetes 里面的 RBAC 机制来实现了，下面记录了我的操作流程。

创建 Namespace

1

kubectl create namespace postgresql

首先没有 Namespace 的话需要创建一个 Namespace，这里我创建的是 postgresql，大家可以自行修改。

创建 ServiceAccount

接下来需要创建一个 ServiceAccount，yaml 如下：

1
2
3
4
5

apiVersion: v1
kind: ServiceAccount
metadata:
  name: postgresql
  namespace: postgresql

创建 Role

然后还要创建一个 Role，来控制相应的权限，yaml 如下：

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

kind: Role
apiVersion: rbac.authorization.k8s.io/v1beta1
metadata:
  name: postgresql
  namespace: postgresql
rules:
  - apiGroups: ["", "extensions", "apps"]
    resources: ["*"]
    verbs: ["*"]
  - apiGroups: ["batch"]
    resources:
      - jobs
      - cronjobs
    verbs: ["*"]

这里由于 Role 是 Namespace 级别的，所以只能在特定 Namespace 下生效，这里我要让授予本 Namespace 下的所有权限，这里 rules 就添加了所有的 API类型、资源类型和操作类型。

创建 RoleBinding

最后需要将 Role 和 ServiceAccount 绑定起来，yaml 如下：

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

kind: RoleBinding
apiVersion: rbac.authorization.k8s.io/v1beta1
metadata:
  name: postgresql
  namespace: postgresql
subjects:
  - kind: ServiceAccount
    name: postgresql
    namespace: postgresql
roleRef:
  apiGroup: rbac.authorization.k8s.io
  kind: Role
  name: postgresql

创建 kubeconfig 文件

现在我们执行上述 yaml 文件之后，ServiceAccount 其实就已经创建好了，它会对应一个 secret，我们来看下详情，执行：

1

kubectl get serviceaccount postgresql -n postgresql -o yaml

好，运行结果类似如下：

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

apiVersion: v1
kind: ServiceAccount
metadata:
  annotations:
    kubectl.kubernetes.io/last-applied-configuration: |
      {"apiVersion":"v1","kind":"ServiceAccount","metadata":{"annotations":{},"name":"postgresql","namespace":"postgresql"}}
  creationTimestamp: "2020-07-30T16:10:38Z"
  name: postgresql
  namespace: postgresql
  resourceVersion: "17800240"
  selfLink: /api/v1/namespaces/postgresql/serviceaccounts/postgresql
  uid: 6327db1f-6a93-4f1e-b988-31842989bbbc
secrets:
- name: postgresql-token-v26k7

这里它实际关联了一个 secret，叫做 postgresql-token-v26k7，这里面就隐藏了 ServiceAccount 的 token 和证书。 好，那么我们就可以利用这个 secret 来制作 kubeconfig 文件了，命令如下：

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

server=https://your-server:443
name=postgresql-token-v26k7
namespace=postgresql

ca=$(kubectl get secret/$name -n $namespace -o jsonpath='{.data.ca.crt}')
token=$(kubectl get secret/$name -n $namespace -o jsonpath='{.data.token}' | base64 --decode)

echo "apiVersion: v1
kind: Config
clusters:
- name: test
  cluster:
    certificate-authority-data: ${ca}
    server: ${server}
contexts:
- name: test
  context:
    cluster: test
    user: postgresql
current-context: test
users:
- name: postgresql
  user:
    token: ${token}
" > postgresql.kubeconfig

这里我们需要指定三个变量：

• server：就是 Kubernetes Server API 的地址
• name：就是 ServiceAccount 对应的 secret
• namespace：就是当前操作的 Namespace

运行之后就会生成一个 portgresql.kubeconfig 文件。

使用

那么怎么使用呢？很简单，设置下环境变量切换下就好了。

1

export KUBECONFIG=postgresql.kubeconfig

这里我们就将 KUBECONFIG 设置了下，这样再执行 kubectl 就会读取到当前的 kubeconfig 文件，就会生效了。 这时候我们来测试下：

1

kubectl get nodes

运行结果如下：

1

Error from server (Forbidden): nodes is forbidden: User "system:serviceaccount:postgresql:postgresql" cannot list resource "nodes" in API group "" at the cluster scope

可以看到这里就提示没有列出节点的权限了。 然后我们操作下 postgresql 下的权限试试：

1

kubectl get svc -n postgresql

运行结果如下：

1
2
3
4
5

NAME                  TYPE        CLUSTER-IP     EXTERNAL-IP   PORT(S)    AGE
postgresql            ClusterIP   10.0.193.137   <none>        5432/TCP   9d
postgresql-headless   ClusterIP   None           <none>        5432/TCP   9d
postgresql-metrics    ClusterIP   10.0.60.88     <none>        9187/TCP   9d
postgresql-read       ClusterIP   10.0.236.184   <none>        5432/TCP   9d

这里就可以看到对 postgresql 这个命名空间的操作就成功了。 到此为止我们就成功实现了特定 Namespace 的限制，大功告成。

之前也写过不少关于爬虫的博客了，比如我拿一个案例来写了一篇博客，当时写的时候好好的，结果过了一段时间这个页面改版了，甚至直接下线了，那这篇案例就废掉了。 另外如果拿别人的站或者 App 来做案例的话，比较容易触犯到对方的利益，风险比较高，比如把某个站的 JavaScript 逆向方案公布出来，比如把某个 App 的逆向方案公布出来。如果此时此刻没有对方联系你的话，一个很大原因可能是规模太小了别人没注意到，但不代表以后不会。我还是选择爱护自己的羽毛，关于逆向实际网站和 App 的案例我都不会发的。在这种情况下比较理想的方案是自建案例，只用这个案例讲明白对应的知识点就可以了。 所以，为此，这段时间我也在写一些爬虫相关的案例，比如：

• 无反爬的服务端渲染网站
• 带有参数加密的 SPA 网站
• 各类形式验证码网站
• 反 WebDriver 网站
• 字体反爬等网站
• 模拟登录网站
• App 案例，如代理检测，SSL Pining 等

今天发布一下。

案例列表

本案例平台自爬数据、自建页面、自接反爬，案例稳定后永不过期，适合教学与练习。

SSR 网站

• ssr1：猫眼电影数据网站，数据通过服务端渲染，适合基本爬虫练习。
• ssr2：HTTPS 无效证书网站，适合做跳过验证 HTTPS 案例。
• ssr3：HTTP Basic Authentication 网站，适合做 HTTP 认证案例，用户名密码均为 admin。
• ssr4：每个响应增加了 5 秒延迟，适合测试慢速网站爬取或做爬取速度测试，减少本身网速干扰。

SPA 网站

• spa1：猫眼电影数据网站，数据通过 Ajax 加载，页面动态渲染，适合 Ajax 分析和动态页面渲染爬取。
• spa2：猫眼电影数据网站，数据通过 Ajax 加载，数据接口参数加密且有时间限制，适合动态页面渲染爬取或 JavaScript 逆向分析。
• spa3：猫眼电影数据网站，数据通过 Ajax 加载，无页码翻页，适合 Ajax 分析和动态页面渲染抓取。
• spa4：新闻网站索引，数据通过 Ajax 加载，无页码翻页，适合 Ajax 分析和动态页面渲染抓取以及智能页面提取分析。
• spa5：豆瓣图书网站，数据通过 Ajax 加载，有翻页，无反爬，适合大批量动态页面渲染抓取。
• spa6：电影数据网站，数据通过 Ajax 加载，数据接口参数加密且有时间限制，源码经过混淆，适合 JavaScript 逆向分析。

验证码网站

• captcha1：对接滑动拼图验证码，适合滑动拼图验证码分析处理。
• captcha2：对接图标点选验证码，适合图标点选验证码分析处理。
• captcha3：对接图文点选验证码，适合图文点选验证码分析处理。
• captcha4：对接语序分析验证码，适合语序分析验证码分析处理。
• captcha5：对接空间推理验证码，适合空间推理验证码分析处理。
• captcha6：对接九宫格识图验证码，适合九宫格识图验证码分析处理。