云迈博客

您现在的位置是:首页 > 后端开发 > Python > 正文

Python

[实战篇]Python+Selenium模拟浏览器登录网易邮箱

wsinbol2020-12-17Python450
运行环境操作系统:Windows语言环境:Python浏览器:Chrome浏览器驱动:Selenium设计思路从全局角度考虑,多线程的方式肯定是更快的、效率更高的,但是受限于IP池的质量、尽可能使

运行环境

操作系统:Windows
语言环境:Python
浏览器:Chrome
浏览器驱动:Selenium

设计思路

从全局角度考虑,多线程的方式肯定是更快的、效率更高的,但是受限于IP池的质量、尽可能使用可靠的IP、避免浏览器频繁关闭开启等因素,最终选择单线程的方式。多线程的方式等作者了解更透彻些再行补上吧。

那么,在实现整个需求的过程中,就必须满足以下几点:

  1. 1.账户登录失败就要重新发起请求,前一个账户未登录成功不执行后面账户的登录;
  2. 2.尽可能使用可靠的IP,前一个账户登录成功后该IP可以继续给后续的账户使用,直到该IP不可用;
  3. 3.登录成功后需要保持窗口不自动关闭,登录失败的窗口要自动关闭

编码过程

程序本身难度不大,但是还是花了点时间的。其中犯了个严重的错误就是代理IP的继承问题,想要下一次的请求使用上一次的IP,那么代理IP就必须用一个全局变量保存起来,但我在赋值的过程中将

browser.close()
self.proxy = proxy
self.login(email, password)

这3行代码的顺序写错了,增加了调试的时间。

代码实例

from multiprocessing import Process
from multiprocessing import Pool
import os, time, random
from selenium import webdriver
from time import sleep
import sys
import re

class Netyi:

    def __init__(self, use_proxy=False,proxy_list=[]):
        if use_proxy:
            headers = {
                'User-Agent':'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/80.0.3987.163 Safari/537.36'
            }

            proxy_str = requests.get('爬虫代理地址',headers=headers)

            if proxy_str.status_code == 200:
                proxy_list = proxy_str.text.split(' ')
                if len(proxy_list) < 10:
                    print('请求IP池失败')
                    exit()
                else:
                    print('请求IP池成功')
            else:
                print('请求IP池失败')
                exit()

        self.proxy_list = proxy_list
        self.use_proxy = use_proxy
        self.proxy = None

    def read(self):
        # p = Pool(4)
        with open('./netyi.txt') as f:
            line = f.readline()
            while line:
                email,password = line.split('----')
                account = re.findall(r'(.*?)@163.com', email)[0]
                # p.apply_async(self.login, args=(email,password))
                self.login(email,password)
                line = f.readline()

        # p.close()
        # p.join()


    def login(self, email, password):

        print('-'*100)
        print('当前正在登陆邮箱:',email)
        print('-'*100)

        option = webdriver.ChromeOptions()
        option.add_experimental_option("detach",True)

        if self.use_proxy:
            print('-'*100)
            print('last IP:',self.proxy)
            print('-'*100)

            if self.proxy is None:
                proxy = self.proxy_list.pop()
            else:
                print('*'*100)
                print('使用了上次的IP')
                print('*'*100)

                proxy = self.proxy

            print('-'*100)
            print('当前IP:',proxy)
            print('-'*100)

            option.add_argument('--proxy-server=http://' + proxy)

        try:

            if self.use_proxy:
                self.proxy = proxy

            browser = webdriver.Chrome(chrome_options=option)
            browser.get("https://mail.163.com/")
            browser.implicitly_wait(3)

            browser.switch_to.frame(browser.find_element_by_xpath('/html/body/div[2]/div[3]/div/div[3]/div[4]/div[1]/div[1]/iframe'))

            browser.find_element_by_xpath('/html/body/div[2]/div[2]/div[2]/form/div/div[1]/div[2]/input').send_keys(email)
            browser.find_element_by_xpath('/html/body/div[2]/div[2]/div[2]/form/div/div[3]/div[2]/input[2]').send_keys(password)
            browser.find_element_by_xpath('/html/body/div[2]/div[2]/div[2]/form/div/div[8]/a').click()

        except:
            browser.close()
            if self.use_proxy:
                self.proxy = None
            # 这行代码一定要写在最后面
            self.login(email, password)

if __name__=='__main__':
    flag = input('是否使用代理模式?是回复1,否回复2,默认为否。请选择:')
    if flag == '1':
        use_proxy = True
    else:
        use_proxy = False
    N = Netyi(use_proxy=use_proxy)
    N.read()

注意事项

1.对于邮箱的操作未结束前不可关闭 Powershell(命令行) 窗口,否则无论是登录成功的还是未登录的浏览器窗口都会关闭;

2.不要关闭正在执行的浏览器窗口,已经登录成功的窗口可以操作

多进程版本

在写单进程的过程中还是尝试着写了下多进程版本,最突出的问题就是IP池的请求问题,通过 apply_async 方法下发任务后,每一次执行都会重新请求IP池。这个有点不能接受,抽空看下 Scipy 源码看他是如何提高IP利用率的吧。而不使用代理情况下,用本地IP去请求执行效率还是非常可观的。(完)

发表评论

评论列表

  • 这篇文章还没有收到评论,赶紧来抢沙发吧~