AttributeError: 'CustomTimedRotatingFileHandler' object has no attribute 'owner_id'

<think>我们遇到了一个AttributeError: 'CustomTimedRotatingFileHandler' object has no attribute 'owner_id'。这个错误通常发生在尝试访问一个不存在的属性时。 在之前的代码中，我们在`CustomTimedRotatingFileHandler`的`__init__`方法中设置了`owner_id`和`group_id`属性。但是，错误提示没有`owner_id`属性，说明在访问该属性时，它还没有被正确初始化。 可能的原因： 1. 在`__init__`方法中，我们尝试获取`owner`和`group`对应的ID，但如果发生了异常（比如用户或组不存在），我们设置了`owner_id`和`group_id`为当前用户/组，但可能异常处理代码有误导致属性没有正确创建？ 2. 在调用`__init__`方法时，可能由于某种原因跳过了设置这些属性的代码。 然而，查看之前提供的`CustomTimedRotatingFileHandler`代码，我们在`__init__`中确实有设置这些属性。但注意，在父类的`__init__`中可能会调用`_open`方法，而我们的`_open`方法中使用了`owner_id`和`group_id`属性。如果父类的`__init__`调用了`_open`，那么此时`owner_id`和`group_id`可能还没有被初始化（因为我们的`__init__`中先调用了父类的`__init__`，然后才设置这些属性）。 因此，问题在于：父类的`__init__`方法中可能会调用`_open`（例如，如果`delay`参数为False，则立即打开文件），而我们的`_open`方法需要`owner_id`和`group_id`属性，但此时我们还没有设置这些属性（因为我们在父类`__init__`之后才设置）。 解决方案： 1. 调整初始化顺序：先设置`owner_id`和`group_id`，再调用父类的`__init__`。 2. 或者在`_open`方法中，如果属性还不存在，则延迟设置（但这样可能复杂）。 我们选择第一种：先初始化`owner_id`和`group_id`，再调用父类的`__init__`。 修改`CustomTimedRotatingFileHandler`的`__init__`方法： 原代码： super(CustomTimedRotatingFileHandler, self).__init__(...) 然后设置owner_id和group_id 改为： 先设置owner_id和group_id，再调用super 同样，在`CustomRotatingFileHandler`中也采用相同的调整。 下面我们重新提供修正后的两个处理器代码。 注意：由于之前两个处理器都有相同的问题，所以都需要调整初始化顺序。 另外，我们还需要确保在父类初始化之前，这些属性已经存在，因为父类初始化可能会调用`_open`（它需要这些属性）。 修正后的`CustomTimedRotatingFileHandler`：</think>### 问题分析：AttributeError 错误修复 您遇到的 `AttributeError: 'CustomTimedRotatingFileHandler' object has no attribute 'owner_id'` 错误通常发生在以下情况： 1. **初始化顺序问题**：父类的 `__init__` 方法可能在子类属性初始化之前就被调用 2. **异常处理问题**：当 `pwd.getpwnam()` 或 `grp.getgrnam()` 抛出异常时，属性可能没有被正确设置 3. **继承问题**：父类方法调用了子类中需要这些属性的方法 ### 修复后的完整代码 以下是修复后的 `CustomTimedRotatingFileHandler` 和 `CustomRotatingFileHandler` 实现： ```python import logging import logging.handlers import os import re import stat import pwd import grp import time class CustomTimedRotatingFileHandler(logging.handlers.TimedRotatingFileHandler): """ 修复后的自定义时间轮转日志处理器： 1. 修复了 owner_id/group_id 属性初始化问题 2. 按天轮询日志文件（格式：YYYY-mm-dd_HH-MM-SS） 3. 设置文件权限为 664 4. 修改文件所有者为 appadmin 用户和组 """ def __init__(self, filename, when='D', interval=1, backupCount=0, encoding=None, delay=False, utc=False, owner='appadmin', group='appadmin'): # 先初始化所有者和组信息 self.owner = owner self.group = group # 获取用户和组的ID try: self.owner_id = pwd.getpwnam(owner).pw_uid self.group_id = grp.getgrnam(group).gr_gid except KeyError as e: # 如果找不到用户/组，使用当前用户/组 self.owner_id = os.getuid() self.group_id = os.getgid() logging.error("User or group not found: %s. Using current UID/GID: %s/%s", str(e), self.owner_id, self.group_id) except Exception as e: # 其他异常处理 self.owner_id = os.getuid() self.group_id = os.getgid() logging.error("Error setting owner/group: %s. Using current UID/GID: %s/%s", str(e), self.owner_id, self.group_id) # 确保属性存在 if not hasattr(self, 'owner_id'): self.owner_id = os.getuid() if not hasattr(self, 'group_id'): self.group_id = os.getgid() # 调用父类构造函数 super(CustomTimedRotatingFileHandler, self).__init__( filename, when, interval, backupCount, encoding, delay, utc) # 设置按天轮询的文件后缀格式 if self.when == 'D' or self.when == 'MIDNIGHT': self.suffix = "%Y-%m-%d_%H-%M-%S" self.extMatch = re.compile(r"^\d{4}-\d{2}-\d{2}_\d{2}-\d{2}-\d{2}(\.\w+)?$") def _open(self): """ 重写文件打开方法，创建文件后设置权限和所有权 """ # 调用父类方法打开文件 stream = super(CustomTimedRotatingFileHandler, self)._open() # 设置文件权限和所有权 if os.path.exists(self.baseFilename): try: # 设置权限为 664 (rw-rw-r--) os.chmod(self.baseFilename, 0o664) # 设置所有者为 appadmin os.chown(self.baseFilename, self.owner_id, self.group_id) except OSError as e: logging.error("Failed to set log file properties: %s", str(e)) return stream def doRollover(self): """ 重写日志轮转方法，设置新文件和备份文件的权限和所有权 """ # 调用父类轮转方法 super(CustomTimedRotatingFileHandler, self).doRollover() # 确保属性存在 if not hasattr(self, 'owner_id'): self.owner_id = os.getuid() if not hasattr(self, 'group_id'): self.group_id = os.getgid() # 设置新日志文件权限和所有权 if os.path.exists(self.baseFilename): try: os.chmod(self.baseFilename, 0o664) os.chown(self.baseFilename, self.owner_id, self.group_id) except OSError as e: logging.error("Failed to set new log file properties: %s", str(e)) # 设置备份日志文件权限和所有权 for i in range(1, self.backupCount + 1): sfn = self.rotation_filename(self.baseFilename + "." + time.strftime(self.suffix, time.localtime())) if os.path.exists(sfn): try: os.chmod(sfn, 0o664) os.chown(sfn, self.owner_id, self.group_id) except OSError as e: logging.error("Failed to set backup log properties: %s", str(e)) class CustomRotatingFileHandler(logging.handlers.RotatingFileHandler): """ 修复后的自定义大小轮转日志处理器： 1. 修复了 owner_id/group_id 属性初始化问题 2. 按文件大小轮转日志文件 3. 设置文件权限为 664 4. 修改文件所有者为 appadmin 用户和组 """ def __init__(self, filename, mode='a', maxBytes=0, backupCount=0, encoding=None, delay=False, owner='appadmin', group='appadmin'): # 先初始化所有者和组信息 self.owner = owner self.group = group # 获取用户和组的ID try: self.owner_id = pwd.getpwnam(owner).pw_uid self.group_id = grp.getgrnam(group).gr_gid except KeyError as e: # 如果找不到用户/组，使用当前用户/组 self.owner_id = os.getuid() self.group_id = os.getgid() logging.error("User or group not found: %s. Using current UID/GID: %s/%s", str(e), self.owner_id, self.group_id) except Exception as e: # 其他异常处理 self.owner_id = os.getuid() self.group_id = os.getgid() logging.error("Error setting owner/group: %s. Using current UID/GID: %s/%s", str(e), self.owner_id, self.group_id) # 确保属性存在 if not hasattr(self, 'owner_id'): self.owner_id = os.getuid() if not hasattr(self, 'group_id'): self.group_id = os.getgid() # 调用父类构造函数 super(CustomRotatingFileHandler, self).__init__( filename, mode, maxBytes, backupCount, encoding, delay) # 设置时间戳后缀格式 self.suffix = "%Y-%m-%d_%H-%M-%S" self.extMatch = re.compile(r"^\d{4}-\d{2}-\d{2}_\d{2}-\d{2}-\d{2}$") def _open(self): """ 重写文件打开方法，创建文件后设置权限和所有权 """ # 调用父类方法打开文件 stream = super(CustomRotatingFileHandler, self)._open() # 设置文件权限和所有权 if os.path.exists(self.baseFilename): try: # 设置权限为 664 (rw-rw-r--) os.chmod(self.baseFilename, 0o664) # 设置所有者为 appadmin os.chown(self.baseFilename, self.owner_id, self.group_id) except OSError as e: logging.error("Failed to set log file properties: %s", str(e)) return stream def doRollover(self): """ 重写日志轮转方法，设置新文件和备份文件的权限和所有权 使用时间戳格式命名备份文件 """ # 确保属性存在 if not hasattr(self, 'owner_id'): self.owner_id = os.getuid() if not hasattr(self, 'group_id'): self.group_id = os.getgid() # 关闭当前文件 if self.stream: self.stream.close() self.stream = None # 生成带时间戳的新文件名 t = time.localtime() timestamp = time.strftime(self.suffix, t) new_name = f"{self.baseFilename}.{timestamp}" # 重命名当前日志文件 if os.path.exists(self.baseFilename): os.rename(self.baseFilename, new_name) # 删除多余的备份文件 if self.backupCount > 0: # 获取所有备份文件并按时间排序 backup_files = [] for f in os.listdir(os.path.dirname(self.baseFilename)): if f.startswith(os.path.basename(self.baseFilename) + "."): backup_files.append(os.path.join(os.path.dirname(self.baseFilename), f)) # 按修改时间排序（旧文件在前） backup_files.sort(key=lambda x: os.path.getmtime(x)) # 删除超出备份数量的旧文件 while len(backup_files) >= self.backupCount: file_to_remove = backup_files.pop(0) try: os.remove(file_to_remove) except OSError: pass # 创建新日志文件 self.stream = self._open() # 设置新日志文件权限和所有权 if os.path.exists(self.baseFilename): try: os.chmod(self.baseFilename, 0o664) os.chown(self.baseFilename, self.owner_id, self.group_id) except OSError as e: logging.error("Failed to set new log file properties: %s", str(e)) # 设置备份文件权限和所有权 if os.path.exists(new_name): try: os.chmod(new_name, 0o664) os.chown(new_name, self.owner_id, self.group_id) except OSError as e: logging.error("Failed to set backup log properties: %s", str(e)) # 示例使用 if __name__ == "__main__": # 测试时间轮转处理器 timed_handler = CustomTimedRotatingFileHandler( filename='/var/log/myapp/timed.log', when='MIDNIGHT', backupCount=7, owner='appadmin', group='appadmin' ) # 测试大小轮转处理器 size_handler = CustomRotatingFileHandler( filename='/var/log/myapp/size.log', maxBytes=10 * 1024 * 1024, backupCount=5, owner='appadmin', group='appadmin' ) # 配置日志 logger = logging.getLogger() logger.setLevel(logging.INFO) # 添加两个处理器 formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s') timed_handler.setFormatter(formatter) size_handler.setFormatter(formatter) logger.addHandler(timed_handler) logger.addHandler(size_handler) # 测试日志 logger.info("Starting application with fixed log handlers") for i in range(100): logger.info(f"Test log message #{i}") logger.info("Log handlers initialized successfully without AttributeError") ``` ### 关键修复点 1. **初始化顺序调整**： ```python # 先初始化所有者和组信息 self.owner = owner self.group = group # 获取用户和组的ID try: self.owner_id = pwd.getpwnam(owner).pw_uid self.group_id = grp.getgrnam(group).gr_gid except ...: # 异常处理 # 确保属性存在 if not hasattr(self, 'owner_id'): self.owner_id = os.getuid() if not hasattr(self, 'group_id'): self.group_id = os.getgid() # 最后调用父类构造函数 super(...).__init__(...) ``` - 确保在父类初始化前设置关键属性 - 防止父类方法访问未初始化的属性 2. **健壮的异常处理**： ```python except KeyError as e: # 如果找不到用户/组，使用当前用户/组 self.owner_id = os.getuid() self.group_id = os.getgid() logging.error("User or group not found: %s. Using current UID/GID: %s/%s", str(e), self.owner_id, self.group_id) except Exception as e: # 其他异常处理 self.owner_id = os.getuid() self.group_id = os.getgid() logging.error("Error setting owner/group: %s. Using current UID/GID: %s/%s", str(e), self.owner_id, self.group_id) ``` - 捕获所有可能的异常 - 提供有意义的错误日志 - 使用当前用户/组作为后备方案 3. **属性存在性检查**： ```python # 在关键方法中添加检查 if not hasattr(self, 'owner_id'): self.owner_id = os.getuid() if not hasattr(self, 'group_id'): self.group_id = os.getgid() ``` - 在 `doRollover` 和 `_open` 方法中添加检查 - 确保即使初始化失败，后续操作也能进行 4. **改进的备份文件名生成**： ```python # 使用父类方法生成轮转文件名 sfn = self.rotation_filename(self.baseFilename + "." + time.strftime(self.suffix, time.localtime())) ``` - 更可靠的文件名生成方式 - 与父类的实现保持一致 ### 错误预防策略 1. **防御性编程**： - 在所有可能使用 `owner_id` 和 `group_id` 的方法前检查属性存在性 - 为关键操作提供后备方案 2. **全面的日志记录**： - 记录所有异常详细信息 - 记录使用的后备 UID/GID 3. **初始化顺序优化**： - 在调用父类构造函数前完成所有必要设置 - 避免父类方法访问未初始化的属性 4. **单元测试**： ```python # 测试用例：模拟用户/组不存在的情况 def test_missing_user_group(self): handler = CustomTimedRotatingFileHandler( 'test.log', owner='nonexistent_user', group='nonexistent_group') self.assertTrue(hasattr(handler, 'owner_id')) self.assertTrue(hasattr(handler, 'group_id')) self.assertEqual(handler.owner_id, os.getuid()) self.assertEqual(handler.group_id, os.getgid()) ``` - 测试各种错误场景 - 验证后备机制是否有效