"""Fail-closed logging boundary for data derived from external inputs.

Use :func:`get_safe_logger` in modules whose correlation fields can be
influenced by an HTTP request, webhook, provider response, or imported row.
The adapter preserves useful identifiers while making every record single-line,
redacting common credential/PII shapes, and suppressing raw exception tracebacks
whose frames or exception arguments may contain signed URLs and request bodies.
"""

from __future__ import annotations

import logging
import re
import secrets
from collections.abc import Mapping
from decimal import Decimal
from itertools import islice
from typing import Any

from cryptography.hazmat.primitives.ciphers.aead import AESSIV

_CONTROL_CHARS_RE = re.compile(r"[\x00-\x1f\x7f-\x9f\u2028\u2029]")
_EMAIL_LOCAL_CHARS = frozenset(
    "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789._%+-"
)
_EMAIL_DOMAIN_CHARS = frozenset(
    "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789.-"
)
_PHONE_RE = re.compile(r"(?<!\d)(\+?\d[\d ()-]{7,30}\d)(?!\d)")
_JWT_RE = re.compile(r"\beyJ[A-Za-z0-9_-]{8,}\.[A-Za-z0-9_-]{8,}\.[A-Za-z0-9_-]{8,}\b")
_NAMED_SECRET_PREFIX_RE = re.compile(
    r"(?i)(?<![\w-])[\"']?(proxy[_-]?authorization|authorization|"
    r"access[_-]?token|refresh[_-]?token|private[_-]?token|"
    r"session[_-]?token|auth[_-]?token|id[_-]?token|token|"
    r"client[_-]?secret|app[_-]?secret|api[_-]?key|access[_-]?key|"
    r"x[_-]?api[_-]?key|aws[_-]?secret[_-]?access[_-]?key|"
    r"webhook[_-]?secret|signing[_-]?key|encryption[_-]?key|"
    r"secret[_-]?key|private[_-]?key|credential(?:s)?|password|passwd|"
    r"database[_-]?url|connection[_-]?string|dsn|set[_-]?cookie|cookie|"
    r"signature|secret)"
    r"[\"']?\s{0,10000}[:=]\s{0,10000}"
)
_BEARER_RE = re.compile(r"(?i)\bbearer\s+[^\s,;]+")
_SENSITIVE_KEY_PARTS = frozenset(
    {
        "authorization",
        "credential",
        "credentials",
        "cookie",
        "dsn",
        "passwd",
        "password",
        "secret",
        "signature",
        "token",
    }
)
_SENSITIVE_COMPACT_KEYS = frozenset(
    {
        "accesskey",
        "apikey",
        "clientsecret",
        "connectionstring",
        "databaseurl",
        "encryptionkey",
        "privatekey",
        "proxyauthorization",
        "setcookie",
        "signingkey",
        "webhooksecret",
    }
)
_SENSITIVE_COMPACT_SUFFIXES = (
    "accesskey",
    "apikey",
    "authorization",
    "cookie",
    "credential",
    "credentials",
    "dsn",
    "passwd",
    "password",
    "secret",
    "signature",
    "token",
)
_URL_USERINFO_RE = re.compile(
    r"(?i)\b([a-z][a-z0-9+.-]{1,20}://)([^/\s?#@]+)@"
)
_URL_QUERY_RE = re.compile(r"(?i)\b((?:https?|s3|gs)://[^\s?#]+)\?[^\s#]*")


class _ExternalHttpLogFilter(logging.Filter):
    """Sanitize third-party transport records before any handler sees them."""

    def filter(self, record: logging.LogRecord) -> bool:
        try:
            rendered = record.getMessage()
        except Exception:  # noqa: BLE001 - fail closed at the log boundary
            rendered = "external HTTP log formatting failed"
        record.msg = sanitize_log_text(rendered)
        record.args = ()
        record.exc_info = None
        record.exc_text = None
        record.stack_info = None
        return True


def configure_external_http_log_safety() -> None:
    """Protect httpx/httpcore logs from signed URLs and auth headers.

    The worker intentionally runs at INFO, and httpx includes the complete
    request URL at that level. Cloud APIs commonly authenticate in the query
    string, so every transport record must cross the same redaction boundary
    as application logs. Registration is idempotent for import/reload safety.
    """
    for logger_name in ("httpx", "httpcore"):
        external_logger = logging.getLogger(logger_name)
        if not any(
            isinstance(existing, _ExternalHttpLogFilter)
            for existing in external_logger.filters
        ):
            external_logger.addFilter(_ExternalHttpLogFilter())
