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

import asyncio
import logging
import re
import ssl
from dataclasses import dataclass, field
from os import environ
from pathlib import Path
from typing import Iterable, Mapping, Sequence
from urllib.parse import unquote, urlsplit

import httpx

logger = logging.getLogger("stealth")

OUTBOUND_PROXY_URL: str | None = environ.get("DEVPLACE_OUTBOUND_PROXY_URL", "").strip() or None


def _supported_accept_encoding() -> str:
    encodings = ["gzip", "deflate"]
    try:
        import brotli  # noqa: F401
        encodings.append("br")
    except ImportError:
        logger.debug("brotli not installed; not advertising 'br'")
    try:
        import zstandard  # noqa: F401
        encodings.append("zstd")
    except ImportError:
        logger.debug("zstandard not installed; not advertising 'zstd'")
    return ", ".join(encodings)


ACCEPT_ENCODING: str = _supported_accept_encoding()


CHROME_CIPHER_SUITES: str = ":".join(
    [
        "TLS_AES_128_GCM_SHA256",
        "TLS_AES_256_GCM_SHA384",
        "TLS_CHACHA20_POLY1305_SHA256",
        "ECDHE-ECDSA-AES128-GCM-SHA256",
        "ECDHE-RSA-AES128-GCM-SHA256",
        "ECDHE-ECDSA-AES256-GCM-SHA384",
        "ECDHE-RSA-AES256-GCM-SHA384",
        "ECDHE-ECDSA-CHACHA20-POLY1305",
        "ECDHE-RSA-CHACHA20-POLY1305",
        "ECDHE-RSA-AES128-SHA",
        "ECDHE-RSA-AES256-SHA",
        "AES128-GCM-SHA256",
        "AES256-GCM-SHA384",
        "AES128-SHA",
        "AES256-SHA",
    ]
)

ALPN_PROTOCOLS: tuple[str, ...] = ("h2", "http/1.1")

DEFAULT_CHROME_VERSION: str = "146"

DEFAULT_CHUNK_SIZE: int = 65536

IN_MEMORY_DOWNLOAD_CAP: int = 100 * 1024 * 1024

IMAGE_EXTENSIONS: tuple[str, ...] = (
    ".jpg",
    ".jpeg",
    ".png",
    ".gif",
    ".webp",
    ".avif",
    ".bmp",
    ".svg",
    ".ico",
    ".tiff",
)

MAX_FILENAME_LENGTH: int = 200

MAX_CONCURRENT_DOWNLOADS: int = 64


@dataclass(frozen=True)
class ChromeProfile:
    version: str = DEFAULT_CHROME_VERSION
    platform: str = "Windows"
    platform_token: str = "Windows NT 10.0; Win64; x64"
    accept_language: str = "en-US,en;q=0.9"

    @property
    def user_agent(self) -> str:
        return (
            f"Mozilla/5.0 ({self.platform_token}) AppleWebKit/537.36 "
            f"(KHTML, like Gecko) Chrome/{self.version}.0.0.0 Safari/537.36"
        )

    @property
    def sec_ch_ua(self) -> str:
        return (
            f'"Google Chrome";v="{self.version}", '
            f'"Chromium";v="{self.version}", '
            f'"Not)A;Brand";v="24"'
        )

    @classmethod
    def windows(cls, version: str = DEFAULT_CHROME_VERSION) -> "ChromeProfile":
        return cls(version=version, platform="Windows", platform_token="Windows NT 10.0; Win64; x64")

    @classmethod
    def macos(cls, version: str = DEFAULT_CHROME_VERSION) -> "ChromeProfile":
        return cls(
            version=version,
            platform="macOS",
            platform_token="Macintosh; Intel Mac OS X 10_15_7",
        )

    @classmethod
    def linux(cls, version: str = DEFAULT_CHROME_VERSION) -> "ChromeProfile":
        return cls(version=version, platform="Linux", platform_token="X11; Linux x86_64")


@dataclass
class DownloadResult:
    url: str
    path: Path | None
    status_code: int | None
    bytes_written: int
    ok: bool
    error: str | None = None


@dataclass
class BytesResult:
    url: str
    final_url: str
    status_code: int
    content_type: str
    data: bytes
    truncated: bool
    error: str | None = None

    def as_dict(self) -> dict[str, object]:
        return {
            "url": self.url,
            "final_url": self.final_url,
            "status_code": self.status_code,
            "content_type": self.content_type,
            "data": self.data,
            "truncated": self.truncated,
            "error": self.error,
        }


