"""Safe HTTP client used for pages, thumbnails and direct media downloads.

All requests perform structural URL checks, DNS/IP checks and redirect
revalidation. The connector uses :class:`ValidatingResolver`, so the actual
socket target is an address that passed the SSRF policy.
"""

from __future__ import annotations

import asyncio
import logging
import re
from collections.abc import AsyncIterator, Callable
from dataclasses import dataclass
from datetime import UTC, datetime
from email.utils import parsedate_to_datetime
from pathlib import Path
from urllib.parse import urljoin

import aiofiles
import aiohttp
from bs4.dammit import UnicodeDammit

from app.errors import (
    AccessDeniedError,
    ContentTooLargeError,
    DownloadCancelledError,
    InvalidMediaError,
    NetworkError,
    NotFoundError,
    PermanentSiteError,
    RateLimitedError,
)
from app.services.url_validator import NetworkGuard, ValidatingResolver
from app.utils.files import sniff_media_header

logger = logging.getLogger(__name__)

_CONTENT_RANGE_RE = re.compile(r"bytes\s+(\d+)-(\d+)/(\d+|\*)", re.IGNORECASE)
_HTML_TYPES = {
    "text/html",
    "application/xhtml+xml",
    "application/xml",
    "text/xml",
}
_NON_MEDIA_TYPES = {
    "text/html",
    "application/xhtml+xml",
    "application/xml",
    "text/xml",
    "application/json",
    "text/plain",
}
_MEDIA_CHUNK_SIZE = 64 * 1024
_PROBE_SIZE = 64 * 1024


@dataclass(frozen=True, slots=True)
class FetchedPage:
    requested_url: str
    final_url: str
    status: int
    content_type: str
    html: str
    size: int


@dataclass(frozen=True, slots=True)
class DownloadMetadata:
    final_url: str
    status: int
    content_type: str
    total_size: int | None
    etag: str | None
    last_modified: str | None
    resumed: bool


@dataclass(frozen=True, slots=True)
class StreamResult:
    final_url: str
    status: int
    content_type: str
    size: int
    resumed: bool
    etag: str | None
    last_modified: str | None
    total_size: int | None


@dataclass(frozen=True, slots=True)
class MediaStreamInfo:
    """Metadata obtained without writing media bytes to disk."""

    requested_url: str
    final_url: str
    status: int
    content_type: str
    total_size: int | None
    mime_type: str
    extension: str
    container: str


def parse_retry_after(value: str | None) -> float | None:
    if not value:
        return None
    value = value.strip()
    try:
        return max(0.0, float(value))
    except ValueError:
        pass
    try:
        parsed = parsedate_to_datetime(value)
        if parsed.tzinfo is None:
            parsed = parsed.replace(tzinfo=UTC)
        return max(0.0, (parsed - datetime.now(UTC)).total_seconds())
    except (TypeError, ValueError):
        return None


def _decode_html(raw: bytes, charset: str | None) -> str:
    if charset:
        try:
            return raw.decode(charset, errors="replace")
        except LookupError:
            pass
    try:
        return raw.decode("utf-8")
    except UnicodeDecodeError:
        pass
    dammit = UnicodeDammit(raw, is_html=True)
    if dammit.unicode_markup:
        return dammit.unicode_markup
    return raw.decode("utf-8", errors="replace")


