一种基于pyppeteer的新浪微博登录方式

微博搜索采集时，默认情况下只显示当前页数据。如果搜索的关键词是热词，当前页数据的时间范围可能只有三五分钟。所以，如果要把数据采集全，则必须登录。

在大批量采集时，必须使用账号构建cookie池，并根据cookie有效期实时更新已过期的cookie，下面主要实现基于Pyppeteer的微博登录，供大家参考。

新浪微博登录DEMO主类：

import asyncio, time
from com.fy.plugs.browser.pyppeteer.PyppeteerBrowser import PyppeteerBrowser
from com.fy.utils.date.DateUtils import Date_Utils
class WeiBoLogin:
    def __init__(self):
        self.pb=PyppeteerBrowser()
        self.du=Date_Utils()

    def login(self):
        url="https://passport.weibo.cn/signin/login?entry=mweibo&res=wel&wm=3349&r=https%3A%2F%2Fm.weibo.cn%2F"
        userDataDir="d://pyppeteer" + str(self.du.getCurrentTimeStr_Year())
        asyncio.get_event_loop() .run_until_complete(self.pb.getbrowser(False, userDataDir))
        asyncio.get_event_loop() .run_until_complete(self.pb.open(url, 60))
        time.sleep(10)
        asyncio.get_event_loop() .run_until_complete(self.pb.inputKw(None, "#loginName", "用户名"))
        time.sleep(1) 
        asyncio.get_event_loop() .run_until_complete(self.pb.inputKw(None, "#loginPassword", "密码"))
        time.sleep(1)
        eles=asyncio.get_event_loop() .run_until_complete(self.pb.getElementsByXpaths(None, '//*[@id="loginAction"]'))
        asyncio.get_event_loop() .run_until_complete(self.pb.clickByEle(eles[0]))
        time.sleep(100)

if __name__=='__main__':
    sbl=WeiBoLogin()
    sbl.login()

Pyppeteer公共类：

import asyncio, tkinter, traceback 
import time
from pyppeteer import launch
from com.fy.utils.http.UserAgentUtils import UserAgentUtils
from com.fy.utils.hash.HashUtils import Hash_Utils
from com.fy.utils.file.FileUtils import File_Utils 
class PyppeteerBrowser:
    def __init__(self):
        self.hash=Hash_Utils()
        self.url=None
        self.ua=UserAgentUtils()
    
    #"""使用tkinter获取屏幕大小"""）
    def screen_size(self):
        tk=tkinter.Tk()
        width=tk.winfo_screenwidth()
        height=tk.winfo_screenheight()
        tk.quit()
        return width, height

    async def getbrowser(self, headless=False, userDataDir=None):
       
        args=[ "--start-maximized", '--no-sandbox', "--disable-infobars" , "--log-level=3"]
        parameters={}
        if userDataDir==None:
            parameters={'headless': headless, #是否打开浏览器；False:打开浏览器；True：进程中运行；
                                       'args': args,
                                        'dumpio': True  #'dumpio': True：解决chromium浏览器多开页面卡死问题。
                                        }
        else:
            parameters={'headless': headless, #是否打开浏览器；False:打开浏览器；True：进程中运行；
                                       'args': args,
                                       "userDataDir": userDataDir,
                                        'dumpio': True   #'dumpio': True：解决chromium浏览器多开页面卡死问题。
                                        }
        #注意：同一个用户目录（userDataDir）不能被两个chrome进程使用，如果你要多开，记得分别指定用户目录。否则会报编码错误。
        self.browser=await launch(parameters)

        self.page=await self.browser.newPage()#在此浏览器上创建新页面并返回其对象。
        
        width, height=self.screen_size()
        # 设置网页可视区域大小
        await self.page.setViewport({
            "width": width,
            "height": height
        })

        # 是否启用JS，enabled设为False，则无渲染效果
        await self.page.setJavaScriptEnabled(enabled=True)

        #设置请求头userAgent
        await self.page.setUserAgent(self.ua.getheaders())
        
        await self.preventCheckWebdriver(self.page)
        print("构造浏览器对象完毕....", self.page)
        
    #获取当前操作的界面  
    async def getPage(self): 
        return  self.page

    #获取当前page对象的链接；
    async def getCurUrl(self, page):
        if page==None:
            page=self.page 
        return  await page.url

    #打开一个新的界面；）
    async def getnewpage(self): 
        return  await self.browser.newPage()

    #获取当前操作的界面重新加载    
    async def reload(self): 
        await self.page.reload()   

    #当前操作界面返回    
    async def goBack(self): 
        await self.page.goBack() 

    #获取当前操作的界面的URL   
    async def getPageUrl(self): 
        await self.page.url() 

    #打开连接；
    async def open(self, url, timeout=60):
        try:
            if url==None:
                print("当前传入的【url】不能为空，参数错误！！")
            self.url=url
            print("打开网页：" + (url))      
            self.res=await self.page.goto(url, options={'timeout':int(timeout * 1000)})#打开连接；
            await asyncio.sleep(1)#强行等待3秒
            status=await self.res.status
            curUrl=await self.page.url
            await self.preventCheckWebdriver(self.page)
            return  status, curUrl
        except:return  404, None

