爬虫之异步相关
涂寐 Lv4

前言

实验代码不想按分类处理,单独作为一个模块,其中笔记较多,对着代码更好理解

目的

  • 在爬虫中使用异步实现高性能的数据爬取操作

    异步方式

  1. 多线程|多进程
  • 优点:为阻塞操作单独开启线程|进程,实现异步
  • 劣势:不能无限制开启多线程|多进程
  1. 线程池|进程池
  • 优点:降低系统对进程|进程创建销毁的频率,进而降低系统内存开销
  • 劣势:池中线程数|进程数有限
  1. 单线程 + 异步协程
  • even_loop:事件循环对象。其中可写入待循环方法(协程对象)。
  • coroutine:协程对象。使用 async 关键字定义方法,该方法被调用时返回一个协程对象,而不是调用后立即执行;协程对象被事件循环调用。
  • task:任务对象。对协程对象再次封装,内含任务各状态。
  • future:将执行或未执行的任务,与 task 无本质差别。
  • async:定义一个协程。
  • await:用来挂起阻塞方法的执行。

    selenium 模块使用

  1. selenium 模块和爬虫间关联?
  • 便捷获取网站中动态数据
  • 编写模拟登录
  1. 什么是 selenium 模块?
  • 基于浏览器自动化的一个 python 模块
  1. selenium 模块使用步骤
  • 环境安装:pip install selenium
  • 浏览器驱动程序下载
  • 实例化浏览器对象
    • from selenium import webdriver
    • from selenium.webdriver.chrome.service import Service
    • service = Service(executable_path=’./chromedriver.exe’)
    • driver = webdriver.Chrome(service=service)
  • 编写基于浏览器自动化的操作代码
    • 发起请求:get(URL))
    • 标签定位:find_element() 结合 By 类
    • 标签交互:send_keys(‘xxx’)
    • 执行 js 脚本:excute_script(‘js代码’)
    • 界面回退:back()
    • 界面前进:forward()
    • 关闭浏览器:quit()
  • selenium 模块处理 iframe
    • 待定位标签位于 iframe 标签中,使用 switch_to.frame() 方法改变定位作用域
    • 动作链,需导入:from selenium.webdriver import ActionChains
      • 长按点击动作:click_and_hold(draggable_id)
      • 拖动操作:move_by_offset(x, y)
      • 立即执行操作:perform()
      • 释放动作链:release()
  • options 类相关方法
    • binary_location:设置 Chrome 二进制文件位置
    • add_argument:添加启动参数
    • add_extension,add_encoded_extension:添加扩展应用
    • add_experimental_option:添加实验性质的设置参数
    • debugger_address:设置调试器地址

      代码实例

      以下部分为代码实例讲解部分,内含单条语句注释说明

同步爬虫测试

首先,感受一下同步爬虫

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
import requests

# 同步爬虫测试
# 由于阻塞操作,同步爬虫仅在执行完一个请求后才能开始执行另一个请求

headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/96.0.4664.110 Safari/537.36 Edg/96.0.1054.62'
}
urls = [
'https://downsc.chinaz.net/Files/DownLoad/jianli/202112/jianli16623.rar',
'https://downsc.chinaz.net/Files/DownLoad/jianli/202112/jianli16611.rar',
'https://downsc.chinaz.net/Files/DownLoad/jianli/202112/jianli16615.rar'
]


class RequestGet(object):

def get_content(self, url):
print('正在爬取:' + url)
# get() 方法为阻塞方法,当执行完当前操作才能执行其他操作
response = requests.get(url=url, headers=headers)
# 根据响应码执行下步操作,若为 200,则返回二进制数据
if response.status_code == 200:
return response.content

def parse_content(self, content):
print('数据长度:', len(content))


if __name__ == "__main__":
requestGet = RequestGet()
for url in urls:
content = requestGet.get_content(url)
requestGet.parse_content(content)

线程池的基本使用

主要进行同步爬虫、单线程串行、线程池的使用效果

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
import time
# 导入线程池模块需要的类
from multiprocessing.dummy import Pool


def get_page(str):

# 模拟单线程串行方式操作
print("正在下载" + str)
time.sleep(2)
print('下载成功' + str)


name_list = ['xiaozi', 'aa','bb','cc']

start_time = time.time()

# 1. 模拟单线程串行方式操作
#
# for i in range(len(name_list)):
# get_page(name_list[i])

