util

T = typing.TypeVar('T') module-attribute

__registered__instances__: Set[Browser] = set() module-attribute

logger = logging.getLogger(__name__) module-attribute

cdp_get_module(domain)

get cdp module by given string

Parameters:

Name	Type	Description	Default
domain	Union[str, ModuleType]		required

Returns:

Type	Description

Source code in zendriver/core/util.py

def cdp_get_module(domain: Union[str, types.ModuleType]) -> Any:
    """
    get cdp module by given string

    :param domain:
    :return:
    :rtype:
    """
    import importlib

    if isinstance(domain, types.ModuleType):
        # you get what you ask for
        domain_mod = domain
    else:
        try:
            if domain in ("input",):
                domain = "input_"

            #  fallback if someone passes a str
            domain_mod = getattr(cdp, domain)
            if not domain_mod:
                raise AttributeError
        except AttributeError:
            try:
                domain_mod = importlib.import_module(domain)
            except ModuleNotFoundError:
                raise ModuleNotFoundError(
                    "could not find cdp module from input '%s'" % domain
                )
    return domain_mod

circle(x, y=None, radius=10, num=10, dir=0)

a generator will calculate coordinates around a circle.

Parameters:

Name	Type	Description	Default
x	float	start x position	required
y	float | None	start y position	None
radius	int	size of the circle	10
num	int	the amount of points calculated (higher => slower, more cpu, but more detailed)	10

Returns:

Type	Description

Source code in zendriver/core/util.py

def circle(
    x: float, y: float | None = None, radius: int = 10, num: int = 10, dir: int = 0
) -> typing.Generator[typing.Tuple[float, float], None, None]:
    """
    a generator will calculate coordinates around a circle.

    :param x: start x position
    :param y: start y position
    :param radius: size of the circle
    :param num: the amount of points calculated (higher => slower, more cpu, but more detailed)
    :return:
    :rtype:
    """
    import math

    r = radius
    w = num
    if not y:
        y = x
    a = int(x - r * 2)
    b = int(y - r * 2)
    m = (2 * math.pi) / w
    if dir == 0:
        # regular direction
        ran = 0, w + 1, 1
    else:
        # opposite ?
        ran = w + 1, 0, -1

    for i in range(*ran):
        x = a + r * math.sin(m * i)
        y = b + r * math.cos(m * i)

        yield x, y

compare_target_info(info1, info2)

when logging mode is set to debug, browser object will log when target info is changed. To provide more meaningful log messages, this function is called to check what has actually changed between the 2 (by simple dict comparison). it returns a list of tuples [ ... ( key_which_has_changed, old_value, new_value) ]

Parameters:

Name	Type	Description	Default
info1	TargetInfo | None		required
info2	TargetInfo		required

Returns:

Type	Description

Source code in zendriver/core/util.py

def compare_target_info(
    info1: cdp.target.TargetInfo | None, info2: cdp.target.TargetInfo
) -> List[typing.Tuple[str, typing.Any, typing.Any]]:
    """
    when logging mode is set to debug, browser object will log when target info
    is changed. To provide more meaningful log messages, this function is called to
    check what has actually changed between the 2 (by simple dict comparison).
    it returns a list of tuples [ ... ( key_which_has_changed, old_value, new_value) ]

    :param info1:
    :param info2:
    :return:
    :rtype:
    """
    d1 = info1.__dict__ if info1 else {}
    d2 = info2.__dict__
    return [(k, v, d2[k]) for (k, v) in d1.items() if d2[k] != v]

create_from_undetected_chromedriver(driver) async

create a zendriver.Browser instance from a running undetected_chromedriver.Chrome instance.

Source code in zendriver/core/util.py

async def create_from_undetected_chromedriver(driver: Any) -> Browser:
    """
    create a zendriver.Browser instance from a running undetected_chromedriver.Chrome instance.
    """
    from .config import Config

    conf = Config()

    host, port = driver.options.debugger_address.split(":")
    conf.host, conf.port = host, int(port)

    # create zendriver Browser instance
    browser = await start(conf)

    browser._process_pid = driver.browser_pid
    # stop chromedriver binary
    driver.service.stop()
    driver.browser_pid = -1
    driver.user_data_dir = None
    return browser