_FINGERPRINT_CIPHER = AESSIV(secrets.token_bytes(64))
_FINGERPRINT_CONTEXT = [b"helix-safe-log-fingerprint-v1"]
_CORRELATION_CONTEXT = [b"helix-safe-log-correlation-v1"]
_EVENT_TOKEN_CIPHER: AESSIV | None = AESSIV(secrets.token_bytes(64))
_EVENT_TOKEN_CONTEXT = [b"helix-safe-log-event-token-v1"]
_VALUE_DELIMITERS = frozenset(" \t,;}&")
_REGISTERED_SAFE_LOG_EVENTS: dict[str, frozenset[str]] = {
    "APPROVAL_CANCEL_FAILED": frozenset(
        {"seat_allocation_id", "instance_code", "approval_code", "error_type"}
    ),
    "APPROVAL_LOOKUP_FAILED": frozenset(
        {"seat_allocation_id", "error_type"}
    ),
    "CLOUD_BILLING_DAILY_NO_LIST_VALUE": frozenset({"scope_id", "date"}),
    "CLOUD_BILLING_MTD_NOT_LIST_COMPLETE": frozenset(
        {"scope_id", "start_date", "end_date"}
    ),
    "CLOUD_SCOPE_ALERT_MEASURE_AMBIGUITY": frozenset(
        {"scope_id", "rule_count", "rule_types"}
    ),
    "CLOUD_SCOPE_ALERT_NO_AGGREGATE": frozenset(
        {"scope_id", "rule_type", "threshold", "measure"}
    ),
    "CLOUD_SCOPE_BUDGET_BASIS_MOVED_WITHOUT_RESTATEMENT": frozenset(
        {"scope_id", "from_measure", "to_measure", "budget"}
    ),
    # identity_reconcile bookkeeping. These three used to go through a
    # plain ``logging.getLogger`` in the same module — one of them printed
    # ``hris_person_id`` verbatim, which is precisely the identifier this
    # boundary exists to pseudonymize. Registered 2026-08-25 so the module
    # can drop its bypass logger.
    "IDENTITY_MANAGER_BACKFILL_LINKED": frozenset(
        {"person_id", "hris_person_id", "manager_person_id",
         "manager_external_id"},
    ),
    "IDENTITY_LOGIN_BINDINGS_DEACTIVATED": frozenset(
        {"person_id", "count"},
    ),
    "IDENTITY_ORG_NODES_DEACTIVATED": frozenset(
        {"source_system", "org_node_type", "count", "upstream_active"},
    ),
    "HRIS_CANONICAL_EMAIL_COLLISION_SKIPPED": frozenset(
        {"user_id", "employee_number", "resolved_email"}
    ),
    # A second SF userId whose personIdExternal is already held by a live
    # active person — a concurrent employment, not a new hire. Recorded in
    # identity_hris_user_aliases so every later reference to that userId
    # (managerId / relUserId) resolves to the canonical person.
    "HRIS_CONCURRENT_EMPLOYMENT_ALIASED": frozenset(
        {"user_id", "person_id_external", "canonical_hris_person_id", "event_type"}
    ),
    # Siblings of the above for the case where the userId is not new to
    # Helix but was a PURGED person's primary key (prod 2026-09-04: EU
    # re-minted 100075, which a leaver tombstoned on 2026-04-30 still
    # held). Two codes rather than one boolean field, because every field
    # VALUE here is a hash — whether the alias actually landed has to be
    # readable off the code itself. ``UNALIASABLE`` = the event is still
    # skipped (never resurrect a stranger) but ``record_hris_user_alias``
    # refused the mapping, which it does for a userId a LIVE person owns.
    "HRIS_TOMBSTONE_USERID_CLAIMED": frozenset(
        {"user_id", "canonical_hris_person_id", "event_type",
         "tombstone_person_id"}
    ),
    "HRIS_TOMBSTONE_USERID_UNALIASABLE": frozenset(
        {"user_id", "canonical_hris_person_id", "event_type",
         "tombstone_person_id"}
    ),
    # identity_hris_user_aliases bookkeeping. The raw HRIS userId must not
    # reach the log sink in plaintext (the collision path is covered by a
    # regression asserting exactly that), so these carry pseudonymized
    # fields like every other HRIS event.
    "HRIS_USER_ALIAS_RECORDED": frozenset(
        {"user_id", "person_id", "reason", "person_id_external"}
    ),
    "HRIS_USER_ALIAS_REFUSED": frozenset(
        {"user_id", "person_id", "reason"}
    ),
    "HRIS_USER_ALIAS_REPOINTED": frozenset(
        {"user_id", "person_id", "previous_person_id", "reason"}
    ),
    "HRIS_EMPLOYEE_NUMBER_CONFLICT": frozenset(
        {"employee_number", "conflicting_person_id", "person_id", "current_employee_number"}
    ),
    "HRIS_EMAIL_COLLISION_EXISTING_PRESERVED": frozenset(
        {"user_id", "current_email", "reason"}
    ),
    "HRIS_PLACEHOLDER_EMAIL_ASSIGNED": frozenset(
        {"user_id", "placeholder_email", "reason"}
    ),
    "HRIS_WEBHOOK_RECEIVED": frozenset(
        {"trace_id", "slug", "event_types", "event_count", "body_bytes", "body_sha256"}
    ),
}


