Source code for langchain_community.tools.playwright.utils

"""用于Playwright浏览器工具的实用程序。"""
from __future__ import annotations

import asyncio
from typing import TYPE_CHECKING, Any, Coroutine, List, Optional, TypeVar

if TYPE_CHECKING:
    from playwright.async_api import Browser as AsyncBrowser
    from playwright.async_api import Page as AsyncPage
    from playwright.sync_api import Browser as SyncBrowser
    from playwright.sync_api import Page as SyncPage


async def aget_current_page(browser: AsyncBrowser) -> AsyncPage:
    """异步获取浏览器的当前页面。

参数：
    browser: 要获取当前页面的浏览器（AsyncBrowser）。

返回：
    AsyncPage: 当前页面。
"""
    if not browser.contexts:
        context = await browser.new_context()
        return await context.new_page()
    context = browser.contexts[0]  # Assuming you're using the default browser context
    if not context.pages:
        return await context.new_page()
    # Assuming the last page in the list is the active one
    return context.pages[-1]


def get_current_page(browser: SyncBrowser) -> SyncPage:
    """获取浏览器的当前页面。
参数：
    browser：要获取当前页面的浏览器。

返回：
    SyncPage：当前页面。
"""
    if not browser.contexts:
        context = browser.new_context()
        return context.new_page()
    context = browser.contexts[0]  # Assuming you're using the default browser context
    if not context.pages:
        return context.new_page()
    # Assuming the last page in the list is the active one
    return context.pages[-1]


def create_async_playwright_browser(
    headless: bool = True, args: Optional[List[str]] = None
) -> AsyncBrowser:
    """创建一个异步的Playwright浏览器。

参数：
    headless：是否在无头模式下运行浏览器。默认为True。
    args：传递给browser.chromium.launch的参数。

返回：
    AsyncBrowser：Playwright浏览器。
"""
    from playwright.async_api import async_playwright

    browser = run_async(async_playwright().start())
    return run_async(browser.chromium.launch(headless=headless, args=args))


def create_sync_playwright_browser(
    headless: bool = True, args: Optional[List[str]] = None
) -> SyncBrowser:
    """创建一个playwright浏览器。

参数：
    headless: 是否在无头模式下运行浏览器。默认为True。
    args: 传递给browser.chromium.launch的参数。

返回：
    SyncBrowser: playwright浏览器。
"""
    from playwright.sync_api import sync_playwright

    browser = sync_playwright().start()
    return browser.chromium.launch(headless=headless, args=args)


T = TypeVar("T")


def run_async(coro: Coroutine[Any, Any, T]) -> T:
    """运行一个异步协程。

参数:
    coro: 要运行的协程。Coroutine[Any, Any, T]

返回:
    T: 协程的结果。
"""
    event_loop = asyncio.get_event_loop()
    return event_loop.run_until_complete(coro)