filter_recurse(doc, predicate)

test each child using predicate(child), and return the first child of which predicate(child) == True

Parameters:

Name	Type	Description	Default
doc	Node	the cdp.dom.Node object or :py:class:zendriver.Element	required
predicate	Callable[[Node], bool]	a function which takes a node as first parameter and returns a boolean, where True means include	required

Source code in zendriver/core/util.py

def filter_recurse(
    doc: cdp.dom.Node, predicate: Callable[[cdp.dom.Node], bool]
) -> cdp.dom.Node | None:
    """
    test each child using predicate(child), and return the first child of which predicate(child) == True

    :param doc: the cdp.dom.Node object or :py:class:`zendriver.Element`
    :param predicate: a function which takes a node as first parameter and returns a boolean, where True means include

    """
    if not hasattr(doc, "children"):
        raise TypeError("object should have a .children attribute")

    if doc and doc.children:
        for child in doc.children:
            if predicate(child):
                # if predicate is True
                return child
            if child.shadow_roots:
                shadow_root_result = filter_recurse(child.shadow_roots[0], predicate)
                if shadow_root_result:
                    return shadow_root_result
            result = filter_recurse(child, predicate)
            if result:
                return result

    return None

filter_recurse_all(doc, predicate)

test each child using predicate(child), and return all children for which predicate(child) == True

Parameters:

Name	Type	Description	Default
doc	T	the cdp.dom.Node object or :py:class:zendriver.Element	required
predicate	Union[Callable[[Node], bool], Callable[[Element], bool]]	a function which takes a node as first parameter and returns a boolean, where True means include	required

Returns:

Type	Description

Source code in zendriver/core/util.py

def filter_recurse_all(
    doc: T, predicate: Union[Callable[[cdp.dom.Node], bool], Callable[[Element], bool]]
) -> List[T]:
    """
    test each child using predicate(child), and return all children for which predicate(child) == True

    :param doc: the cdp.dom.Node object or :py:class:`zendriver.Element`
    :param predicate: a function which takes a node as first parameter and returns a boolean, where True means include
    :return:
    :rtype:
    """
    if not hasattr(doc, "children"):
        raise TypeError("object should have a .children attribute")
    out = []
    if doc and doc.children:
        for child in doc.children:
            if predicate(child):
                # if predicate is True
                out.append(child)
            if child.shadow_roots is not None:
                out.extend(filter_recurse_all(child.shadow_roots[0], predicate))
            out.extend(filter_recurse_all(child, predicate))

    return out

free_port()

Determines a free port using sockets.

Source code in zendriver/core/util.py

def free_port() -> int:
    """
    Determines a free port using sockets.
    """
    import socket

    free_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    free_socket.bind(("127.0.0.1", 0))
    free_socket.listen(5)
    port: int = free_socket.getsockname()[1]
    free_socket.close()
    return port

get_registered_instances()

Source code in zendriver/core/util.py

def get_registered_instances() -> Set[Browser]:
    return __registered__instances__

html_from_tree(tree, target) async

Source code in zendriver/core/util.py

async def html_from_tree(
    tree: Union[cdp.dom.Node, Element], target: zendriver.Tab
) -> str:
    if not hasattr(tree, "children"):
        raise TypeError("object should have a .children attribute")
    out = ""
    if tree and tree.children:
        for child in tree.children:
            if isinstance(child, Element):
                out += await child.get_html()
            elif isinstance(child, cdp.dom.Node):
                out += await target.send(
                    cdp.dom.get_outer_html(backend_node_id=child.backend_node_id)
                )
            else:
                out += child

            if not isinstance(child, str):
                out += await html_from_tree(child, target)
    return out

loop()

Source code in zendriver/core/util.py

@deprecated(
    version="0.5.1", reason="Use asyncio functions directly instead, e.g. asyncio.run"
)
def loop() -> AbstractEventLoop:
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    return loop