def _safe_text(value: object) -> str:
    """Convert an arbitrary value without allowing its formatter to escape.

    Provider SDK objects sometimes implement ``__str__`` themselves.  A
    broken or hostile implementation must not make the logging boundary raise
    an exception whose arguments are then emitted by an outer error handler.
    """
    try:
        return str(value)
    except Exception:  # noqa: BLE001 - this is the fail-closed formatting boundary
        type_name = re.sub(r"[^A-Za-z0-9_.-]", "_", type(value).__name__)[:64]
        return f"<unprintable:{type_name or 'object'}>"


def _fingerprint(value: str) -> str:
    # AES-SIV is deterministic under the process-local secret, so operators
    # retain short-lived correlation without exposing a dictionary-testable
    # plain hash. Only the first 64 tag bits are emitted, making the logged
    # token non-decryptable even to a reader who later obtains the key.
    # Public helpers and registered events can receive provider-controlled
    # values. Bound before encryption so a multi-megabyte webhook field cannot
    # turn a warning log into an unbounded allocation / CPU spike. Values that
    # share the first 10k characters intentionally share a safety token.
    sealed = _FINGERPRINT_CIPHER.encrypt(
        value[:10_000].encode("utf-8", errors="replace"),
        _FINGERPRINT_CONTEXT,
    )
    return sealed[:8].hex()


def _correlation_fingerprint(value: str) -> str:
    """Return a bounded 128-bit token for high-cardinality log joins."""
    sealed = _FINGERPRINT_CIPHER.encrypt(
        value[:10_000].encode("utf-8", errors="replace"),
        _CORRELATION_CONTEXT,
    )
    return sealed[:16].hex()


def _event_token(*, logger_name: str, level: int, message: str) -> str:
    """Return a domain-separated 128-bit token, never the message itself.

    Tokens are deterministic only for this process key and domain.  Including
    logger and level prevents the same text in distinct operational domains
    from correlating accidentally.  Any key/cipher failure returns a constant
    marker so an outer error handler can never expose the message.
    """
    try:
        if _EVENT_TOKEN_CIPHER is None:
            raise RuntimeError("event token key unavailable")
        payload = f"{logger_name[:256]}\0{level}\0{message[:10_000]}".encode(
            "utf-8",
            errors="replace",
        )
        sealed = _EVENT_TOKEN_CIPHER.encrypt(payload, _EVENT_TOKEN_CONTEXT)
        return sealed[:16].hex()
    except Exception:  # noqa: BLE001 - cryptographic failure must fail closed
        return "unavailable"


def _redact_emails(text: str) -> str:
    """Replace email-shaped spans with a fingerprint in linear time."""
    pieces: list[str] = []
    cursor = 0
    search_from = 0
    while True:
        at = text.find("@", search_from)
        if at < 0:
            break
        local_start = at
        while local_start > cursor and text[local_start - 1] in _EMAIL_LOCAL_CHARS:
            local_start -= 1
        domain_end = at + 1
        while domain_end < len(text) and text[domain_end] in _EMAIL_DOMAIN_CHARS:
            domain_end += 1
        domain = text[at + 1 : domain_end]
        tld = domain.rsplit(".", 1)[-1] if "." in domain else ""
        if (
            local_start < at
            and not domain.startswith((".", "-"))
            and not domain.endswith((".", "-"))
            and len(tld) >= 2
            and tld.isalpha()
        ):
            pieces.append(text[cursor:local_start])
            email = text[local_start:domain_end]
            pieces.append(f"<email:{_fingerprint(email.lower())}>")
            cursor = domain_end
            search_from = domain_end
        else:
            search_from = at + 1
    pieces.append(text[cursor:])
    return "".join(pieces)


