异步爬虫和selenium模块

浅浅的花香味﹌ 2023-08-17 16:53 107阅读 0赞

## asyncio模块 ##

import asyncio
    
    
    # 标识是一个协程函数
    async def demo():  # 协程方法
        print("start")
        await asyncio.sleep(1)  # 阻塞
        print("end")
    
    
    # 第三方操作demo
    loop = asyncio.get_event_loop()  # 创建一个事件循环
    loop.run_until_complete(demo())  # 把demo任务放到事件循环中执行
    
    # 结果：
    　　start
    　　end

import asyncio
    
    # 启动多个任务，没有返回值
    
    async def demo():
        print("start")
        await asyncio.sleep(1)
        print("end")
    
    
    loop = asyncio.get_event_loop()
    wait_obj = asyncio.wait([demo(), demo(), demo(), demo()])
    loop.run_until_complete(wait_obj)
    
    # 结果：
        start
        start
        start
        start
        end
        end
        end
        end

import asyncio
    
    # 启动多个任务并且有返回值
    async def demo():
        print("start")
        await asyncio.sleep(1)
        print("end")
        return 123
    
    
    loop = asyncio.get_event_loop()
    t1 = loop.create_task(demo())
    t2 = loop.create_task(demo())
    t3 = loop.create_task(demo())
    tasks = [t1, t2, t3]
    wait_obj = asyncio.wait([t1, t2])
    loop.run_until_complete(wait_obj)
    for t in tasks:print(t.result())
    
    # 结果
        start
        start
        start
        end
        end
        end
        123
        123
        123

import asyncio
    
    # 谁先回来先取谁的结果
    async def demo(i):
        print("start")
        await asyncio.sleep(10 - i)
        print("end")
        return i, 123
    
    
    async def main():
        task_l = []
        for i in range(10):
            task = asyncio.ensure_future(demo(i))
            task_l.append(task)
        for ret in asyncio.as_completed(task_l):
            res = await ret
            print(res)
    
    
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())
    
    # 结果：
        start
        start
        start
        start
        start
        start
        start
        start
        start
        start
        end
        (9, 123)
        end
        (8, 123)
        end
        (7, 123)
        end
        (6, 123)
        end
        (5, 123)
        end
        (4, 123)
        end
        (3, 123)
        end
        (2, 123)
        end
        (1, 123)
        end
        (0, 123)

#### 总结： ####

# await 后面跟阻塞事件，协程函数从这里要切换出去，还能保证一会切回来
    # await 必须写在async函数里，async函数是协程函数（类似生成器）
    # loop 是一个事件循环
    # 所有的协程执行，调度都离不开loop

## 单线程+多任务异步协程 ##

> \- 意义：提升爬取数据的效率
> 
> \- 实现异步爬虫的方式：
> 
> 　　- 多线程/多进程（不建议）
> 
> 　　- 线程池（可适当使用）
> 
> 　　- 单线程+多任务异步协程（推荐使用）
> 
> \- 概念
> 
> 　　- 协程：协程对象。
> 
> 　　　- 可以使用async关键字修饰的一个函数的定义（特殊的函数），当该函数被调用后，就可以返回一个协程对象。
> 
> 　　　（ 当函数调用后，函数内部的实现语句不会被立即执行。）
> 
> 　　- 任务对象：
> 
> 　　　　- 本质上就是对协程对象的进一步封装。
> 
> 　　　　- 可以给任务对象绑定一个回调
> 
> 　　　　　　- add\_done\_callback(callback)
> 
> 　　- 事件循环（EventLoop）：无限的循环对象
> 
> 　　　　- 我们必须将任务对象注册到事件循环对象中，然后开启事件循环对象。
> 
> 　　　　- 事件循环对象在执行任务对象的时候是基于异步的。
> 
> 　　- await async
> 
> 　　- 注意事项：
> 
> 　　　　- 保证特殊函数内部不可以出现不支持异步模块对应的代码
> 
> 　　　　- 在特殊函数内部遇到阻塞操作必须使用await关键字对其进行手动挂起
> 
> 　　　　- 如果想要将多个任务对象注册到事件循环中，必须将多个任务对象封装到一个列表中，然后将列表注册到事件循环中（也必须使用await方法将列表中的任务对象进行挂起）