# 2. 模拟线程池方式操作
#
# 实例化线程池对象
pool = Pool(4)
# 列表元素传入 get_page() 中处理
pool.map(get_page, name_list)
#
# # 此处为比对同步爬虫
# for a in name_list:
# get_page(a)

end_time = time.time()
print('%d second' %(end_time - start_time))

线程池爬取梨视频

梨视频网站现在通过 ajax 实时加载视频,且改变了 ajax 响应包的视频 URL,具体看下面的代码
再bb一句,下载量大时对比线程池耗时效果更明显

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
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
import re
import time
from multiprocessing.dummy import Pool

import requests
from lxml import etree


# 爬取梨视频的生活频道视频
# 注意:进程池处理的式阻塞且耗时的操作

class Video(object):
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/96.0.4664.110 Safari/537.36 Edg/96.0.1054.62'
}
# 空列表,存储所有的视频名和视屏链接
urls = []

def video_url(self):
url = 'https://www.pearvideo.com/category_130'
page_text = requests.get(url=url, headers=self.headers).text
tree = etree.HTML(page_text)
# 找到聚集点
li_list = tree.xpath('//ul[@id="categoryList"]/li')
for li in li_list:
# 精确查找每一个符合值,并进行拼接成详情页完整URL
# https://www.pearvideo.com/video_1749618
detail_url = 'https://www.pearvideo.com/' + li.xpath('./div/a/@href')[0]
name = li.xpath('./div/a/div[2]/text()')[0] + '.mp4'
# 去除某些符号
# name = li.xpath('./div/a/div[2]/text()')[0].replace(' | ', '').replace('·', '').replace(',', '').replace('、', '')
# 输出文件名,及其URL
print(name + " --> " + detail_url)
# 当直接点击网页上的某个视频,将在新页面打开并播放
# 若复制 URL 后打开,则需要点击后播放,判断:原页面增加跳转并播放的点击事件
# 比较发现,视频的真实地址通过 AJAX 动态加载,如下:
# {'resultCode': '1', 'resultMsg': 'success', 'reqId': 'bfd8d987-4766-45f7-944c-36669facb5bc', 'systemTime': '1642779887007', 'videoInfo': {'playSta': '1', 'video_image': 'https://image1.pearvideo.com/cont/20220112/14494912-124035-1.png', 'videos': {'hdUrl': '', 'hdflvUrl': '', 'sdUrl': '', 'sdflvUrl': '', 'srcUrl': 'https://video.pearvideo.com/mp4/third/20220112/1642779887007-14494912-124007-hd.mp4'}}}
# detail_page_text = requests.session().get(detail_url, headers=headers).text
# 尝试直接拿 ajax 的请求链接访问,文章已下线
# 对比请求包不同,请求头中少了 Referer 参数,用于标识该请求的链接来源
# 尝试过先访问详情页,保持 cookie 下再访问 ajax 页面,无效
# 拿到视频详情页id
ex = 'video_(.*)'
video_id = re.findall(ex, detail_url)[0]
# print(detail_url)
# print(vedio_id)
# 对视频详情页 ajax 发起请求
ajax_url = 'https://www.pearvideo.com/videoStatus.jsp?contId={false_id}&mrd=0.6923628803380188'.format(
false_id=video_id)
# print(ajax_url)
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/96.0.4664.110 Safari/537.36 Edg/96.0.1054.62',
'Referer': detail_url
}
ajax_page_json = requests.get(url=ajax_url, headers=headers).json()
# print(ajax_page_json)
# 编写正则表达式
# ea = "', 'srcUrl': '(.*?)'}}}"
# 最后发现: ajax 的响应数据中得到的链接有些需要替换,自行对照,如下:
# 可访问:https://video.pearvideo.com/mp4/third/20220120/cont-1750332-15498275-092138-hd.mp4
# ajax: https://video.pearvideo.com/mp4/third/20220120/1642775491037-15498275-092138-hd.mp4
# 修改正则,实现视频url的拼接
ea = "', 'srcUrl': '.*?(-.*?)'}}}"
eb = ", 'srcUrl': '(.*/)"
# 此处需要使用 str() 方法将 ajax_page_json 对象强转为字符串
video_url_a = re.findall(ea, str(ajax_page_json))[0]
video_url_b = re.findall(eb, str(ajax_page_json))[0]
video_url = video_url_b + 'cont-' + video_id + video_url_a
# print(video_url_a)
# print(video_url_b)
print(video_url)
# 获取到数据写入字典进行存储
video_dict = {
'name': name,
'url': video_url
}
# 列表嵌套字典,每个视频名和视频链接作为一个字典写入列表 urls
urls.append(video_dict)
return urls

