pyppeteer使用笔记

1、手动下载chromium驱动，或者在本地运行官网例子时会自动下载，但是这种方式会很慢，会默认下载安装至

C:\Users\Administrator\AppData\Local\pyppeteer\pyppeteer\local-chromium

2、以命令的方式启动浏览器，远程连接的端口为9222

chrome.exe --disable-infobars --remote-debugging-port=9222

3、启动之后通过http://localhost:9222/json得到调试信息，其中webSocketDebuggerUrl为pyppeteer连接的ws地址

import asyncio
import random

from conf import config
from db import db_utils
from pyppeteer.launcher import connect

async def init_page(url):
    '''
    初始化page对象
    :param url: 各网站登录的地址
    :return:
    '''
    # browser = await launch(headless=False, autoClose=False, args=[f'--window-size={width},{height}'])
    browser = await connect({'browserWSEndpoint':'ws://localhost:9222/devtools/page/CAAFBEE0FB710CC59580EEC57BA5AEC5'})
    page = await browser.newPage()
    await page.setUserAgent('Mozilla/5.0 (Windows NT 6.1; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) '
                            'Chrome/78.0.3904.70 Safari/537.36')
    await page.setViewport({'width': 1920, 'height': 1080})
    # 防止淘宝反爬虫机制检测到
    await page.evaluateOnNewDocument('() =>{ Object.defineProperties(navigator,{ webdriver:{ get: () => false } }) }')
    await page.goto(url)
    return page

async def login_tb():
    '''
    淘宝模拟登录
    '''
    page = await init_page('https://login.taobao.com/')
    # 检测是否已经切换到密码登录方式，没有就点击到密码登录
    to_login = await page.J('.J_Quick2Static')
    if to_login:
        if await to_login.isIntersectingViewport():
            await page.click('.J_Quick2Static')
    # 如果已经登陆过，用户名文本框会保存上次登录的用户名，防止重复输入
    user_name = await page.Jeval('#TPL_username_1','node => node.value')
    if user_name != config.tb_name:
        await page.type('#TPL_username_1', config.tb_name, {'delay': input_time_random() - 50})
    await page.type('#TPL_password_1', config.tb_pwd, {'delay': input_time_random()})
    await page.click('#J_SubmitStatic')
    cookies = await get_cookie(page)
    param_key = 'cookies_tb'
    sql = f"update sys_parameter set param_value = '{cookies}',update_time = now() where param_key = '{param_key}'"
    db_utils.commit_sql(sql)
    await page.close()

async def login_1688():
    '''
    1688模拟登录
    '''
    page = await init_page('https://login.1688.com/member/signin.htm')
    # 如果登录过了，输入登录地址会自动直接跳到1688主页
    if 'login.1688.com' in page.url:
        fr = page.frames[1]
        # 检测是否已经切换到密码登录方式，没有就点击到密码登录
        to_login = await fr.J('.J_Quick2Static')
        if to_login:
            if await to_login.isIntersectingViewport():
                await fr.click('.J_Quick2Static')
        # 如果已经登陆过，用户名文本框会保存上次登录的用户名，防止重复输入
        user_name = await fr.Jeval('#TPL_username_1', 'node => node.value')
        if user_name != config.tb_name:
            await fr.type('#TPL_username_1', config.tb_name, {'delay': input_time_random() - 50})
        await fr.type('#TPL_password_1', config.tb_pwd, {'delay': input_time_random()})
        await fr.click('#J_SubmitStatic')

    cookies = await get_cookie(page)
    param_key = 'cookies_1688'
    sql = f"update sys_parameter set param_value = '{cookies}',update_time = now() where param_key = '{param_key}'"
    db_utils.commit_sql(sql)
    # await page.close()

def input_time_random():
    '''
    获取随机数，用于模拟人工输入的速度
    :return:
    '''
    return random.randint(100, 151)

async def get_cookie(page):
    '''
    获取页面cookies
    :param page:
    :return:
    '''
    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
    return cookies

def run(source):
    '''
    根据参数来执行模拟哪个网站的登录
    :param source: tb-淘宝,1688-1688
    :return:
    '''
    if source == 'tb':
        asyncio.new_event_loop().run_until_complete(login_tb())
    elif source == '1688':
        asyncio.new_event_loop().run_until_complete(login_1688())
    else:
        return 'param error'
    return 'success'

if __name__ == '__main__':
    asyncio.get_event_loop().run_until_complete(login_1688())