"""Logging utility functions for Sphinx."""

from __future__ import annotations

import logging
import logging.handlers
import os.path
from collections import defaultdict
from contextlib import contextmanager, nullcontext
from typing import TYPE_CHECKING

from docutils import nodes
from docutils.utils import get_source_line

from sphinx._cli.util.colour import colourise
from sphinx.errors import SphinxWarning

if TYPE_CHECKING:
    from collections.abc import Iterator, Mapping, Sequence, Set
    from typing import IO, Any, NoReturn

    from docutils.nodes import Node

    from sphinx.application import Sphinx


NAMESPACE = 'sphinx'
VERBOSE = 15

LEVEL_NAMES: defaultdict[str, int] = defaultdict(
    lambda: logging.WARNING,
    {
        'CRITICAL': logging.CRITICAL,
        'SEVERE': logging.CRITICAL,
        'ERROR': logging.ERROR,
        'WARNING': logging.WARNING,
        'INFO': logging.INFO,
        'VERBOSE': VERBOSE,
        'DEBUG': logging.DEBUG,
    },
)

VERBOSITY_MAP: defaultdict[int, int] = defaultdict(
    lambda: logging.NOTSET,
    {
        0: logging.INFO,
        1: VERBOSE,
        2: logging.DEBUG,
    },
)

COLOR_MAP: dict[int, str] = {
    logging.ERROR: 'darkred',
    logging.WARNING: 'red',
    logging.DEBUG: 'darkgray',
}


def getLogger(name: str) -> SphinxLoggerAdapter:
    """Get logger wrapped by :class:`sphinx.util.logging.SphinxLoggerAdapter`.

    Sphinx logger always uses ``sphinx.*`` namespace to be independent from
    settings of root logger.  It ensures logging is consistent even if a
    third-party extension or imported application resets logger settings.

    Example usage::

        >>> from sphinx.util import logging
        >>> logger = logging.getLogger(__name__)
        >>> logger.info('Hello, this is an extension!')
        Hello, this is an extension!
    """
    # add sphinx prefix to name forcely
    logger = logging.getLogger(NAMESPACE + '.' + name)
    # Forcely enable logger
    logger.disabled = False
    # wrap logger by SphinxLoggerAdapter
    return SphinxLoggerAdapter(logger, {})


def convert_serializable(records: list[logging.LogRecord]) -> None:
    """Convert LogRecord serializable."""
    for r in records:
        # extract arguments to a message and clear them
        r.msg = r.getMessage()
        r.args = ()

        location = getattr(r, 'location', None)
        if isinstance(location, nodes.Node):
            r.location = get_node_location(location)


class SphinxLogRecord(logging.LogRecord):
    """Log record class supporting location"""

    prefix = ''
    location: Any = None

    def getMessage(self) -> str:
        message = super().getMessage()
        location = getattr(self, 'location', None)
        if location:
            message = f'{location}: {self.prefix}{message}'
        elif self.prefix not in message:
            message = self.prefix + message

        return message


class SphinxInfoLogRecord(SphinxLogRecord):
    """Info log record class supporting location"""

    prefix = ''  # do not show any prefix for INFO messages


class SphinxWarningLogRecord(SphinxLogRecord):
    """Warning log record class supporting location"""

    @property
    def prefix(self) -> str:  # type: ignore[override]
        if self.levelno >= logging.CRITICAL:
            return 'CRITICAL: '
        elif self.levelno >= logging.ERROR:
            return 'ERROR: '
        else:
            return 'WARNING: '