remove_from_tree(tree, node)

Source code in zendriver/core/util.py

def remove_from_tree(tree: cdp.dom.Node, node: cdp.dom.Node) -> cdp.dom.Node:
    if not hasattr(tree, "children"):
        raise TypeError("object should have a .children attribute")

    if tree and tree.children:
        for child in tree.children:
            if child.backend_node_id == node.backend_node_id:
                tree.children.remove(child)
            remove_from_tree(child, node)
    return tree

start(config=None, *, user_data_dir=None, headless=False, browser_executable_path=None, browser='auto', browser_args=None, sandbox=True, lang=None, host=None, port=None, expert=None, user_agent=None, **kwargs) async

helper function to launch a browser. it accepts several keyword parameters. conveniently, you can just call it bare (no parameters) to quickly launch an instance with best practice defaults. note: this should be called await start()

Parameters:

Name	Type	Description	Default
user_data_dir	Optional[PathLike]		None
headless	Optional[bool]		False
browser_executable_path	Optional[PathLike]		None
browser_args	Optional[List[str]]	["--some-chromeparam=somevalue", "some-other-param=someval"]	None
sandbox	Optional[bool]	default True, but when set to False it adds --no-sandbox to the params, also when using linux under a root user, it adds False automatically (else chrome won't start	True
lang	Optional[str]	language string	None
port	Optional[int]	if you connect to an existing debuggable session, you can specify the port here if both host and port are provided, zendriver will not start a local chrome browser!	None
host	Optional[str]	if you connect to an existing debuggable session, you can specify the host here if both host and port are provided, zendriver will not start a local chrome browser!	None
expert	Optional[bool]	when set to True, enabled "expert" mode. This conveys, the inclusion of parameters: --disable-web-security ----disable-site-isolation-trials, as well as some scripts and patching useful for debugging (for example, ensuring shadow-root is always in "open" mode)	None
user_agent	Optional[str]	if set, this will be used as the user agent for the browser.	None

Returns:

Type	Description
Browser

Source code in zendriver/core/util.py

async def start(
    config: Optional[Config] = None,
    *,
    user_data_dir: Optional[PathLike] = None,
    headless: Optional[bool] = False,
    browser_executable_path: Optional[PathLike] = None,
    browser: BrowserType = "auto",
    browser_args: Optional[List[str]] = None,
    sandbox: Optional[bool] = True,
    lang: Optional[str] = None,
    host: Optional[str] = None,
    port: Optional[int] = None,
    expert: Optional[bool] = None,
    user_agent: Optional[str] = None,
    **kwargs: Any,
) -> Browser:
    """
    helper function to launch a browser. it accepts several keyword parameters.
    conveniently, you can just call it bare (no parameters) to quickly launch an instance
    with best practice defaults.
    note: this should be called ```await start()```

    :param user_data_dir:

    :param headless:

    :param browser_executable_path:

    :param browser_args: ["--some-chromeparam=somevalue", "some-other-param=someval"]

    :param sandbox: default True, but when set to False it adds --no-sandbox to the params, also
    when using linux under a root user, it adds False automatically (else chrome won't start

    :param lang: language string

    :param port: if you connect to an existing debuggable session, you can specify the port here
                 if both host and port are provided, zendriver will not start a local chrome browser!

    :param host: if you connect to an existing debuggable session, you can specify the host here
                 if both host and port are provided, zendriver will not start a local chrome browser!

    :param expert:  when set to True, enabled "expert" mode.
                    This conveys, the inclusion of parameters: --disable-web-security ----disable-site-isolation-trials,
                    as well as some scripts and patching useful for debugging (for example, ensuring shadow-root is always in "open" mode)

    :param user_agent: if set, this will be used as the user agent for the browser.

    :return:
    """
    if not config:
        config = Config(
            user_data_dir,
            headless,
            browser_executable_path,
            browser,
            browser_args,
            sandbox,
            lang,
            host=host,
            port=port,
            expert=expert,
            user_agent=user_agent,
            **kwargs,
        )
    from .browser import Browser

    return await Browser.create(config)