from requests import Response
import json
from datetime import datetime, timezone
from typing import Optional
from urllib.parse import urlparse


# RFC 9110 §5.6.7: HTTP-date "IMF-fixdate" format used by `Retry-After`.
_HTTP_DATE_FMT = "%a, %d %b %Y %H:%M:%S GMT"


def _parse_retry_after(value: Optional[str]) -> Optional[float]:
    """
    Parse an HTTP `Retry-After` header into a delay in seconds.

    Returns `None` if missing or unparseable. Per RFC 9110, the value is
    either a non-negative integer of seconds or an IMF-fixdate HTTP-date.
    """
    if not value:
        return None
    try:
        return max(float(value), 0.0)
    except ValueError:
        pass
    try:
        retry_at = datetime.strptime(value, _HTTP_DATE_FMT).replace(tzinfo=timezone.utc)
    except ValueError:
        return None
    return max((retry_at - datetime.now(timezone.utc)).total_seconds(), 0.0)


class FelderaError(Exception):
    """
    Generic class for Feldera error handling
    """

    def __init__(self, message: str) -> None:
        self.message = message
        super().__init__(self.message)

    def __str__(self) -> str:
        return f"FelderaError. Error message: {self.message}"


class FelderaAPIError(FelderaError):
    """Error sent by Feldera API"""

    def __init__(self, error: str, request: Response) -> None:
        self.status_code = request.status_code
        self.error = error
        self.error_code = None
        self.message = None
        self.details = None
        self.retry_after: Optional[float] = _parse_retry_after(
            request.headers.get("Retry-After")
        )

        err_msg = ""

        if request.text:
            try:
                json_data = json.loads(request.text)

                self.error_code = json_data.get("error_code")
                if self.error_code:
                    err_msg += f"\nError Code: {self.error_code}"
                self.message = json_data.get("message")
                if self.message:
                    err_msg += f"\nMessage: {self.message}"
                self.details = json_data.get("details")
                if self.details:
                    err_msg += f"\nDetails: {self.details}"
            except Exception:
                self.message = request.text
                err_msg += request.text

        err_msg += f"\nResponse Status: {request.status_code}"

        if int(request.status_code) == 401:
            parsed = urlparse(request.request.url)

            auth_err = f"\nAuthorization error at '{parsed.scheme}://{parsed.hostname}': "
            auth = request.request.headers.get("Authorization")
            if auth is None:
                err_msg += f"{auth_err}no credential provided"
            else:
                # The credential may be an API key or a JWT/bearer token; do not
                # assume which. The server `message` above carries the specifics.
                err_msg += f"{auth_err}credential rejected (invalid or expired token / API key)"

        err_msg = err_msg.strip()

        super().__init__(err_msg)


# Compatibility alias: the RFC and some docs refer to this as `FelderaApiError`.
FelderaApiError = FelderaAPIError


class FelderaTimeoutError(FelderaError):
    """Error when Feldera operation takes longer than expected"""

    def __init__(self, err: str) -> None:
        super().__init__(f"Timeout connecting to Feldera: {err}")


class FelderaCommunicationError(FelderaError):
    """Error when connection to Feldera"""

    def __init__(self, err: str) -> None:
        super().__init__(f"Cannot connect to Feldera API: {err}")