def resource_kind_for_url(url: str) -> str:
    path = urlsplit(url).path.lower()
    return "image" if path.endswith(IMAGE_EXTENSIONS) else "empty"


CONSENT_GATE_MARKERS: tuple[str, ...] = ("privacy gate", "privacy-gate", "cmpproperties", "consent.js")

CONSENT_GATE_CALLBACK_RE = re.compile(r"decodeURIComponent\(\s*['\"]([^'\"]+)['\"]\s*\)")


def detect_consent_gate(html: str) -> str | None:
    lowered = html.lower()
    if not any(marker in lowered for marker in CONSENT_GATE_MARKERS):
        return None
    match = CONSENT_GATE_CALLBACK_RE.search(html)
    if not match:
        return None
    candidate = unquote(match.group(1))
    if not candidate.startswith(("http://", "https://")):
        return None
    return candidate


def build_chrome_ssl_context() -> ssl.SSLContext:
    context = ssl.create_default_context()
    context.check_hostname = True
    context.verify_mode = ssl.CERT_REQUIRED
    context.minimum_version = ssl.TLSVersion.TLSv1_2
    context.maximum_version = ssl.TLSVersion.TLSv1_3
    context.set_ciphers(CHROME_CIPHER_SUITES)
    context.set_alpn_protocols(list(ALPN_PROTOCOLS))
    context.options |= ssl.OP_NO_COMPRESSION
    logger.debug("Built Chrome-aligned SSL context with %d ciphers", CHROME_CIPHER_SUITES.count(":") + 1)
    return context


_SHARED_SSL_CONTEXT: ssl.SSLContext | None = None

DEFAULT_PROFILE: ChromeProfile = ChromeProfile.windows()

try:
    from devplacepy.curl_transport import CurlTransport, IMPERSONATE_TARGET

    CURL_AVAILABLE: bool = True
    logger.info("Stealth transport: curl_cffi impersonation active (%s)", IMPERSONATE_TARGET)
except ImportError as error:
    CurlTransport = None
    IMPERSONATE_TARGET = ""
    CURL_AVAILABLE = False
    logger.warning("curl_cffi unavailable (%s); falling back to httpx stealth transport", error)


def curl_active() -> bool:
    return CURL_AVAILABLE


def chrome_ssl_context() -> ssl.SSLContext:
    global _SHARED_SSL_CONTEXT
    if _SHARED_SSL_CONTEXT is None:
        _SHARED_SSL_CONTEXT = build_chrome_ssl_context()
    return _SHARED_SSL_CONTEXT


def chrome_base_headers(profile: ChromeProfile | None = None) -> dict[str, str]:
    active = profile or DEFAULT_PROFILE
    return {
        "sec-ch-ua": active.sec_ch_ua,
        "sec-ch-ua-mobile": "?0",
        "sec-ch-ua-platform": f'"{active.platform}"',
        "user-agent": active.user_agent,
        "accept-encoding": ACCEPT_ENCODING,
        "accept-language": active.accept_language,
    }


def configured_proxy_url() -> str | None:
    try:
        from devplacepy.database import get_setting

        configured = get_setting("outbound_proxy_url", "").strip()
    except ImportError:
        configured = ""
    return configured or OUTBOUND_PROXY_URL


def stealth_transport(
    profile: ChromeProfile | None = None,
    *,
    proxy: str | None = None,
    **kwargs: object,
) -> httpx.AsyncBaseTransport:
    effective_proxy = proxy or configured_proxy_url()
    if CURL_AVAILABLE:
        return CurlTransport(proxy=effective_proxy)
    return httpx.AsyncHTTPTransport(verify=chrome_ssl_context(), http2=True, proxy=effective_proxy, **kwargs)


def stealth_async_client(
    *,
    profile: ChromeProfile | None = None,
    headers: Mapping[str, str] | None = None,
    transport: httpx.AsyncBaseTransport | None = None,
    proxy: str | None = None,
    **kwargs: object,
) -> httpx.AsyncClient:
    merged_headers: dict[str, str] = {} if CURL_AVAILABLE else chrome_base_headers(profile)
    if headers:
        merged_headers.update({key.lower(): value for key, value in headers.items()})
    if transport is None:
        transport = stealth_transport(profile, proxy=proxy)
    return httpx.AsyncClient(
        transport=transport,
        trust_env=True,
        headers=merged_headers,
        **kwargs,
    )


