from functools import lru_cache
from re import compile as re_compile

from curl_cffi.requests import Response as CurlResponse
from playwright._impl._errors import Error as PlaywrightError
from playwright.sync_api import Page as SyncPage, Response as SyncResponse
from playwright.async_api import Page as AsyncPage, Response as AsyncResponse

from scrapling.core.utils import log
from .custom import Response, StatusText
from scrapling.core._types import Dict, List, Optional

__CHARSET_RE__ = re_compile(r"charset=([\w-]+)")


class ResponseFactory:
    """
    Factory class for creating `Response` objects from various sources.

    This class provides multiple static and instance methods for building standardized `Response` objects
    from diverse input sources such as Playwright responses, asynchronous Playwright responses,
    and raw HTTP request responses. It supports handling response histories, constructing the proper
    response objects, and managing encoding, headers, cookies, and other attributes.
    """

    @classmethod
    @lru_cache(maxsize=16)
    def __extract_browser_encoding(cls, content_type: str | None, default: str = "utf-8") -> str:
        """Extract browser encoding from headers.
        Ex: from header "content-type: text/html; charset=utf-8" -> "utf-8
        """
        if content_type:
            # Because Playwright can't do that by themselves like all libraries for some reason :3
            match = __CHARSET_RE__.search(content_type)
            return match.group(1) if match else default
        return default

    @classmethod
    def _process_response_history(cls, first_response: SyncResponse, parser_arguments: Dict) -> list[Response]:
        """Process response history to build a list of `Response` objects"""
        history: list[Response] = []
        current_request = first_response.request.redirected_from

        try:
            while current_request:
                try:
                    current_response = current_request.response()
                    history.insert(
                        0,
                        Response(
                            **{
                                "url": current_request.url,
                                # using current_response.text() will trigger "Error: Response.text: Response body is unavailable for redirect responses"
                                "content": "",
                                "status": current_response.status if current_response else 301,
                                "reason": (current_response.status_text or StatusText.get(current_response.status))
                                if current_response
                                else StatusText.get(301),
                                "encoding": cls.__extract_browser_encoding(
                                    current_response.headers.get("content-type", "")
                                )
                                if current_response
                                else "utf-8",
                                "cookies": tuple(),
                                "headers": current_response.all_headers() if current_response else {},
                                "request_headers": current_request.all_headers(),
                                **parser_arguments,
                            }
                        ),
                    )
                except Exception as e:  # pragma: no cover
                    log.error(f"Error processing redirect: {e}")
                    break

                current_request = current_request.redirected_from
        except Exception as e:  # pragma: no cover
            log.error(f"Error processing response history: {e}")

        return history

    @classmethod
    def from_playwright_response(
        cls,
        page: Optional[SyncPage],
        first_response: SyncResponse,
        final_response: Optional[SyncResponse],
        parser_arguments: Dict,
        meta: Optional[Dict] = None,
        xhr_captured: Optional[List[SyncResponse]] = None,
        collect_history: bool = True,
    ) -> Response:
        """
        Transforms a Playwright response into an internal `Response` object, encapsulating
        the page's content, response status, headers, and relevant metadata.

        The function handles potential issues, such as empty or missing final responses,
        by falling back to the first response if necessary. Encoding and status text
        are also derived from the provided response headers or reasonable defaults.
        Additionally, the page content and cookies are extracted for further use.

        :param page: A synchronous Playwright `Page` instance that represents the current browser page. Required to retrieve the page's URL, cookies, and content.
        :param final_response: The last response received for the given request from the Playwright instance. Typically used as the main response object to derive status, headers, and other metadata.
        :param first_response: An earlier or initial Playwright `Response` object that may serve as a fallback response in the absence of the final one.
        :param parser_arguments: A dictionary containing additional arguments needed for parsing or further customization of the returned `Response`. These arguments are dynamically unpacked into
            the `Response` object.
        :param meta: Additional meta data to be saved with the response.
        :param xhr_captured: Optional list of captured Playwright XHR/fetch responses to convert and attach to the returned Response.
        :param collect_history: Optional boolean indicating whether to collect redirections history or not.
        :return: A fully populated `Response` object containing the page's URL, content, status, headers, cookies, and other derived metadata.
        :rtype: Response
        """
        # In case we didn't catch a document type somehow
        final_response = final_response if final_response else first_response
        if not final_response:
            raise ValueError("Failed to get a response from the page")

        encoding = cls.__extract_browser_encoding(final_response.headers.get("content-type", ""))
        # PlayWright API sometimes give empty status text for some reason!
        status_text = final_response.status_text or StatusText.get(final_response.status)

        history = cls._process_response_history(first_response, parser_arguments) if collect_history else []
        try:
            if page and "html" in final_response.all_headers().get("content-type", ""):
                page_content = cls._get_page_content(page).encode("utf-8")
            else:
                page_content = final_response.body()
        except Exception as e:  # pragma: no cover
            log.error(f"Error getting page content: {e}")
            page_content = b""

        response = Response(
            **{
                "url": page.url if page else first_response.url,
                "content": page_content,
                "status": final_response.status,
                "reason": status_text,
                "encoding": encoding,
                "cookies": tuple(dict(cookie) for cookie in page.context.cookies()) if page else {},
                "headers": first_response.all_headers(),
                "request_headers": first_response.request.all_headers(),
                "history": history,
                "meta": meta,
                **parser_arguments,
            }
        )
        if xhr_captured:
            response.captured_xhr = [
                cls.from_playwright_response(None, p, None, {}, collect_history=False) for p in xhr_captured
            ]
        return response

    @classmethod
    async def _async_process_response_history(
        cls, first_response: AsyncResponse, parser_arguments: Dict
    ) -> list[Response]:
        """Process response history to build a list of `Response` objects"""
        history: list[Response] = []
        current_request = first_response.request.redirected_from

        try:
            while current_request:
                try:
                    current_response = await current_request.response()
                    history.insert(
                        0,
                        Response(
                            **{
                                "url": current_request.url,
                                # using current_response.text() will trigger "Error: Response.text: Response body is unavailable for redirect responses"
                                "content": "",
                                "status": current_response.status if current_response else 301,
                                "reason": (current_response.status_text or StatusText.get(current_response.status))
                                if current_response
                                else StatusText.get(301),
                                "encoding": cls.__extract_browser_encoding(
                                    current_response.headers.get("content-type", "")
                                )
                                if current_response
                                else "utf-8",
                                "cookies": tuple(),
                                "headers": await current_response.all_headers() if current_response else {},
                                "request_headers": await current_request.all_headers(),
                                **parser_arguments,
                            }
                        ),
                    )
                except Exception as e:  # pragma: no cover
                    log.error(f"Error processing redirect: {e}")
                    break

                current_request = current_request.redirected_from
        except Exception as e:  # pragma: no cover
            log.error(f"Error processing response history: {e}")

        return history

    @classmethod
    def _get_page_content(cls, page: SyncPage, max_retries: int = 20) -> str:
        """
        A workaround for the Playwright issue with `page.content()` on Windows. Ref.: https://github.com/microsoft/playwright/issues/16108
        :param page: The page to extract content from.
        :param max_retries: Maximum number of retry attempts before raising `RuntimeError`.
        :return:
        """
        for _ in range(max_retries):
            try:
                return page.content() or ""
            except PlaywrightError:
                page.wait_for_timeout(500)
        raise RuntimeError(f"Failed to retrieve the page content after retrying for {max_retries * 500}ms.")

    @classmethod
    async def _get_async_page_content(cls, page: AsyncPage, max_retries: int = 20) -> str:
        """
        A workaround for the Playwright issue with `page.content()` on Windows. Ref.: https://github.com/microsoft/playwright/issues/16108
        :param page: The page to extract content from.
        :param max_retries: Maximum number of retry attempts before raising `RuntimeError`.
        :return:
        """
        for _ in range(max_retries):
            try:
                return (await page.content()) or ""
            except PlaywrightError:
                await page.wait_for_timeout(500)
        raise RuntimeError(f"Failed to retrieve the page content after retrying for {max_retries * 500}ms.")

    @classmethod
    async def from_async_playwright_response(
        cls,
        page: Optional[AsyncPage],
        first_response: AsyncResponse,
        final_response: Optional[AsyncResponse],
        parser_arguments: Dict,
        meta: Optional[Dict] = None,
        xhr_captured: Optional[List[AsyncResponse]] = None,
        collect_history: bool = True,
    ) -> Response:
        """
        Transforms a Playwright response into an internal `Response` object, encapsulating
        the page's content, response status, headers, and relevant metadata.

        The function handles potential issues, such as empty or missing final responses,
        by falling back to the first response if necessary. Encoding and status text
        are also derived from the provided response headers or reasonable defaults.
        Additionally, the page content and cookies are extracted for further use.

        :param page: An asynchronous Playwright `Page` instance that represents the current browser page. Required to retrieve the page's URL, cookies, and content.
        :param final_response: The last response received for the given request from the Playwright instance. Typically used as the main response object to derive status, headers, and other metadata.
        :param first_response: An earlier or initial Playwright `Response` object that may serve as a fallback response in the absence of the final one.
        :param parser_arguments: A dictionary containing additional arguments needed for parsing or further customization of the returned `Response`. These arguments are dynamically unpacked into
            the `Response` object.
        :param meta: Additional meta data to be saved with the response.
        :param xhr_captured: Optional list of captured async Playwright XHR/fetch responses to convert and attach to the returned Response.
        :param collect_history: Optional boolean indicating whether to collect redirections history or not.

        :return: A fully populated `Response` object containing the page's URL, content, status, headers, cookies, and other derived metadata.
        :rtype: Response
        """
        # In case we didn't catch a document type somehow
        final_response = final_response if final_response else first_response
        if not final_response:
            raise ValueError("Failed to get a response from the page")

        encoding = cls.__extract_browser_encoding(final_response.headers.get("content-type", ""))
        # PlayWright API sometimes give empty status text for some reason!
        status_text = final_response.status_text or StatusText.get(final_response.status)

        history = await cls._async_process_response_history(first_response, parser_arguments) if collect_history else []
        try:
            if page and "html" in (await final_response.all_headers()).get("content-type", ""):
                page_content = (await cls._get_async_page_content(page)).encode("utf-8")
            else:
                page_content = await final_response.body()
        except Exception as e:  # pragma: no cover
            log.error(f"Error getting page content in async: {e}")
            page_content = b""

        response = Response(
            **{
                "url": page.url if page else first_response.url,
                "content": page_content,
                "status": final_response.status,
                "reason": status_text,
                "encoding": encoding,
                "cookies": tuple(dict(cookie) for cookie in await page.context.cookies()) if page else {},
                "headers": await first_response.all_headers(),
                "request_headers": await first_response.request.all_headers(),
                "history": history,
                "meta": meta,
                **parser_arguments,
            }
        )
        if xhr_captured:
            response.captured_xhr = [
                await cls.from_async_playwright_response(None, p, None, {}, collect_history=False) for p in xhr_captured
            ]
        return response

    @staticmethod
    def from_http_request(response: CurlResponse, parser_arguments: Dict, meta: Optional[Dict] = None) -> Response:
        """Takes `curl_cffi` response and generates `Response` object from it.

        :param response: `curl_cffi` response object
        :param parser_arguments: Additional arguments to be passed to the `Response` object constructor.
        :param meta: Optional metadata dictionary to attach to the Response.
        :return: A `Response` object that is the same as `Selector` object except it has these added attributes: `status`, `reason`, `cookies`, `headers`, and `request_headers`
        """
        return Response(
            **{
                "url": response.url,
                "content": response.content,
                "status": response.status_code,
                "reason": response.reason,
                "encoding": response.encoding or "utf-8",
                "cookies": dict(response.cookies),
                "headers": dict(response.headers),
                "request_headers": dict(response.request.headers) if response.request else {},
                "method": response.request.method if response.request else "GET",
                "history": response.history,  # https://github.com/lexiforest/curl_cffi/issues/82
                "meta": meta,
                **parser_arguments,
            }
        )