def log_fingerprint(value: object, *, label: str = "value") -> str:
    """Return a process-scoped, non-reversible token for correlation."""
    # Labels are normally literals, but this helper is a public boundary and
    # must remain safe if a caller passes request/provider data by mistake.
    safe_label = re.sub(r"[^A-Za-z0-9_.-]", "_", _safe_text(label))[:32] or "value"
    return f"<{safe_label}:{_fingerprint(_safe_text(value))}>"


def _redact_private_key(text: str) -> str:
    """Fail closed from a private-key PEM header through its end marker."""
    search_from = 0
    while True:
        lowered = text.lower()
        begin = lowered.find("-----begin ", search_from)
        if begin < 0:
            return text
        header_end = lowered.find("-----", begin + len("-----begin "))
        if header_end < 0:
            return text
        header_end += len("-----")
        if "private key" not in lowered[begin:header_end]:
            search_from = begin + len("-----begin ")
            continue
        end_start = lowered.find("-----end ", header_end)
        end = len(text)
        if end_start >= 0:
            end_label = lowered.find("private key-----", end_start, end_start + 100)
            if end_label >= 0:
                end = end_label + len("private key-----")
        text = f"{text[:begin]}<private-key-redacted>{text[end:]}"
        search_from = begin + len("<private-key-redacted>")


def _redact_url_credentials(text: str) -> str:
    """Remove URL userinfo and complete query strings, including signed URLs."""
    text = _URL_USERINFO_RE.sub(r"\1<userinfo-redacted>@", text)
    return _URL_QUERY_RE.sub(r"\1?<query-redacted>", text)


def _is_sensitive_key(value: object) -> bool:
    """Recognise common secret field names across snake/kebab/camel styles."""
    key = _safe_text(value).strip().lower()
    parts = {part for part in re.split(r"[^a-z0-9]+", key) if part}
    compact = "".join(character for character in key if character.isalnum())
    return (
        bool(parts & _SENSITIVE_KEY_PARTS)
        or compact in _SENSITIVE_COMPACT_KEYS
        or compact.endswith(_SENSITIVE_COMPACT_SUFFIXES)
    )


def _redact_named_secrets(text: str, *, preserve_placeholders: bool = False) -> str:
    """Redact unquoted, JSON, and Python-repr secret field values linearly."""
    pieces: list[str] = []
    cursor = 0
    for match in _NAMED_SECRET_PREFIX_RE.finditer(text):
        if match.start() < cursor:
            continue
        pieces.append(text[cursor : match.end()])
        value_start = match.end()
        if preserve_placeholders and re.match(
            r"%(?!%)(?:\([^)]{1,100}\))?[-+#0 ]{0,10}\d{0,10}"
            r"(?:\.\d{1,10})?[hlL]?[diouxXeEfFgGcrsa]",
            text[value_start:],
        ):
            cursor = value_start
            continue
        value_end = value_start
        if value_start < len(text) and text[value_start] in {'"', "'"}:
            quote = text[value_start]
            value_end = value_start + 1
            escaped = False
            while value_end < len(text):
                character = text[value_end]
                value_end += 1
                if escaped:
                    escaped = False
                elif character == "\\":
                    escaped = True
                elif character == quote:
                    break
        else:
            while value_end < len(text) and text[value_end] not in _VALUE_DELIMITERS:
                value_end += 1
        pieces.append("<redacted>")
        cursor = value_end
    pieces.append(text[cursor:])
    return "".join(pieces)