class SphinxLoggerAdapter(logging.LoggerAdapter[logging.Logger]):
    """LoggerAdapter allowing ``type`` and ``subtype`` keywords."""

    KEYWORDS = ['type', 'subtype', 'location', 'nonl', 'color', 'once']

    def log(  # type: ignore[override]
        self, level: int | str, msg: str, *args: Any, **kwargs: Any
    ) -> None:
        if isinstance(level, int):
            super().log(level, msg, *args, **kwargs)
        else:
            levelno = LEVEL_NAMES[level]
            super().log(levelno, msg, *args, **kwargs)

    def verbose(self, msg: str, *args: Any, **kwargs: Any) -> None:
        self.log(VERBOSE, msg, *args, **kwargs)

    def process(self, msg: str, kwargs: dict[str, Any]) -> tuple[str, dict[str, Any]]:  # type: ignore[override]
        extra = kwargs.setdefault('extra', {})
        for keyword in self.KEYWORDS:
            if keyword in kwargs:
                extra[keyword] = kwargs.pop(keyword)

        return msg, kwargs

    def handle(self, record: logging.LogRecord) -> None:
        self.logger.handle(record)

    def warning(  # type: ignore[override]
        self,
        msg: object,
        *args: object,
        type: str | None = None,
        subtype: str | None = None,
        location: str | tuple[str | None, int | None] | Node | None = None,
        nonl: bool = True,
        color: str | None = None,
        once: bool = False,
        **kwargs: Any,
    ) -> None:
        """Log a sphinx warning.

        It is recommended to include a ``type`` and ``subtype`` for warnings as
        these can be displayed to the user using :confval:`show_warning_types`
        and used in :confval:`suppress_warnings` to suppress specific warnings.

        It is also recommended to specify a ``location`` whenever possible
        to help users in correcting the warning.

        :param msg: The message, which may contain placeholders for ``args``.
        :param args: The arguments to substitute into ``msg``.
        :param type: The type of the warning.
        :param subtype: The subtype of the warning.
        :param location: The source location of the warning's origin,
            which can be a string (the ``docname`` or ``docname:lineno``),
            a tuple of ``(docname, lineno)``,
            or the docutils node object.
        :param nonl: Whether to append a new line terminator to the message.
        :param color: A color code for the message.
        :param once: Do not log this warning,
            if a previous warning already has same ``msg``, ``args`` and ``once=True``.
        """
        return super().warning(
            msg,
            *args,
            type=type,
            subtype=subtype,
            location=location,
            nonl=nonl,
            color=color,
            once=once,
            **kwargs,
        )


class WarningStreamHandler(logging.StreamHandler['SafeEncodingWriter']):
    """StreamHandler for warnings."""

    pass


class NewLineStreamHandler(logging.StreamHandler['SafeEncodingWriter']):
    """StreamHandler which switches line terminator by record.nonl flag."""

    def emit(self, record: logging.LogRecord) -> None:
        try:
            self.acquire()
            if getattr(record, 'nonl', False):
                # skip appending terminator when nonl=True
                self.terminator = ''
            super().emit(record)
        finally:
            self.terminator = '\n'
            self.release()


class MemoryHandler(logging.handlers.BufferingHandler):
    """Handler buffering all logs."""

    buffer: list[logging.LogRecord]

    def __init__(self) -> None:
        super().__init__(-1)

    def shouldFlush(self, record: logging.LogRecord) -> bool:
        return False  # never flush

    def flush(self) -> None:
        # suppress any flushes triggered by importing packages that flush
        # all handlers at initialization time
        pass

    def flushTo(self, logger: logging.Logger) -> None:
        self.acquire()
        try:
            for record in self.buffer:
                logger.handle(record)
            self.buffer = []
        finally:
            self.release()

    def clear(self) -> list[logging.LogRecord]:
        buffer, self.buffer = self.buffer, []
        return buffer


@contextmanager
def pending_warnings() -> Iterator[logging.Handler]:
    """Context manager to postpone logging warnings temporarily.

    Similar to :func:`pending_logging`.
    """
    logger = logging.getLogger(NAMESPACE)
    memhandler = MemoryHandler()
    memhandler.setLevel(logging.WARNING)

    try:
        handlers = []
        for handler in logger.handlers[:]:
            if isinstance(handler, WarningStreamHandler):
                logger.removeHandler(handler)
                handlers.append(handler)

        logger.addHandler(memhandler)
        yield memhandler
    finally:
        logger.removeHandler(memhandler)

        for handler in handlers:
            logger.addHandler(handler)

        memhandler.flushTo(logger)


@contextmanager
def suppress_logging() -> Iterator[MemoryHandler]:
    """Context manager to suppress logging all logs temporarily.

    For example::

        >>> with suppress_logging():
        >>>     logger.warning('Warning message!')  # suppressed
        >>>     some_long_process()
        >>>
    """
    logger = logging.getLogger(NAMESPACE)
    memhandler = MemoryHandler()

    try:
        handlers = []
        for handler in logger.handlers[:]:
            logger.removeHandler(handler)
            handlers.append(handler)

        logger.addHandler(memhandler)
        yield memhandler
    finally:
        logger.removeHandler(memhandler)

        for handler in handlers:
            logger.addHandler(handler)


@contextmanager
def pending_logging() -> Iterator[MemoryHandler]:
    """Context manager to postpone logging all logs temporarily.

    For example::

        >>> with pending_logging():
        >>>     logger.warning('Warning message!')  # not flushed yet
        >>>     some_long_process()
        >>>
        Warning message!  # the warning is flushed here
    """
    logger = logging.getLogger(NAMESPACE)
    try:
        with suppress_logging() as memhandler:
            yield memhandler
    finally:
        memhandler.flushTo(logger)


