app.case_logging

  1"""Case-scoped application logging helpers.
  2
  3Routes Python :mod:`logging` records to per-case log files so that each
  4forensic case maintains its own ``logs/application.log``.  This is separate
  5from the forensic audit trail (``audit.jsonl``) and is intended for
  6developer-facing diagnostic output.
  7
  8The mechanism works by attaching a :class:`logging.Handler` to the root
  9logger for every active case.  A :class:`~contextvars.ContextVar` tracks
 10which case ID is active on the current thread / coroutine so that log
 11records are routed to the correct file without requiring callers to pass
 12case references explicitly.
 13
 14Typical usage::
 15
 16    register_case_log_handler(case_id, case_dir)
 17    with case_log_context(case_id):
 18        logging.getLogger("app").info("Parsing started")
 19
 20Attributes:
 21    CASE_LOGS_DIRNAME: Subdirectory name for case log files.
 22    CASE_LOG_FILENAME: Name of the per-case application log file.
 23    CASE_LOG_FORMAT: :mod:`logging` format string for case log entries.
 24"""
 25
 26from __future__ import annotations
 27
 28from contextlib import contextmanager
 29from contextvars import ContextVar, Token
 30import logging
 31from pathlib import Path
 32import threading
 33from typing import Iterator
 34
 35__all__ = [
 36    "push_case_log_context",
 37    "pop_case_log_context",
 38    "case_log_context",
 39    "register_case_log_handler",
 40    "unregister_case_log_handler",
 41    "unregister_all_case_log_handlers",
 42]
 43
 44CASE_LOGS_DIRNAME = "logs"
 45CASE_LOG_FILENAME = "application.log"
 46CASE_LOG_FORMAT = "%(asctime)s %(levelname)s [%(name)s] %(message)s"
 47
 48_ACTIVE_CASE_ID: ContextVar[str | None] = ContextVar("aift_active_case_id", default=None)
 49_HANDLER_LOCK = threading.RLock()
 50_CASE_HANDLERS: dict[str, logging.Handler] = {}
 51
 52
 53class _CasePathLogHandler(logging.Handler):
 54    """Logging handler that appends formatted records to a file path.
 55
 56    The file is opened and closed for each record to avoid holding file
 57    descriptors open across long-lived operations.
 58
 59    Args:
 60        log_path: Destination file path for log output.
 61    """
 62
 63    def __init__(self, log_path: Path) -> None:
 64        super().__init__()
 65        self.log_path = Path(log_path)
 66
 67    def emit(self, record: logging.LogRecord) -> None:
 68        """Write a single formatted log record to :attr:`log_path`."""
 69        try:
 70            rendered = self.format(record)
 71            self.log_path.parent.mkdir(parents=True, exist_ok=True)
 72            with self.log_path.open("a", encoding="utf-8") as stream:
 73                stream.write(f"{rendered}\n")
 74        except OSError:
 75            return
 76
 77
 78class _CaseLogFilter(logging.Filter):
 79    """Logging filter that passes only records matching a specific case ID.
 80
 81    If the record does not carry a ``case_id`` attribute, the filter
 82    falls back to the context variable set via :func:`push_case_log_context`.
 83
 84    Args:
 85        case_id: The case identifier this filter accepts.
 86    """
 87
 88    def __init__(self, case_id: str) -> None:
 89        super().__init__()
 90        self.case_id = case_id
 91
 92    def filter(self, record: logging.LogRecord) -> bool:
 93        """Return *True* when the record belongs to this filter's case."""
 94        record_case_id = getattr(record, "case_id", None)
 95        if not record_case_id:
 96            record_case_id = _ACTIVE_CASE_ID.get()
 97            if record_case_id:
 98                setattr(record, "case_id", record_case_id)
 99        return str(record_case_id or "").strip() == self.case_id