import asyncio
    
    # 定义一个特殊的函数
    async def test(num):
        print(num)
    
    
    # 返回一个协程对象，并且函数内部的语句不会立即执行
    c = test(10)
    print(c)
    '''
    
    '''
    async def test(num):
        print(num)
    
    
    c = test(10)
    
    # 根据协程对象封装一个任务对象
    task = asyncio.ensure_future(c)
    print(task)

import asyncio
    import time
    
    async def request(url):
        print("正在请求：", url)
        time.sleep(2)
        print("请求完毕！", url)
    
    
    c1 = request("www.baidu.com")
    
    task_A = asyncio.ensure_future(c1)
    
    # 创建一个事件循环对象
    loop = asyncio.get_event_loop()
    # 将任务对象注册到该对象中并且启动事件循环
    loop.run_until_complete(task_A)

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

import asyncio
    import time
    
    
    async def request(url):
        print('正在请求：', url)
        time.sleep(2)
        print("请求完毕!", url)
    
        return url
    
    
    # 定义一个任务对象的回调函数
    # task参数表示的就是该函数被绑定的那个任务对象
    def task_callback(task):
        print("i am task_callback()")
        print(task.result())
        # task.result()返回的就是任务对象对应的特殊函数内部的返回值
    
    
    c = request("www.xxx.com")
    
    task = asyncio.ensure_future(c)
    task.add_done_callback(task_callback)
    
    loop = asyncio.get_event_loop()
    loop.run_until_complete(task)

任务对象绑定回调

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

import asyncio
    import time
    
    start_time = time.time()
    
    
    # 在特殊函数内部不可以实现不支持异步模块相关的代码
    async def request(url):
        print("正在请求：", url)
        # time.sleep(2)  # time模块不支持异步
        await asyncio.sleep(2)  # 阻塞操作必须使用await关键字进行挂起
        print("请求完毕!", url)
    
        return url
    
    
    urls = [
        "www.1.com",
        "www.2.com",
        "www.3.com"
    ]
    
    
    def task_callback(task):
        print(task.result())
    
    
    tasks = []  # 多任务列表：存放多个任务对象
    for url in urls:
        c = request(url)
        task = asyncio.ensure_future(c)
        task.add_done_callback(task_callback)
        tasks.append(task)  # 将多个任务对象放到一个任务列表中
    
    loop = asyncio.get_event_loop()
    # 多任务注册
    # wait就是将任务列表中的任务对象进行挂起
    loop.run_until_complete(asyncio.wait(tasks))
    
    print(time.time() - start_time)

多任务异步协程

多任务异步爬虫测试

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

from flask import Flask
    from time import sleep
    
    app = Flask(__name__)
    
    
    @app.route('/test1')
    def test1():
        sleep(2)
        return 'test 01'
    
    
    @app.route('/test2')
    def test2():
        sleep(2)
        return 'test 02'
    
    
    @app.route('/test3')
    def test3():
        sleep(2)
        return 'test 03'
    
    
    app.run()

flask-server

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

import asyncio
    import time
    import requests
    
    start_time = time.time()
    
    
    # 在特殊函数的内部不可以出现不支持异步模块的相关代码，requests这里就不可以使用了
    async def request(url):
        print("正在请求：", url)
        response = requests.get(url)  # 无法完成异步
        return response.text
    
    
    urls = [
        "http://127.0.0.1:5000/test1",
        "http://127.0.0.1:5000/test2",
        "http://127.0.0.1:5000/test3"
    ]
    
    
    def parse(task):
        page_text = task.result()
        print(page_text + ",请求到的数据！！！")
    
    
    tasks = []
    for url in urls:
        c = request(url)
        task = asyncio.ensure_future(c)
        task.add_done_callback(parse)
        tasks.append(task)
    
    loop = asyncio.get_event_loop()
    loop.run_until_complete(asyncio.wait(tasks))
    
    print(time.time() - start_time)

