# retoor <retoor@molodetz.nl>
from __future__ import annotations

import asyncio
import logging
from dataclasses import dataclass
from types import TracebackType
from typing import Optional
from urllib.parse import urlparse

from devplacepy.services.jobs.isslop.acquisition.domcapture import (
    DOM_EXTRACT_SCRIPT,
    DOM_INIT_SCRIPT,
    DomSnapshot,
)
from devplacepy.services.jobs.isslop.config import (
    CONSOLE_SAMPLE_CAP,
    DOM_CAPTURE_TIMEOUT_MS,
    HEADER_VALUE_CAP,
)

logger = logging.getLogger(__name__)

BROWSER_USER_AGENT: str = (
    "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36"
)
BROWSER_VIEWPORT: dict[str, int] = {"width": 1440, "height": 900}
BROWSER_LOCALE: str = "en-US"
BROWSER_TIMEZONE: str = "America/New_York"

LAUNCH_ARGS: list[str] = [
    "--no-sandbox",
    "--disable-dev-shm-usage",
    "--disable-blink-features=AutomationControlled",
    "--disable-gpu",
    "--no-first-run",
    "--no-default-browser-check",
    "--disable-extensions",
]

NAVIGATION_TIMEOUT_MS: int = 30000
RENDER_SETTLE_MS: int = 1200
FETCH_TIMEOUT_MS: int = 25000
BROWSER_CLOSE_TIMEOUT_S: float = 15.0
LAUNCH_TIMEOUT_S: float = 60.0


@dataclass(frozen=True)
class RenderResult:
    url: str
    final_url: str
    status: int
    html: str
    content_type: str
    ok: bool
    error: Optional[str] = None


@dataclass(frozen=True)
class FetchResult:
    url: str
    status: int
    body: bytes
    content_type: str
    ok: bool
    error: Optional[str] = None


def browser_available() -> bool:
    try:
        import playwright.async_api  # noqa: F401
        import playwright_stealth  # noqa: F401
    except ImportError as error:
        logger.info("Stealth browser unavailable, will use HTTP client: %s", error)
        return False
    return True