100
101
102def push_case_log_context(case_id: str | None) -> Token[str | None]:
103    """Bind a case ID to the current execution context for log routing.
104
105    Args:
106        case_id: Case identifier to set, or *None* to clear.
107
108    Returns:
109        A :class:`~contextvars.Token` that can be passed to
110        :func:`pop_case_log_context` to restore the previous value.
111    """
112    normalized = str(case_id).strip() if case_id is not None else None
113    return _ACTIVE_CASE_ID.set(normalized or None)
114
115
116def pop_case_log_context(token: Token[str | None]) -> None:
117    """Restore the previous case log context.
118
119    Args:
120        token: Token returned by a prior :func:`push_case_log_context` call.
121    """
122    _ACTIVE_CASE_ID.reset(token)
123
124
125@contextmanager
126def case_log_context(case_id: str | None) -> Iterator[None]:
127    """Context manager that binds *case_id* for logs emitted within its scope.
128
129    Args:
130        case_id: Case identifier to bind, or *None* to clear.
131
132    Yields:
133        Nothing.  The context variable is restored on exit.
134    """
135    token = push_case_log_context(case_id)
136    try:
137        yield
138    finally:
139        pop_case_log_context(token)
140
141
142def register_case_log_handler(case_id: str, case_dir: str | Path) -> Path:
143    """Attach a per-case file handler to the root logger.
144
145    Creates a ``logs/application.log`` file inside *case_dir* and installs
146    a :class:`_CasePathLogHandler` with a :class:`_CaseLogFilter` so that
147    only records matching *case_id* are written there.  Duplicate
148    registrations for the same case are silently ignored.
149
150    Args:
151        case_id: Non-empty case identifier string.
152        case_dir: Path to the case directory where logs will be stored.
153
154    Returns:
155        The :class:`~pathlib.Path` to the created log file.
156
157    Raises:
158        ValueError: If *case_id* is empty or whitespace-only.
159    """
160    normalized_case_id = str(case_id).strip()
161    if not normalized_case_id:
162        raise ValueError("case_id must be a non-empty string.")
163
164    root_logger = logging.getLogger()
165    app_logger = logging.getLogger("app")
166    logs_dir = Path(case_dir) / CASE_LOGS_DIRNAME
167    logs_dir.mkdir(parents=True, exist_ok=True)
168    log_path = logs_dir / CASE_LOG_FILENAME
169
170    with _HANDLER_LOCK:
171        existing = _CASE_HANDLERS.get(normalized_case_id)
172        if existing is not None:
173            return log_path
174
175        handler = _CasePathLogHandler(log_path)
176        handler.setLevel(logging.INFO)
177        handler.setFormatter(logging.Formatter(CASE_LOG_FORMAT))
178        handler.addFilter(_CaseLogFilter(normalized_case_id))
179        root_logger.addHandler(handler)
180        _CASE_HANDLERS[normalized_case_id] = handler
181
182        if app_logger.level == logging.NOTSET or app_logger.level > logging.INFO:
183            app_logger.setLevel(logging.INFO)
184
185    return log_path
186
187
188def unregister_case_log_handler(case_id: str) -> None:
189    """Detach and close the file handler for *case_id* if it exists.
190
191    Args:
192        case_id: Case identifier whose handler should be removed.
193    """
194    normalized_case_id = str(case_id).strip()
195    if not normalized_case_id:
196        return
197
198    with _HANDLER_LOCK:
199        handler = _CASE_HANDLERS.pop(normalized_case_id, None)
200        if handler is None:
201            return
202        root_logger = logging.getLogger()
203        root_logger.removeHandler(handler)
204        handler.close()
205
206
207def unregister_all_case_log_handlers() -> None:
208    """Detach and close all case file handlers."""
209    with _HANDLER_LOCK:
210        handlers = list(_CASE_HANDLERS.values())
211        _CASE_HANDLERS.clear()
212        root_logger = logging.getLogger()
213        for handler in handlers:
214            root_logger.removeHandler(handler)
215            handler.close()