多任务异步爬虫测试

## aiohttp模块 ##

> 是一个支持异步的网络请求模块

安装：

pip install aiohttp

import asyncio
    import aiohttp
    
    # 简单的基本架构
    async def request(url):
        with aiohttp.ClientSession() as s:
            # s.get/post和requests中的get/post用法几乎一样，除了代理，s.get中如果使用代理：proxy="http://ip:port"
            with s.get(url) as response:
                # 获取字符串形式的响应：response.text()
                # 获取byte类型的响应：response.read()
                page_text = response.text()
                return page_text

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

import asyncio
    import time
    import aiohttp
    
    start_time = time.time()
    
    # 在当前架构的基础上进行细节补充
    # 细节1：在没一个with前加上async关键字
    # 细节2：在get方法前和response.text()前加上await关键字进行收到挂起操作
    async def request(url):
        async with aiohttp.ClientSession() as s:
            async with await s.get(url) as response:
                page_text = await response.text()
                return page_text
    
    urls = []
    # 测试更多的请求（最多511）
    for i in range(511):
        urls.append('http://127.0.0.1:5000/test1')
    
    
    def parse(task):
        page_text = task.result()
        print(page_text, "，请求到的数据！！！")
    
    
    tasks = []
    for url in urls:
        c = request(url)
        task = asyncio.ensure_future(c)
        task.add_done_callback(parse)
        tasks.append(task)
    
    loop = asyncio.get_event_loop()
    loop.run_until_complete(asyncio.wait(tasks))
    
    print(time.time() - start_time)

同样也是使用flask服务器测试

## selenium模块 ##

> \- 概念：就是一基于浏览器自动化的模块
> 
> \- selenium和爬虫之间的关联
> 
> 　　- 很便捷的捕获动态加载的数据（可见即可得）
> 
> 　　- 实现模拟登陆
> 
> \- 使用
> 
> 　　- 环境安装：
> 
> 　　　　- pip install selenium
> 
> 　　　　- 下载一个浏览器的驱动程序
> 
> 　　　　　　- 谷歌驱动下载：http://chromedriver.storage.googleapis.com/index.html
> 
> 　　　　　　- 驱动程序和浏览器版本的映射关系：http://blog.csdn.net/huilan\_same/article/details/51896672
> 
> 　　　　- 创建一个浏览器对象

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

from selenium import webdriver
    from time import sleep
    
    # 后面是你的浏览器驱动位置，记得前面加r'','r'是防止字符转义的
    driver = webdriver.Chrome(r'chromedriver.exe')
    # 用get打开百度页面
    driver.get("http://www.baidu.com")
    # 查找页面的“设置”选项，并进行点击
    driver.find_elements_by_link_text('设置')[0].click()
    sleep(2)
    # # 打开设置后找到“搜索设置”选项，设置为每页显示50条
    driver.find_elements_by_link_text('搜索设置')[0].click()
    sleep(2)
    
    # 选中每页显示50条
    m = driver.find_element_by_id('nr')
    sleep(2)
    m.find_element_by_xpath('//*[@id="nr"]/option[3]').click()
    m.find_element_by_xpath('.//option[3]').click()
    sleep(2)
    
    # 点击保存设置
    driver.find_elements_by_class_name("prefpanelgo")[0].click()
    sleep(2)
    
    # 处理弹出的警告页面   确定accept() 和 取消dismiss()
    driver.switch_to_alert().accept()
    sleep(2)
    # 找到百度的输入框，并输入 美女
    driver.find_element_by_id('kw').send_keys('美女')
    sleep(2)
    # 点击搜索按钮
    driver.find_element_by_id('su').click()
    sleep(2)
    # 在打开的页面中找到“Selenium - 开源中国社区”，并打开这个页面
    driver.find_elements_by_link_text('美女_百度图片')[0].click()
    sleep(3)
    
    # 关闭浏览器
    driver.quit()