def video_data(self, dict):
url = dict['url']
# print(url)
time.sleep(19)
data = requests.get(url=url, headers=self.headers).content
print(dict['name'] + '下载中……')
# 持久化存储
with open('./梨视频/' + dict['name'], 'wb') as fp:
fp.write(data)
print(dict['name'] + '下载完成!!!')


if __name__ == '__main__':
video = Video()
urls = []
urls = video.video_url()
# print(urls)
start = time.time()
# 通过进程池对视频数据进行请求
pool = Pool(4)
pool.map(video.video_data, urls)
# 同步爬虫比较
# for url in urls:
# video.video_data(url)
end = time.time()
print('全部下载完成……')
# 上面加了个延时,直观点
print(end - start)

单任务协程

要理解难的,肯定先理解其下较为简单的,别说了,看源码

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
import asyncio


# 协程单任务基本使用
# 使用 async 关键字声明为异步函数
async def request(url):
print('正在请求的url为:' + url)
print('请求成功:' + url)
return url


# 调用协程,返回一个协程对象
c = request('0xtlu.me')


# # 创建一个事件循环对象
# loop = asyncio.get_event_loop()
# # 将协程对象注册到 loop 中,再启动 loop
# loop.run_until_complete(c)

# task 的使用
# loop = asyncio.get_event_loop()
# # 基于 loop 创建 task 对象
# task = loop.create_task(c)
# 打印 task 对象状态
# # print(task)
# # 注册协程对象并启动 loop
# loop.run_until_complete(task)
# # 再次打印 task 对象状态
# print(task)

# future 的使用
# loop = asyncio.get_event_loop()
# task = asyncio.ensure_future(c)
# print(task)
# loop.run_until_complete(task)
# print(task)

# 绑定回调
# 回调函数
def callback_func(task):
# result() 返回入伍对象中封装的协程对象对应函数(request())的返回值
print(task.result())

# 创建事件对象
loop = asyncio.get_event_loop()
# 创建任务对象
task = asyncio.ensure_future(c)
# 使用回调函数(add_done_callback)绑定任务对象(task)
task.add_done_callback(callback_func)
# 运行任务对象
loop.run_until_complete(task)

多任务协程

单任务学完了,看多任务呗

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
import asyncio
import time


# 协程多任务使用

async def request(url):
print('正在下载' + url)
# 异步协程中出现同步模块相关代码(time.sleep(2)),则无法异步
# time.sleep(2)
# 当asyncio 中遇到阻塞操作必需手动挂起(使用 await 关键字)
# await 用来用来声明程序的挂起
await asyncio.sleep(2)
print('下载完毕' + url)


urls = ['www.baidu.com', '0xtlu.me', 'www.google.com', 'www.sogou.com']

# 存放任务列表
tasks = []
# 开始事件
start = time.time()

for url in urls:
c = request(url)
task = asyncio.ensure_future(c)
tasks.append(task)

loop = asyncio.get_event_loop()
# 人物列表封装到 wait 中
loop.run_until_complete(asyncio.wait(tasks))

print(time.time() - start)

多任务协程实验

这是一个难过的例子,有同步模块,看不到多任务异步协程效果

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
import asyncio
import time

import requests

# 多任务异步协程实现
# 由于含同步模块代码,未实现

start = time.time()
urls = ['https://0xtlu.me/article/c2461216.html', 'https://0xtlu.me/article/82b683e3.html',
'https://0xtlu.me/article/c86c0afc.html']


async def get_page(url):
print('正在下载' + url)
# requests.get 为基于同步请求的模块,但现在需要的是给予异步的网络请求模块进行指定 url 的请求发送,即 aiohttp
response = requests.get(url=url)
print('下载完毕' + response.text)


tasks = []

for url in urls:
c = get_page(url)
task = asyncio.ensure_future(c)
tasks.append(task)

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

end = time.time()

print('耗时:' + str(end - start))

aiohttp实现多任务异步协程

上个题目的解决方案:通过 aiohttp 模块解决同步模块 requests,让异步协程正确运行

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
import asyncio
import time

import aiohttp
import requests

# aiohttp实现多任务异步协程

# 环境安装
# pip install aiohttp
# 需要使用 aiohttp 模块的 ClientSession 对象