async def preventCheckWebdriver(self, page):
        if page==None:
            page=self.page
        await page.evaluate('''()=>{ Object.defineProperties(navigator,{ webdriver:{ get: ()=> undefined } }) }''')  # 以下为插入中间js，将淘宝会为了检测浏览器而调用的js修改其结果。
        await page.evaluate('''()=>{ window.navigator.chrome={ runtime: {},  }; }''')
        await page.evaluate('''()=>{ Object.defineProperty(navigator, 'languages', { get: ()=> ['en-US', 'en'] }); }''')
        await page.evaluate('''()=>{ Object.defineProperty(navigator, 'plugins', { get: ()=> [1, 2, 3, 4, 5,6], }); }''')

    async def closeBrowser(self, browser):
        if browser==None:
            browser=self.browser
        try:
            await browser.close()
        except:pass

    async def closePage(self, page):
        if page==None:
            page=self.page
        await page.close()
        
    async def closeNumPage(self, number:"号码从0开始"):
        pages=await self.browser.pages()
        await pages[number].close()
        return True
        
    async def retainLastPage(self):
        pages=await self.browser.pages()
        num=0
        for  page in pages:
            if num !=(len(pages) - 1):
                await page.close()
            else:
                self.page=page
            num +=1
        
    async def gerReponseStatus(self):
        try:return await self.res.status  # 响应状态
        except:return 200

    async def screenshot(self, page):
        hashCode=self.hash.getMd5Hash(self.url)
        if page==None:
            page=self.page
        await page.screenshot({'path': './screenshots/' + str(hashCode) + '.png'})

    async def getHeader(self):
        return await self.res.headers  # 响应头；

    async def scrollToButtom(self, page):
        if page==None:
            page=self.page
        await page.evaluate('window.scrollBy(0, document.body.scrollHeight)')

    async def getCookies(self, page):
        if page==None:
            page=self.page
        return await page.cookies()

    async def getCookieStr(page):
        if page==None:
            page=self.page
        cookies_list=await page.cookies()
        cookies=''
        for cookie in cookies_list:
            str_cookie='{0}={1};'
            str_cookie=str_cookie.format(cookie.get('name'), cookie.get('value'))
            cookies +=str_cookie
        try:print(cookies)
        except:pass
        return cookies
    
    async def setCookies(self, page, cookies):
        if page==None:
            page=self.page
        return await page.setCookie(*cookies)
    
    async def getHtml(self, page):
        if page==None:
            page=self.page
        return (await page.content())

    async def getCurPageTitle(self, page):
        if page==None:
            page=self.page
        return (await page.title())

    async def getElementFieldValue(self, page, element, field):
        if element==None:
            print("当前传入的【element】不能为空，参数错误！！")
            return None
        if field==None:
            print("当前传入的【field】不能为空，参数错误！！")
            return None
        if page==None:
            page=self.page
        if str(type(element))=="<class 'list'>":
            print("当前传入的【element】不是单个对象，为list集合，参数错误！！")
            return None
        fieldValue=(await element.getProperty(field)).jsonValue()
        return  fieldValue

    async def getPageWidthHight(self, page):
        if page==None:
            page=self.page
        return  await page.evaluate('''()=> {
                    return {
                        width: document.documentElement.clientWidth,
                        height: document.documentElement.clientHeight,
                        deviceScaleFactor: window.devicePixelRatio,
                    }
                }''')

    async def getCurBrowserAllPages(self):
        return await self.browser.pages()

    async def getElementsByXpaths(self, page, xpath:'如：//div[@class="title-box"]/a'):
        if xpath==None:
            print("当前传入的【xpath】不能为空，参数错误！！")
            return None
        if page==None:
            page=self.page
        try:elemList=await page.xpath(xpath)
        except:
            print("获取xpath路径为【" + str(xpath) + "】的标签对象异常...")
        return elemList#返回类型为：list集合；

    async def getPageText(self, page):
        if page==None:
            page=self.page
        '''Pyppeteer的evaluate()方法只使用JavaScript字符串，该字符串可以是函数也可以是表达式，
           Pyppeteer会进行自动判断。但有时会判断错误，如果字符串被判断成了函数，并且报错，
             可以添加选项force_expr=True，强制Pyppeteer作为表达式处理。'''
        return await page.evaluate('document.body.textContent', force_expr=True)
        
    async def getElementText(self, page, element):
        if element==None:
            print("当前传入的【element】不能为空，参数错误！！")
            return None
        if page==None:
            page=self.page
        if str(type(element))=="<class 'list'>":
            print("当前传入的【element】不是单个对象，为list集合，参数错误！！")
            return None
        return  await page.evaluate('(element)=> element.textContent', element)
    
    async def getElementBySelector(self, page , selector):
        if selector==None:
            print("当前传入的【selector】不能为空，参数错误！！")
            return None
        if page==None:
            page=self.page
        return  await page.querySelector(selector)
    
    async def inputKw(self, page, selector:"如：'input#kw.s_ipt':获取input标签中id='kw',class='s_ipt'的对象。不可用xpath路径", kw:'待输入的关键词'):
        if kw==None:
            return None
        if selector==None:
            return None
        if page==None:
            page=self.page
        try:print(selector, kw)
        except:pass
        await page.type(selector, kw)
        return None

    async def clickElement(self, page, selector:"如：'input#kw.s_ipt':获取input标签中id='kw',class='s_ipt'的对象。。不可用xpath路径"):
        if selector==None:
            print("当前传入的【selector】不能为空，参数错误！！")
        if page==None:
            page=self.page
        await page.click(selector)#如果selector获取的对象是list集合，则执行第一个元素的点击；

    async def removeInputValue(self, page, idValue):
        if idValue==None:
            print("当前传入的【idValue】不能为空，参数错误！！")
        if page==None:
            page=self.page
        await page.evaluate("document.querySelector('#" + str(idValue) + "').value=''")
        print("清空【" + str(idValue) + "】的内容")

    async def clickByEle(self, ele):
        if ele==None:
            return
        return await ele.click()
            
    async def getLastPage(self):
        pages=await self.browser.pages()
        return pages[-1]
            
    async def getPageTotal(self):
        pages=await self.browser.pages()
        return len(pages)
            
    async def getFirstPage(self):
        pages=await self.browser.pages()
        return pages[0]
            
    async def getAllFrames(self, page):
        if page==None:
            page=self.page
        return  await page.frames

    async def getScreenshotByEle(self, page, ele, screenshotFilePath:"目前测试只有.png图片可正常生成，jpg异常；"):
        picture=''
        try:
            fu=File_Utils(None)
            fu=File_Utils(fu.getParentDir(screenshotFilePath))
            if not fu.exists(fu.getParentDir(screenshotFilePath)):fu.makeDirs()#如果图片的保存目录不存在，则创建；
            time.sleep(3)
            try:
                for _ in range(6):
                    clip=await ele.boundingBox()
                    picture=base64.b64encode(await page.screenshot({
                        'path': screenshotFilePath, # 图片路径, 不指定就不保存
                        'clip': clip # 指定图片位置,大小
                    }))
                    if picture !='':
                        break
            except Exception as e:
                print(traceback.print_exc())
        except Exception as e:
            print(traceback.print_exc())
        return picture