def sanitize_log_text(value: object, *, max_len: int = 1000) -> str:
    """Return a bounded, one-line, credential-safe representation.

    Email addresses retain only a stable hash so operators can correlate two
    records without copying PII into the log store. Credential-shaped values
    are replaced entirely. The explicit CR/LF replacements are intentionally
    kept at this boundary so static analysis can verify CWE-117 neutralisation.
    """
    bounded_len = max(0, min(max_len, 10_000))
    text = _safe_text(value)[:bounded_len]
    text = text.replace("\r\n", " ").replace("\n", " ").replace("\r", " ")
    text = _CONTROL_CHARS_RE.sub("?", text)
    text = _redact_private_key(text)
    text = _redact_url_credentials(text)
    text = _JWT_RE.sub("<jwt-redacted>", text)
    text = _BEARER_RE.sub("Bearer <redacted>", text)
    text = _redact_named_secrets(text)
    text = _redact_emails(text)
    text = _PHONE_RE.sub(
        lambda match: f"<phone:{_fingerprint(match.group(0))}>",
        text,
    )
    return text[:bounded_len]


def _sanitize_log_template(value: object, *, max_len: int = 1000) -> str:
    """Sanitize literal content without changing ``%`` placeholders."""
    bounded_len = max(0, min(max_len, 10_000))
    text = _safe_text(value)[:bounded_len]
    text = text.replace("\r\n", " ").replace("\n", " ").replace("\r", " ")
    text = _CONTROL_CHARS_RE.sub("?", text)
    text = _redact_private_key(text)
    text = _redact_url_credentials(text)
    text = _JWT_RE.sub("<jwt-redacted>", text)
    text = _BEARER_RE.sub("Bearer <redacted>", text)
    text = _redact_named_secrets(text, preserve_placeholders=True)
    text = _redact_emails(text)
    text = _PHONE_RE.sub(
        lambda match: f"<phone:{_fingerprint(match.group(0))}>",
        text,
    )
    return text[:bounded_len]


def _sanitize_structured(value: Any, *, depth: int = 0) -> Any:
    if value is None or isinstance(value, (bool, int, float, Decimal)):
        return value
    if depth >= 5:
        return "<nested-value-redacted>"
    if isinstance(value, Mapping):
        sanitized: dict[str, Any] = {}
        try:
            entries = islice(value.items(), 51)
            for index, (key, item) in enumerate(entries):
                if index >= 50:
                    sanitized["<truncated>"] = True
                    break
                safe_key = sanitize_log_text(key, max_len=100)
                sanitized[safe_key] = (
                    "<redacted>"
                    if _is_sensitive_key(key)
                    else _sanitize_structured(item, depth=depth + 1)
                )
        except Exception:  # noqa: BLE001 - provider containers must fail closed
            sanitized["<container-error>"] = "<redacted>"
        return sanitized
    if isinstance(value, (list, tuple, set, frozenset)):
        sanitized_items: list[Any] = []
        try:
            for index, item in enumerate(islice(value, 51)):
                if index >= 50:
                    sanitized_items.append("<truncated>")
                    break
                sanitized_items.append(_sanitize_structured(item, depth=depth + 1))
        except Exception:  # noqa: BLE001 - provider containers must fail closed
            sanitized_items.append("<container-error:redacted>")
        return sanitized_items
    return sanitize_log_text(value)


def _sanitize_value(value: Any) -> Any:
    # Preserve numeric values for ``%d`` / ``%f`` format strings. Everything
    # else becomes text before it crosses the logging boundary, including
    # exceptions, provider payload containers, UUIDs, and enum-like objects.
    if value is None or isinstance(value, (bool, int, float, Decimal)):
        return value
    if isinstance(value, Mapping):
        # Keep a mapping object so standard ``%(name)s`` templates retain
        # their logging semantics after every key/value has crossed the
        # recursive boundary.
        return _sanitize_structured(value)
    if isinstance(value, (list, tuple, set, frozenset)):
        return sanitize_log_text(_sanitize_structured(value))
    return sanitize_log_text(value)


