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 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)
    # conn = await browser.getPage(callback=action_printer)
    conn = await browser.getPage()

    print("[- GO TO GOOGLE ... -]")
    await conn.Page.navigate("https://www.google.com", )
    print("[- EMULATE INPUT TEXT ... -]")

    input_node = await conn.DOM.querySelector("[type=search]")
    await input_node.click()
    await asyncio.sleep(1)
    await conn.Input.insertText("github PieceOfGood")
    await asyncio.sleep(1)

    await conn.extend.action.sendKeyEvent(KeyEvents.enter)
    await asyncio.sleep(1)

    # submit_button_selector = "div:not([jsname])>center>[type=submit]:not([jsaction])"
    # submit_button = await conn.DOM.querySelector(submit_button_selector)
    # await submit_button.click()

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

    print("[- WAIT FOR CLOSE PAGE ... -]")
    # ? Пока соединение существует, цикл выполняется.
    await conn.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 conn.addListener(
    test_func,  # ! слушатель
    {"name": "test", "value": True}  # ! bind_arg
)

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

await conn.Page.navigate(html)

Headless

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

import asyncio
from aio_dt_protocol import 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 -]")
    conn = await browser.waitFirstTab()
    print("[- GO TO GOOGLE -]")
    await conn.Page.navigate("https://www.google.com")

    print("[- MAKE SCREENSHOT -]")
    await async_util_call(
        save_img_as, "google.png", await conn.Page.makeScreenshot()
    )

    print("[- CLOSE BROWSER -]")
    await conn.Browser.close()
    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-1.0.2.tar.gz (95.6 kB view details)

Uploaded Source

File details

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

File metadata

  • Download URL: aio_dt_protocol-1.0.2.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-1.0.2.tar.gz
Algorithm Hash digest
SHA256 2034b4c959d53cecacb08b85aea3f2336cf62ad6101722e0166b60f400275c22
MD5 a15d6ed7172066e12af109eba4326499
BLAKE2b-256 3d8d0da99280d7353de945ee8dd56e53276e27fd4b2d926e33d5e616dcd54932

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