browser.py

"""CDP-Driver is based on NoDriver"""
from __future__ import annotations
import asyncio
import atexit
import fasteners
import http.cookiejar
import json
import logging
import os
import pathlib
import pickle
import psutil
import re
import shutil
import time
import urllib.parse
import urllib.request
import warnings
from collections import defaultdict
from contextlib import suppress
from seleniumbase import config as sb_config
from seleniumbase.fixtures import constants
from seleniumbase.fixtures import shared_utils
from typing import List, Optional, Set, Tuple, Union
import mycdp as cdp
from . import cdp_util as util
from . import tab
from ._contradict import ContraDict
from .config import PathLike, Config, is_posix
from .connection import Connection

logger = logging.getLogger(__name__)


def get_registered_instances():
    return __registered__instances__


def deconstruct_browser():
    for _ in __registered__instances__:
        if not _.stopped:
            _.stop(deconstruct=True)
        max_attempts = 5
        for attempt in range(max_attempts):
            try:
                if _.config and not _.config.uses_custom_data_dir:
                    if os.path.exists(_.config.user_data_dir):
                        shutil.rmtree(
                            _.config.user_data_dir, ignore_errors=False
                        )
                    if not os.path.exists(_.config.user_data_dir):
                        break
                    else:
                        time.sleep(0.12)
            except FileNotFoundError:
                break
            except (PermissionError, OSError) as e:
                if attempt == max_attempts - 1:
                    logger.debug(
                        "Problem removing data dir %s\n"
                        "Consider checking whether it's there "
                        "and remove it by hand\nerror: %s"
                        % (_.config.user_data_dir, e)
                    )
                    break
                time.sleep(0.12)
                continue
        logging.debug("Temp profile %s was removed." % _.config.user_data_dir)


