鲁达 在微博搜索采集时，默认情况下只显示当前页数据。如果搜索的关键词是热词，当前页数据的时间范围可能只有三五分钟。所以，如果要把数据采集全，则必须登录。
在大批量采集时，必须使用账号构建cookie池，并根据cookie有效期实时更新已过期的cookie，下面主要实现基于Pyppeteer的微博登录，供大家参考。
新浪微博登录DEMO主类：
import asyncio, time from com. import PyppeteerBrowser from com. import Date_Utils class WeiBoLogin: def __init__(self): = PyppeteerBrowser() = Date_Utils() def login(self): url = ";res=wel&wm=3349&r=https%3A%2F%2Fm.weibo.cn%2F" userDataDir = "d://pyppeteer" + str(.getCurrentTimeStr_Year()) a() .run_until_complete(.getbrowser(False, userDataDir)) a() .run_until_complete(.open(url, 60)) (10) a() .run_until_complete(.inputKw(None, "#loginName", "用户名")) (1) a() .run_until_complete(.inputKw(None, "#loginPassword", "密码")) (1) eles = a() .run_until_complete(.getElementsByXpaths(None, '//*[@id="loginAction"]')) a() .run_until_complete(.clickByEle(eles[0])) (100) if __name__ == '__main__': sbl = WeiBoLogin() () Pyppeteer公共类：
import asyncio, tkinter, traceback import time from pyppeteer import launch from com. import UserAgentUtils from com. import Hash_Utils from com. import File_Utils class PyppeteerBrowser: def __init__(self): = Hash_Utils() = None = UserAgentUtils() #"""使用tkinter获取屏幕大小"""） def screen_size(self): tk = () width = () height = () () 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进程使用，如果你要多开，记得分别指定用户目录。否则会报编码错误。 = await launch(parameters) = await .newPage()#在此浏览器上创建新页面并返回其对象。 width, height = () # 设置网页可视区域大小 await .setViewport({ "width": width, "height": height }) # 是否启用JS，enabled设为False，则无渲染效果 await .setJavaScriptEnabled(enabled=True) #设置请求头userAgent await .setUserAgent(.getheaders()) await () print("构造浏览器对象完毕....", ) #获取当前操作的界面 async def getPage(self): return #获取当前page对象的链接； async def getCurUrl(self, page): if page == None: page = return await #打开一个新的界面；） async def getnewpage(self): return await .newPage() #获取当前操作的界面重新加载 async def reload(self): await .reload() #当前操作界面返回 async def goBack(self): await .goBack() #获取当前操作的界面的URL async def getPageUrl(self): await .url() #打开连接； async def open(self, url, timeout=60): try: if url == None: print("当前传入的【url】不能为空，参数错误！！") = url print("打开网页：" + (url)) = await .goto(url, options={'timeout':int(timeout * 1000)})#打开连接； await a(1)#强行等待3秒 status = await .status curUrl = await .url await () return status, curUrl except:return 404, None async def preventCheckWebdriver(self, page): if page == None: page = await ('''() =>{ Object.defineProperties(navigator,{ webdriver:{ get: () => undefined } }) }''') # 以下为插入中间js，将淘宝会为了检测浏览器而调用的js修改其结果。 await ('''() =>{ window.naviga = { runtime: {}, }; }''') await ('''() =>{ Object.defineProperty(navigator, 'languages', { get: () => ['en-US', 'en'] }); }''') await ('''() =>{ Object.defineProperty(navigator, 'plugins', { get: () => [1, 2, 3, 4, 5,6], }); }''') async def closeBrowser(self, browser): if browser == None: browser = try: await brow() except:pass async def closePage(self, page): if page == None: page = await () async def closeNumPage(self, number:"号码从0开始"): pages = await .pages() await pages[number].close() return True async def retainLastPage(self): pages = await .pages() num = 0 for page in pages: if num != (len(pages) - 1): await () else: = page num += 1 async def gerReponseStatus(self): try:return await .status # 响应状态 except:return 200 async def screenshot(self, page): hashCode = .getMd5Hash() if page == None: page = await ({'path': './screenshots/' + str(hashCode) + '.png'}) async def getHeader(self): return await .headers # 响应头； async def scrollToButtom(self, page): if page == None: page = await ('window.scrollBy(0, document.body.scrollHeight)') async def getCookies(self, page): if page == None: page = return await () async def getCookieStr(page): if page == None: page = cookies_list = await () cookies = '' for cookie in cookies_list: str_cookie = '{0}={1};' str_cookie = ('name'), cookie.get('value')) cookies += str_cookie try:print(cookies) except:pass return cookies async def setCookies(self, page, cookies): if page == None: page = return await (*cookies) async def getHtml(self, page): if page == None: page = return (await ()) async def getCurPageTitle(self, page): if page == None: page = return (await ()) 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 = 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 = return await ('''() => { return { width: document.documen, height: document.documen, deviceScaleFactor: window.devicePixelRatio, } }''') async def getCurBrowserAllPages(self): return await .pages() async def getElementsByXpaths(self, page, xpath:'如：//div[@class="title-box"]/a'): if xpath == None: print("当前传入的【xpath】不能为空，参数错误！！") return None if page == None: page = try:elemList = await (xpath) except: print("获取xpath路径为【" + str(xpath) + "】的标签对象异常...") return elemList#返回类型为：list集合； async def getPageText(self, page): if page == None: page = '''Pyppeteer的evaluate()方法只使用JavaScript字符串，该字符串可以是函数也可以是表达式， Pyppeteer会进行自动判断。但有时会判断错误，如果字符串被判断成了函数，并且报错， 可以添加选项force_expr=True，强制Pyppeteer作为表达式处理。''' return await ('document.body.textContent', force_expr=True) async def getElementText(self, page, element): if element == None: print("当前传入的【element】不能为空，参数错误！！") return None if page == None: page = if str(type(element)) == "<class 'list'>": print("当前传入的【element】不是单个对象，为list集合，参数错误！！") return None return await ('(element) => element.textContent', element) async def getElementBySelector(self, page , selector): if selector == None: print("当前传入的【selector】不能为空，参数错误！！") return None if page == None: page = return await (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 = try:print(selector, kw) except:pass await (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 = await (selector)#如果selector获取的对象是list集合，则执行第一个元素的点击； async def removeInputValue(self, page, idValue): if idValue == None: print("当前传入的【idValue】不能为空，参数错误！！") if page == None: page = await ("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 .pages() return pages[-1] async def getPageTotal(self): pages = await .pages() return len(pages) async def getFirstPage(self): pages = await .pages() return pages[0] async def getAllFrames(self, page): if page == None: page = return await async def getScreenshotByEle(self, page, ele, screenshotFilePath:"目前测试只有.png图片可正常生成，jpg异常；"): picture = '' try: fu = File_Utils(None) fu = File_Util(screenshotFilePath)) if not fu.exist(screenshotFilePath)):fu.makeDirs()#如果图片的保存目录不存在，则创建； (3) try: for _ in range(6): clip = await ele.boundingBox() picture = ba(await ({ 'path': screenshotFilePath, # 图片路径, 不指定就不保存 'clip': clip # 指定图片位置,大小 })) if picture != '': break except Exception as e: prin()) except Exception as e: prin()) return picture注意事项：
测试过程中发现，基于PC端的登录界面，在Pyppeteer浏览器中，登录按钮无法使用。但是手机端登录界面可以正常登录