| from __future__ import annotations | |
| import os | |
| import sys | |
| from pathlib import Path | |
| from loguru import logger | |
| _configured_log_path: Path | None = None | |
| def configure_logging(storage_dir: Path) -> Path: | |
| global _configured_log_path | |
| logs_dir = (storage_dir / "logs").resolve() | |
| logs_dir.mkdir(parents=True, exist_ok=True) | |
| log_path = logs_dir / "service.log" | |
| if _configured_log_path == log_path: | |
| return log_path | |
| _configured_log_path = log_path | |
| level = (os.getenv("LOG_LEVEL") or "INFO").strip().upper() | |
| rotation = (os.getenv("LOG_ROTATION") or "1 day").strip() | |
| retention = (os.getenv("LOG_RETENTION") or "14 days").strip() | |
| compression = (os.getenv("LOG_COMPRESSION") or "zip").strip() | |
| fmt = "{time:YYYY-MM-DD HH:mm:ss.SSS} | {level: <8} | {name}:{function}:{line} - {message} | {extra}" | |
| logger.remove() | |
| logger.add( | |
| sys.stderr, | |
| level=level, | |
| format=fmt, | |
| enqueue=True, | |
| backtrace=False, | |
| diagnose=False, | |
| ) | |
| logger.add( | |
| str(log_path), | |
| level=level, | |
| format=fmt, | |
| enqueue=True, | |
| backtrace=False, | |
| diagnose=False, | |
| rotation=rotation, | |
| retention=retention, | |
| compression=compression, | |
| ) | |
| logger.info(f"logging_initialized path={log_path} rotation={rotation} retention={retention}") | |
| return log_path | |