如何使用Playwright（Python）实现Dolphin Anty自动化？

我尝试使用Python结合Playwright工具自动化Dolphin Anty，但自动化流程在浏览器打开后便终止。经排查发现，需手动打开新标签并关闭初始标签才能使自动化正常运行。请问如何实现完全无人干预的自动化？

import asyncio
import requests
from playwright.async_api import async_playwright
import DolphinController


async def run(playwright):
    session = requests.session()
    auth_token = DolphinController.authentication("username", "password")
    headers = {
        "Authorization": f"Bearer {auth_token}",
        "Content-Type": "application/json"
    }
    session.headers = headers
    current_profile_id = DolphinController.create_profile_and_send_id(session)
    automation_json = DolphinController.start_automation(session, profile_id=current_profile_id)
    dolphin_port = automation_json["automation"]["port"]
    dolphin_ws_endpoint = automation_json["automation"]["wsEndpoint"]
    chromium = playwright.chromium
    browser = await chromium.connect_over_cdp(f"ws://127.0.0.1:{dolphin_port}{dolphin_ws_endpoint}")
    context = browser.contexts[0]
    await context.close()
    context = await browser.new_context()
    page = context.pages[0]
    await page.goto("http://google.com")
    # other actions...
    await browser.close()


async def main():
    async with async_playwright() as playwright:
        await run(playwright)


asyncio.run(main())

DolphinController.py（无需关注，仅用于连接） #

import json
import random
import requests

platforms_array = ["macos", "windows", "linux"]
vendors_array = ["Google Inc. (Intel)", "Google Inc. (NVIDIA)", "Google Inc. (AMD)", "Google Inc.",
                 "Google Inc. (Microsoft)", "Google Inc. (Unknown)", "Intel Inc.",
                 "Google Inc. (NVIDIA Corporation)"]
renderers_array = {"Google Inc. (Intel)": [
    "ANGLE (Intel, Intel(R) HD Graphics Family Direct3D11 vs_5_0 ps_5_0, D3D11)",
    "ANGLE (Intel, Intel(R) UHD Graphics Direct3D11 vs_5_0 ps_5_0, D3D11)"
  ],
  "Google Inc. (NVIDIA)": [
    "ANGLE (NVIDIA, NVIDIA GeForce GTX 1060 6GB Direct3D11 vs_5_0 ps_5_0, D3D11-30.0.14.7247)",
    "ANGLE (NVIDIA, NVIDIA GeForce GTX 1050 Ti Direct3D11 vs_5_0 ps_5_0, D3D11-27.21.14.6172)"
  ],
  "Google Inc. (AMD)": [
    "ANGLE (AMD, AMD Radeon(TM) Vega 10 Graphics Direct3D11 vs_5_0 ps_5_0, D3D11)",
    "ANGLE (AMD, AMD Radeon(TM) Graphics Direct3D11 vs_5_0 ps_5_0, D3D11-30.0.13044.0)"
  ],
  "Google Inc.": [
    "ANGLE (Intel(R) UHD Graphics Direct3D11 vs_5_0 ps_5_0)",
    "ANGLE (Intel(R) UHD Graphics 620 Direct3D11 vs_5_0 ps_5_0)"
  ],
  "Google Inc. (Microsoft)": [
    "ANGLE (Microsoft, Microsoft Basic Render Driver Direct3D11 vs_5_0 ps_5_0, D3D11-10.0.19041.546)",
    "ANGLE (Microsoft, Microsoft Basic Render Driver Direct3D11 vs_5_0 ps_5_0, D3D11-6.3.9600.16505)"
  ],
  "Google Inc. (Unknown)": [
    "ANGLE (Unknown, Qualcomm(R) Adreno(TM) 630 GPU Direct3D11 vs_5_0 ps_5_0, D3D11)",
    "ANGLE (Unknown, Qualcomm(R) Adreno(TM) 630 GPU Direct3D11 vs_5_0 ps_5_0, D3D11-25.18.10500.0)"
  ],
  "Intel Inc.": [
    "Intel Iris OpenGL Engine"
  ],
  "Google Inc. (NVIDIA Corporation)": [
    "ANGLE (NVIDIA Corporation, GeForce RTX 3070/PCIe/SSE2, OpenGL 4.5.0 NVIDIA 461.40)"
  ]}
cpu = [2, 4, 6, 8, 12, 16]
memory = [2, 4, 8]


def authentication(username: str, password: str):
    options = {
        "url": "https://anty-api.com/auth/login",
        "data": {
            "username": username,
            "password": password
        }
    }
    response = requests.post(options["url"], data=options["data"])
    response_json = response.json()
    # print(str(response_json))
    if "token" in response_json:
        return response_json["token"]
    else:
        raise Exception("Authentication error, check your Internet connection and config data")