class SafeLogger:
    """Small logger facade that sanitizes before entering ``logging``.

    This intentionally does not subclass ``Logger``/``LoggerAdapter``: the
    standard logging call is confined to :meth:`_emit`, after sanitization,
    which also gives static data-flow analysis an explicit trust boundary.
    """

    def __init__(self, logger: logging.Logger):
        self._logger = logger

    def _emit(
        self,
        level: int,
        msg: object,
        *args: object,
        **kwargs: Any,
    ) -> None:
        if not self._logger.isEnabledFor(level):
            return
        kwargs = dict(kwargs)
        # A raw traceback can include request/provider values in frame locals
        # or exception args, so this boundary never forwards it. Callers retain
        # trace ids and the sanitized exception value/type for correlation.
        kwargs["exc_info"] = False
        kwargs["stack_info"] = False
        if isinstance(kwargs.get("extra"), Mapping):
            kwargs["extra"] = _sanitize_structured(kwargs["extra"])
        if args:
            template = _sanitize_log_template(msg)
            safe_args = tuple(_sanitize_value(arg) for arg in args)
            # Render inside the boundary, then sanitize the completed text.
            # Sanitizing arguments alone is insufficient for the common
            # ``password=%s`` shape: the argument has no key context until
            # formatting joins it to the template.
            format_args: object = (
                safe_args[0]
                if len(safe_args) == 1 and isinstance(safe_args[0], Mapping)
                else safe_args
            )
            try:
                rendered = template % format_args
            except (KeyError, TypeError, ValueError):
                # Do not let malformed attacker-influenced format strings
                # emit a logging-internal traceback or raw argument repr.
                rendered = f"{template} <formatting-error>"
            safe_message = sanitize_log_text(rendered)
        else:
            template = sanitize_log_text(msg)
            safe_args = ()
            safe_message = sanitize_log_text(msg)
        # Arbitrary message text never enters the logging sink, even after
        # redaction. Emit a template token (which identifies the call site), an
        # instance token, and per-argument process-scoped fingerprints. The
        # latter close the observability gap of an instance-only token: the
        # same trace/resource/error value can now be correlated across event
        # templates without exposing the value itself.
        token = _event_token(
            logger_name=self._logger.name,
            level=level,
            message=safe_message,
        )
        template_token = _event_token(
            logger_name=self._logger.name,
            level=level,
            message=template,
        )
        arg_tokens = ",".join(
            f"{index}:{_correlation_fingerprint(_safe_text(value))}"
            for index, value in enumerate(safe_args)
        ) or "-"
        self._logger.log(
            level,
            "event_token_v1=%s template_token_v1=%s arg_tokens_v1=%s",
            token,
            template_token,
            arg_tokens,
            **kwargs,
        )

    def log(self, level: int, msg: object, *args: object, **kwargs: Any) -> None:
        self._emit(level, msg, *args, **kwargs)

    def event(
        self,
        event_code: str,
        fields: Mapping[str, object] | None = None,
        *,
        level: int = logging.INFO,
    ) -> None:
        """Emit a registered readable code with pseudonymized field values.

        Both event and field names come from the explicit registry above;
        callers cannot smuggle arbitrary text into the sink.  Every value is
        represented only by a domain-separated 128-bit token.
        """
        allowed_fields = _REGISTERED_SAFE_LOG_EVENTS.get(event_code)
        if allowed_fields is None:
            raise ValueError("unregistered safe logging event")
        provided = fields or {}
        if not isinstance(provided, Mapping) or not set(provided).issubset(
            allowed_fields
        ):
            raise ValueError("unregistered safe logging event field")
        if not self._logger.isEnabledFor(level):
            return

        parts = [event_code]
        for field_name in sorted(allowed_fields & set(provided)):
            # Field tokens deliberately exclude event/logger/level so the
            # same registered field value correlates across event codes. The
            # field name remains the domain separator: a person_id token is
            # not accidentally joinable to an unrelated budget value.
            token = _correlation_fingerprint(
                f"field={field_name}\0"
                f"value={sanitize_log_text(provided[field_name])}"
            )
            parts.append(f"{field_name}={token}")
        rendered_event = " ".join(parts)
        self._logger.log(level, rendered_event, exc_info=False, stack_info=False)

    def debug(self, msg: object, *args: object, **kwargs: Any) -> None:
        self._emit(logging.DEBUG, msg, *args, **kwargs)

    def info(self, msg: object, *args: object, **kwargs: Any) -> None:
        self._emit(logging.INFO, msg, *args, **kwargs)

    def warning(self, msg: object, *args: object, **kwargs: Any) -> None:
        self._emit(logging.WARNING, msg, *args, **kwargs)

    def error(self, msg: object, *args: object, **kwargs: Any) -> None:
        self._emit(logging.ERROR, msg, *args, **kwargs)

    def exception(self, msg: object, *args: object, **kwargs: Any) -> None:
        self._emit(logging.ERROR, msg, *args, **kwargs)

    def critical(self, msg: object, *args: object, **kwargs: Any) -> None:
        self._emit(logging.CRITICAL, msg, *args, **kwargs)


def get_safe_logger(name: str) -> SafeLogger:
    """Return a standard named logger protected by the safe-data adapter."""
    return SafeLogger(logging.getLogger(name))