from time import time
from asyncio import sleep as asyncio_sleep, Lock

from camoufox import DefaultAddons
from playwright.sync_api._generated import Page
from playwright.sync_api import (
    Frame,
    BrowserContext,
    Playwright,
    Response as SyncPlaywrightResponse,
)
from playwright.async_api._generated import Page as AsyncPage
from playwright.async_api import (
    Frame as AsyncFrame,
    Playwright as AsyncPlaywright,
    Response as AsyncPlaywrightResponse,
    BrowserContext as AsyncBrowserContext,
)
from playwright._impl._errors import Error as PlaywrightError
from camoufox.pkgman import installed_verstr as camoufox_version
from camoufox.utils import launch_options as generate_launch_options

from ._page import PageInfo, PagePool
from scrapling.parser import Selector
from scrapling.core._types import Any, cast, Dict, List, Optional, Callable, TYPE_CHECKING
from scrapling.engines.toolbelt.fingerprints import get_os_name
from ._validators import validate, PlaywrightConfig, CamoufoxConfig
from ._config_tools import _compiled_stealth_scripts, _launch_kwargs, _context_kwargs
from scrapling.engines.toolbelt.navigation import intercept_route, async_intercept_route

__ff_version_str__ = camoufox_version().split(".", 1)[0]


class SyncSession:
    def __init__(self, max_pages: int = 1):
        self.max_pages = max_pages
        self.page_pool = PagePool(max_pages)
        self._max_wait_for_page = 60
        self.playwright: Playwright | Any = None
        self.context: BrowserContext | Any = None
        self._closed = False

    def start(self):
        pass

    def close(self):  # pragma: no cover
        """Close all resources"""
        if self._closed:
            return

        if self.context:
            self.context.close()
            self.context = None

        if self.playwright:
            self.playwright.stop()
            self.playwright = None  # pyright: ignore

        self._closed = True

    def __enter__(self):
        self.start()
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.close()

    def _get_page(
        self,
        timeout: int | float,
        extra_headers: Optional[Dict[str, str]],
        disable_resources: bool,
    ) -> PageInfo[Page]:  # pragma: no cover
        """Get a new page to use"""

        # No need to check if a page is available or not in sync code because the code blocked before reaching here till the page closed, ofc.
        assert self.context is not None, "Browser context not initialized"
        page = self.context.new_page()
        page.set_default_navigation_timeout(timeout)
        page.set_default_timeout(timeout)
        if extra_headers:
            page.set_extra_http_headers(extra_headers)

        if disable_resources:
            page.route("**/*", intercept_route)

        if getattr(self, "stealth", False):
            for script in _compiled_stealth_scripts():
                page.add_init_script(script=script)

        page_info = self.page_pool.add_page(page)
        page_info.mark_busy()
        return page_info

    def get_pool_stats(self) -> Dict[str, int]:
        """Get statistics about the current page pool"""
        return {
            "total_pages": self.page_pool.pages_count,
            "busy_pages": self.page_pool.busy_count,
            "max_pages": self.max_pages,
        }

    @staticmethod
    def _wait_for_networkidle(page: Page | Frame, timeout: Optional[int] = None):
        """Wait for the page to become idle (no network activity) even if there are never-ending requests."""
        try:
            page.wait_for_load_state("networkidle", timeout=timeout)
        except (PlaywrightError, Exception):
            pass

    def _wait_for_page_stability(self, page: Page | Frame, load_dom: bool, network_idle: bool):
        page.wait_for_load_state(state="load")
        if load_dom:
            page.wait_for_load_state(state="domcontentloaded")
        if network_idle:
            self._wait_for_networkidle(page)

    @staticmethod
    def _create_response_handler(page_info: PageInfo[Page], response_container: List) -> Callable:
        """Create a response handler that captures the final navigation response.

        :param page_info: The PageInfo object containing the page
        :param response_container: A list to store the final response (mutable container)
        :return: A callback function for page.on("response", ...)
        """

        def handle_response(finished_response: SyncPlaywrightResponse):
            if (
                finished_response.request.resource_type == "document"
                and finished_response.request.is_navigation_request()
                and finished_response.request.frame == page_info.page.main_frame
            ):
                response_container[0] = finished_response

        return handle_response


