You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

每日午夜无异常时无法创建新空日志文件,寻求技术解决方案

问题分析与解决方案

你的核心问题是自定义的RosCanTimedRotatingFileHandler原本为5分钟轮转设计,但当前配置为每日午夜轮转时,逻辑冲突导致无法自动生成新日志文件。以下是具体问题点和修复方案:

关键问题点

  1. 轮转逻辑冲突:自定义Handler的__init__和doRollover方法都是针对5分钟间隔编写的,当使用when='midnight'时,时间计算逻辑完全错误,破坏了午夜轮转的时间点。
  2. 文件名处理混乱:初始化时日志文件名已包含5分钟时间戳,后续轮转时又通过split('-')[0]截断文件名,导致原文件被重命名为无后缀的基础名,新文件名生成逻辑混乱。
  3. 异常静默吞噬:类定义内的try-except会覆盖__init__和doRollover的异常,导致问题无法被及时排查。

解决方案一:使用原生Handler实现每日轮转

如果仅需要每日午夜自动生成新日志,直接使用Python原生的TimedRotatingFileHandler即可,无需自定义子类:

import logging
import logging.handlers as handlers
import os
from datetime import datetime

LOGGER = logging.getLogger('tractor_init_config_manager')
LOGGER.setLevel(logging.INFO)

# 定义日志格式
FORMATTER = logging.Formatter(
    '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
LOG_PATH = "/opt/monarch/debug/"
# 使用不带时间戳的基础文件名,让Handler自动生成后缀
BASE_FILE_NAME = "ros_can_logs/integration.log"
FULL_LOG_PATH = os.path.join(LOG_PATH, BASE_FILE_NAME)

def folder_exists(file_path):
    """检查目录是否存在,不存在则创建"""
    try:
        path = os.path.dirname(file_path)
        os.makedirs(path, exist_ok=True)
        # 初始化空日志文件(如果不存在)
        if not os.path.exists(file_path):
            with open(file_path, 'w') as f:
                f.write("Initiating the logging...\n")
    except Exception as err:
        print(err)

folder_exists(FULL_LOG_PATH)

# 配置每日午夜轮转
LOG_HANDLER = handlers.TimedRotatingFileHandler(
    FULL_LOG_PATH,
    when='midnight',
    interval=1,
    backupCount=10,
    encoding='utf-8',
    delay=False
)
# 设置日志文件后缀格式
LOG_HANDLER.suffix = "%Y-%m-%d.log"
LOG_HANDLER.setLevel(logging.INFO)
LOG_HANDLER.setFormatter(FORMATTER)

LOGGER.addHandler(LOG_HANDLER)

def info(msg):
    LOGGER.info(str(msg))

def error(msg):
    LOGGER.error(str(msg))

解决方案二:修复自定义Handler兼容两种轮转模式

如果需要同时保留5分钟轮转和每日午夜轮转的能力,修改自定义Handler的分支逻辑:

import codecs
import glob
import logging
import logging.handlers as handlers
import os
import re
import time
from datetime import datetime
import syslog

class RosCanTimedRotatingFileHandler(handlers.TimedRotatingFileHandler):
    """支持5分钟轮转和每日午夜轮转的日志Handler"""
    def __init__(self, *args, **kwargs):
        self.target_when = kwargs.get('when', 'M')
        super().__init__(*args, **kwargs)
        
        if self.target_when == 'M' and self.interval == 5:
            # 5分钟轮转专属配置
            self.suffix = '%Y-%m-%d-%H-%M-%S.log'
            self.extMatch = re.compile(r"^\d{4}-\d{2}-\d{2}-\d{2}-\d{2}-\d{2}\.log$")
            # 计算下一个5分钟整点的轮转时间
            now = datetime.now()
            next_rollover = now.replace(minute=(now.minute - now.minute %5)+5, second=0, microsecond=0)
            self.rolloverAt = time.mktime(next_rollover.timetuple())
        elif self.target_when == 'midnight' and self.interval ==1:
            # 每日午夜轮转专属配置
            self.suffix = '%Y-%m-%d.log'
            self.extMatch = re.compile(r"^\d{4}-\d{2}-\d{2}\.log$")
            # 使用原生午夜轮转时间计算逻辑
            pass

    def doRollover(self):
        try:
            if self.target_when == 'M' and self.interval ==5:
                # 5分钟轮转处理逻辑
                time_t = self.rolloverAt - self.interval*60
                time_tuple = time.localtime(time_t)
                base_name = self.baseFilename.rsplit('-', 1)[0] if '-' in self.baseFilename else self.baseFilename
                dfn = f"{base_name}-{time.strftime(self.suffix, time_tuple)}"
                
                if os.path.exists(dfn):
                    os.remove(dfn)
                if os.path.exists(self.baseFilename):
                    os.rename(self.baseFilename, dfn)
                
                # 生成新的5分钟对齐文件名
                cur_time = datetime.now()
                current_timestamp = cur_time.replace(
                    minute=(cur_time.minute - cur_time.minute %5), second=0, microsecond=0
                ).strftime("%Y-%m-%d-%H-%M-%S")
                self.baseFilename = f"{base_name}-{current_timestamp}.log"
                
                # 清理超过备份数量的旧日志
                if self.backupCount >0:
                    files_list = glob.glob(f"{base_name}-20*")
                    if len(files_list) > self.backupCount:
                        files_list.sort()
                        os.remove(files_list[0])
            elif self.target_when == 'midnight' and self.interval ==1:
                # 每日午夜轮转处理逻辑,复用原生逻辑并确保空文件生成
                super().doRollover()
                # 确保新日志文件被创建(即使无日志写入)
                if not os.path.exists(self.baseFilename):
                    self.stream = codecs.open(self.baseFilename, 'w', self.encoding) if self.encoding else open(self.baseFilename, 'w')
            
            # 重新打开日志流
            self.stream = codecs.open(self.baseFilename, 'w', self.encoding) if self.encoding else open(self.baseFilename, 'w')
            
            # 计算下一次轮转时间
            current_time = time.time()
            new_rollover_at = self.computeRollover(current_time)
            while new_rollover_at <= current_time:
                new_rollover_at += self.interval if self.target_when != 'M' else self.interval*60
            self.rolloverAt = new_rollover_at
        except Exception as err:
            syslog.syslog(f"error in giles_log file of sync service: {err}")

# 初始化日志
LOGGER = logging.getLogger('tractor_init_config_manager')
LOGGER.setLevel(logging.INFO)

FORMATTER = logging.Formatter(
    '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
LOG_PATH = "/opt/monarch/debug/"
BASE_FILE_NAME = "ros_can_logs/integration.log"
FULL_LOG_PATH = os.path.join(LOG_PATH, BASE_FILE_NAME)

def folder_exists(file_path):
    try:
        path = os.path.dirname(file_path)
        os.makedirs(path, exist_ok=True)
        if not os.path.exists(file_path):
            with open(file_path, 'w') as f:
                f.write("Initiating the logging...\n")
    except Exception as err:
        print(err)

folder_exists(FULL_LOG_PATH)

LOG_HANDLER = RosCanTimedRotatingFileHandler(
    FULL_LOG_PATH,
    when='midnight',
    interval=1,
    backupCount=10,
    delay=False
)
LOG_HANDLER.setLevel(logging.INFO)
LOG_HANDLER.setFormatter(FORMATTER)

LOGGER.addHandler(LOG_HANDLER)

def info(msg):
    LOGGER.info(str(msg))

def error(msg):
    LOGGER.error(str(msg))

内容的提问来源于stack exchange,提问作者tushar aloorkar

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.07.09 06:17:02