LQAgentPlatform/foundation/observability/logger/weekly_handler.py

import os
import logging
from datetime import datetime, timedelta
from logging.handlers import RotatingFileHandler
from foundation.infrastructure.tracing.trace_context import trace_filter


def _get_week_range(dt: datetime) -> tuple:
    """获取 dt 所在周的周一和周日日期"""
    monday = dt - timedelta(days=dt.weekday())
    sunday = monday + timedelta(days=6)
    return monday, sunday


def _get_week_dir_name(dt: datetime) -> str:
    """生成周目录名: 2026-W19_0505-0511"""
    monday, sunday = _get_week_range(dt)
    iso_cal = dt.isocalendar()
    week_num = iso_cal[1]
    return f"{monday.year}-W{week_num:02d}_{monday.strftime('%m%d')}-{sunday.strftime('%m%d')}"


class WeeklyRotatingFileHandler(logging.Handler):
    """按周创建目录、按天创建日志文件的 Handler"""

    def __init__(self, base_dir: str, module_name: str, level_name: str,
                 file_max_mb: int = 10, backup_count: int = 5):
        super().__init__()
        self.base_dir = base_dir
        self.module_name = module_name
        self.level_name = level_name
        self.file_max_bytes = file_max_mb * 1024 * 1024
        self.backup_count = backup_count

        self._current_week_dir = None
        self._current_date_str = None
        self._current_handler = None

    def _resolve_handler(self) -> RotatingFileHandler:
        now = datetime.now()
        week_dir = _get_week_dir_name(now)
        date_str = now.strftime('%m%d')

        if week_dir != self._current_week_dir or date_str != self._current_date_str:
            if self._current_handler:
                self._current_handler.close()
            self._current_week_dir = week_dir
            self._current_date_str = date_str

            dir_path = os.path.join(self.base_dir, week_dir)
            os.makedirs(dir_path, exist_ok=True)

            filename = os.path.join(
                dir_path,
                f"{date_str}_{self.module_name}_{self.level_name}.log"
            )
            self._current_handler = RotatingFileHandler(
                filename=filename,
                mode='a',
                maxBytes=self.file_max_bytes,
                backupCount=self.backup_count,
                encoding='utf-8',
                delay=True
            )
            self._current_handler.setFormatter(self.formatter)
            self._current_handler.addFilter(trace_filter)

        return self._current_handler

    def emit(self, record):
        try:
            handler = self._resolve_handler()
            if handler:
                handler.emit(record)
        except Exception:
            self.handleError(record)


def create_weekly_file_handlers(module_name: str, base_dir: str,
                                file_max_mb: int = 10, backup_count: int = 5) -> list:
    """为模块创建按周组织的日志 Handler 列表"""
    level_map = {
        'debug': logging.DEBUG,
        'info': logging.INFO,
        'warning': logging.WARNING,
        'error': logging.ERROR,
        'critical': logging.CRITICAL,
    }

    handlers = []
    for level_name, level_value in level_map.items():
        h = WeeklyRotatingFileHandler(
            base_dir=base_dir,
            module_name=module_name,
            level_name=level_name,
            file_max_mb=file_max_mb,
            backup_count=backup_count,
        )
        h.setLevel(level_value)
        h.addFilter(lambda record, lvl=level_value: record.levelno >= lvl)
        handlers.append(h)
    return handlers