爬虫之异步相关
涂寐 Lv5
  2022-01-26 00:00:42 2022-01-26 00:00   2022-02-24 21:47:06    

前言

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

目的

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

异步方式

  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
  • 本文链接：article/d35d1b94.html
  • 版权声明：本博客所有文章除特别声明外，均采用 BY-NC-SA 许可协议。转载请注明出处！
 评论
  1. 1. 前言
  2. 2. 目的
  3. 3. 异步方式
  4. 4. selenium 模块使用
  5. 5. 代码实例
    1. 5.1. 同步爬虫测试
    2. 5.2. 线程池的基本使用
    3. 5.3. 线程池爬取梨视频
    4. 5.4. 单任务协程
    5. 5.5. 多任务协程
    6. 5.6. 多任务协程实验
    7. 5.7. aiohttp实现多任务异步协程
    8. 5.8. selenium模块基本使用
    9. 5.9. selenium模块其他使用
    10. 5.10. selenium模块__iframe__动作链
    11. 5.11. 未实现登录QQ空间
    12. 5.12. 无可视化界面与规避反反爬
    13. 5.13. 登录12306
  6. 6. 后记