【自动化系列】plawright过某音登录验证

import httpximport randomimport asyncioimport ddddocrfrom PIL import Imagefrom io import BytesIOfrom loguru import loggerfrom playwright.async_api import Page, Response, async_playwrightclass AwemeBrowser:    def __init__(self, session: str, headless: bool = False):        self.x = 0        self.y = ""        self.url1 = ""        self.url2 = ""        self.tag = False        self._session = session        self._headless = headless        self.ocr = ddddocr.DdddOcr(det=False, ocr=False, show_ad=False)    async def login(self, page: Page, username) -> None:        page.on("response", self.on_response)        await page.goto('aHR0cHM6Ly9kb3VqaWEuZG91eWluLmNvbS9sb2dpbg==')        await page.wait_for_timeout(self.wait_random)        if "扫码登录" not in await page.content():            logger.info("账号={}, 正常在线".format(username))        else:            await page.click('li:has-text("验证码登录")')            # 清空输入框            await page.evaluate('document.querySelector(".web-login-normal-input__input").value=""')            await page.evaluate('document.querySelector(".web-login-button-input__input").value=""')            # 账号密码登录            await page.wait_for_timeout(self.wait_random)            await page.type(selector=".web-login-normal-input__input", text=username, delay=100)            # await page.type(selector=".web-login-button-input__input", text=password, delay=100)            await page.wait_for_timeout(self.wait_random)            await page.click('span:has-text("获取验证码")')            await page.wait_for_timeout(self.wait_random)            num = 1            while True:                logger.warning("当前滑动次数: {}".format(num))                distance = await self.get_distance()                tracks = self.get_tracks(distance)                new_frame = page.frame_locator('#captcha_container > iframe')                slide = new_frame.locator('.captcha-slider-btn')                box = await slide.bounding_box()                await page.mouse.move(box["x"] + box["width"] / 2, box["y"] + box["height"] / 2)                await page.mouse.down()                x = box["x"] + box["width"] / 2                for track in tracks:                    x += track                    await page.mouse.move(x, box["y"])                    await asyncio.sleep(random.uniform(0.005, 0.02))                await page.mouse.up()                await page.wait_for_timeout(self.wait_random)                num = num + 1                if num > 5 or self.tag:                    break                await page.wait_for_timeout(self.wait_random)    @property    def wait_random(self):        return random.randint(1000, 3000)    async def on_response(self, response: Response):        if "https://verify.zijieapi.com/captcha/get" in response.url:            result = await response.json()            data = result.get("data")            picture = data.get("question")            self.y = picture.get("tip_y")            self.url1 = picture.get("url1")            self.url2 = picture.get("url2")        if "https://verify.zijieapi.com/captcha/verify" in response.url:            result = await response.json()            if result.get("code") == 200:                logger.success("滑块通过, res={}".format(result))                self.tag = True    async def get_distance(self):        async with httpx.AsyncClient() as session:            response_1 = await session.get(self.url1)            response_2 = await session.get(self.url2)        image = Image.open(BytesIO(response_1.content))        width, height = image.size        sub_images = []        for i in range(6):            left = (i * width) // 6            right = ((i + 1) * width) // 6            sub_image = image.crop((left, 0, right, height))            sub_images.append(sub_image)        order = [1, 5, 4, 2, 0, 3]        new_images = [None] * 6        for i in range(6):            new_images[order[i]] = sub_images[i]        merged_image = Image.new('RGB', (width, height))        for i in range(6):            left = (i * width) // 6            right = ((i + 1) * width) // 6            merged_image.paste(new_images[i], (left, 0))        # 保存图片        merged_image.save('logs/images/captcha_restored.png')        img_byte = BytesIO()        merged_image.save(img_byte, format='JPEG')        url1_content = img_byte.getvalue()        ocr_response = self.ocr.slide_match(response_2.content, url1_content)        x = ocr_response['target'][0]        logger.info(f"滑动的距离为: {x}, 高度为: {self.y}")        return int(int(x) / 550 * 340) - 5    def get_tracks(self, distance):        track = []        current = 0        mid = distance * 4 / 5        t = 0.2        v = 1        while current < distance:            if current < mid:                a = 4            else:                a = -3            v0 = v            v = v0 + a * t            move = v0 * t + 1 / 2 * a * t * t            current += move            track.append(round(move))        return track    async def start(self, username):        async with async_playwright() as pwt:            browser = await pwt.chromium.launch_persistent_context(                headless=self._headless,                user_data_dir=self._session,                args=["--start-maximized", '--disable-blink-features=AutomationControlled'],                no_viewport=True  # 窗口最大化            )            await browser.add_init_script(path="logs/js/stealth.min.js")            page = await browser.new_page()            await self.login(page=page, username=username)async def main(username):    b = AwemeBrowser(session=f"logs/session/{username}")    await b.start(username)asyncio.run(main("18888888281"))