Skip to main content

Asynchronous wrapper over Chromium browser debugger protocol.

Project description

Асинхронная обёртка над протоколом отладчика браузера Chromium.

Установка

pip install aio-dt-protocol

Имеет одну зависимость: https://github.com/aaugustin/websockets

И так как всё общение через протокол основано на формате JSON, по желанию можно установить ujson: https://github.com/ultrajson/ultrajson

Примеры:

import asyncio
from aio_dt_protocol import BrowserEx as Browser
from aio_dt_protocol import find_instances
from aio_dt_protocol.Data import KeyEvents

DEBUG_PORT: int = 9222
BROWSER_NAME: str = "chrome"


async def main() -> None:
    # ? Если на указанном порту есть запущенный браузер, происходит подключение.
    if browser_instances := find_instances(DEBUG_PORT, BROWSER_NAME):
        browser = Browser(debug_port=DEBUG_PORT, browser_pid=browser_instances[DEBUG_PORT])
        msg = f"[- CONNECT TO EXIST BROWSER ON {DEBUG_PORT} PORT -]"

    # ? Иначе, запуск нового браузера.
    else:
        browser = Browser(
            debug_port=DEBUG_PORT, browser_exe=BROWSER_NAME
        )
        msg = f"[- LAUNCH NEW BROWSER ON {DEBUG_PORT} PORT -]"
    print(msg)

    # ? Будет печатать в консоль всё, что приходит по соединению со страницей.
    # ? Полезно при разработке.
    # async def action_printer(data: dict) -> None:
    #     print(data)
    # page = await browser.GetPage(callback=action_printer)
    page = await browser.GetPage()

    print("[- GO TO GOOGLE ... -]")
    await page.Navigate("https://www.google.com")

    input_node = await page.QuerySelector("input")
    await input_node.Click()
    await asyncio.sleep(1)
    await page.action.InsertText("github PieceOfGood")
    await asyncio.sleep(1)
    await page.action.SendKeyEvent(KeyEvents.enter)
    await asyncio.sleep(1)

    submit_button_selector = "div:not([jsname])>center>[type=submit]:not([jsaction])"

    submit_button = await page.QuerySelector(submit_button_selector)
    await submit_button.Click()

    # ? Или выполнить клик используя JS
    # click_code = f"""\
    # document.querySelector("{submit_button_selector}").click();
    # """
    # await page.InjectJS(click_code)

    print("[- WAIT FOR CLOSE PAGE ... -]")
    # ? Пока соединение существует, цикл выполняется.
    await page.WaitForClose()
    print("[- DONE -]")


if __name__ == '__main__':
    asyncio.run(main())

На страницу можно легко зарегистрировать слушателей, которые будут вызываться на стороне клиентского(Python) кода. Для этого необходимо выполнить в браузере JavaScript, передав в console.info() JSON-строку из JavaScript-объекта с двумя обязательными полями func_name - имя вызываемой python-функции и args - список аргументов, которые будут ей переданы. Например:

    html = """
    <html lang="ru">
    <head>
        <meta charset="utf-8" />
        <title>Test application</title>
    </head>
    <body>
        <button id="knopka">Push me</button>
    </body>
    <script>
        const btn = document.querySelector('#knopka');
        btn.addEventListener('click', () => {
            console.info(JSON.stringify({
                 func_name: "test_func",
                 args: [1, "test"]
            }))
        });
    </script>
    </html>"""

    # ? number и text будут переданы из браузера, а bind_arg указан при регистрации
    async def test_func(number: int, text: str, bind_arg: dict) -> None:
        print(f"[- test_func -] Called with args:\n\tnumber: {number}\n\ttext: {text}\n\tbing_arg: {bind_arg}")

    await page.AddListener(
        test_func,                          # ! слушатель
        {"name": "test", "value": True}     # ! bind_arg
    )

    # ? Если ожидается внушительный функционал прикрутить к странице, то это можно
    # ? сделать за один раз.
    # await page.AddListeners(
    #     (test_func, [ {"name": "test", "value": True} ]),
    #     # (any_awaitable1, [1, 2, 3])
    #     # (any_awaitable2, [])
    # )

    await page.Navigate(html)

Headless

Чтобы запустить браузер в безголовом режиме, нужно передать аргументу принимающему путь к папке профиля пустую строку.

import asyncio
from aio_dt_protocol import BrowserEx as Browser
from aio_dt_protocol.utils import save_img_as, async_util_call


async def main() -> None:
    print("[- HEADLESS RUN -]")
    browser = Browser(profile_path="")
    print("[- WAITING PAGE -]")
    page = await browser.WaitFirstTab()
    print("[- GO TO GOOGLE -]")
    await page.Navigate("https://www.google.com")
    
    print("[- MAKE SCREENSHOT -]")
    await async_util_call(
        save_img_as, "google.png", await page.MakeScreenshot()
    )
    
    print("[- CLOSE BROWSER -]")
    await page.CloseBrowser()
    print("[- DONE -]")

if __name__ == '__main__':
    asyncio.run(main())

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

aio_dt_protocol-0.9.5.tar.gz (94.7 kB view details)

Uploaded Source

File details

Details for the file aio_dt_protocol-0.9.5.tar.gz.

File metadata

  • Download URL: aio_dt_protocol-0.9.5.tar.gz
  • Upload date:
  • Size: 94.7 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/4.0.2 CPython/3.11.3

File hashes

Hashes for aio_dt_protocol-0.9.5.tar.gz
Algorithm Hash digest
SHA256 8e64b59699b959b4e979ba5a3a06ca987650cbddf8745fa7a4e310b8b5b7bb50
MD5 c30abb149ce836b0613fdbba1ddb165a
BLAKE2b-256 04846d94eccfb0a9677ccccdb2845e94187a710a96d3f147a0aec554f7561837

See more details on using hashes here.

Supported by

AWS AWS Cloud computing and Security Sponsor Datadog Datadog Monitoring Fastly Fastly CDN Google Google Download Analytics Microsoft Microsoft PSF Sponsor Pingdom Pingdom Monitoring Sentry Sentry Error logging StatusPage StatusPage Status page