class Browser:
    """
    The Browser object is the "root" of the hierarchy
    and contains a reference to the browser parent process.
    There should usually be only 1 instance of this.
    All opened tabs, extra browser screens,
    and resources will not cause a new Browser process,
    but rather create additional :class:`Tab` objects.
    So, besides starting your instance and first/additional tabs,
    you don't actively use it a lot under normal conditions.
    Tab objects will represent and control:
     - tabs (as you know them)
     - browser windows (new window)
     - iframe
     - background processes
    Note:
    The Browser object is not instantiated by __init__
    but using the asynchronous :meth:`Browser.create` method.
    Note:
    In Chromium based browsers, there is a parent process which keeps
    running all the time, even if there are no visible browser windows.
    Sometimes it's stubborn to close it, so make sure that after using
    this library, the browser is correctly and fully closed/exited/killed.
    """
    _process: asyncio.subprocess.Process
    _process_pid: int
    _http: HTTPApi = None
    _cookies: CookieJar = None
    config: Config
    connection: Connection

    @classmethod
    async def create(
        cls,
        config: Config = None,
        *,
        user_data_dir: PathLike = None,
        headless: bool = False,
        incognito: bool = False,
        guest: bool = False,
        browser_executable_path: PathLike = None,
        browser_args: List[str] = None,
        sandbox: bool = True,
        host: str = None,
        port: int = None,
        **kwargs,
    ) -> Browser:
        """Entry point for creating an instance."""
        if not config:
            config = Config(
                user_data_dir=user_data_dir,
                headless=headless,
                incognito=incognito,
                guest=guest,
                browser_executable_path=browser_executable_path,
                browser_args=browser_args or [],
                sandbox=sandbox,
                host=host,
                port=port,
                **kwargs,
            )
        try:
            instance = cls(config)
            await instance.start()
        except Exception:
            time.sleep(0.15)
            instance = cls(config)
            await instance.start()
        return instance

    def __init__(self, config: Config, **kwargs):
        """
        Constructor. To create a instance, use :py:meth:`Browser.create(...)`
        :param config:
        """
        try:
            asyncio.get_running_loop()
        except RuntimeError:
            raise RuntimeError(
                "{0} objects of this class are created "
                "using await {0}.create()".format(
                    self.__class__.__name__
                )
            )
        self.config = config
        self.targets: List = []
        self.info = None
        self._target = None
        self._process = None
        self._process_pid = None
        self._keep_user_data_dir = None
        self._is_updating = asyncio.Event()
        self.connection: Connection = None
        logger.debug("Session object initialized: %s" % vars(self))

    @property
    def websocket_url(self):
        return self.info.webSocketDebuggerUrl

    @property
    def main_tab(self) -> tab.Tab:
        """Returns the target which was launched with the browser."""
        return sorted(
            self.targets, key=lambda x: x.type_ == "page", reverse=True
        )[0]

    @property
    def tabs(self) -> List[tab.Tab]:
        """Returns the current targets which are of type "page"."""
        tabs = filter(lambda item: item.type_ == "page", self.targets)
        return list(tabs)

    @property
    def cookies(self) -> CookieJar:
        if not self._cookies:
            self._cookies = CookieJar(self)
        return self._cookies

    @property
    def stopped(self):
        if self._process and self._process.returncode is None:
            return False
        return True

    async def wait(self, time: Union[float, int] = 1) -> Browser:
        """Wait for <time> seconds. Important to use,
        especially in between page navigation.
        :param time:
        """
        return await asyncio.sleep(time, result=self)

    sleep = wait
    """Alias for wait"""
    async def _handle_target_update(
        self,
        event: Union[
            cdp.target.TargetInfoChanged,
            cdp.target.TargetDestroyed,
            cdp.target.TargetCreated,
            cdp.target.TargetCrashed,
        ],
    ):
        """This is an internal handler which updates the targets
        when Chrome emits the corresponding event."""
        if isinstance(event, cdp.target.TargetInfoChanged):
            target_info = event.target_info
            current_tab = next(
                filter(
                    lambda item: item.target_id == target_info.target_id,
                    self.targets,
                )
            )
            current_target = current_tab.target
            if logger.getEffectiveLevel() <= 10:
                changes = util.compare_target_info(
                    current_target, target_info
                )
                changes_string = ""
                for change in changes:
                    key, old, new = change
                    changes_string += f"\n{key}: {old} => {new}\n"
                logger.debug(
                    "Target #%d has changed: %s"
                    % (self.targets.index(current_tab), changes_string)
                )
                current_tab.target = target_info
        elif isinstance(event, cdp.target.TargetCreated):
            target_info: cdp.target.TargetInfo = event.target_info
            websocket_url = (
                f"ws://{self.config.host}:{self.config.port}"
                f"/devtools/{target_info.type_ or 'page'}"
                f"/{target_info.target_id}"
            )
            async with tab.Tab(
                websocket_url=websocket_url,
                target=target_info,
                browser=self
            ) as new_target:
                self.targets.append(new_target)
                logger.debug(
                    "Target #%d created => %s"
                    % (len(self.targets), new_target)
                )
        elif isinstance(event, cdp.target.TargetDestroyed):
            current_tab = next(
                filter(
                    lambda item: item.target_id == event.target_id,
                    self.targets,
                )
            )
            logger.debug(
                "Target removed. id # %d => %s"
                % (self.targets.index(current_tab), current_tab)
            )
            self.targets.remove(current_tab)

    def is_running(self):
        """Check if the browser is still running."""
        if not getattr(self, "_process_pid", None):
            return False
        # Not good enough: return psutil.pid_exists(self._process_pid)
        try:
            process = psutil.Process(self._process_pid)
            # Check for PID recycling: Is this actually our process?
            if hasattr(self, "_process_create_time"):
                if process.create_time() != self._process_create_time:
                    return False  # The PID was reused by a different process!
            # If the process is a zombie, assume the browser isn't running
            return process.status() != psutil.STATUS_ZOMBIE
        except psutil.NoSuchProcess:
            return False
        except Exception:
            return None

    def get_rd_host(self):
        return self.config.host

    def get_rd_port(self):
        return self.config.port

    def get_rd_url(self):
        """Returns the remote-debugging URL, which is used for
        allowing the Playwright integration to launch stealthy.
        Also sets an environment variable to hide this warning:
        Deprecation: "url.parse() behavior is not standardized".
        (github.com/microsoft/playwright-python/issues/3016)"""
        os.environ["NODE_NO_WARNINGS"] = "1"
        host = self.config.host
        port = self.config.port
        return f"http://{host}:{port}"

    def get_endpoint_url(self):
        return self.get_rd_url()

    def get_port(self):
        return self.get_rd_port()

    async def set_auth(self, username, password, tab):
        async def auth_challenge_handler(event: cdp.fetch.AuthRequired):
            await tab.send(
                cdp.fetch.continue_with_auth(
                    request_id=event.request_id,
                    auth_challenge_response=cdp.fetch.AuthChallengeResponse(
                        response="ProvideCredentials",
                        username=username,
                        password=password,
                    ),
                )
            )

        async def req_paused(event: cdp.fetch.RequestPaused):
            await tab.send(
                cdp.fetch.continue_request(request_id=event.request_id)
            )

        tab.add_handler(
            cdp.fetch.RequestPaused,
            lambda event: asyncio.create_task(req_paused(event)),
        )

        tab.add_handler(
            cdp.fetch.AuthRequired,
            lambda event: asyncio.create_task(auth_challenge_handler(event)),
        )

        await tab.send(cdp.fetch.enable(handle_auth_requests=True))

    async def get(
        self,
        url="about:blank",
        new_tab: bool = False,
        new_window: bool = False,
        **kwargs,
    ) -> tab.Tab:
        """Top level get. Utilizes the first tab to retrieve given url.
        Convenience function known from selenium.
        This function detects when DOM events have fired during navigation.
        :param url: The URL to navigate to
        :param new_tab: Open new tab
        :param new_window: Open new window
        :return: Page
        """
        await asyncio.sleep(0.005)
        if url and ":" not in url:
            url = "https://" + url
        if new_tab or new_window:
            # Create new target using the browser session.
            target_id = await self.connection.send(
                cdp.target.create_target(
                    url, new_window=new_window, enable_begin_frame_control=True
                )
            )
            connection: tab.Tab = next(
                filter(
                    lambda item: (
                        item.type_ == "page" and item.target_id == target_id
                    ),
                    self.targets,
                )
            )
            connection.browser = self
        else:
            try:
                # Most recently opened tab
                connection = self.targets[-1]
                await connection.sleep(0.005)
            except Exception:
                # First tab from browser.tabs
                connection: tab.Tab = next(
                    filter(lambda item: item.type_ == "page", self.targets)
                )
                await connection.sleep(0.005)
        _cdp_timezone = None
        _cdp_user_agent = ""
        _cdp_locale = None
        _cdp_platform = None
        _cdp_disable_csp = None
        _cdp_geolocation = None
        _cdp_mobile_mode = None
        _cdp_recorder = None
        _cdp_ad_block = None
        if getattr(sb_config, "_cdp_timezone", None):
            _cdp_timezone = sb_config._cdp_timezone
        if getattr(sb_config, "_cdp_user_agent", None):
            _cdp_user_agent = sb_config._cdp_user_agent
        if getattr(sb_config, "_cdp_locale", None):
            _cdp_locale = sb_config._cdp_locale
        if getattr(sb_config, "_cdp_platform", None):
            _cdp_platform = sb_config._cdp_platform
        if getattr(sb_config, "_cdp_geolocation", None):
            _cdp_geolocation = sb_config._cdp_geolocation
        if getattr(sb_config, "_cdp_mobile_mode", None):
            _cdp_mobile_mode = sb_config._cdp_mobile_mode
        if getattr(sb_config, "ad_block_on", None):
            _cdp_ad_block = sb_config.ad_block_on
        if getattr(sb_config, "disable_csp", None):
            _cdp_disable_csp = sb_config.disable_csp
        if "timezone" in kwargs:
            _cdp_timezone = kwargs["timezone"]
        elif "tzone" in kwargs:
            _cdp_timezone = kwargs["tzone"]
        if "user_agent" in kwargs:
            _cdp_user_agent = kwargs["user_agent"]
        elif "agent" in kwargs:
            _cdp_user_agent = kwargs["agent"]
        if "locale" in kwargs:
            _cdp_locale = kwargs["locale"]
        elif "lang" in kwargs:
            _cdp_locale = kwargs["lang"]
        elif "locale_code" in kwargs:
            _cdp_locale = kwargs["locale_code"]
        if "platform" in kwargs:
            _cdp_platform = kwargs["platform"]
        elif "plat" in kwargs:
            _cdp_platform = kwargs["plat"]
        if "disable_csp" in kwargs:
            _cdp_disable_csp = kwargs["disable_csp"]
        if "geolocation" in kwargs:
            _cdp_geolocation = kwargs["geolocation"]
        elif "geoloc" in kwargs:
            _cdp_geolocation = kwargs["geoloc"]
        if "ad_block" in kwargs:
            _cdp_ad_block = kwargs["ad_block"]
        if "mobile" in kwargs:
            _cdp_mobile_mode = kwargs["mobile"]
        if "recorder" in kwargs:
            _cdp_recorder = kwargs["recorder"]
        headless = self.config.headless
        await connection.sleep(0.01)
        await connection.send(cdp.network.enable())
        await connection.sleep(0.01)
        if _cdp_timezone:
            await connection.set_timezone(_cdp_timezone)
        if _cdp_locale:
            await connection.set_locale(_cdp_locale)
        if _cdp_user_agent or _cdp_locale or _cdp_platform or headless:
            if (
                (headless and not _cdp_user_agent)
                or (_cdp_user_agent and "Headless" in _cdp_user_agent)
            ):
                agent = await self.main_tab.evaluate("navigator.userAgent")
                if agent and "Headless" in agent:
                    _cdp_user_agent = agent.replace("Headless", "")
            await connection.set_user_agent(
                user_agent=_cdp_user_agent,
                accept_language=_cdp_locale,
                platform=_cdp_platform,
            )
        if _cdp_ad_block:
            await connection.send(cdp.network.set_blocked_urls(
                urls=[
                    "*.cloudflareinsights.com*",
                    "*.googlesyndication.com*",
                    "*.googletagmanager.com*",
                    "*.google-analytics.com*",
                    "*.amazon-adsystem.com*",
                    "*.adsafeprotected.com*",
                    "*.m.media-amazon.com*",
                    "*.ads.linkedin.com*",
                    "*.casalemedia.com*",
                    "*.doubleclick.net*",
                    "*.admanmedia.com*",
                    "*.quantserve.com*",
                    "*.ads.simpli.fi*",
                    "*.fastclick.net*",
                    "*.snigelweb.com*",
                    "*.bidswitch.net*",
                    "*.360yield.com*",
                    "*.adthrive.com*",
                    "*.pubmatic.com*",
                    "*.id5-sync.com*",
                    "*.moatads.com*",
                    "*.dotomi.com*",
                    "*.adsrvr.org*",
                    "*.atmtd.com*",
                    "*.liadm.com*",
                    "*.loopme.me*",
                    "*.adnxs.com*",
                    "*.openx.net*",
                    "*.tapad.com*",
                    "*.3lift.com*",
                    "*.turn.com*",
                    "*.2mdn.net*",
                    "*.cpx.to*",
                    "*.ad.gt*",
                ]
            ))
        if _cdp_geolocation:
            await connection.set_geolocation(_cdp_geolocation)
        if _cdp_disable_csp:
            await connection.send(cdp.page.set_bypass_csp(enabled=True))
        if _cdp_mobile_mode:
            await connection.send(
                cdp.emulation.set_device_metrics_override(
                    width=412, height=732, device_scale_factor=3, mobile=True
                )
            )
        # (The code below is for the Chrome 142 extension fix)
        if (
            getattr(sb_config, "_cdp_proxy", None)
            and "@" in sb_config._cdp_proxy
            and "auth" not in kwargs
        ):
            username_and_password = sb_config._cdp_proxy.split("@")[0]
            proxy_user = username_and_password.split(":")[0]
            proxy_pass = username_and_password.split(":")[1]
            if (
                hasattr(self.main_tab, "_last_auth")
                and self.main_tab._last_auth == username_and_password
            ):
                pass  # Auth was already set
            else:
                self.main_tab._last_auth = username_and_password
                await self.set_auth(proxy_user, proxy_pass, self.tabs[0])
                time.sleep(0.25)
        if "auth" in kwargs and kwargs["auth"] and ":" in kwargs["auth"]:
            username_and_password = kwargs["auth"]
            proxy_user = username_and_password.split(":")[0]
            proxy_pass = username_and_password.split(":")[1]
            if (
                hasattr(self.main_tab, "_last_auth")
                and self.main_tab._last_auth == username_and_password
            ):
                pass  # Auth was already set
            else:
                self.main_tab._last_auth = username_and_password
                await self.set_auth(proxy_user, proxy_pass, self.tabs[0])
                time.sleep(0.25)
        await connection.sleep(0.15)
        frame_id, loader_id, *_ = await connection.send(
            cdp.page.navigate(url)
        )
        major_browser_version = None
        try:
            major_browser_version = (
                int(self.info["Browser"].split("/")[-1].split(".")[0])
            )
        except Exception:
            pass
        if (
            _cdp_recorder
            and (
                not hasattr(sb_config, "browser")
                or (
                    sb_config.browser == "chrome"
                    and (
                        not major_browser_version
                        or major_browser_version >= 142
                    )
                )
            )
        ):
            # (The code below is for the Chrome 142 extension fix)
            from seleniumbase.js_code.recorder_js import recorder_js
            recorder_code = (
                """window.onload = function() { %s };""" % recorder_js
            )
            await connection.send(
                cdp.page.add_script_to_evaluate_on_new_document(recorder_code)
            )
            await connection.sleep(0.25)
            await self.wait(0.05)
            await connection.send(cdp.runtime.evaluate(recorder_js))
        # Update the frame_id on the tab
        connection.frame_id = frame_id
        connection.browser = self
        # Give settings time to take effect
        await connection.sleep(0.25)
        await self.wait(0.05)
        return connection

    async def start(self=None) -> Browser:
        """Launches the actual browser."""
        if not self:
            warnings.warn(
                "Use ``await Browser.create()`` to create a new instance!"
            )
            return
        if self._process or self._process_pid:
            if self._process.returncode is not None:
                return await self.create(config=self.config)
            warnings.warn(
                "Ignored! This call has no effect when already running!"
            )
            return
        # self.config.update(kwargs)
        connect_existing = False
        if self.config.host is not None and self.config.port is not None:
            connect_existing = True
        else:
            self.config.host = "127.0.0.1"
            self.config.port = util.free_port()
        if not connect_existing:
            logger.debug(
                "BROWSER EXECUTABLE PATH: %s"
                % self.config.browser_executable_path,
            )
            if not pathlib.Path(self.config.browser_executable_path).exists():
                raise FileNotFoundError(
                    (
                        """
                    ---------------------------------------
                    Could not determine browser executable.
                    ---------------------------------------
                    Browser must be installed in the default location / path!
                    If you are sure about the browser executable,
                    set it using `browser_executable_path='{}` parameter."""
                    ).format(
                        "/path/to/your/browser/executable"
                        if is_posix
                        else "c:/path/to/your/browser.exe"
                    )
                )
        if getattr(self.config, "_extensions", None):
            self.config.add_argument(
                "--load-extension=%s"
                % ",".join(str(_) for _ in self.config._extensions)
            )
        exe = self.config.browser_executable_path
        params = self.config()
        logger.debug(
            "Starting\n\texecutable :%s\n\narguments:\n%s",
            exe,
            "\n\t".join(params),
        )
        if not connect_existing:
            self._process: asyncio.subprocess.Process = (
                await asyncio.create_subprocess_exec(
                    exe,
                    *params,
                    stdin=asyncio.subprocess.PIPE,
                    stdout=asyncio.subprocess.PIPE,
                    stderr=asyncio.subprocess.PIPE,
                    close_fds=is_posix,
                )
            )
            await asyncio.sleep(0.05)
            self._process_pid = self._process.pid
            try:
                self._process_create_time = (
                    psutil.Process(self._process_pid).create_time()
                )
            except Exception:
                self._process_create_time = None
        await asyncio.sleep(0.05)
        self._http = HTTPApi((self.config.host, self.config.port))
        await asyncio.sleep(0.05)
        get_registered_instances().add(self)
        await asyncio.sleep(0.15)
        max_attempts = 20
        for attempt in range(max_attempts):
            try:
                self.info = ContraDict(
                    await self._http.get("version"), silent=True
                )
            except (Exception,):
                if attempt == max_attempts - 1:
                    logger.debug("Could not start", exc_info=True)
                else:
                    await self.sleep(0.2)
            else:
                break
        if not self.info:
            chromium = "Chromium"
            if getattr(sb_config, "_cdp_browser", None):
                chromium = sb_config._cdp_browser
                chromium = chromium[0].upper() + chromium[1:]
            message = "Failed to connect to the browser"
            if not exe or not os.path.exists(exe):
                message = (
                    "%s executable not found. Is it installed?" % chromium
                )
            dash_len = len(message)
            dashes = "-" * dash_len
            raise Exception(
                """
                %s
                %s
                %s
                """ % (dashes, message, dashes)
            )
        await asyncio.sleep(0.03)
        self.connection = Connection(
            self.info.webSocketDebuggerUrl, browser=self
        )
        await asyncio.sleep(0.03)
        if self.config.autodiscover_targets:
            logger.debug("Enabling autodiscover targets")
            self.connection.handlers[cdp.target.TargetInfoChanged] = [
                self._handle_target_update
            ]
            self.connection.handlers[cdp.target.TargetCreated] = [
                self._handle_target_update
            ]
            self.connection.handlers[cdp.target.TargetDestroyed] = [
                self._handle_target_update
            ]
            self.connection.handlers[cdp.target.TargetCrashed] = [
                self._handle_target_update
            ]
            await self.connection.send(
                cdp.target.set_discover_targets(discover=True)
            )
        await self
        # self.connection.handlers[cdp.inspector.Detached] = [self.stop]
        # return self

    async def grant_permissions(
        self,
        permissions: List[str] | str,
        origin: Optional[str] = None,
    ):
        """Grant specific permissions to the current window.
        Applies to all origins if no origin is specified."""
        if isinstance(permissions, str):
            permissions = [permissions]
        await self.connection.send(
            cdp.browser.grant_permissions(permissions, origin)
        )

    async def grant_all_permissions(self):
        """
        Grant permissions for:
            audioCapture
            backgroundSync
            backgroundFetch
            clipboardReadWrite
            clipboardSanitizedWrite
            displayCapture
            durableStorage
            geolocation
            idleDetection
            localFonts
            midi
            midiSysex
            nfc
            notifications
            paymentHandler
            periodicBackgroundSync
            sensors
            storageAccess
            topLevelStorageAccess
            videoCapture
            wakeLockScreen
            wakeLockSystem
            windowManagement
        """
        permissions = [
            "audioCapture",
            "backgroundSync",
            "backgroundFetch",
            "clipboardReadWrite",
            "clipboardSanitizedWrite",
            "displayCapture",
            "durableStorage",
            "geolocation",
            "idleDetection",
            "localFonts",
            "midi",
            "midiSysex",
            "nfc",
            "notifications",
            "paymentHandler",
            "periodicBackgroundSync",
            "sensors",
            "storageAccess",
            "topLevelStorageAccess",
            "videoCapture",
            "wakeLockScreen",
            "wakeLockSystem",
            "windowManagement",
        ]
        await self.connection.send(cdp.browser.grant_permissions(permissions))

    async def reset_permissions(self):
        """Reset permissions for all origins on the current window."""
        await self.connection.send(cdp.browser.reset_permissions())

    async def tile_windows(self, windows=None, max_columns: int = 0):
        import math
        try:
            import mss
        except Exception:
            pip_find_lock = fasteners.InterProcessLock(
                constants.PipInstall.FINDLOCK
            )
            with pip_find_lock:  # Prevent issues with multiple processes
                shared_utils.pip_install("mss")
            import mss
        m = mss.mss()
        screen, screen_width, screen_height = 3 * (None,)
        if m.monitors and len(m.monitors) >= 1:
            screen = m.monitors[0]
            screen_width = screen["width"]
            screen_height = screen["height"]
        if not screen or not screen_width or not screen_height:
            warnings.warn("No monitors detected!")
            return
        await self
        distinct_windows = defaultdict(list)
        if windows:
            tabs = windows
        else:
            tabs = self.tabs
        for _tab in tabs:
            window_id, bounds = await _tab.get_window()
            distinct_windows[window_id].append(_tab)
        num_windows = len(distinct_windows)
        req_cols = max_columns or int(num_windows * (19 / 6))
        req_rows = int(num_windows / req_cols)
        while req_cols * req_rows < num_windows:
            req_rows += 1
        box_w = math.floor((screen_width / req_cols) - 1)
        box_h = math.floor(screen_height / req_rows)
        distinct_windows_iter = iter(distinct_windows.values())
        grid = []
        for x in range(req_cols):
            for y in range(req_rows):
                try:
                    tabs = next(distinct_windows_iter)
                except StopIteration:
                    continue
                if not tabs:
                    continue
                tab = tabs[0]
                try:
                    pos = [x * box_w, y * box_h, box_w, box_h]
                    grid.append(pos)
                    await tab.set_window_size(*pos)
                except Exception:
                    logger.info(
                        "Could not set window size. Exception => ",
                        exc_info=True,
                    )
                    continue
        return grid

    async def _get_targets(self) -> List[cdp.target.TargetInfo]:
        info = await self.connection.send(
            cdp.target.get_targets(), _is_update=True
        )
        return info

    async def update_targets(self):
        targets: List[cdp.target.TargetInfo]
        targets = await self._get_targets()
        for t in targets:
            for existing_tab in self.targets:
                existing_target = existing_tab.target
                if existing_target.target_id == t.target_id:
                    existing_tab.target.__dict__.update(t.__dict__)
                    break
            else:
                self.targets.append(
                    Connection(
                        (
                            f"ws://{self.config.host}:{self.config.port}"
                            f"/devtools/page"  # All types are "page"
                            f"/{t.target_id}"
                        ),
                        target=t,
                        browser=self,
                    )
                )
        await asyncio.sleep(0)

    async def __aenter__(self):
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        if exc_type and exc_val:
            raise exc_type(exc_val)

    def __iter__(self):
        self._i = self.tabs.index(self.main_tab)
        return self

    def __reversed__(self):
        return reversed(list(self.tabs))

    def __next__(self):
        try:
            return self.tabs[self._i]
        except IndexError:
            del self._i
            raise StopIteration
        except AttributeError:
            del self._i
            raise StopIteration
        finally:
            if hasattr(self, "_i"):
                if self._i != len(self.tabs):
                    self._i += 1
                else:
                    del self._i

    def stop(self, deconstruct=False):
        if (
            not hasattr(sb_config, "_closed_connection_ids")
            or not isinstance(sb_config._closed_connection_ids, list)
        ):
            sb_config._closed_connection_ids = []
        connection_id = None
        with suppress(Exception):
            connection_id = self.connection.websocket.id.hex
        close_success = False
        try:
            if self.connection:
                asyncio.get_event_loop().create_task(self.connection.aclose())
                logger.debug(
                    "Closed connection using get_event_loop().create_task()"
                )
        except RuntimeError:
            if self.connection:
                try:
                    asyncio.run(self.connection.aclose())
                    logger.debug("Closed the connection using asyncio.run()")
                except Exception:
                    pass
        for _ in range(3):
            try:
                if connection_id not in sb_config._closed_connection_ids:
                    self._process.terminate()
                    logger.debug(
                        "Terminated browser with pid %d successfully."
                        % self._process.pid
                    )
                    if connection_id:
                        sb_config._closed_connection_ids.append(connection_id)
                        close_success = True
                    break
            except (Exception,):
                try:
                    self._process.kill()
                    logger.debug(
                        "Killed browser with pid %d successfully."
                        % self._process.pid
                    )
                    break
                except (Exception,):
                    try:
                        if hasattr(self, "browser_process_pid"):
                            os.kill(self._process_pid, 15)
                            logger.debug(
                                "Killed browser with pid %d "
                                "using signal 15 successfully."
                                % self._process.pid
                            )
                            break
                    except (TypeError,):
                        logger.info("TypeError", exc_info=True)
                        pass
                    except (PermissionError,):
                        logger.info(
                            "Browser already stopped, "
                            "or no permission to kill. Skip."
                        )
                        pass
                    except (ProcessLookupError,):
                        logger.info("ProcessLookupError")
                        pass
                    except (Exception,):
                        raise
            self._process = None
            self._process_pid = None
        if (
            hasattr(sb_config, "_xvfb_users")
            and isinstance(sb_config._xvfb_users, int)
            and close_success
            and hasattr(sb_config, "_virtual_display")
            and sb_config._virtual_display
        ):
            sb_config._xvfb_users -= 1
            if sb_config._xvfb_users < 0:
                sb_config._xvfb_users = 0
        if (
            shared_utils.is_linux()
            and (
                hasattr(sb_config, "_virtual_display")
                and sb_config._virtual_display
                and hasattr(sb_config._virtual_display, "stop")
            )
            and sb_config._xvfb_users == 0
        ):
            try:
                sb_config._virtual_display.stop()
                sb_config._virtual_display = None
                sb_config.headless_active = False
            except AttributeError:
                pass
            except Exception:
                pass
        if (
            deconstruct
            and connection_id
            and connection_id in sb_config._closed_connection_ids
        ):
            sb_config._closed_connection_ids.remove(connection_id)

    def quit(self):
        self.stop()

    def __await__(self):
        # return ( asyncio.sleep(0)).__await__()
        return self.update_targets().__await__()

    def __del__(self):
        pass