class AsyncSession:
    def __init__(self, max_pages: int = 1):
        self.max_pages = max_pages
        self.page_pool = PagePool(max_pages)
        self._max_wait_for_page = 60
        self.playwright: AsyncPlaywright | Any = None
        self.context: AsyncBrowserContext | Any = None
        self._closed = False
        self._lock = Lock()

    async def start(self):
        pass

    async def close(self):
        """Close all resources"""
        if self._closed:  # pragma: no cover
            return

        if self.context:
            await self.context.close()
            self.context = None  # pyright: ignore

        if self.playwright:
            await self.playwright.stop()
            self.playwright = None  # pyright: ignore

        self._closed = True

    async def __aenter__(self):
        await self.start()
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        await self.close()

    async def _get_page(
        self,
        timeout: int | float,
        extra_headers: Optional[Dict[str, str]],
        disable_resources: bool,
    ) -> PageInfo[AsyncPage]:  # pragma: no cover
        """Get a new page to use"""
        if TYPE_CHECKING:
            assert self.context is not None, "Browser context not initialized"

        async with self._lock:
            # If we're at max capacity after cleanup, wait for busy pages to finish
            if self.page_pool.pages_count >= self.max_pages:
                start_time = time()
                while time() - start_time < self._max_wait_for_page:
                    await asyncio_sleep(0.05)
                    if self.page_pool.pages_count < self.max_pages:
                        break
                else:
                    raise TimeoutError(
                        f"No pages finished to clear place in the pool within the {self._max_wait_for_page}s timeout period"
                    )

            page = await self.context.new_page()
            page.set_default_navigation_timeout(timeout)
            page.set_default_timeout(timeout)
            if extra_headers:
                await page.set_extra_http_headers(extra_headers)

            if disable_resources:
                await page.route("**/*", async_intercept_route)

            if getattr(self, "stealth", False):
                for script in _compiled_stealth_scripts():
                    await page.add_init_script(script=script)

            return self.page_pool.add_page(page)

    def get_pool_stats(self) -> Dict[str, int]:
        """Get statistics about the current page pool"""
        return {
            "total_pages": self.page_pool.pages_count,
            "busy_pages": self.page_pool.busy_count,
            "max_pages": self.max_pages,
        }

    @staticmethod
    async def _wait_for_networkidle(page: AsyncPage | AsyncFrame, timeout: Optional[int] = None):
        """Wait for the page to become idle (no network activity) even if there are never-ending requests."""
        try:
            await page.wait_for_load_state("networkidle", timeout=timeout)
        except (PlaywrightError, Exception):
            pass

    async def _wait_for_page_stability(self, page: AsyncPage | AsyncFrame, load_dom: bool, network_idle: bool):
        await page.wait_for_load_state(state="load")
        if load_dom:
            await page.wait_for_load_state(state="domcontentloaded")
        if network_idle:
            await self._wait_for_networkidle(page)

    @staticmethod
    def _create_response_handler(page_info: PageInfo[AsyncPage], response_container: List) -> Callable:
        """Create an async response handler that captures the final navigation response.

        :param page_info: The PageInfo object containing the page
        :param response_container: A list to store the final response (mutable container)
        :return: A callback function for page.on("response", ...)
        """

        async def handle_response(finished_response: AsyncPlaywrightResponse):
            if (
                finished_response.request.resource_type == "document"
                and finished_response.request.is_navigation_request()
                and finished_response.request.frame == page_info.page.main_frame
            ):
                response_container[0] = finished_response

        return handle_response


class DynamicSessionMixin:
    def __validate__(self, **params):
        if "__max_pages" in params:
            params["max_pages"] = params.pop("__max_pages")

        config = validate(params, model=PlaywrightConfig)

        self._max_pages = config.max_pages
        self._headless = config.headless
        self._hide_canvas = config.hide_canvas
        self._disable_webgl = config.disable_webgl
        self._real_chrome = config.real_chrome
        self._stealth = config.stealth
        self._google_search = config.google_search
        self._wait = config.wait
        self._proxy = config.proxy
        self._locale = config.locale
        self._extra_headers = config.extra_headers
        self._useragent = config.useragent
        self._timeout = config.timeout
        self._cookies = config.cookies
        self._disable_resources = config.disable_resources
        self._cdp_url = config.cdp_url
        self._network_idle = config.network_idle
        self._load_dom = config.load_dom
        self._wait_selector = config.wait_selector
        self._init_script = config.init_script
        self._wait_selector_state = config.wait_selector_state
        self._extra_flags = config.extra_flags
        self._selector_config = config.selector_config
        self._timezone_id = config.timezone_id
        self._additional_args = config.additional_args
        self._page_action = config.page_action
        self._user_data_dir = config.user_data_dir
        self._headers_keys = {header.lower() for header in self._extra_headers.keys()} if self._extra_headers else set()
        self.__initiate_browser_options__()

    def __initiate_browser_options__(self):
        if TYPE_CHECKING:
            assert isinstance(self._proxy, tuple)

        if not self._cdp_url:
            # `launch_options` is used with persistent context
            self.launch_options = dict(
                _launch_kwargs(
                    self._headless,
                    self._proxy,
                    self._locale,
                    tuple(self._extra_headers.items()) if self._extra_headers else tuple(),
                    self._useragent,
                    self._real_chrome,
                    self._stealth,
                    self._hide_canvas,
                    self._disable_webgl,
                    self._timezone_id,
                    tuple(self._extra_flags) if self._extra_flags else tuple(),
                )
            )
            self.launch_options["extra_http_headers"] = dict(self.launch_options["extra_http_headers"])
            self.launch_options["proxy"] = dict(self.launch_options["proxy"]) or None
            self.launch_options["user_data_dir"] = self._user_data_dir
            self.launch_options.update(cast(Dict, self._additional_args))
            self.context_options = dict()
        else:
            # while `context_options` is left to be used when cdp mode is enabled
            self.launch_options = dict()
            self.context_options = dict(
                _context_kwargs(
                    self._proxy,
                    self._locale,
                    tuple(self._extra_headers.items()) if self._extra_headers else tuple(),
                    self._useragent,
                    self._stealth,
                )
            )
            self.context_options["extra_http_headers"] = dict(self.context_options["extra_http_headers"])
            self.context_options["proxy"] = dict(self.context_options["proxy"]) or None
            self.context_options.update(cast(Dict, self._additional_args))


