| """ | |
| 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) | |