def stealth_sync_client(
    *,
    profile: ChromeProfile | None = None,
    headers: Mapping[str, str] | None = None,
    proxy: str | None = None,
    **kwargs: object,
) -> httpx.Client:
    merged_headers = chrome_base_headers(profile)
    if headers:
        merged_headers.update({key.lower(): value for key, value in headers.items()})
    return httpx.Client(
        http2=True,
        verify=chrome_ssl_context(),
        trust_env=True,
        headers=merged_headers,
        proxy=proxy or configured_proxy_url(),
        **kwargs,
    )


def slugify_filename(name: str) -> str:
    name = unquote(name).strip()
    name = name.replace("\\", "/").split("/")[-1]
    name = name.split("?")[0].split("#")[0]
    name = re.sub(r"[^A-Za-z0-9._-]+", "-", name).strip("-._")
    if not name:
        name = "download"
    if len(name) > MAX_FILENAME_LENGTH:
        stem, _, suffix = name.rpartition(".")
        if stem and len(suffix) <= 16:
            keep = MAX_FILENAME_LENGTH - len(suffix) - 1
            name = f"{stem[:keep]}.{suffix}"
        else:
            name = name[:MAX_FILENAME_LENGTH]
    return name


def resolve_destination(directory: Path, url: str, override_name: str | None) -> Path:
    directory = directory.resolve()
    raw_name = override_name if override_name else Path(urlsplit(url).path).name
    safe_name = slugify_filename(raw_name)
    candidate = (directory / safe_name).resolve()
    if directory != candidate.parent:
        raise ValueError(f"Refusing path traversal for {url!r} -> {candidate}")
    return candidate