class StealthySessionMixin:
    def __validate__(self, **params):
        if "__max_pages" in params:
            params["max_pages"] = params.pop("__max_pages")

        config: CamoufoxConfig = validate(params, model=CamoufoxConfig)

        self._max_pages = config.max_pages
        self._headless = config.headless
        self._block_images = config.block_images
        self._disable_resources = config.disable_resources
        self._block_webrtc = config.block_webrtc
        self._allow_webgl = config.allow_webgl
        self._network_idle = config.network_idle
        self._load_dom = config.load_dom
        self._humanize = config.humanize
        self._solve_cloudflare = config.solve_cloudflare
        self._wait = config.wait
        self._timeout = config.timeout
        self._page_action = config.page_action
        self._wait_selector = config.wait_selector
        self._init_script = config.init_script
        self._addons = config.addons
        self._wait_selector_state = config.wait_selector_state
        self._cookies = config.cookies
        self._google_search = config.google_search
        self._extra_headers = config.extra_headers
        self._proxy = config.proxy
        self._os_randomize = config.os_randomize
        self._disable_ads = config.disable_ads
        self._geoip = config.geoip
        self._selector_config = config.selector_config
        self._additional_args = config.additional_args
        self._user_data_dir = config.user_data_dir
        self._headers_keys = {header.lower() for header in self._extra_headers.keys()} if self._extra_headers else set()
        self.__initiate_browser_options__()

    def __initiate_browser_options__(self):
        """Initiate browser options."""
        self.launch_options: Dict[str, Any] = generate_launch_options(
            **{
                "geoip": self._geoip,
                "proxy": dict(self._proxy) if self._proxy and isinstance(self._proxy, tuple) else self._proxy,
                "addons": self._addons,
                "exclude_addons": [] if self._disable_ads else [DefaultAddons.UBO],
                "headless": self._headless,
                "humanize": True if self._solve_cloudflare else self._humanize,
                "i_know_what_im_doing": True,  # To turn warnings off with the user configurations
                "allow_webgl": self._allow_webgl,
                "block_webrtc": self._block_webrtc,
                "block_images": self._block_images,  # Careful! it makes some websites don't finish loading at all like stackoverflow even in headful mode.
                "os": None if self._os_randomize else get_os_name(),
                "user_data_dir": self._user_data_dir,
                "ff_version": __ff_version_str__,
                "firefox_user_prefs": {
                    # This is what enabling `enable_cache` does internally, so we do it from here instead
                    "browser.sessionhistory.max_entries": 10,
                    "browser.sessionhistory.max_total_viewers": -1,
                    "browser.cache.memory.enable": True,
                    "browser.cache.disk_cache_ssl": True,
                    "browser.cache.disk.smart_size.enabled": True,
                },
                **cast(Dict, self._additional_args),
            }
        )

    @staticmethod
    def _detect_cloudflare(page_content: str) -> str | None:
        """
        Detect the type of Cloudflare challenge present in the provided page content.

        This function analyzes the given page content to identify whether a specific
        type of Cloudflare challenge is present. It checks for three predefined
        challenge types: non-interactive, managed, and interactive. If a challenge
        type is detected, it returns the corresponding type as a string. If no
        challenge type is detected, it returns None.

        Args:
            page_content (str): The content of the page to analyze for Cloudflare
                challenge types.

        Returns:
            str: A string representing the detected Cloudflare challenge type, if
                found. Returns None if no challenge matches.
        """
        challenge_types = (
            "non-interactive",
            "managed",
            "interactive",
        )
        for ctype in challenge_types:
            if f"cType: '{ctype}'" in page_content:
                return ctype

        # Check if turnstile captcha is embedded inside the page (Usually inside a closed Shadow iframe)
        selector = Selector(content=page_content)
        if selector.css('script[src*="challenges.cloudflare.com/turnstile/v"]'):
            return "embedded"

        return None