97 lines
3.1 KiB
Python
97 lines
3.1 KiB
Python
import logging
|
|
import json
|
|
from logging.handlers import RotatingFileHandler
|
|
from pathlib import Path
|
|
from datetime import datetime
|
|
|
|
class JsonFormatter(logging.Formatter):
|
|
"""
|
|
Custom JSON log formatter for structured logging.
|
|
"""
|
|
def format(self, record: logging.LogRecord) -> str:
|
|
log_object = {
|
|
"timestamp": datetime.utcfromtimestamp(record.created).strftime("%Y-%m-%dT%H:%M:%SZ"),
|
|
"level": record.levelname,
|
|
"message": record.getMessage(),
|
|
"module": record.module,
|
|
"funcName": record.funcName,
|
|
"line": record.lineno,
|
|
}
|
|
|
|
# Include any extra fields added via logger.info("msg", extra={...})
|
|
if hasattr(record, "extra") and isinstance(record.extra, dict):
|
|
log_object.update(record.extra)
|
|
|
|
# Include exception info if present
|
|
if record.exc_info:
|
|
log_object["exception"] = self.formatException(record.exc_info)
|
|
|
|
return json.dumps(log_object, ensure_ascii=False)
|
|
|
|
def setup_logger(log_folder: Path, log_file_name: str = "conversion.log", level=logging.INFO) -> logging.Logger:
|
|
"""
|
|
Sets up a logger that prints to console and writes to a rotating JSON log file.
|
|
"""
|
|
log_folder.mkdir(parents=True, exist_ok=True)
|
|
log_file = log_folder / log_file_name
|
|
|
|
logger = logging.getLogger("conversion_logger")
|
|
logger.setLevel(level)
|
|
logger.propagate = False # Prevent double logging
|
|
|
|
# Formatters
|
|
text_formatter = logging.Formatter(
|
|
"%(asctime)s [%(levelname)s] %(message)s (%(module)s:%(lineno)d)",
|
|
datefmt="%Y-%m-%d %H:%M:%S"
|
|
)
|
|
json_formatter = JsonFormatter()
|
|
|
|
# Console handler (human-readable)
|
|
console_handler = logging.StreamHandler()
|
|
console_handler.setFormatter(text_formatter)
|
|
console_handler.setLevel(level)
|
|
|
|
# File handler (JSON logs)
|
|
file_handler = RotatingFileHandler(log_file, maxBytes=5 * 1024 * 1024, backupCount=3, encoding="utf-8")
|
|
file_handler.setFormatter(json_formatter)
|
|
file_handler.setLevel(level)
|
|
|
|
# Add handlers only once
|
|
if not logger.handlers:
|
|
logger.addHandler(console_handler)
|
|
logger.addHandler(file_handler)
|
|
|
|
return logger
|
|
|
|
|
|
def setup_failure_logger(log_folder: Path) -> logging.Logger:
|
|
"""
|
|
Setup a dedicated logger for encoding failures.
|
|
Returns a logger that writes to logs/failure.log
|
|
"""
|
|
log_folder.mkdir(parents=True, exist_ok=True)
|
|
log_file = log_folder / "failure.log"
|
|
|
|
logger = logging.getLogger("failure_logger")
|
|
logger.setLevel(logging.WARNING)
|
|
|
|
# Prevent duplicate handlers
|
|
if logger.handlers:
|
|
return logger
|
|
|
|
# Simple text formatter for failure log
|
|
formatter = logging.Formatter(
|
|
"%(asctime)s | %(message)s",
|
|
datefmt="%Y-%m-%d %H:%M:%S"
|
|
)
|
|
|
|
# File handler only
|
|
file_handler = RotatingFileHandler(log_file, maxBytes=5 * 1024 * 1024, backupCount=3, encoding="utf-8")
|
|
file_handler.setFormatter(formatter)
|
|
file_handler.setLevel(logging.WARNING)
|
|
|
|
logger.addHandler(file_handler)
|
|
logger.propagate = False
|
|
|
|
return logger
|