| """ |
| Structured logging configuration with PII redaction. |
| Uses structlog for JSON-formatted log output. |
| """ |
| import logging |
| import re |
| import structlog |
|
|
|
|
| _EMAIL_RE = re.compile(r"[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+") |
| _PHONE_RE = re.compile(r"\b\d{3}[-.]?\d{3}[-.]?\d{4}\b") |
|
|
|
|
| def _redact_pii(_logger, _method, event_dict): |
| """Redact emails and phone numbers from log messages.""" |
| msg = event_dict.get("event", "") |
| if isinstance(msg, str): |
| msg = _EMAIL_RE.sub("[REDACTED_EMAIL]", msg) |
| msg = _PHONE_RE.sub("[REDACTED_PHONE]", msg) |
| event_dict["event"] = msg |
| return event_dict |
|
|
|
|
| def setup_logging(log_level: str = "INFO"): |
| structlog.configure( |
| processors=[ |
| structlog.contextvars.merge_contextvars, |
| structlog.stdlib.filter_by_level, |
| structlog.stdlib.add_logger_name, |
| structlog.stdlib.add_log_level, |
| structlog.processors.TimeStamper(fmt="iso"), |
| _redact_pii, |
| structlog.processors.StackInfoRenderer(), |
| structlog.processors.format_exc_info, |
| structlog.processors.JSONRenderer(), |
| ], |
| wrapper_class=structlog.stdlib.BoundLogger, |
| context_class=dict, |
| logger_factory=structlog.stdlib.LoggerFactory(), |
| cache_logger_on_first_use=True, |
| ) |
| logging.basicConfig(format="%(message)s", level=getattr(logging, log_level)) |
|
|
|
|
| def get_logger(name: str = __name__): |
| return structlog.get_logger(name) |
|
|