skip_warningiserror = nullcontext  # Deprecate in Sphinx 10


@contextmanager
def prefixed_warnings(prefix: str) -> Iterator[None]:
    """Context manager to prepend prefix to all warning log records temporarily.

    For example::

        >>> with prefixed_warnings("prefix:"):
        >>>     logger.warning('Warning message!')  # => prefix: Warning message!

    .. versionadded:: 2.0
    """
    logger = logging.getLogger(NAMESPACE)
    warning_handler = None
    for handler in logger.handlers:
        if isinstance(handler, WarningStreamHandler):
            warning_handler = handler
            break
    else:
        # warning stream not found
        yield
        return

    prefix_filter = None
    for _filter in warning_handler.filters:
        if isinstance(_filter, MessagePrefixFilter):
            prefix_filter = _filter
            break

    if prefix_filter:
        # already prefixed
        try:
            previous = prefix_filter.prefix
            prefix_filter.prefix = prefix
            yield
        finally:
            prefix_filter.prefix = previous
    else:
        # not prefixed yet
        prefix_filter = MessagePrefixFilter(prefix)
        try:
            warning_handler.addFilter(prefix_filter)
            yield
        finally:
            warning_handler.removeFilter(prefix_filter)


class LogCollector:
    def __init__(self) -> None:
        self.logs: list[logging.LogRecord] = []

    @contextmanager
    def collect(self) -> Iterator[None]:
        with pending_logging() as memhandler:
            yield

            self.logs = memhandler.clear()


class InfoFilter(logging.Filter):
    """Filter error and warning messages."""

    def filter(self, record: logging.LogRecord) -> bool:
        return record.levelno < logging.WARNING


class _RaiseOnWarningFilter(logging.Filter):
    """Raise exception if a warning is emitted."""

    def filter(self, record: logging.LogRecord) -> NoReturn:
        try:
            message = record.msg % record.args
        except (TypeError, ValueError):
            message = record.msg  # use record.msg itself
        if location := getattr(record, 'location', ''):
            message = f'{location}:{message}'
        if record.exc_info is not None:
            raise SphinxWarning(message) from record.exc_info[1]
        raise SphinxWarning(message)


def is_suppressed_warning(
    warning_type: str, sub_type: str, suppress_warnings: Set[str] | Sequence[str]
) -> bool:
    """Check whether the warning is suppressed or not."""
    if warning_type is None or len(suppress_warnings) == 0:
        return False
    suppressed_warnings = frozenset(suppress_warnings)
    if warning_type in suppressed_warnings:
        return True
    if f'{warning_type}.*' in suppressed_warnings:
        return True
    return f'{warning_type}.{sub_type}' in suppressed_warnings


class WarningSuppressor(logging.Filter):
    """Filter logs by `suppress_warnings`."""

    def __init__(self, app: Sphinx) -> None:
        self.app = app
        super().__init__()

    def filter(self, record: logging.LogRecord) -> bool:
        type = getattr(record, 'type', '')
        subtype = getattr(record, 'subtype', '')

        try:
            suppress_warnings = self.app.config.suppress_warnings
        except AttributeError:
            # config is not initialized yet (ex. in conf.py)
            suppress_warnings = ()

        if is_suppressed_warning(type, subtype, suppress_warnings):
            return False
        else:
            self.app._warncount += 1
            return True


class MessagePrefixFilter(logging.Filter):
    """Prepend prefix to all log records."""

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

    def filter(self, record: logging.LogRecord) -> bool:
        if self.prefix:
            record.msg = self.prefix + ' ' + record.msg
        return True


class OnceFilter(logging.Filter):
    """Show the message only once."""

    def __init__(self, name: str = '') -> None:
        super().__init__(name)
        self.messages: dict[
            str, list[tuple[object, ...] | Mapping[str, object] | None]
        ] = {}

    def filter(self, record: logging.LogRecord) -> bool:
        once = getattr(record, 'once', '')
        if not once:
            return True
        else:
            params = self.messages.setdefault(record.msg, [])
            if record.args in params:
                return False

            params.append(record.args)
            return True


