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.3.tar.gz (95.6 kB view details)

Uploaded Source

File details

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

File metadata

  • Download URL: aio_dt_protocol-0.9.3.tar.gz
  • Upload date:
  • Size: 95.6 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.3.tar.gz
Algorithm Hash digest
SHA256 ede49b27214bb8540e289b02b65dd26352de56fb7dbabfcc1d055ca42a134449
MD5 cfe57e5f17cf47c475bbe7d379ac265d
BLAKE2b-256 75bfa2c53d5c1b60ffd221f159d2fc95e824e3fbd78406f80ad3a2ed3f8a0d68

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