def get_new_user_agent(platform: str, session: requests.Session):
    version_number = random.randint(101, 107)
    options = {
        "url": f"https://anty-api.com/fingerprints/useragent?browser_type=anty&browser_version=107&" +
               f"platform={platform}"
    }
    response = session.get(options["url"]).json()
    # print(response)
    if "data" in response:
        return response["data"]
    else:
        raise Exception("Can't get new user agent, something went wrong")


def create_profile_and_send_id(session: requests.Session):
    null = None
    # set current parameters for browser profile
    current_vendor = random.choice(vendors_array)
    current_renderer = random.choice(renderers_array[current_vendor])
    current_platform = random.choice(platforms_array)
    current_user_agent = get_new_user_agent(platform=current_platform, session=session)
    # set options for browser profile
    options = {
        "url": "https://anty-api.com/browser_profiles",
        "data": {
            "platform": current_platform,
            "browserType": "anty",
            "useragent": {
                "mode": "manual",
                "value": current_user_agent
            },
            "webrtc": {
                "mode": "altered",
                "ipAddress": null
            },
            "canvas": {
                "mode": "real"
            },
            "webgl": {
                "mode": "real"
            },
            "webglInfo": {
                "mode": "manual",
                "vendor": current_vendor,
                "renderer": current_renderer
            },
            "geolocation": {
                "mode": "real",
                "latitude": null,
                "longitude": null
            },
            "cpu": {
                "mode": "manual",
                "value": random.choice(cpu)
            },
            "memory": {
                "mode": "manual",
                "value": random.choice(memory)
            },
            "timezone": {
                "mode": "auto",
                "value": null
            },
            "locale": {
                "mode": "auto",
                "value": null
            },
            "name": "Current Bot",
            "tags": [
                ""
            ],
            "mainWebsite": "",
            "notes": {
                "content": null,
                "color": "blue",
                "style": "text",
                "icon": null
            },
            "proxy": null,
            "statusId": 0,
            "doNotTrack": False
        }
    }
    response = session.post(options["url"], data=json.dumps(options["data"]))
    # print(response.json())
    profile_id = get_last_browser_profile(session=session)
    session.close()
    return profile_id['id']


def get_last_browser_profile(session: requests.Session):
    options = {
        "url": "https://anty-api.com/browser_profiles"
    }
    response = session.get(options["url"]).json()
    if "data" in response:
        return response["data"][0]
    else:
        raise Exception("Can't get list of browser profiles")


def delete_browser_profile(session: requests.Session, profile_id: str):
    options = {
        'url': "https://anty-api.com/browser_profiles/"
    }
    response = session.delete(options["url"] + str(profile_id))
    # print(response.json())


def start_automation(session, profile_id: str):
    options = {
        "url": f"http://localhost:3001/v1.0/browser_profiles/{profile_id}/start?automation=1"
    }
    response = session.get(options['url']).json()
    return response

直接关闭初始上下文会导致Playwright与Dolphin Anty的连接异常，正确的逻辑是模拟手动操作：保留初始上下文，创建新标签页后关闭默认的初始标签页。修改后的run函数如下：

async def run(playwright):
    session = requests.session()
    auth_token = DolphinController.authentication("username", "password")
    headers = {
        "Authorization": f"Bearer {auth_token}",
        "Content-Type": "application/json"
    }
    session.headers = headers
    current_profile_id = DolphinController.create_profile_and_send_id(session)
    automation_json = DolphinController.start_automation(session, profile_id=current_profile_id)
    dolphin_port = automation_json["automation"]["port"]
    dolphin_ws_endpoint = automation_json["automation"]["wsEndpoint"]
    chromium = playwright.chromium
    browser = await chromium.connect_over_cdp(f"ws://127.0.0.1:{dolphin_port}{dolphin_ws_endpoint}")
    
    # 获取初始上下文和默认打开的页面
    context = browser.contexts[0]
    initial_page = context.pages[0]
    
    # 创建新标签页（模拟手动打开新标签）
    new_page = await context.new_page()
    
    # 关闭初始页面
    await initial_page.close()
    
    # 使用新页面执行后续操作
    await new_page.goto("http://google.com")
    # other actions...
    
    await browser.close()

关键改动说明 #

• 不再销毁初始上下文，而是直接操作上下文内的标签页
• 通过context.new_page()创建新标签页，完全模拟手动操作流程
• 关闭初始页面后，使用新页面进行后续自动化操作，保持连接稳定

内容的提问来源于stack exchange，提问作者VOLiSEE