class ChromeStealthClient:

    def __init__(
        self,
        profile: ChromeProfile | None = None,
        *,
        http2: bool = True,
        timeout: float = 30.0,
        follow_redirects: bool = True,
        proxy: str | None = None,
        trust_env: bool = True,
        max_connections: int = 100,
        max_keepalive_connections: int = 20,
    ) -> None:
        self.profile = profile or ChromeProfile.windows()
        self._ssl_context = build_chrome_ssl_context()
        limits = httpx.Limits(
            max_connections=max_connections,
            max_keepalive_connections=max_keepalive_connections,
        )
        self._client = httpx.AsyncClient(
            http2=http2,
            verify=self._ssl_context,
            timeout=httpx.Timeout(timeout),
            follow_redirects=follow_redirects,
            limits=limits,
            proxy=proxy,
            trust_env=trust_env,
            headers=self._base_headers(),
        )
        logger.info(
            "ChromeStealthClient ready: Chrome %s on %s, http2=%s, proxy=%s",
            self.profile.version,
            self.profile.platform,
            http2,
            bool(proxy),
        )

    def _base_headers(self) -> dict[str, str]:
        return {
            "sec-ch-ua": self.profile.sec_ch_ua,
            "sec-ch-ua-mobile": "?0",
            "sec-ch-ua-platform": f'"{self.profile.platform}"',
            "user-agent": self.profile.user_agent,
            "accept-encoding": ACCEPT_ENCODING,
            "accept-language": self.profile.accept_language,
        }

    def _navigation_headers(self, referer: str | None) -> dict[str, str]:
        headers = {
            "upgrade-insecure-requests": "1",
            "accept": (
                "text/html,application/xhtml+xml,application/xml;q=0.9,"
                "image/avif,image/webp,image/apng,*/*;q=0.8,"
                "application/signed-exchange;v=b3;q=0.7"
            ),
            "sec-fetch-site": "same-origin" if referer else "none",
            "sec-fetch-mode": "navigate",
            "sec-fetch-user": "?1",
            "sec-fetch-dest": "document",
        }
        if referer:
            headers["referer"] = referer
        return headers

    def _resource_headers(self, dest: str, referer: str | None) -> dict[str, str]:
        accept = {
            "image": "image/avif,image/webp,image/apng,image/svg+xml,image/*,*/*;q=0.8",
            "empty": "*/*",
        }.get(dest, "*/*")
        headers = {
            "accept": accept,
            "sec-fetch-site": "same-origin" if referer else "cross-site",
            "sec-fetch-mode": "no-cors" if dest == "image" else "cors",
            "sec-fetch-dest": dest,
        }
        if referer:
            headers["referer"] = referer
        return headers

    def _api_headers(self, referer: str | None) -> dict[str, str]:
        headers = {
            "accept": "application/json, text/plain, */*",
            "sec-fetch-site": "same-origin" if referer else "cross-site",
            "sec-fetch-mode": "cors",
            "sec-fetch-dest": "empty",
        }
        if referer:
            headers["referer"] = referer
        return headers

    @staticmethod
    def _json_or_error(response: httpx.Response) -> dict[str, object]:
        if response.status_code >= 400:
            return {
                "error": f"HTTP {response.status_code}",
                "status_code": response.status_code,
                "body": response.text[:5000],
            }
        try:
            return response.json()
        except ValueError:
            return {
                "error": "Invalid JSON response",
                "status_code": response.status_code,
                "body": response.text[:5000],
            }

    async def __aenter__(self) -> "ChromeStealthClient":
        return self

    async def __aexit__(self, *exc: object) -> None:
        await self.aclose()

    async def aclose(self) -> None:
        await self._client.aclose()
        logger.debug("ChromeStealthClient closed")

    async def _send(self, method: str, url: str, headers: dict[str, str], **kwargs: object) -> httpx.Response:
        try:
            return await self._client.request(method, url, headers=headers, **kwargs)
        except httpx.DecodingError:
            logger.warning("content decode failed for %s; retrying without br/zstd", url)
            retry_headers = dict(headers)
            retry_headers["accept-encoding"] = "gzip, deflate"
            return await self._client.request(method, url, headers=retry_headers, **kwargs)

    async def request(
        self,
        method: str,
        url: str,
        *,
        referer: str | None = None,
        headers: Mapping[str, str] | None = None,
        **kwargs: object,
    ) -> httpx.Response:
        merged = self._navigation_headers(referer)
        if headers:
            merged.update({key.lower(): value for key, value in headers.items()})
        logger.info("%s %s", method.upper(), url)
        response = await self._send(method, url, merged, **kwargs)
        logger.debug("%s %s -> %s (%s)", method.upper(), url, response.status_code, response.http_version)
        return response

    async def get(self, url: str, *, referer: str | None = None, **kwargs: object) -> httpx.Response:
        return await self.request("GET", url, referer=referer, **kwargs)

    async def post(self, url: str, *, referer: str | None = None, **kwargs: object) -> httpx.Response:
        return await self.request("POST", url, referer=referer, **kwargs)

    def stream(
        self,
        method: str,
        url: str,
        *,
        referer: str | None = None,
        dest: str = "empty",
        headers: Mapping[str, str] | None = None,
        **kwargs: object,
    ):
        merged = self._resource_headers(dest, referer)
        if headers:
            merged.update({key.lower(): value for key, value in headers.items()})
        return self._client.stream(method, url, headers=merged, **kwargs)

    async def download_file(
        self,
        url: str,
        destination: str | Path,
        *,
        filename: str | None = None,
        referer: str | None = None,
        chunk_size: int = DEFAULT_CHUNK_SIZE,
        dest_type: str = "empty",
    ) -> DownloadResult:
        directory = Path(destination)
        await asyncio.to_thread(directory.mkdir, parents=True, exist_ok=True)
        try:
            target = resolve_destination(directory, url, filename)
        except ValueError as error:
            logger.error("Destination rejected for %s: %s", url, error)
            return DownloadResult(url=url, path=None, status_code=None, bytes_written=0, ok=False, error=str(error))

        written = 0
        try:
            async with self.stream("GET", url, referer=referer, dest=dest_type) as response:
                response.raise_for_status()
                handle = await asyncio.to_thread(target.open, "wb")
                try:
                    async for chunk in response.aiter_bytes(chunk_size):
                        await asyncio.to_thread(handle.write, chunk)
                        written += len(chunk)
                finally:
                    await asyncio.to_thread(handle.close)
            logger.info("Downloaded %s -> %s (%d bytes)", url, target, written)
            return DownloadResult(
                url=url,
                path=target,
                status_code=response.status_code,
                bytes_written=written,
                ok=True,
            )
        except httpx.HTTPStatusError as error:
            logger.error("HTTP %s downloading %s", error.response.status_code, url)
            return DownloadResult(
                url=url,
                path=None,
                status_code=error.response.status_code,
                bytes_written=written,
                ok=False,
                error=f"HTTP {error.response.status_code}",
            )
        except httpx.HTTPError as error:
            logger.error("Transport error downloading %s: %s", url, error)
            return DownloadResult(url=url, path=None, status_code=None, bytes_written=written, ok=False, error=str(error))

    async def download_files(
        self,
        urls: Iterable[str] | Mapping[str, str],
        destination: str | Path,
        *,
        concurrency: int = 8,
        referer: str | None = None,
        chunk_size: int = DEFAULT_CHUNK_SIZE,
        dest_type: str = "empty",
    ) -> list[DownloadResult]:
        if isinstance(urls, Mapping):
            jobs: Sequence[tuple[str, str | None]] = [(url, name) for url, name in urls.items()]
        else:
            jobs = [(url, None) for url in urls]

        bounded = max(1, min(concurrency, MAX_CONCURRENT_DOWNLOADS))
        semaphore = asyncio.Semaphore(bounded)
        logger.info("Starting bulk download of %d files with concurrency %d", len(jobs), bounded)

        async def worker(url: str, name: str | None) -> DownloadResult:
            async with semaphore:
                return await self.download_file(
                    url,
                    destination,
                    filename=name,
                    referer=referer,
                    chunk_size=chunk_size,
                    dest_type=dest_type,
                )

        results = await asyncio.gather(*(worker(url, name) for url, name in jobs))
        succeeded = sum(1 for result in results if result.ok)
        logger.info("Bulk download finished: %d succeeded, %d failed", succeeded, len(results) - succeeded)
        return list(results)

    async def download_bytes(
        self,
        url: str,
        *,
        referer: str | None = None,
        max_bytes: int = IN_MEMORY_DOWNLOAD_CAP,
        chunk_size: int = DEFAULT_CHUNK_SIZE,
        dest_type: str | None = None,
    ) -> BytesResult:
        kind = dest_type if dest_type else resource_kind_for_url(url)
        buffer = bytearray()
        try:
            async with self.stream("GET", url, referer=referer, dest=kind) as response:
                content_type = response.headers.get("content-type", "")
                async for chunk in response.aiter_bytes(chunk_size):
                    buffer.extend(chunk)
                    if len(buffer) > max_bytes:
                        break
                final_url = str(response.url)
            truncated = len(buffer) > max_bytes
            data = bytes(buffer[:max_bytes])
            logger.debug("download_bytes %s -> %d bytes (status %s)", url, len(data), response.status_code)
            return BytesResult(
                url=url,
                final_url=final_url,
                status_code=response.status_code,
                content_type=content_type,
                data=data,
                truncated=truncated,
            )
        except httpx.HTTPError as error:
            logger.error("download_bytes failed for %s: %s", url, error)
            return BytesResult(
                url=url,
                final_url=url,
                status_code=0,
                content_type="",
                data=b"",
                truncated=False,
                error=str(error),
            )

    async def post_json(
        self,
        url: str,
        payload: object,
        *,
        headers: Mapping[str, str] | None = None,
        referer: str | None = None,
        timeout: float | None = None,
    ) -> dict[str, object]:
        request_headers = self._api_headers(referer)
        if headers:
            request_headers.update({key.lower(): value for key, value in headers.items()})
        kwargs: dict[str, object] = {"json": payload}
        if timeout is not None:
            kwargs["timeout"] = httpx.Timeout(timeout)
        logger.info("POST(json) %s", url)
        try:
            response = await self._send("POST", url, request_headers, **kwargs)
        except httpx.HTTPError as error:
            logger.error("post_json transport error for %s: %s", url, error)
            return {"error": f"{type(error).__name__}: {error}"}
        return self._json_or_error(response)

    async def get_json(
        self,
        url: str,
        *,
        referer: str | None = None,
        timeout: float | None = None,
    ) -> dict[str, object]:
        kwargs: dict[str, object] = {}
        if timeout is not None:
            kwargs["timeout"] = httpx.Timeout(timeout)
        logger.info("GET(json) %s", url)
        try:
            response = await self._send("GET", url, self._api_headers(referer), **kwargs)
        except httpx.HTTPError as error:
            logger.error("get_json transport error for %s: %s", url, error)
            return {"error": f"{type(error).__name__}: {error}"}
        return self._json_or_error(response)


__all__ = [
    "ChromeStealthClient",
    "ChromeProfile",
    "DownloadResult",
    "BytesResult",
    "build_chrome_ssl_context",
    "chrome_ssl_context",
    "chrome_base_headers",
    "stealth_transport",
    "stealth_async_client",
    "stealth_sync_client",
    "curl_active",
    "resource_kind_for_url",
    "slugify_filename",
    "detect_consent_gate",
    "configured_proxy_url",
]