start = time.time()
urls = ['https://0xtlu.me/article/c2461216.html', 'https://0xtlu.me/article/82b683e3.html',
'https://0xtlu.me/article/c86c0afc.html']


async def get_page(url):
print('正在下载' + url)
async with aiohttp.ClientSession() as session:
# aiohttp 可以使用操作
# get() 或 post()
# headers,params/data,proxy='http://ip:port/'
async with await session.get(url) as response:
# text() 方法返回字符串形式的响应数据
# read() 方法返回二进制形式的响应数据
# json() 方法返回 json 对象的响应数据
# 注:获取响应数据前需要使用 await 关键字手动挂起
page_text = await response.text()

print('下载完毕' + page_text)


tasks = []

for url in urls:
c = get_page(url)
task = asyncio.ensure_future(c)
tasks.append(task)

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

end = time.time()

print('耗时:' + str(end - start))

selenium模块基本使用

之前看到过谷歌自动化的效果,现在使用 selenium 模块发现原来是这个家伙

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
from selenium import webdriver
from selenium.webdriver.chrome.service import Service
from lxml import etree
from time import sleep

# 爬取药监局企业名称
# http://scxk.nmpa.gov.cn:81/xk/

# 实例化一个浏览器对象(传入浏览器的驱动程序)
#
# 老方法
# 能运行,但由于版本更新时,executable_path 方法被重构到 Service 函数中而引起弃用警告
# DeprecationWarning: executable_path has been deprecated, please pass in a Service object
# driver = webdriver.Chrome(executable_path='./chromedriver.exe')
#
# 新方法
service = Service(executable_path='./chromedriver.exe')
driver = webdriver.Chrome(service=service)
#
#
# 指定 URL 发起搜索请求
driver.get('http://scxk.nmpa.gov.cn:81/xk/')
# 打印网页标题
# print(driver.title)

#
# 通过 page_source 方法获取网页源码
page_text = driver.page_source
# print(page_text)
# 对网页源码进行数据解析
tree = etree.HTML(page_text)
li_list = tree.xpath('//ul[@id="gzlist"]/li')
# print(li_list)
for li in li_list:
name = li.xpath('./dl/@title')[0]
print(name)
#
#
# 暂停6s后关闭浏览器
sleep(6)
driver.close()

selenium模块其他使用

上一个标题是 selenium 模块的基本使用,现在看下它的元素定位、数据交互、前进后退等功能

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
from time import sleep
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.chrome.service import Service

service = Service(executable_path='./chromedriver.exe')
driver = webdriver.Chrome(service=service)

driver.get('https://www.taobao.com/')

# 进行标签定位
#
# 不推荐使用 find——element_by_id() 方法
# q_id = driver.find_element_by_id('q')
#
# 通过 By 类定位页面上 id 为 'q' 的元素
# By是 selenium 中内置的一个 class ,在该 class 中有各种方法来定位元素
q_id = driver.find_element(By.ID, 'q')
# print(q_id)
#
# 搜索框写入数据实现交互
q_id.send_keys('iphone')
#
# 实现点击搜索按钮的效果
# 获取按钮元素
button = driver.find_element(By.CSS_SELECTOR, '.btn-search')
# 实现点击事件,由于没登陆不给搜,直接添砖登录界面
button.click()
#
#
# 执行一组 js 脚本
# 使界面滚动到最底部
driver.execute_script('window.scrollTo(0,document.body.scrollHeight)')
#
#
# 实现回退效果:回到上一标签页
# 先访问下自己的博客
driver.get('https://0xtlu.github.io')
# 延时2s后回退淘宝登录界面
sleep(2)
driver.back()
# 继续延时2s后前进博客首页
sleep(2)
driver.forward()


#
#
sleep(6)
driver.quit()

selenium模块__iframe__动作链

登陆界面一般都是存在验证的,通常是通过 iframe 标签实现
当切换到 iframe 标签作用域时,需要实现一系列的动作——动作链的使用

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
# 利用菜鸟教程的练习模块完成拖动效果的代码实现
# https://www.runoob.com/try/try.php?filename=jqueryui-api-droppable
# 运行感受,这是搜索引擎问题还是什么,界面加载不出来|未连接到互联网


from time import sleep

from selenium.webdriver import ActionChains
from selenium.webdriver.chrome.service import Service
from selenium import webdriver
from selenium.webdriver.common.by import By