class StealthBrowser:
    def __init__(self, tab_concurrency: int = 4) -> None:
        self._tab_concurrency = max(1, tab_concurrency)
        self._stealth_manager = None
        self._playwright = None
        self._browser = None
        self._context = None
        self._tab_semaphore = asyncio.Semaphore(self._tab_concurrency)

    async def __aenter__(self) -> "StealthBrowser":
        await self._start()
        return self

    async def __aexit__(
        self,
        exc_type: Optional[type[BaseException]],
        exc: Optional[BaseException],
        traceback: Optional[TracebackType],
    ) -> None:
        await self.aclose()

    async def _start(self) -> None:
        from playwright.async_api import async_playwright
        from playwright_stealth import Stealth

        stealth = Stealth()
        self._stealth_manager = stealth.use_async(async_playwright())
        self._playwright = await asyncio.wait_for(self._stealth_manager.__aenter__(), timeout=LAUNCH_TIMEOUT_S)
        self._browser = await self._launch_browser()
        self._context = await self._browser.new_context(
            user_agent=BROWSER_USER_AGENT,
            viewport=BROWSER_VIEWPORT,
            locale=BROWSER_LOCALE,
            timezone_id=BROWSER_TIMEZONE,
            ignore_https_errors=True,
            java_script_enabled=True,
        )
        self._context.set_default_navigation_timeout(NAVIGATION_TIMEOUT_MS)
        self._context.set_default_timeout(NAVIGATION_TIMEOUT_MS)
        logger.info("Stealth browser ready (tab concurrency %d)", self._tab_concurrency)

    async def _launch_browser(self):
        errors: list[str] = []
        for channel in ("chrome", None):
            try:
                kwargs: dict[str, object] = {"headless": True, "args": LAUNCH_ARGS}
                if channel:
                    kwargs["channel"] = channel
                browser = await asyncio.wait_for(
                    self._playwright.chromium.launch(**kwargs), timeout=LAUNCH_TIMEOUT_S
                )
                logger.info("Chromium launched (channel=%s)", channel or "bundled")
                return browser
            except Exception as error:  # noqa: BLE001 - launch failures are heterogeneous, we retry the next channel
                errors.append(f"{channel or 'bundled'}: {error}")
                logger.warning("Browser launch failed on channel %s: %s", channel or "bundled", error)
        raise RuntimeError(f"Could not launch any Chromium build: {'; '.join(errors)}")

    async def _goto_and_settle(self, page: object, url: str) -> object:
        from playwright.async_api import TimeoutError as PlaywrightTimeout

        response = await page.goto(url, wait_until="domcontentloaded", timeout=NAVIGATION_TIMEOUT_MS)
        try:
            await page.wait_for_load_state("networkidle", timeout=RENDER_SETTLE_MS * 3)
        except PlaywrightTimeout:
            await page.wait_for_timeout(RENDER_SETTLE_MS)
        return response

    async def _build_render_result(self, page: object, response: object, url: str) -> RenderResult:
        html = await page.content()
        status = response.status if response else 0
        content_type = ""
        if response is not None:
            headers = await response.all_headers()
            content_type = headers.get("content-type", "")
        final_url = page.url
        return RenderResult(
            url=url,
            final_url=final_url,
            status=status,
            html=html,
            content_type=content_type or "text/html",
            ok=bool(html) and status < 400,
        )

    async def render(self, url: str) -> RenderResult:
        from playwright.async_api import Error as PlaywrightError
        from playwright.async_api import TimeoutError as PlaywrightTimeout

        async with self._tab_semaphore:
            page = None
            try:
                page = await self._context.new_page()
                response = await self._goto_and_settle(page, url)
                return await self._build_render_result(page, response, url)
            except (PlaywrightTimeout, PlaywrightError) as error:
                logger.warning("Render failed for %s: %s", url, error)
                return RenderResult(url, url, 0, "", "", False, str(error))
            finally:
                if page is not None:
                    try:
                        await page.close()
                    except Exception as error:  # noqa: BLE001 - a failed tab close must never abort the crawl
                        logger.debug("Tab close error for %s: %s", url, error)

    async def _evaluate_dom_and_screenshot(self, page: object) -> tuple[dict, Optional[bytes]]:
        from playwright.async_api import Error as PlaywrightError
        from playwright.async_api import TimeoutError as PlaywrightTimeout

        dom: dict = {}
        try:
            dom = await page.evaluate(DOM_EXTRACT_SCRIPT)
        except (PlaywrightTimeout, PlaywrightError) as error:
            logger.warning("DOM evaluate failed: %s", error)
            dom = {}
        screenshot_bytes: Optional[bytes] = None
        try:
            screenshot_bytes = await page.screenshot(full_page=False)
        except (PlaywrightTimeout, PlaywrightError) as error:
            logger.warning("Screenshot capture failed: %s", error)
            screenshot_bytes = None
        return dom if isinstance(dom, dict) else {}, screenshot_bytes

    async def capture(self, url: str) -> DomSnapshot:
        from playwright.async_api import Error as PlaywrightError
        from playwright.async_api import TimeoutError as PlaywrightTimeout

        async with self._tab_semaphore:
            page = None
            try:
                page = await self._context.new_page()
                console_warnings: list[str] = []
                console_errors: list[str] = []

                def _on_console(message: object) -> None:
                    try:
                        kind = message.type
                        text = message.text[:HEADER_VALUE_CAP]
                    except Exception as error:
                        logger.debug("Console handler error: %s", error)
                        return
                    if kind == "warning" and len(console_warnings) < CONSOLE_SAMPLE_CAP:
                        console_warnings.append(text)
                    elif kind == "error" and len(console_errors) < CONSOLE_SAMPLE_CAP:
                        console_errors.append(text)

                page.on("console", _on_console)

                response_headers: dict[str, str] = {}
                resource_hosts: list[str] = []
                seen_hosts: set[str] = set()
                captured_main = False

                def _on_response(response: object) -> None:
                    nonlocal captured_main
                    try:
                        hostname = urlparse(response.url).hostname
                        if hostname and hostname not in seen_hosts:
                            seen_hosts.add(hostname)
                            resource_hosts.append(hostname)
                        if not captured_main and response.frame == page.main_frame:
                            captured_main = True
                            for key, value in response.headers.items():
                                response_headers[key] = str(value)[:HEADER_VALUE_CAP]
                    except Exception as error:
                        logger.debug("Response handler error: %s", error)

                page.on("response", _on_response)
                await page.add_init_script(DOM_INIT_SCRIPT)
                response = await self._goto_and_settle(page, url)
                render = await self._build_render_result(page, response, url)
                dom: dict = {}
                screenshot_bytes: Optional[bytes] = None
                try:
                    dom, screenshot_bytes = await asyncio.wait_for(
                        self._evaluate_dom_and_screenshot(page),
                        timeout=DOM_CAPTURE_TIMEOUT_MS / 1000,
                    )
                except asyncio.TimeoutError as error:
                    logger.warning("DOM capture timed out for %s: %s", url, error)
                return DomSnapshot(
                    render=render,
                    dom=dom,
                    console_warnings=console_warnings,
                    console_errors=console_errors,
                    response_headers=response_headers,
                    resource_hosts=resource_hosts,
                    screenshot_bytes=screenshot_bytes,
                )
            except (PlaywrightTimeout, PlaywrightError) as error:
                logger.warning("Capture failed for %s: %s", url, error)
                return DomSnapshot(render=RenderResult(url, url, 0, "", "", False, str(error)))
            finally:
                if page is not None:
                    try:
                        await page.close()
                    except Exception as error:
                        logger.debug("Tab close error for %s: %s", url, error)

    async def fetch(self, url: str, max_bytes: int) -> FetchResult:
        from playwright.async_api import Error as PlaywrightError

        try:
            response = await self._context.request.get(url, timeout=FETCH_TIMEOUT_MS)
            body = await response.body()
            if len(body) > max_bytes:
                body = body[:max_bytes]
            content_type = response.headers.get("content-type", "")
            return FetchResult(url, response.status, body, content_type, response.status < 400)
        except (PlaywrightError, asyncio.TimeoutError) as error:
            logger.warning("Fetch failed for %s: %s", url, error)
            return FetchResult(url, 0, b"", "", False, str(error))

    async def aclose(self) -> None:
        for closer, label in (
            (self._context, "context"),
            (self._browser, "browser"),
        ):
            if closer is not None:
                try:
                    await asyncio.wait_for(closer.close(), timeout=BROWSER_CLOSE_TIMEOUT_S)
                except Exception as error:  # noqa: BLE001 - cleanup must be exhaustive
                    logger.warning("Error closing %s: %s", label, error)
        self._context = None
        self._browser = None
        if self._stealth_manager is not None:
            try:
                await asyncio.wait_for(
                    self._stealth_manager.__aexit__(None, None, None), timeout=BROWSER_CLOSE_TIMEOUT_S
                )
            except Exception as error:  # noqa: BLE001 - playwright teardown must not leak
                logger.warning("Error stopping playwright: %s", error)
        self._stealth_manager = None
        self._playwright = None
        logger.info("Stealth browser closed")
