Spaces:
Sleeping
Sleeping
| from __future__ import annotations | |
| import logging | |
| import sys | |
| from collections.abc import Mapping | |
| from typing import Final, TextIO | |
| __all__: tuple[str, ...] = ( | |
| "get_logger", | |
| "stage_metric", | |
| ) | |
| _REPRO_SAFE_FORMAT: Final[str] = "%(levelname)s %(name)s %(message)s" | |
| class _DeterministicFormatter(logging.Formatter): | |
| """A ``Formatter`` that never renders wall-clock time into a record.""" | |
| def formatTime( # noqa: N802 — overrides logging.Formatter's mixed-case API. | |
| self, record: logging.LogRecord, datefmt: str | None = None | |
| ) -> str: | |
| """Return an empty string; repro-relevant lines carry no timestamp.""" | |
| return "" | |
| class _DeterministicStreamHandler(logging.StreamHandler[TextIO]): | |
| """Marker subclass so :func:`get_logger` can detect prior setup idempotently.""" | |
| def get_logger(name: str, *, level: int = logging.INFO) -> logging.Logger: | |
| """Return a stdlib logger configured for deterministic, timestamp-free output. | |
| Idempotent: calling this repeatedly for the same ``name`` reuses the | |
| already-attached handler instead of stacking duplicate handlers, and | |
| disables propagation to the root logger so output is not duplicated by a | |
| caller's own root configuration. | |
| Args: | |
| name: The logger name, conventionally the calling module's stage id | |
| (e.g. ``"R3"`` or ``"redstack.engines.scoring"``). | |
| level: The minimum level this logger emits at. | |
| Returns: | |
| A configured :class:`logging.Logger`. | |
| """ | |
| logger = logging.getLogger(name) | |
| logger.setLevel(level) | |
| if not any(isinstance(h, _DeterministicStreamHandler) for h in logger.handlers): | |
| handler = _DeterministicStreamHandler(sys.stderr) | |
| handler.setFormatter(_DeterministicFormatter(_REPRO_SAFE_FORMAT)) | |
| logger.addHandler(handler) | |
| logger.propagate = False | |
| return logger | |
| def _render_fields(fields: Mapping[str, object]) -> str: | |
| """Render structured fields as ``key=value`` pairs in ascending key order. | |
| Sorting by key — rather than trusting call-site or dict insertion order — | |
| is what makes the rendered line stable across runs and Python versions. | |
| """ | |
| return " ".join(f"{key}={fields[key]!r}" for key in sorted(fields)) | |
| def stage_metric( | |
| logger: logging.Logger, stage: str, event: str, **fields: object | |
| ) -> None: | |
| """Emit one deterministic structured record: ``[stage] event key=value ...``. | |
| This is the per-stage metric capture surface: stages report facts | |
| (``stage_metric(log, "R6", "ranked", honeypot_count=0, size=100)``) rather | |
| than hand-formatting strings, so every emitted line has a stable shape. | |
| Args: | |
| logger: A logger obtained from :func:`get_logger`. | |
| stage: The stage identifier the metric belongs to (e.g. ``"R6"``). | |
| event: A short, stable event name (e.g. ``"ranked"``, ``"started"``). | |
| **fields: Arbitrary structured payload; rendered sorted by key. | |
| """ | |
| rendered = _render_fields(fields) | |
| message = f"[{stage}] {event}" + (f" {rendered}" if rendered else "") | |
| logger.info(message) | |