from __future__ import annotations

import urllib.parse
import urllib.request
from dataclasses import dataclass
from typing import Mapping

from govoplan_core.security.outbound_http import (
    bounded_response_bytes,
    build_outbound_http_opener,
    validate_outbound_http_url,
)


@dataclass(frozen=True, slots=True)
class HttpFetchResponse:
    status: int
    headers: dict[str, str]
    body: bytes

    def text(self, encoding: str = "utf-8") -> str:
        return self.body.decode(encoding)


def is_http_url(value: str) -> bool:
    try:
        validate_http_url(value)
    except ValueError:
        return False
    return True


def validate_http_url(value: str, *, label: str = "URL") -> str:
    parsed = urllib.parse.urlparse(str(value).strip())
    if parsed.scheme not in {"http", "https"} or not parsed.netloc:
        raise ValueError(f"{label} must be an absolute HTTP(S) URL.")
    if parsed.username or parsed.password:
        raise ValueError(f"{label} must not include embedded credentials.")
    return urllib.parse.urlunparse(parsed)


def fetch_http(
    url: str,
    *,
    timeout: float = 15,
    label: str = "URL",
    method: str = "GET",
    headers: Mapping[str, str] | None = None,
    max_bytes: int | None = None,
) -> HttpFetchResponse:
    validated_url = validate_outbound_http_url(url, label=label)
    request = urllib.request.Request(  # noqa: S310 - URL is restricted to validated HTTP(S).
        validated_url,
        headers=dict(headers or {}),
        method=method,
    )
    opener = build_outbound_http_opener(_PolicyRedirectHandler(label=label))
    with opener.open(request, timeout=timeout) as response:  # noqa: S310 - URL and every redirect are policy-validated.  # nosec B310  # nosemgrep: python.lang.security.audit.dynamic-urllib-use-detected.dynamic-urllib-use-detected
        response_headers = dict(response.headers.items())
        return HttpFetchResponse(
            status=int(getattr(response, "status", 0)),
            headers=response_headers,
            body=bounded_response_bytes(
                response,
                headers=response_headers,
                max_bytes=max_bytes,
                label=f"{label} response",
            ),
        )


def fetch_http_text(
    url: str,
    *,
    timeout: float = 15,
    label: str = "URL",
    method: str = "GET",
    headers: Mapping[str, str] | None = None,
    encoding: str = "utf-8",
    max_bytes: int | None = None,
) -> str:
    return fetch_http(url, timeout=timeout, label=label, method=method, headers=headers, max_bytes=max_bytes).text(encoding)


class _PolicyRedirectHandler(urllib.request.HTTPRedirectHandler):
    def __init__(self, *, label: str) -> None:
        super().__init__()
        self._label = label

    def redirect_request(self, req, fp, code, msg, headers, newurl):  # type: ignore[no-untyped-def]
        candidate = validate_outbound_http_url(newurl, label=f"{self._label} redirect")
        previous = urllib.parse.urlparse(req.full_url)
        redirected = urllib.parse.urlparse(candidate)
        if previous.scheme.lower() == "https" and redirected.scheme.lower() != "https":
            return None
        new_request = super().redirect_request(req, fp, code, msg, headers, candidate)
        if new_request is not None and _http_origin(previous) != _http_origin(redirected):
            for header in ("Authorization", "Proxy-Authorization", "Cookie", "Cookie2"):
                new_request.remove_header(header)
        return new_request


def _http_origin(parsed: urllib.parse.ParseResult) -> tuple[str, str, int]:
    scheme = parsed.scheme.lower()
    return scheme, (parsed.hostname or "").lower(), parsed.port or (443 if scheme == "https" else 80)