注意事项：

测试过程中发现，基于PC端的登录界面，在Pyppeteer浏览器中，登录按钮无法使用。但是手机端登录界面可以正常登录

午已经说完了CSS文本样式，接着说下CSS的引入方式，包含内部样式表(嵌入式)，行内样式表(行内式)和外部样式表(链接式)。

内部样式表(内嵌样式表)是写道HTML页面内部，是将所有的CSS代码抽取出来，单独放到一个<style>?标签中。

以前的语法展示:

<style>

? ?div {

? ?color:pink;

font-size:12px;

</style>

将所有的样式，都放到<style>标签中

展示如下:

对应的代码如下:

<!DOCTYPE html>

<head>

<title>CSS引入方式-内部样式表</title>

<style>

div {

/* 文本颜色属性 */

color: blue;

/* 文本行间距 */

line-height: 32px;

/* 让文本在中间展示 */

text-align: center;

}

p {

/* 段落首行收缩2个字 */

text-indent: 2em;

}

a {

/* 文本不展示下划线 */

text-decoration: none;

}

</style>

</head>

<body>

<p>10月15日晚，甘肃省天水市妇女联合会官方微博发布通报称，经过多方劝解，孩子父亲毛某已于10月14日将孩子送到母亲范某身边。目前孩子健康状况良好，范某情绪稳定，毛某也认识到因自己一时冲动造成的严重后果，并保证不再有过激行为。</p>

<div>一片祥云，我自横刀向天笑</div>

文本不展示下划线<br/>

<a href="www.baidu.com">跳转到百度页面</a>

</body>

</html>

我向您讲授了如何向 html 页面添加 JavaScript，使得网站的动态性和交互性更强。

你已经学习了如何创建对事件的响应，验证表单，以及如何根据不同的情况运行不同的脚本。

你也学到了如何创建和使用对象，以及如何使用 JavaScript 的内置对象。

如需更多关于 JavaScript 的信息和知识，请参阅我的 JavaScript 实例和 JavaScript 参考手册。

现在已经你已经学习了 JavaScript，接下来该学习什么呢？

下一步应该学习 HTML DOM 和 DHTML。

如果你希望学习关于服务器端脚本的知识，那么下一步应该学习 ASP,PHP, .Net。

HTML DOM

HTML DOM 定义了访问和操作 HTML 文档的标准方法。 HTML DOM 独立于平台和语言，可被任何编程语言使用，比如 Java、JavaScript 和 VBscript。

jQuery

jQuery 是一个 JavaScript 库。

jQuery 极大地简化了 JavaScript 编程。

jQuery 很容易学习。

AJAX

AJAX=异步 JavaScript 和 XML。

AJAX 不是一种新的编程语言，而是一种使用现有标准的新方法。

通过与服务器进行数据交换，AJAX 可以在不重新加载整个网页的情况下，对网页的某部分进行更新。

有很多使用 AJAX 的应用程序案例：新浪微博、Google 地图、开心网等等。

ASP / PHP / .NET

和 HTML 文档中的脚本运行于客户端（浏览器）不同，ASP/PHP 文件中的脚本在服务器上运行。

使用 ASP，你可以动态地编辑、改变或者添加网站内容，对由 HTML 表单提交而来的数据进行响应，访问数据或者数据库并向浏览器返回结果，或者定制对不同的用户来说更有帮助的网页。

由于 ASP/PHP 文件返回的是纯粹的 HTML，因此可显示在任何浏览器中。

如您还有不明白的可以在下面与我留言或是与我探讨QQ群308855039，我们一起飞！

在线咨询

上一篇：前端10大开源拖拽排序库汇总，让搭建，更简单
下一篇：BSN开发者手册和帮助手册发布在线版

您的项目需求

*请认真填写需求信息，我们会在24小时内与您取得联系。

整合营销服务商

一种基于pyppeteer的新浪微博登录方式

新浪微博登录DEMO主类：

Pyppeteer公共类：

您的项目需求