账号密码登录
微信安全登录
微信扫描二维码登录

登录后绑定QQ、微信即可实现信息互通

手机验证码登录
找回密码返回
邮箱找回 手机找回
注册账号返回
其他登录方式
分享
  • 收藏
    X
    使用Flask调用selenium后的轮询问题
    27
    0

    本人正在写一个selenium爬虫的网页面板,用flask控制selenium的一些行为。目标网页需要扫描二维码登陆,现在卡在扫描二维码完成登陆后这个阶段,因为想不到如何怎么实现登录后进行状态轮询,并返回到Flask进而触发网页元素变化。具体代码如下:

    from flask import Flask,render_template,request,redirect,url_for
    from Control import control
    from flask import session,escape
    from flask_socketio import SocketIO
    
    app = Flask(__name__)
    app.config['SECRET_KEY'] = 'dqwer235r*tbqew4r1$1232~@'
    test = control()
    socketio = SocketIO(app)
    
    @app.route('/',methods=['GET','POST'])
    def login():
        try:
            if request.method == 'POST':
                uname = request.form['username']
                passwd = request.form['password']
                if uname == 'admin' and passwd == '123456':
                    session['username'] = 'admin'
                    return redirect(url_for('dashboard'))
    
        except Exception as e:
            return render_template('homepage.html',error=e)
    
        return render_template('homepage.html')
    
    @app.route('/dashboard/',methods=['GET','POST'])
    def dashboard():
        if request.method == 'GET'and request.headers.get("Referer") == 'http://127.0.0.1/':
                if 'username' in session:
                    if session['username'] == 'admin':
                        if 'QRstatus' in session and session['QRstatus'] == 'True':
                            if test.ck_login():
                                return render_template('dashboard.html',msg='Logined.',qrSrc=test.ck_login())
                            else:
                                session['QRstatus'] = 'False'
                        else:
                            # TODO: Add not login status
                            return render_template('dashboard.html')
                else:
                    return redirect(url_for('login'))
    
        elif request.method == 'POST':
            if request.form['submit'] == 'Start':
                if 'QRstatus' in session and session['QRstatus'] == 'True':
                    return render_template('dashboard.html',msg='QR code was printed.',qrSrc=test.ck_login())
                msg = 'Started.'
                qrSrc = test.qr()
                session['QRstatus'] = 'True'
                return render_template('dashboard.html',msg=msg,qrSrc=str(qrSrc))
            else:
                msg = 'no'
                return render_template('dashboard.html',msg=msg)
        return render_template('dashboard.html')
    
    if __name__ == '__main__':
        socketio.run(app,port=80)
    
       

    selenium的部分:

    from selenium import webdriver
    from selenium.webdriver.support.ui import WebDriverWait
    from selenium.webdriver.common.by import By
    from selenium.webdriver.support import expected_conditions as EC
    from selenium.common.exceptions import TimeoutException,StaleElementReferenceException
    
    class img_url_match(object):
        def __init__(self,locator,text_):
            self.locator = locator
            self.text = text_
    
        def __call__(self, driver):
            try:
                img_src = EC._find_element(driver,self.locator).get_attribute('src')
                return img_src.startswith(self.text)
            except StaleElementReferenceException:
                return False
    
    
    class control:
        def __init__(self):
            self.driver = webdriver.Chrome()
    
        def qr(self):
            self.driver.get('http://example.com')
            try:
                while True :
                    if not WebDriverWait(self.driver,2).until(img_url_match((By.ID,'js_login_qrcode_img'),'data:image/gif;base64')):
                        continue
                    else:
                        qr_src = self.driver.find_element_by_id('js_login_qrcode_img').get_attribute('src')
                        return qr_src
            except TimeoutException:
                self.driver.refresh()
    
        def ck_login(self):
            if self.driver.current_url == 'http://example.com':
                try:
                    url = WebDriverWait(self.driver,25).until(EC.url_changes('http://example.com'))
                    return 'data:image/gif;base64,' +self.driver.get_screenshot_as_base64()
                except TimeoutException:
                    return False
            elif self.driver.current_url == 'http://example.com/home/':
                return 'data:image/gif;base64,' + self.driver.get_screenshot_as_base64()
            elif self.driver.current_url == ':data:,':
                print('Now the url is :' + self.driver.current_url)
                WebDriverWait(self.driver,5).until(EC.presence_of_element_located((By.ID,'js-ch-member-face')))
                return 'data:image/gif;base64,' + self.driver.get_screenshot_as_base64()
    
    if __name__ == '__main__':
        test = control()
        print(test.qr())

    现在我打算用协程或者subprocess来进行selenium某些状态的轮询,但不知这样会不会启动另一个新的selenium实例,即开启一个新的浏览器?另外使用tornado会不会好些?

    谢谢大家。

    0
    打赏
    收藏
    点击回答
        全部回答
    • 0
    • 薄荷微光都很凉。 普通会员 1楼

      在使用Flask调用Selenium后,可能会遇到轮询问题。这可能是因为Flask的请求请求轮询机制导致的。在这种情况下,你可以尝试以下解决方案:

      1. 避免请求轮询:在调用Selenium之前,最好先检查请求是否已经完成。你可以使用Flask的status_code属性来检查请求的状态码。如果请求状态码不是200,那么你可能需要重新发送请求。

      2. 设置超时时间:在调用Selenium时,你可能需要设置一个超时时间。如果超时时间过短,那么请求可能会频繁地被重新发送,从而导致轮询问题。

      3. 使用代理:如果你需要发送请求到多个URL,那么你可能需要使用代理。你可以使用Flask的httphttps装饰器来实现这个功能。

      4. 使用线程池:如果你有大量的请求需要处理,那么你可能需要使用线程池来分发这些请求。你可以使用Flask的threading装饰器来实现这个功能。

      5. 使用异步编程:如果你的请求需要长时间处理,那么你可能需要使用异步编程。你可以使用Python的asyncio库来实现这个功能。

    更多回答
    扫一扫访问手机版
    • 回到顶部
    • 回到顶部