class SphinxLogRecordTranslator(logging.Filter):
    """Converts a log record to one Sphinx expects

    * Make a instance of SphinxLogRecord
    * docname to path if location given
    * append warning type/subtype to message if :confval:`show_warning_types` is ``True``
    """

    LogRecordClass: type[logging.LogRecord]

    def __init__(self, app: Sphinx) -> None:
        self.app = app
        super().__init__()

    def filter(self, record: SphinxWarningLogRecord) -> bool:  # type: ignore[override]
        if isinstance(record, logging.LogRecord):
            # force subclassing to handle location
            record.__class__ = self.LogRecordClass  # type: ignore[assignment]

        location = getattr(record, 'location', None)
        if isinstance(location, tuple):
            docname, lineno = location
            if docname:
                if lineno:
                    record.location = f'{self.app.env.doc2path(docname)}:{lineno}'
                else:
                    record.location = f'{self.app.env.doc2path(docname)}'
            else:
                record.location = None
        elif isinstance(location, nodes.Node):
            record.location = get_node_location(location)
        elif location and ':' not in location:
            record.location = f'{self.app.env.doc2path(location)}'

        return True


class InfoLogRecordTranslator(SphinxLogRecordTranslator):
    """LogRecordTranslator for INFO level log records."""

    LogRecordClass = SphinxInfoLogRecord


class WarningLogRecordTranslator(SphinxLogRecordTranslator):
    """LogRecordTranslator for WARNING level log records."""

    LogRecordClass = SphinxWarningLogRecord

    def filter(self, record: SphinxWarningLogRecord) -> bool:  # type: ignore[override]
        ret = super().filter(record)

        try:
            show_warning_types = self.app.config.show_warning_types
        except AttributeError:
            # config is not initialized yet (ex. in conf.py)
            show_warning_types = False
        if show_warning_types:
            if log_type := getattr(record, 'type', ''):
                if log_subtype := getattr(record, 'subtype', ''):
                    record.msg += f' [{log_type}.{log_subtype}]'
                else:
                    record.msg += f' [{log_type}]'

        return ret


def get_node_location(node: Node) -> str | None:
    source, line = get_source_line(node)
    if source and line:
        return f'{os.path.abspath(source)}:{line}'
    if source:
        return f'{os.path.abspath(source)}:'
    if line:
        return f'<unknown>:{line}'
    return None


class ColorizeFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        message = super().format(record)
        colour_name = getattr(record, 'color', '')
        if not colour_name:
            colour_name = COLOR_MAP.get(record.levelno, '')
        if not colour_name:
            return message
        try:
            return colourise(colour_name, message)
        except ValueError:
            return message


class SafeEncodingWriter:
    """Stream writer which ignores UnicodeEncodeError silently"""

    def __init__(self, stream: IO[str]) -> None:
        self.stream = stream
        self.encoding = getattr(stream, 'encoding', 'ascii') or 'ascii'

    def write(self, data: str) -> None:
        try:
            self.stream.write(data)
        except UnicodeEncodeError:
            # stream accept only str, not bytes.  So, we encode and replace
            # non-encodable characters, then decode them.
            self.stream.write(
                data.encode(self.encoding, 'replace').decode(self.encoding)
            )

    def flush(self) -> None:
        if hasattr(self.stream, 'flush'):
            self.stream.flush()


class LastMessagesWriter:
    """Stream writer storing last 10 messages in memory to save trackback"""

    def __init__(self, app: Sphinx, stream: IO[str]) -> None:
        self.app = app

    def write(self, data: str) -> None:
        self.app.messagelog.append(data)


def setup(app: Sphinx, status: IO[str], warning: IO[str]) -> None:
    """Setup root logger for Sphinx"""
    logger = logging.getLogger(NAMESPACE)
    logger.setLevel(logging.DEBUG)
    logger.propagate = False

    # clear all handlers
    for handler in logger.handlers[:]:
        logger.removeHandler(handler)

    info_handler = NewLineStreamHandler(SafeEncodingWriter(status))
    info_handler.addFilter(InfoFilter())
    info_handler.addFilter(InfoLogRecordTranslator(app))
    info_handler.setLevel(VERBOSITY_MAP[app.verbosity])
    info_handler.setFormatter(ColorizeFormatter())

    warning_handler = WarningStreamHandler(SafeEncodingWriter(warning))
    if app._exception_on_warning:
        warning_handler.addFilter(_RaiseOnWarningFilter())
    warning_handler.addFilter(WarningSuppressor(app))
    warning_handler.addFilter(WarningLogRecordTranslator(app))
    warning_handler.addFilter(OnceFilter())
    warning_handler.setLevel(logging.WARNING)
    warning_handler.setFormatter(ColorizeFormatter())

    messagelog_handler = logging.StreamHandler(LastMessagesWriter(app, status))
    messagelog_handler.addFilter(InfoFilter())
    messagelog_handler.setLevel(VERBOSITY_MAP[app.verbosity])

    logger.addHandler(info_handler)
    logger.addHandler(warning_handler)
    logger.addHandler(messagelog_handler)