class SafeHttpClient:
    def __init__(
        self,
        guard: NetworkGuard,
        *,
        timeout: int = 30,
        user_agent: str = "TelegramVideoBot/1.0",
        max_redirects: int = 5,
        proxy: str | None = None,
    ) -> None:
        self.guard = guard
        self.timeout = timeout
        self.user_agent = user_agent
        self.max_redirects = max_redirects
        self.proxy = proxy
        self._session: aiohttp.ClientSession | None = None

    @property
    def session(self) -> aiohttp.ClientSession:
        if self._session is None or self._session.closed:
            raise RuntimeError("HTTP client is not started")
        return self._session

    async def start(self) -> None:
        if self._session is not None and not self._session.closed:
            return
        connector = aiohttp.TCPConnector(
            resolver=ValidatingResolver(self.guard),
            ttl_dns_cache=0,
            force_close=False,
            enable_cleanup_closed=True,
            ssl=True,
        )
        self._session = aiohttp.ClientSession(
            connector=connector,
            timeout=aiohttp.ClientTimeout(
                total=self.timeout,
                connect=min(self.timeout, 15),
                sock_read=self.timeout,
            ),
            headers={
                "User-Agent": self.user_agent,
                "Accept": "text/html,application/xhtml+xml,*/*;q=0.8",
                "Accept-Language": "en-US,en;q=0.8,fa;q=0.7",
            },
            trust_env=False,
            cookie_jar=aiohttp.CookieJar(unsafe=False),
            raise_for_status=False,
        )

    async def close(self) -> None:
        if self._session is not None:
            await self._session.close()
            self._session = None

    async def _open(
        self,
        method: str,
        url: str,
        *,
        headers: dict[str, str] | None = None,
        stream_timeout: bool = False,
    ) -> tuple[aiohttp.ClientResponse, str]:
        """Open a request and manually revalidate every redirect target."""

        current = url
        for redirect_count in range(self.max_redirects + 1):
            validated = await self.guard.validate(current, resolve_dns=True)
            timeout = (
                aiohttp.ClientTimeout(
                    total=None,
                    connect=min(self.timeout, 15),
                    sock_read=self.timeout,
                )
                if stream_timeout
                else aiohttp.ClientTimeout(
                    total=self.timeout,
                    connect=min(self.timeout, 15),
                    sock_read=self.timeout,
                )
            )
            try:
                response = await self.session.request(
                    method,
                    validated.url,
                    headers=headers,
                    allow_redirects=False,
                    timeout=timeout,
                    proxy=self.proxy,
                )
            except asyncio.CancelledError:
                raise
            except (aiohttp.ClientError, TimeoutError, OSError) as exc:
                raise NetworkError(
                    f"HTTP request failed ({type(exc).__name__})"
                ) from exc
            if response.status in {301, 302, 303, 307, 308}:
                location = response.headers.get("Location")
                response.close()
                if not location:
                    raise PermanentSiteError("redirect response without Location header")
                if redirect_count >= self.max_redirects:
                    raise PermanentSiteError("too many redirects")
                current = urljoin(validated.url, location)
                continue
            return response, validated.url
        raise PermanentSiteError("too many redirects")

    @staticmethod
    def _raise_media_status(response: aiohttp.ClientResponse) -> None:
        if response.status == 429:
            raise RateLimitedError(parse_retry_after(response.headers.get("Retry-After")))
        if response.status in {401, 402, 403, 407, 451}:
            raise AccessDeniedError(response.status)
        if response.status in {404, 410}:
            raise NotFoundError()
        if response.status >= 500:
            raise NetworkError(f"upstream HTTP {response.status}")
        if not 200 <= response.status < 300:
            raise PermanentSiteError(f"unexpected HTTP status {response.status}")

    @staticmethod
    def _media_response_details(
        response: aiohttp.ClientResponse, max_bytes: int
    ) -> tuple[str, int | None]:
        """Return (normalized content type, total size) and enforce the cap."""

        content_type = (
            (response.headers.get("Content-Type") or "").split(";", 1)[0].strip().lower()
        )
        if content_type.startswith("text/") or content_type in _NON_MEDIA_TYPES:
            raise InvalidMediaError(f"server returned a non-media content type: {content_type}")
        if content_type.startswith(("image/", "audio/")):
            raise InvalidMediaError(f"server returned a non-video content type: {content_type}")

        total_size: int | None = None
        content_range = response.headers.get("Content-Range", "")
        match = _CONTENT_RANGE_RE.fullmatch(content_range.strip())
        if match:
            start = int(match.group(1))
            if start != 0:
                raise PermanentSiteError("media response did not start at byte 0")
            total_raw = match.group(3)
            if total_raw.isdigit():
                total_size = int(total_raw)
        elif response.status != 206:
            length_header = response.headers.get("Content-Length")
            if length_header and length_header.isdigit():
                total_size = int(length_header)

        if total_size is not None and total_size > max_bytes:
            raise ContentTooLargeError("declared media size exceeds configured maximum")
        if total_size == 0:
            raise InvalidMediaError("server returned an empty media response")
        return content_type or "application/octet-stream", total_size

    async def probe_media(
        self,
        url: str,
        *,
        max_bytes: int,
        referer: str | None = None,
    ) -> MediaStreamInfo:
        """Validate a direct media URL and read only a small signature probe.

        No media file is created. The probe checks status, declared size,
        content type and container magic bytes before Telegram is contacted.
        """

        headers = {
            "Accept-Encoding": "identity",
            "Accept": "*/*",
            "Range": f"bytes=0-{_PROBE_SIZE - 1}",
        }
        if referer:
            headers["Referer"] = referer
        response, final_url = await self._open(
            "GET", url, headers=headers, stream_timeout=True
        )
        try:
            self._raise_media_status(response)
            content_type, total_size = self._media_response_details(response, max_bytes)
            header = bytearray()
            limit = min(_PROBE_SIZE, max_bytes)
            try:
                while len(header) < min(512, limit):
                    chunk = await response.content.read(limit - len(header))
                    if not chunk:
                        break
                    header.extend(chunk)
            except (aiohttp.ClientError, TimeoutError, OSError) as exc:
                raise NetworkError(
                    f"failed while probing media ({type(exc).__name__})"
                ) from exc
            if not header:
                raise InvalidMediaError("media response is empty")
            detected = sniff_media_header(bytes(header))
            if detected is None:
                raise InvalidMediaError(
                    "media signature is not a permitted video container"
                )
            mime_type, extension, container = detected
            return MediaStreamInfo(
                requested_url=url,
                final_url=final_url,
                status=response.status,
                content_type=content_type,
                total_size=total_size,
                mime_type=mime_type,
                extension=extension,
                container=container,
            )
        finally:
            response.close()

    async def iter_media(
        self,
        url: str,
        *,
        max_bytes: int,
        cancel_event: asyncio.Event | None = None,
        referer: str | None = None,
    ) -> AsyncIterator[bytes]:
        """Yield validated video bytes directly from the source.

        The response is never written to disk. The hard byte cap is enforced
        both from the declared size and while streaming. The caller is
        responsible for consuming the iterator to completion.
        """

        if cancel_event is not None and cancel_event.is_set():
            raise DownloadCancelledError()
        headers = {
            "Accept-Encoding": "identity",
            "Accept": "*/*",
            # Range keeps hotlink-protected CDNs from treating the request as
            # a browser navigation while still asking for the whole object.
            "Range": "bytes=0-",
        }
        if referer:
            headers["Referer"] = referer
        response, _ = await self._open("GET", url, headers=headers, stream_timeout=True)
        try:
            self._raise_media_status(response)
            _, total_size = self._media_response_details(response, max_bytes)
            written = 0
            prefix = bytearray()
            # A single ``read`` may return fewer than 512 bytes even for a
            # valid container. Accumulate enough bytes for signature detection
            # before sending anything to Telegram.
            while len(prefix) < 512:
                if cancel_event is not None and cancel_event.is_set():
                    raise DownloadCancelledError()
                try:
                    chunk = await response.content.read(_MEDIA_CHUNK_SIZE)
                except (aiohttp.ClientError, TimeoutError, OSError) as exc:
                    raise NetworkError(
                        f"media stream interrupted ({type(exc).__name__})"
                    ) from exc
                if not chunk:
                    break
                prefix.extend(chunk)
                written += len(chunk)
                if written > max_bytes:
                    raise ContentTooLargeError("media exceeded configured maximum")
            if not prefix:
                raise InvalidMediaError("media response is empty")
            if sniff_media_header(bytes(prefix[:512])) is None:
                raise InvalidMediaError("media signature is not a permitted video container")
            yield bytes(prefix)

            while True:
                if cancel_event is not None and cancel_event.is_set():
                    raise DownloadCancelledError()
                try:
                    chunk = await response.content.read(_MEDIA_CHUNK_SIZE)
                except (aiohttp.ClientError, TimeoutError, OSError) as exc:
                    raise NetworkError(
                        f"media stream interrupted ({type(exc).__name__})"
                    ) from exc
                if not chunk:
                    break
                written += len(chunk)
                if written > max_bytes:
                    raise ContentTooLargeError("media exceeded configured maximum")
                yield chunk
            if total_size is not None and written != total_size:
                raise NetworkError(
                    f"media stream ended early ({written}/{total_size} bytes)"
                )
        finally:
            response.close()

    async def fetch_html(self, url: str, *, max_bytes: int) -> FetchedPage:
        response, final_url = await self._open("GET", url)
        try:
            if response.status == 429:
                raise RateLimitedError(parse_retry_after(response.headers.get("Retry-After")))
            if response.status in {401, 402, 403, 407, 451}:
                raise AccessDeniedError(response.status)
            if response.status in {404, 410}:
                raise NotFoundError()
            if response.status >= 500:
                raise NetworkError(f"upstream HTTP {response.status}")
            if not 200 <= response.status < 300:
                raise PermanentSiteError(f"unexpected HTTP status {response.status}")
            content_type = (response.headers.get("Content-Type") or "").split(";", 1)[0].strip().lower()
            length_header = response.headers.get("Content-Length")
            if length_header and length_header.isdigit() and int(length_header) > max_bytes:
                raise ContentTooLargeError("HTML response exceeds configured limit")
            try:
                raw = await response.content.read(max_bytes + 1)
            except (aiohttp.ClientError, TimeoutError, OSError) as exc:
                raise NetworkError(f"failed while reading page ({type(exc).__name__})") from exc
            if len(raw) > max_bytes:
                raise ContentTooLargeError("HTML response exceeds configured limit")
            if not raw:
                raise PermanentSiteError("page is empty")
            charset = response.charset
            text = _decode_html(raw, charset)
            looks_html = text.lstrip()[:200].lower().startswith(
                ("<!doctype", "<html", "<head", "<body", "<?xml")
            )
            if content_type and content_type not in _HTML_TYPES and not (
                content_type.startswith("text/") and looks_html
            ):
                raise PermanentSiteError(f"unsupported page content type: {content_type}")
            if not looks_html and content_type not in _HTML_TYPES:
                raise PermanentSiteError("response does not look like HTML")
            return FetchedPage(
                requested_url=url,
                final_url=final_url,
                status=response.status,
                content_type=content_type or "text/html",
                html=text,
                size=len(raw),
            )
        finally:
            response.close()

    async def fetch_bytes(
        self,
        url: str,
        *,
        max_bytes: int,
        allowed_prefixes: tuple[bytes, ...] = (),
        referer: str | None = None,
    ) -> bytes:
        response, _ = await self._open(
            "GET", url, headers={"Referer": referer} if referer else None
        )
        try:
            if response.status == 429:
                raise RateLimitedError(parse_retry_after(response.headers.get("Retry-After")))
            if response.status in {401, 403, 407, 451}:
                raise AccessDeniedError(response.status)
            if response.status in {404, 410}:
                raise NotFoundError()
            if not 200 <= response.status < 300:
                raise NetworkError(f"upstream HTTP {response.status}")
            length_header = response.headers.get("Content-Length")
            if length_header and length_header.isdigit() and int(length_header) > max_bytes:
                raise ContentTooLargeError()
            raw = await response.content.read(max_bytes + 1)
            if len(raw) > max_bytes or not raw:
                raise ContentTooLargeError()
            if allowed_prefixes and not raw.startswith(allowed_prefixes):
                raise InvalidMediaError("downloaded content has an unexpected signature")
            return raw
        finally:
            response.close()

    async def stream_download(
        self,
        url: str,
        destination: Path,
        *,
        max_bytes: int,
        cancel_event: asyncio.Event | None = None,
        resume_from: int = 0,
        if_range: str | None = None,
        referer: str | None = None,
        on_metadata: Callable[[DownloadMetadata], None] | None = None,
    ) -> StreamResult:
        """Stream into destination with a hard size cap and optional safe resume."""

        offset = max(0, resume_from)
        headers = {"Accept-Encoding": "identity", "Accept": "*/*"}
        if referer:
            headers["Referer"] = referer
        restart_used = False
        while True:
            request_headers = dict(headers)
            # Some CDNs (hotlink protection) drop connections for requests that
            # do not look like a ranged media fetch, so always ask for a range.
            request_headers["Range"] = f"bytes={offset}-"
            if offset > 0 and if_range:
                request_headers["If-Range"] = if_range
            response, final_url = await self._open(
                "GET", url, headers=request_headers, stream_timeout=True
            )
            if response.status == 416 and offset > 0 and not restart_used:
                response.close()
                offset = 0
                restart_used = True
                continue
            if response.status == 429:
                retry_after = parse_retry_after(response.headers.get("Retry-After"))
                response.close()
                raise RateLimitedError(retry_after)
            if response.status in {401, 402, 403, 407, 451}:
                status = response.status
                response.close()
                raise AccessDeniedError(status)
            if response.status in {404, 410}:
                response.close()
                raise NotFoundError()
            if response.status >= 500:
                status = response.status
                response.close()
                raise NetworkError(f"upstream HTTP {status}")
            if not 200 <= response.status < 300:
                status = response.status
                response.close()
                raise PermanentSiteError(f"unexpected HTTP status {status}")

            content_type = (
                (response.headers.get("Content-Type") or "").split(";", 1)[0].strip().lower()
            )
            if content_type in {"text/html", "application/json", "text/plain", "text/xml"}:
                response.close()
                raise InvalidMediaError(
                    f"server returned a non-media content type: {content_type}"
                )

            resumed = response.status == 206 and offset > 0
            if response.status == 206:
                content_range = response.headers.get("Content-Range", "")
                match = _CONTENT_RANGE_RE.fullmatch(content_range.strip())
                start = int(match.group(1)) if match else -1
                if start != offset and offset > 0 and not restart_used:
                    response.close()
                    offset = 0
                    restart_used = True
                    continue
                total_raw = match.group(3) if match else "*"
                total_from_range = int(total_raw) if total_raw.isdigit() else None
            else:
                scratch = response.headers.get("Content-Range", "")
                match = _CONTENT_RANGE_RE.fullmatch(scratch.strip())
                total_from_range = int(match.group(3)) if match and match.group(3).isdigit() else None
                if offset > 0:
                    # Server ignored Range or chose a full representation.
                    offset = 0
                resumed = False

            content_length_header = response.headers.get("Content-Length")
            content_length = (
                int(content_length_header)
                if content_length_header and content_length_header.isdigit()
                else None
            )
            total_size = total_from_range
            if total_size is None and content_length is not None:
                total_size = offset + content_length if resumed else content_length
            if total_size is not None and total_size > max_bytes:
                response.close()
                raise ContentTooLargeError("declared media size exceeds configured maximum")
            if content_length is not None and offset + content_length > max_bytes:
                response.close()
                raise ContentTooLargeError("remaining media bytes exceed configured maximum")

            destination.parent.mkdir(parents=True, exist_ok=True)
            etag = response.headers.get("ETag")
            last_modified = response.headers.get("Last-Modified")
            metadata = DownloadMetadata(
                final_url=final_url,
                status=response.status,
                content_type=content_type,
                total_size=total_size,
                etag=etag,
                last_modified=last_modified,
                resumed=resumed,
            )
            if on_metadata is not None:
                on_metadata(metadata)

            if not resumed:
                await asyncio.to_thread(destination.unlink, missing_ok=True)
            mode = "ab" if resumed else "wb"
            written = offset if resumed else 0
            try:
                async with aiofiles.open(destination, mode) as handle:
                    while True:
                        if cancel_event is not None and cancel_event.is_set():
                            raise DownloadCancelledError()
                        try:
                            chunk = await response.content.read(64 * 1024)
                        except (aiohttp.ClientError, TimeoutError, OSError) as exc:
                            raise NetworkError(
                                f"download interrupted ({type(exc).__name__})"
                            ) from exc
                        if not chunk:
                            break
                        written += len(chunk)
                        if written > max_bytes:
                            raise ContentTooLargeError("media exceeded configured maximum")
                        await handle.write(chunk)
                    await handle.flush()
            finally:
                response.close()
            return StreamResult(
                final_url=final_url,
                status=response.status,
                content_type=content_type,
                size=written,
                resumed=resumed,
                etag=etag,
                last_modified=last_modified,
                total_size=total_size,
            )