Coverage for src / qdrant_loader_core / logging_processors.py: 79%

1"""Logging processors and formatters for structlog.""" 

2 

3from __future__ import annotations 

4 

5import logging 

6import re 

7from typing import Any 

8 

9 

10class CleanFormatter(logging.Formatter): 

11 """Formatter that removes ANSI color codes for clean file output.""" 

12 

13 def format(self, record: logging.LogRecord) -> str: 

14 message = super().format(record) 

15 try: 

16 ansi_escape = re.compile(r"\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])") 

17 return ansi_escape.sub("", message) 

18 except Exception: 

19 return message 

20 

21 

22def redact_processor( 

23 logger: Any, method_name: str, event_dict: dict[str, Any] 

24) -> dict[str, Any]: 

25 """Structlog processor to redact sensitive fields in event_dict.""" 

26 sensitive_keys = { 

27 "api_key", 

28 "llm_api_key", 

29 "authorization", 

30 "Authorization", 

31 "token", 

32 "access_token", 

33 "secret", 

34 "password", 

35 } 

36 

37 def mask(value: str) -> str: 

38 try: 

39 if not isinstance(value, str) or not value: 

40 return "***REDACTED***" 

41 if len(value) <= 8: 

42 return "***REDACTED***" 

43 return value[:2] + "***REDACTED***" + value[-2:] 

44 except Exception: 

45 return "***REDACTED***" 

46 

47 def deep_redact(obj: Any) -> Any: 

48 try: 

49 if isinstance(obj, dict): 

50 return { 

51 k: ( 

52 mask(v) 

53 if k in sensitive_keys and isinstance(v, str) 

54 else deep_redact(v) 

55 ) 

56 for k, v in obj.items() 

57 } 

58 if isinstance(obj, list): 

59 return [deep_redact(i) for i in obj] 

60 return obj 

61 except Exception: 

62 return obj 

63 

64 return deep_redact(event_dict)