selenium的使用

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

from selenium import webdriver
    from time import sleep
    
    bro = webdriver.Chrome(executable_path="chromedriver.exe")
    
    # 发起指定url的请求
    bro.get("https://www.jd.com/")
    
    # 在搜索框中搜索商品
    
    # 可以使用find系列的方法进行标签定位
    search_input = bro.find_element_by_id("key")
    # 在搜索框中写入商品名称
    search_input.send_keys('iphone x')
    sleep(2)
    btn = bro.find_element_by_xpath('//*[@id="search"]/div/div[2]/button')
    btn.click()
    sleep(2)
    
    # 执行js代码让滚轮向下滑动
    bro.execute_script("window.scrollTo(0,document.body.scrollHeight)")
    
    # 爬取数据
    page_text = bro.page_source
    with open("./jd.html", 'w', encoding="utf-8") as f:
        f.write(page_text)
    print(page_text)
    
    sleep(4)
    
    # 关闭浏览器
    bro.quit()

测试jd.com

- 动作链
        - 如果想要触发一系列连续的行为动作

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

from selenium import webdriver
    from selenium.webdriver import ActionChains
    from time import sleep
    
    bro = webdriver.Chrome(executable_path="chromedriver.exe")
    bro.get("https://www.runoob.com/try/try.php?filename=jqueryui-api-droppable")
    
    # 定位要拖动的标签
    # 定位的标签是存在于iframe的子页面中，如果直接使用find做定位，是定位不到
    # target_ele = bro.find_element_by_id("draggable")
    # print(target_ele)  # NoSuchElementException: Message: no such element: Unable to locate element
    
    # 像定位iframe中子页面中的标签必须进行如下操作
    bro.switch_to.frame('iframeResult')  # iframe的id
    target_ele = bro.find_element_by_id("draggable")
    
    # 基于动作链实现滑动操作
    action = ActionChains(bro)
    # 点击且长按
    action.click_and_hold(target_ele)
    # 慢慢滑动，每次滑动17px
    for i in range(5):
        # perform()表示立即执行动作链指定好的动作
        action.move_by_offset(17, 0).perform()
        sleep(0.5)
    
    # 释放动作链对应的资源
    action.release()
    
    sleep(4)
    
    bro.quit()

test菜鸟

![ContractedBlock.gif][] ![ExpandedBlockStart.gif][]

import aiohttp
    import asyncio
    from lxml import etree
    
    all_titles = []
    
    headers = {
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/70.0.3538.77 Safari/537.36'
    
    }
    
    
    async def request(url):
        async with aiohttp.ClientSession() as s:
            async with await s.get(url, headers=headers) as response:
                page_text = await response.text()
                return page_text
    
    
    urls = []
    url = 'http://wz.sun0769.com/index.php/question/questionType?type=4&page=%d'
    for page in range(5):
        u_page = page * 30
        new_url = format(url % u_page)
        urls.append(new_url)
    
    tasks = []
    
    
    def parse(task):
        page_text = task.result()
        page_text = page_text.encode('gb2312', 'ignore').decode('gbk')
        tree = etree.HTML(page_text)
        tr_list = tree.xpath('//*[@id="morelist"]/div/table[2]//tr/td/table//tr')
        for tr in tr_list:
            title = tr.xpath('./td[2]/a[2]/text()')[0]
            print(title)
            all_titles.append(title)
    
    
    for url in urls:
        c = request(url)
        task = asyncio.ensure_future(c)
        task.add_done_callback(parse)
        tasks.append(task)
    
    loop = asyncio.get_event_loop()
    loop.run_until_complete(asyncio.wait(tasks))

案例

转载于:https://www.cnblogs.com/biao-wu/articles/11335189.html

[ContractedBlock.gif]: https://images.cnblogs.com/OutliningIndicators/ContractedBlock.gif
[ExpandedBlockStart.gif]: /images/20230809/58d4dbca837946a6af6a2ac500169070.png