playwright打开指定的CDP端口

import asyncio

import time

import pymysql

import json

import re

import logging

from enum import Enum

from datetime import datetime

from typing import Optional

from traceback import format_exc

# 用patchright替换playwright

from patchright.async_api import async_playwright, Frame

from patchright.async_api import Error as PlaywrightError



logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')

logger = logging.getLogger('GMGN Holders Tag')





class ChallengePlatform(Enum):

    """Cloudflare challenge platform types."""



    JAVASCRIPT = "non-interactive"

    MANAGED = "managed"

    INTERACTIVE = "interactive"





class PumpRanks:

    """

    从GMGN获取热门代币的前100holders，以及其标签，以及其历史战绩

    """

    spider_name = 'gmgn_tags'



    def __init__(self):

        self._timeout = 30



    async def on_response(self, response):

        """

        拦截响应

        数据结构 gmgn.json

        """

        if not response.ok:

            return



        if '/v1/rank/sol/pump_ranks/1h' in response.url:

            logger.info(f'捕获 pump_ranks 数据接口: {response.url}')

            oridata = await response.body()

            format_data = json.loads(oridata)

            data = format_data['data']

            completeds = data['completeds']



            for c in completeds:

                logger.info(f'代币--> {c}')

                crawler_timestamp = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time()))



                address = c['address']

                symbol = c['symbol']

                usd_market_cap = c['usd_market_cap']

                created_timestamp = c['created_timestamp']

                created_timestamp = datetime.fromtimestamp(int(created_timestamp))

                holder_count = c['holder_count']

                top_10_holder_rate = c['top_10_holder_rate']

                twitter = c.get('twitter', '无')

                website = c.get('website', '无')



                # 准备要插入的数据

                data = (

                    address,

                    symbol,

                    usd_market_cap,

                    created_timestamp,

                    holder_count,

                    top_10_holder_rate,

                    twitter,

                    website,

                    crawler_timestamp

                )



                # 执行插入

                insert_sql = """

                INSERT INTO pump_token_info (

                    address, symbol, usd_market_cap, created_timestamp, 

                    holder_count, top_10_holder_rate, twitter, website, crawler_timestamp

                ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s);

                """

                # self.cursor.execute(insert_sql, data)

            # self.connection.commit()



    def _get_turnstile_frame(self, page) -> Optional[Frame]:

        """

        Get the Cloudflare turnstile frame.



        Returns

        -------

        Optional[Frame]

            The Cloudflare turnstile frame.

        """

        frame = page.frame(

            url=re.compile(

                "https://challenges.cloudflare.com/cdn-cgi/challenge-platform/h/[bg]/turnstile"

            ),

        )

        return frame



    async def cookies(self, page) -> Optional[str]:

        """The cookies from the current page."""

        cookies = await page.context.cookies()

        if not cookies:

            return None



        for cookie in cookies:

            if cookie["name"] == "cf_clearance":

                return cookie["value"]

        return None



    async def detect_challenge(self, page) -> Optional[str]:

        """

        Detect the Cloudflare challenge platform on the current page.



        Returns

        -------

        Optional[ChallengePlatform]

            The Cloudflare challenge platform.

        """

        html = await page.content()

        for platform in ChallengePlatform:

            if f"cType: '{platform.value}'" in html:

                return platform.value

        return None



    async def solve_challenge(self, page) -> None:

        """Solve the Cloudflare challenge on the current page."""

        verify_button_pattern = re.compile(

            "Verify (I am|you are) (not a bot|(a )?human)"

        )



        verify_button = page.get_by_role("button", name=verify_button_pattern)

        challenge_spinner = page.locator("#challenge-spinner")

        challenge_stage = page.locator("#challenge-stage")

        start_timestamp = datetime.now()



        cookies = await self.cookies(page)

        challenge_type = await self.detect_challenge(page)



        while (

            cookies is None

            and challenge_type is not None

            and (datetime.now() - start_timestamp).seconds < self._timeout

        ):

            if await challenge_spinner.is_visible():

                await challenge_spinner.wait_for(state="hidden")



            turnstile_frame = self._get_turnstile_frame(page)



            if await verify_button.is_visible():

                await verify_button.click()

                await challenge_stage.wait_for(state="hidden")

            elif turnstile_frame is not None:

                await page.mouse.click(210, 290)

                await challenge_stage.wait_for(state="hidden")



            await page.wait_for_timeout(250)



    async def detect(self, page):

        """

        破解CloudFlare

        """

        clearance_cookie = await self.cookies(page)

        if clearance_cookie is None:

            challenge_platform = await self.detect_challenge(page)



            if challenge_platform is None:

                logging.error("No Cloudflare challenge detected.")

                return



            logging.info(f"Solving Cloudflare challenge [{challenge_platform}]...")



            try:

                await self.solve_challenge(page)

            except PlaywrightError as err:

                logging.error(err)



    async def run_local(self, proxy=None):

        async with async_playwright() as p:

            # 必须得是有头浏览器，否则过不了Cloudflare

            launch_data = {

                "headless": False,

                "proxy": proxy,

                "args": [

                    '--disable-blink-features=AutomationControlled',

                    '--disable-dev-shm-usage',

                    '--no-first-run',

                    '--no-default-browser-check',

                    '--disable-infobars',

                    '--disable-extensions',

                    '--remote-debugging-port=9222',

                    '--disable-features=VizDisplayCompositor'

                ]

            }



            user_agent = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36"

            browser = await p.chromium.launch(**launch_data)



            context = await browser.new_context(user_agent=user_agent)

            timeout = 30

            context.set_default_timeout(timeout * 1000)

            page = await context.new_page()



            # 监听请求流

            page.on('response', self.on_response)

            url = 'https://larkfive.sg.larksuite.com/wiki/Yrh5wmnEji6h4nkjjLflj8BJgrc'



            # 访问目标地址

            await page.goto(url)

            # 过反爬，如果不加就是被block的状态

            await page.reload()

            await asyncio.sleep(10)

            await self.detect(page)



            # 1小时后关闭浏览器

            await page.evaluate("setTimeout(() => window.x = 5, 24 * 60 * 60 * 1000)")  # 1小时后设置 window.x = 5

            await page.wait_for_function("() => window.x > 0", timeout=0)



    async def run_aws(self):

        """

        在AWS服务器启动

        """

        # proxy = self.proxy

        from pyvirtualdisplay import Display

        with Display():

            try:

                await self.run_local()

            except:

                logger.error(f'浏览器异常：{format_exc()}')



    def task(self):

        if env == 'local':

            asyncio.run(self.run_local())

        else:

            asyncio.run(self.run_aws())



    def run(self):

        while True:

            self.task()

            logger.info('浏览器等待下一次启动')

            time.sleep(60)





if __name__ == '__main__':

    env = 'local'

    PumpRanks().run()