Upload core/logging.py with huggingface_hub

core/logging.py

@@ -2,12 +2,60 @@ import contextlib
 import json
 import logging
 import os
+import re
 import sys
 from datetime import datetime, timezone
 from pathlib import Path
 from typing import Any
 
 
+class PIIScrubber:
+    """PII detection and scrubbing utilities"""
+
+    # PII patterns
+    EMAIL_PATTERN = re.compile(r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b")
+    PHONE_PATTERN = re.compile(r"\b\d{3}[-.]?\d{3}[-.]?\d{4}\b")
+    SSN_PATTERN = re.compile(r"\b\d{3}-\d{2}-\d{4}\b")
+    CREDIT_CARD_PATTERN = re.compile(r"\b\d{4}[-\s]?\d{4}[-\s]?\d{4}[-\s]?\d{4}\b")
+    IP_ADDRESS_PATTERN = re.compile(r"\b(?:[0-9]{1,3}\.){3}[0-9]{1,3}\b")
+
+    # Additional patterns for sensitive data
+    BANK_ACCOUNT_PATTERN = re.compile(r"\b\d{8,17}\b")
+    PASSPORT_PATTERN = re.compile(r"\b[A-Z]{1,2}\d{7,8}\b")
+    DRIVER_LICENSE_PATTERN = re.compile(r"\b[A-Z]{1,2}\d{6,8}\b")
+
+    @staticmethod
+    def scrub_pii(text: str, mask_char: str = "***") -> str:
+        """Scrub PII from text using patterns"""
+        if not text or not isinstance(text, str):
+            return text
+
+        scrubbed_text = text
+
+        # Apply all PII patterns
+        patterns = [
+            (PIIScrubber.EMAIL_PATTERN, f"{mask_char}@{mask_char}.com"),
+            (PIIScrubber.PHONE_PATTERN, f"{mask_char}-{mask_char}-{mask_char}"),
+            (PIIScrubber.SSN_PATTERN, f"{mask_char}-{mask_char}-{mask_char}"),
+            (
+                PIIScrubber.CREDIT_CARD_PATTERN,
+                f"{mask_char}-{mask_char}-{mask_char}-{mask_char}",
+            ),
+            (
+                PIIScrubber.IP_ADDRESS_PATTERN,
+                f"{mask_char}.{mask_char}.{mask_char}.{mask_char}",
+            ),
+            (PIIScrubber.BANK_ACCOUNT_PATTERN, mask_char * 8),
+            (PIIScrubber.PASSPORT_PATTERN, f"{mask_char}{mask_char}"),
+            (PIIScrubber.DRIVER_LICENSE_PATTERN, f"{mask_char}{mask_char}"),
+        ]
+
+        for pattern, replacement in patterns:
+            scrubbed_text = pattern.sub(replacement, scrubbed_text)
+
+        return scrubbed_text
+
+
 class JSONFormatter(logging.Formatter):
     """Custom JSON formatter for structured logging"""
 
@@ -33,7 +81,15 @@ class JSONFormatter(logging.Formatter):
         if hasattr(record, "user_id"):
             log_entry["user_id"] = record.user_id
         if hasattr(record, "ip_address"):
-            log_entry["ip_address"] = record.ip_address
+            log_entry["ip_address"] = PIIScrubber.scrub_pii(record.ip_address)
+
+        # Scrub the message itself
+        log_entry["message"] = PIIScrubber.scrub_pii(log_entry["message"])
+
+        # Scrub extra fields
+        for key, value in log_entry.items():
+            if isinstance(value, str) and key not in ["timestamp", "level", "logger"]:
+                log_entry[key] = PIIScrubber.scrub_pii(value)
 
         return json.dumps(log_entry, default=str)
 
@@ -83,7 +139,9 @@ def setup_logging(
 
         from logging.handlers import RotatingFileHandler
 
-        fh = RotatingFileHandler(log_file, maxBytes=max_file_size, backupCount=backup_count, encoding="utf-8")
+        fh = RotatingFileHandler(
+            log_file, maxBytes=max_file_size, backupCount=backup_count, encoding="utf-8"
+        )
         fh.setLevel(numeric_level)
         fh.setFormatter(formatter)
         logger.addHandler(fh)
@@ -239,7 +297,9 @@ def configure_environment_logging():
 
     if env == "development":
         # Development: more verbose, console only, human-readable
-        setup_logging(level="DEBUG", format_type="text", enable_console=True, enable_file=False)
+        setup_logging(
+            level="DEBUG", format_type="text", enable_console=True, enable_file=False
+        )
     elif env == "testing":
         # Testing: capture all logs, minimal output, structured
         setup_logging(