__registered__instances__: Set[Browser] = set()


class CookieJar:
    def __init__(self, browser: Browser):
        self._browser = browser

    async def get_all(
        self, requests_cookie_format: bool = False
    ) -> List[Union[cdp.network.Cookie, "http.cookiejar.Cookie"]]:
        """
        Get all cookies.
        :param requests_cookie_format: when True,
         returns python http.cookiejar.Cookie objects,
         compatible with requests library and many others.
        :type requests_cookie_format: bool
        """
        connection = None
        for _tab in self._browser.tabs:
            if getattr(_tab, "closed", None):
                continue
            connection = _tab
            break
        else:
            connection = self._browser.connection
        cookies = await connection.send(cdp.network.get_cookies())
        if requests_cookie_format:
            import requests.cookies

            return [
                requests.cookies.create_cookie(
                    name=c.name,
                    value=c.value,
                    domain=c.domain,
                    path=c.path,
                    expires=c.expires,
                    secure=c.secure,
                )
                for c in cookies
            ]
        return cookies

    async def set_all(self, cookies: List[cdp.network.CookieParam]):
        """
        Set cookies.
        :param cookies: List of cookies
        """
        connection = None
        for _tab in self._browser.tabs:
            if getattr(_tab, "closed", None):
                continue
            connection = _tab
            break
        else:
            connection = self._browser.connection
        await connection.send(cdp.network.set_cookies(cookies))

    async def save(self, file: PathLike = ".session.dat", pattern: str = ".*"):
        """
        Save all cookies (or a subset, controlled by `pattern`)
        to a file to be restored later.
        :param file:
        :param pattern: regex style pattern string.
                any cookie that has a  domain, key or value field
                which matches the pattern will be included.
            default = ".*"  (all)
            Eg: the pattern "(cf|.com|nowsecure)" will include cookies which:
                - Have a string "cf" (cloudflare)
                - Have ".com" in them, in either domain, key or value field.
                - Contain "nowsecure"
        :type pattern: str
        """
        pattern = re.compile(pattern)
        save_path = pathlib.Path(file).resolve()
        connection = None
        for _tab in self._browser.tabs:
            if getattr(_tab, "closed", None):
                continue
            connection = _tab
            break
        else:
            connection = self._browser.connection
        cookies = await connection.send(cdp.network.get_cookies())
        # if not connection:
        #     return
        # if not connection.websocket:
        #     return
        # if connection.websocket.closed:
        #     return
        cookies = await self.get_all(requests_cookie_format=False)
        included_cookies = []
        for cookie in cookies:
            for match in pattern.finditer(str(cookie.__dict__)):
                logger.debug(
                    "Saved cookie for matching pattern '%s' => (%s: %s)"
                    % (pattern.pattern, cookie.name, cookie.value)
                )
                included_cookies.append(cookie)
                break
        pickle.dump(cookies, save_path.open("w+b"))

    async def load(self, file: PathLike = ".session.dat", pattern: str = ".*"):
        """
        Load all cookies (or a subset, controlled by `pattern`)
        from a file created by :py:meth:`~save_cookies`.
        :param file:
        :param pattern: Regex style pattern string.
               Any cookie that has a  domain, key,
               or value field which matches the pattern will be included.
            Default = ".*"  (all)
            Eg: the pattern "(cf|.com|nowsecure)" will include cookies which:
                - Have a string "cf" (cloudflare)
                - Have ".com" in them, in either domain, key or value field.
                - Contain "nowsecure"
        :type pattern: str
        """
        pattern = re.compile(pattern)
        save_path = pathlib.Path(file).resolve()
        cookies = pickle.load(save_path.open("r+b"))
        included_cookies = []
        connection = None
        for _tab in self._browser.tabs:
            if getattr(_tab, "closed", None):
                continue
            connection = _tab
            break
        else:
            connection = self._browser.connection
        for cookie in cookies:
            for match in pattern.finditer(str(cookie.__dict__)):
                included_cookies.append(cookie)
                logger.debug(
                    "Loaded cookie for matching pattern '%s' => (%s: %s)"
                    % (pattern.pattern, cookie.name, cookie.value)
                )
                break
        await connection.send(cdp.network.set_cookies(included_cookies))

    async def clear(self):
        """
        Clear current cookies.
        Note: This includes all open tabs/windows for this browser.
        """
        connection = None
        for _tab in self._browser.tabs:
            if getattr(_tab, "closed", None):
                continue
            connection = _tab
            break
        else:
            connection = self._browser.connection
        cookies = await connection.send(cdp.network.get_cookies())
        if cookies:
            await connection.send(cdp.storage.clear_cookies())


class HTTPApi:
    def __init__(self, addr: Tuple[str, int]):
        self.host, self.port = addr
        self.api = "http://%s:%d" % (self.host, self.port)

    @classmethod
    def from_target(cls, target):
        ws_url = urllib.parse.urlparse(target.websocket_url)
        inst = cls((ws_url.hostname, ws_url.port))
        return inst

    async def get(self, endpoint: str):
        return await self._request(endpoint)

    async def post(self, endpoint, data):
        return await self._request(endpoint, data)

    async def _request(self, endpoint, method: str = "GET", data: dict = None):
        url = urllib.parse.urljoin(
            self.api, f"json/{endpoint}" if endpoint else "/json"
        )
        if data and method.lower() == "get":
            raise ValueError("GET requests cannot contain data")
        if not url:
            url = self.api + endpoint
        request = urllib.request.Request(url)
        request.method = method
        request.data = None
        if data:
            request.data = json.dumps(data).encode("utf-8")

        response = await asyncio.get_running_loop().run_in_executor(
            None, lambda: urllib.request.urlopen(request, timeout=10)
        )
        return json.loads(response.read())


atexit.register(deconstruct_browser)