service = Service(executable_path='./chromedriver.exe')
driver = webdriver.Chrome(service=service)
driver.get('https://www.runoob.com/try/try.php?filename=jqueryui-api-droppable')

# 存在 iframe 标签,则定位其内部标签需先改变浏览器定位标签的作用域,使用如下方法
# 即,通过 switch_to.frame() 方法将当前定位的主题切换为 iframe 表单的内嵌页面中
# 主要为,通过 id、name、element (定位的某个元素)、索引来指定切换到的某个 iframe 表单
driver.switch_to.frame('iframeResult')
# 再通过 id 获取 iframe 表单内部待拖动的 div 标签
draggable_id = driver.find_element(By.ID, 'draggable')
# print(draggable_id)

# 动作链
# 要求:点击-拖动-放开点击
# 当调用 ActionChains 的方法时,不会立即执行
# 而是将所有的操作按顺序存放到队列中,当调用 perform() 方法时,根据队列依次执行
# 实例化行为事件类 ActionChains,实现键盘鼠标的各个操作
action = ActionChains(driver)
# 实现鼠标左键长按标签事件
action.click_and_hold(draggable_id)
# 移动偏移
for i in range(5):
# 调用 perform() 方法立即执行动作链
# move_by_offset(x, y) 方法参数解析:x(水平方向),y(垂直方向)
action.move_by_offset(50, 0).perform()
sleep(0.3)
# 释放动作链(松开左键)
action.release().perform()

sleep(3)
# 释放 action
action.reset_actions()
driver.quit()

未实现登录QQ空间

登录界面没有改变,但滑块与缺口间位置对齐比较麻烦,暂时不想理解
对了,这里可以学到动作链的使用

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
# 登录QQ空间
from selenium.webdriver.common.by import By
from selenium.webdriver.chrome.service import Service
from selenium import webdriver
from time import sleep
from selenium.webdriver import ActionChains

service = Service(executable_path='./chromedriver.exe')
driver = webdriver.Chrome(service=service)
driver.get('https://qzone.qq.com/')
# iframe 标签中再定位
driver.switch_to.frame('login_frame')
a_tag = driver.find_element(By.ID, 'switcher_plogin')
# 鼠标点击事件
a_tag.click()
#
# 账密定位
username = driver.find_element(By.ID, 'u')
password = driver.find_element(By.ID, 'p')
# 清空原数据
username.clear()
password.clear()
# 账密写入
username.send_keys('yourUser')
password.send_keys('yourPasswd!')
# 点击登录事件
login = driver.find_element(By.ID, 'login_button')
sleep(2)
seli = login.click()
print(seli)
sleep(6)
# 自动化多了,提示:由于你的帐号存在异常,需要进行手机验证,请使用QQ手机版扫码进行登录。
# 滑块验证
# 切换定位作用域
driver.switch_to.frame('tcaptcha_iframe')
# 找到滑块
slideBlock_id = driver.find_element(By.ID, 'slideBlock')
# 实例化行为事件
action = ActionChains(driver)
# 点击且长按的动作链
action.click_and_hold(slideBlock_id)
# x 坐标值来源,浏览器控制台观察,拖动滑块时 img 标签的 style 属性中的 left 会改变,看重合时数值减去初始值即可
action.move_by_offset(183, 0).perform()
#
#
# 说一下问题,缺口位置不同导致滑块移动距离差异
# 另一种方法,先安装下面两个模块
# pip install opencv-python
# pip install np
# 看了看,步骤有点……不想复制,有兴趣自己了解吧
# https://www.cnblogs.com/yxh-zixishi/p/14514493.html
sleep(3)
action.release().perform()

sleep(36)
driver.quit()

无可视化界面与规避反反爬

主要通过设置 chrome 的启动参数实现无可视化界面(无头浏览器)
反反爬机制主要为屏蔽自动化的某些特征,如自动化控制提示、禁止启用 blink 功能和嵌入js等等

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
# 实现无可视化界面和规避检测
from selenium.webdriver.chrome.options import Options
from selenium.webdriver import ChromeOptions
from selenium.webdriver.chrome.service import Service
from selenium import webdriver
from time import sleep

# 无可视化界面(无头浏览器)
# options 是一个方便控制 chrome 启动时属性的类
# 实例化 options 类
option = Options()
# 设置启动参数
# 设置非可视化界面,非可视化 linux 中需设置非可视化才能正常运行
option.add_argument('--headless')
# 根据谷歌文档说明,该语句用于规避bug
option.add_argument('--disable-gpu')
# 添加 UA (需要为 chrome 浏览器的 UA )
option.add_argument(
'user-agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/97.0.4692.99 Safari/537.36"')

