|
1 | 1 | import logging
|
2 | 2 | import os
|
3 | 3 | from pathlib import Path
|
| 4 | +from logging.handlers import RotatingFileHandler |
| 5 | + |
4 | 6 |
|
5 | 7 | class IgnoreLogChangeDetectedFilter(logging.Filter):
|
6 | 8 | def filter(self, record: logging.LogRecord):
|
7 | 9 | return "Detected file change in" not in record.getMessage()
|
8 | 10 |
|
| 11 | + |
9 | 12 | def setup_logging(format: str = None):
|
10 | 13 | """
|
11 |
| - Configure logging for the application. |
12 |
| - Reads LOG_LEVEL and LOG_FILE_PATH from environment (defaults: INFO, logs/application.log). |
13 |
| - Ensures log directory exists, and configures both file and console handlers. |
| 14 | + Configure logging for the application with log rotation. |
| 15 | +
|
| 16 | + Environment variables: |
| 17 | + LOG_LEVEL: Log level (default: INFO) |
| 18 | + LOG_FILE_PATH: Path to log file (default: logs/application.log) |
| 19 | + LOG_MAX_SIZE: Max size in MB before rotating (default: 10MB) |
| 20 | + LOG_BACKUP_COUNT: Number of backup files to keep (default: 5) |
| 21 | +
|
| 22 | + Ensures log directory exists, prevents path traversal, and configures |
| 23 | + both rotating file and console handlers. |
14 | 24 | """
|
15 | 25 | # Determine log directory and default file path
|
16 | 26 | base_dir = Path(__file__).parent
|
17 | 27 | log_dir = base_dir / "logs"
|
18 | 28 | log_dir.mkdir(parents=True, exist_ok=True)
|
19 | 29 | default_log_file = log_dir / "application.log"
|
20 | 30 |
|
21 |
| - # Get log level and file path from environment |
| 31 | + # Get log level from environment |
22 | 32 | log_level_str = os.environ.get("LOG_LEVEL", "INFO").upper()
|
23 | 33 | log_level = getattr(logging, log_level_str, logging.INFO)
|
24 |
| - log_file_path = Path(os.environ.get( |
25 |
| - "LOG_FILE_PATH", str(default_log_file))) |
26 | 34 |
|
27 |
| - # ensure log_file_path is within the project's logs directory to prevent path traversal |
| 35 | + # Get log file path |
| 36 | + log_file_path = Path(os.environ.get("LOG_FILE_PATH", str(default_log_file))) |
| 37 | + |
| 38 | + # Secure path check: must be inside logs/ directory |
28 | 39 | log_dir_resolved = log_dir.resolve()
|
29 | 40 | resolved_path = log_file_path.resolve()
|
30 | 41 | if not str(resolved_path).startswith(str(log_dir_resolved) + os.sep):
|
31 |
| - raise ValueError( |
32 |
| - f"LOG_FILE_PATH '{log_file_path}' is outside the trusted log directory '{log_dir_resolved}'" |
33 |
| - ) |
34 |
| - # Ensure parent dirs exist for the log file |
| 42 | + raise ValueError(f"LOG_FILE_PATH '{log_file_path}' is outside the trusted log directory '{log_dir_resolved}'") |
| 43 | + |
| 44 | + # Ensure parent directories exist |
35 | 45 | resolved_path.parent.mkdir(parents=True, exist_ok=True)
|
36 | 46 |
|
37 |
| - # Configure logging handlers and format |
38 |
| - logging.basicConfig( |
39 |
| - level=log_level, |
40 |
| - format = format or "%(asctime)s - %(levelname)s - %(name)s - %(filename)s:%(lineno)d - %(message)s", |
41 |
| - handlers=[ |
42 |
| - logging.FileHandler(resolved_path), |
43 |
| - logging.StreamHandler() |
44 |
| - ], |
45 |
| - force=True |
46 |
| - ) |
47 |
| - |
48 |
| - # Ignore log file's change detection |
49 |
| - for handler in logging.getLogger().handlers: |
50 |
| - handler.addFilter(IgnoreLogChangeDetectedFilter()) |
| 47 | + # Get max log file size (default: 10MB) |
| 48 | + try: |
| 49 | + max_mb = int(os.environ.get("LOG_MAX_SIZE", 10)) # 10MB default |
| 50 | + max_bytes = max_mb * 1024 * 1024 |
| 51 | + except (TypeError, ValueError): |
| 52 | + max_bytes = 10 * 1024 * 1024 # fallback to 10MB on error |
51 | 53 |
|
52 |
| - # Initial debug message to confirm configuration |
| 54 | + # Get backup count (default: 5) |
| 55 | + try: |
| 56 | + backup_count = int(os.environ.get("LOG_BACKUP_COUNT", 5)) |
| 57 | + except ValueError: |
| 58 | + backup_count = 5 |
| 59 | + |
| 60 | + # Configure format |
| 61 | + log_format = format or "%(asctime)s - %(levelname)s - %(name)s - %(filename)s:%(lineno)d - %(message)s" |
| 62 | + |
| 63 | + # Create handlers |
| 64 | + file_handler = RotatingFileHandler(resolved_path, maxBytes=max_bytes, backupCount=backup_count, encoding="utf-8") |
| 65 | + console_handler = logging.StreamHandler() |
| 66 | + |
| 67 | + # Set format for both handlers |
| 68 | + formatter = logging.Formatter(log_format) |
| 69 | + file_handler.setFormatter(formatter) |
| 70 | + console_handler.setFormatter(formatter) |
| 71 | + |
| 72 | + # Add filter to suppress "Detected file change" messages |
| 73 | + file_handler.addFilter(IgnoreLogChangeDetectedFilter()) |
| 74 | + console_handler.addFilter(IgnoreLogChangeDetectedFilter()) |
| 75 | + |
| 76 | + # Apply logging configuration |
| 77 | + logging.basicConfig(level=log_level, handlers=[file_handler, console_handler], force=True) |
| 78 | + |
| 79 | + # Log configuration info |
53 | 80 | logger = logging.getLogger(__name__)
|
54 |
| - logger.debug(f"Log level set to {log_level_str}, log file: {resolved_path}") |
| 81 | + logger.debug( |
| 82 | + f"Logging configured: level={log_level_str}, " |
| 83 | + f"file={resolved_path}, max_size={max_bytes} bytes, " |
| 84 | + f"backup_count={backup_count}" |
| 85 | + ) |
0 commit comments