# 开启开发者模式
# 实现规避检测
# 屏蔽自动化受控提示和开发者提示
option.add_experimental_option('excludeSwitches', ['enable-automation', 'load-extension'])

service = Service(executable_path='./chromedriver.exe')
# DeprecationWarning: use options instead of chrome_options
# chrome_options 已弃用
# driver = webdriver.Chrome(service=service, chrome_options=option)
# 改用 options
# driver = webdriver.Chrome(service=service, options=option)
# 添加规避
driver = webdriver.Chrome(service=service, options=option)

driver.get('https://www.baidu.com/')
print(driver.page_source)

sleep(6)
driver.quit()

登录12306

哒哒,这就是看规避效果的时候了。没有进行无可视化界面处理哦

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
80
81
82
83
84
85
86
# 12306

from time import sleep
from selenium.webdriver.chrome.service import Service
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver import ActionChains
from selenium.webdriver.chrome.options import Options

option = Options()

# 屏蔽 webdriver 使用特征,绕过特征验证
# 也可以通过向页面嵌入 js 代码规避检测,使用到再说
# 屏蔽自动化受控提示和开发者提示
option.add_experimental_option('excludeSwitches', ['enable-automation', 'load-extension'])
# 禁止启用 blink 运行时的功能
option.add_argument('--disable-blink-features=AutomationControlled')
# 禁用自动扩展(插件)
option.add_experimental_option('useAutomationExtension', False)
# 屏蔽'保存密码'提示框
prefs = {}
prefs["credentials_enable_service"] = False
prefs["profile.password_manager_enabled"] = False
option.add_experimental_option("prefs", prefs)
service = Service(executable_path='./chromedriver.exe')
driver = webdriver.Chrome(service=service, options=option)

# 嵌入 js 代码进行规避
# chrome 高版本后该规避不知道还行不行(用不出来),这里不做肯定
# with open('./js.js') as f:
# js = f.read()
# driver.execute_cdp_cmd("Page.addScriptToEvaluateOnNewDocument", {
# "source": js
# }))
#
# 利用 Chrome 开发工具协议,正确从 selenium 启动的 Chrome 浏览器中隐藏 window.navigator.webdriver 关键字
# 单步测试可以使用
# 对比了下,应该是我的 js 文件中没写相关的语句导致无法实现规避效果
driver.execute_cdp_cmd("Page.addScriptToEvaluateOnNewDocument", {
"source": """
Object.defineProperty(navigator, 'webdriver', {
get: () => undefined
})
"""
})

driver.get('https://kyfw.12306.cn/otn/resources/login.html')
username = driver.find_element(By.ID, 'J-userName')
password = driver.find_element(By.ID, 'J-password')
sleep(1)
username.clear()
password.clear()
sleep(2)
username.send_keys('yourUser')
sleep(1)
password.send_keys('yourPasswd')
sleep(2)
login = driver.find_element(By.ID, 'J-login')
sleep(2)
login.click()
sleep(2)
slide = driver.find_element(By.ID, 'nc_1_n1z')
sleep(2)
print(slide)
action = ActionChains(driver)
action.click_and_hold(slide)
# 直接滑到尽头,这个比较简单

action.move_by_offset(300, 0).perform()
action.release().perform()
sleep(1)
# action.click_and_hold(slide)
# for i in range(5):
# sleep(1)
# action.move_by_offset(60, 0).perform()
# sleep(1)
# action.release().perform()
# 拖动完成后报错:
# 哎呀,出错了,点击刷新再来一次(error:EQaZmr)
# 感觉像机器验证
# 确认了 webdriver 必作为反爬虫特征处理,so 不给你登
sleep(10)
# 休眠一下,拿到当前页面源码
print(driver.page_source)
action.reset_actions()
driver.quit()

后记

写完了,写过一遍了,懂了吗?没有
为什么做笔记,方便之后的回顾呀

  • 本文标题:爬虫之异步相关
  • 本文作者:涂寐
  • 创建时间:2022-01-26 00:00:42
  • 本文链接:https://0xtlu.github.io/article/d35d1b94.html
  • 版权声明:本博客所有文章除特别声明外,均采用 BY-NC-SA 许可协议。转载请注明出处!
 评论