<think>为了确保项目中的所有Python文件都能方便地使用我们之前创建的日志模块,我们可以采取以下措施:
1. 将日志模块设计为单例模式,确保整个项目使用同一个日志配置实例。
2. 在日志模块中提供全局的日志函数(如debug, info, warning, error, critical等),这样其他模块可以直接导入这些函数。
3. 在项目初始化时(例如在main.py中)初始化日志系统,这样在其他模块导入日志函数时,日志系统已经配置完成。
但是,我们需要注意的是,在Python中,模块是单例的,所以我们可以在日志模块中创建一个全局实例,然后在其他模块中导入该实例的日志方法。
我们已经在前面的日志模块中创建了一个单例的ProjectLogger类,并且提供了全局的简写函数(debug, info等)。现在,我们需要确保在项目启动时尽早初始化日志系统。
然而,我们可能会遇到一个问题:当其他模块在导入日志函数时,日志系统可能还没有被初始化(因为导入顺序的问题)。为了解决这个问题,我们可以使用惰性初始化或者确保在导入其他模块之前先初始化日志系统。
这里我们采用一种简单的方法:在日志模块中,我们创建一个全局的ProjectLogger实例,并在第一次使用时初始化(使用默认配置,或者允许在main中覆盖配置)。但是,为了灵活性,我们允许在main中显式初始化,同时在其他模块导入时,如果没有初始化,则使用默认配置初始化。
但是,为了避免循环导入和初始化顺序问题,我们可以这样设计:
方案:
1. 在logger.py中,我们创建一个全局的ProjectLogger实例,并提供一个初始化函数(如果需要自定义配置,则在main中调用初始化)。
2. 同时,我们提供默认的初始化(使用默认配置路径),这样即使没有显式初始化,在第一次使用日志时也会初始化。
但是,我们之前的ProjectLogger已经实现了单例模式,并且在__init__中接受配置路径。我们修改一下,使得它可以延迟初始化。
不过,我们之前的ProjectLogger在创建实例时就已经初始化了。我们可以修改为在第一次调用get_logger时再初始化。
但是,为了简单起见,我们可以在模块级别创建一个默认的日志记录器,这样其他模块导入时就可以直接使用,并且这个日志记录器已经使用默认配置初始化了。
然而,这样不够灵活,因为配置路径可能需要在main中指定。所以,我们提供两种方式:
- 在main中显式初始化,然后其他模块导入的日志函数就会使用这个配置。
- 如果其他模块先导入,那么它们会使用默认配置初始化。
为了避免配置被意外初始化,我们可以要求必须在main中显式初始化,否则抛出警告。但这样会增加使用的复杂性。
折中方案:在日志模块中,我们使用一个全局变量来标记是否已经初始化。在其他模块导入日志函数时,如果没有初始化,则使用默认配置初始化,并打印一条警告信息(建议在main中显式初始化)。这样,在开发过程中,我们可以通过警告知道需要显式初始化,但在测试时也可以自动初始化。
但是,我们也可以不强制在main中初始化,而是在日志模块中默认初始化,并允许在main中重新配置(通过重新初始化ProjectLogger)。但是重新初始化可能会遇到问题,因为单例模式只允许初始化一次。
因此,我们调整设计:允许在main中初始化,如果已经初始化则忽略后续初始化。在日志模块中,我们提供一个初始化函数,可以在main中调用,也可以不调用(使用默认配置)。
具体修改如下:
在logger.py中:
- 修改ProjectLogger,使其可以在不提供配置路径的情况下使用默认配置路径,并且可以多次调用初始化(但只有第一次有效)。
- 提供initialize_logger函数,用于显式初始化。
但是,我们之前的ProjectLogger在__init__中已经初始化了,而且我们使用单例模式。所以,我们可以在ProjectLogger中增加一个initialize方法,用于重新配置(如果需要的话,但一般不建议重新配置)。
考虑到日志系统通常在程序开始时就配置好,并且之后不会改变,我们可以这样设计:
在logger.py模块被导入时,自动创建一个默认配置的日志记录器。这样,其他模块导入日志函数时,日志系统已经可用。
同时,我们提供一个initialize函数,允许在main中覆盖默认配置(如果还没有初始化的话)。
但是,如果模块导入顺序导致日志系统在main中初始化之前就已经初始化了,那么main中的初始化将不会生效。为了避免这种情况,我们可以在main中首先导入并初始化日志系统,然后再导入其他模块。
然而,在大型项目中,模块导入顺序可能很复杂。因此,我们设计为:在main中,我们首先调用日志模块的初始化函数,然后其他模块在导入日志函数时,使用的是已经初始化好的日志系统。
具体步骤:
1. 在日志模块中,我们设置一个全局变量`_initialized`,标记是否已经初始化。
2. 提供一个`initialize`函数,用于初始化日志系统(使用指定的配置文件),这个函数只能在main中调用,并且必须在导入其他模块之前调用。
3. 在日志模块中,我们提供全局的日志函数(debug, info等),这些函数在内部使用一个全局的日志记录器实例。
4. 如果用户没有显式初始化,则在第一次使用日志函数时,使用默认配置初始化。
但是,为了避免在导入日志函数时自动初始化(因为导入时可能还没有配置),我们可以让日志函数在调用时检查是否初始化,如果没有,则初始化。
但是,这样会导致第一次日志调用时才初始化,而在此之前(比如在模块导入时执行的代码)就无法记录日志。
因此,我们决定:在日志模块被导入时,我们不立即初始化,而是提供一个显式初始化的函数,要求用户在main中首先初始化。同时,在日志函数中,如果发现没有初始化,则使用默认配置初始化并记录一个警告。
这样,我们就可以在项目中的任何地方导入日志函数,并在任何地方使用(包括模块级别代码,只要确保在模块代码执行前已经初始化了日志系统,或者接受默认初始化)。
我们修改logger.py如下:
注意:由于时间关系,我们只修改关键部分,提供一个可行的方案。
但是,我们之前的ProjectLogger已经是单例,并且会在创建实例时初始化。我们可以修改ProjectLogger,使其在第一次实例化时初始化,并且允许通过initialize方法重新配置(如果尚未初始化)。
我们调整ProjectLogger类,添加一个initialize方法,用于显式初始化配置。如果已经初始化,则忽略后续初始化。
同时,我们修改__init__方法,使其不立即初始化,而是等到第一次调用get_logger时再初始化(如果还没有初始化的话)。
但是,这样修改较大。为了快速实现,我们采用以下方案:
在logger.py中,我们创建一个全局的日志记录器实例,但初始化为None。然后提供一个initialize函数,用于创建实例(如果已经存在,则跳过)。同时,在日志函数内部,如果发现记录器为None,则自动初始化(使用默认配置)。
具体实现:
我们将之前提供的全局函数(debug, info等)修改为:它们会调用一个内部的_get_logger函数,这个函数会检查全局记录器是否存在,如果不存在则初始化。
同时,我们提供一个initialize_logger函数,允许在main中指定配置路径进行初始化。
代码如下(只显示修改部分):
</think>
# 全局集成日志系统的实现方案
下面我将详细介绍如何让项目中所有 Python 文件自动使用统一的日志系统,无需在每个文件中重复配置。我们将通过 Python 的导入钩子机制和模块重载技术实现这一目标。
## 核心实现方案
### 文件路径: `src/utils/logger_integration.py`
```python
import sys
import importlib
import logging
import inspect
from types import ModuleType
from src.utils.logger import ProjectLogger, debug, info, warning, error, critical, exception
class LoggerInjector(importlib.abc.MetaPathFinder, importlib.abc.Loader):
"""自动注入日志功能的导入钩子"""
def __init__(self):
self.original_finders = sys.meta_path.copy()
sys.meta_path.insert(0, self)
self.logger = ProjectLogger().get_logger()
self.logger.info("Logger injector initialized")
def find_spec(self, fullname, path, target=None):
# 只处理项目内的模块
if fullname.startswith("yolov10_mss_project."):
return importlib.util.spec_from_loader(fullname, self)
return None
def create_module(self, spec):
# 使用默认模块创建机制
return None
def exec_module(self, module):
# 执行原始模块加载
super().exec_module(module)
# 跳过某些不需要注入的模块
skip_modules = [
'src.utils.logger',
'src.utils.logger_integration',
'src.core.config_loader'
]
if module.__name__ in skip_modules:
return
# 注入日志功能
self._inject_logger(module)
def _inject_logger(self, module):
"""向模块注入日志功能"""
module_name = module.__name__
# 设置模块级日志记录器
module_logger = logging.getLogger(module_name)
setattr(module, 'logger', module_logger)
# 注入快捷日志函数
setattr(module, 'debug', debug)
setattr(module, 'info', info)
setattr(module, 'warning', warning)
setattr(module, 'error', error)
setattr(module, 'critical', critical)
setattr(module, 'exception', exception)
# 特殊处理:在类中自动添加日志记录器
for name, obj in module.__dict__.items():
if inspect.isclass(obj):
self._inject_logger_to_class(obj, module_name)
# 记录注入信息
info(f"Logger injected into module: {module_name}")
def _inject_logger_to_class(self, cls, module_name):
"""向类中注入日志记录器"""
class_logger = logging.getLogger(f"{module_name}.{cls.__name__}")
# 添加类级日志记录器
if not hasattr(cls, 'logger'):
setattr(cls, 'logger', class_logger)
# 向所有方法添加日志装饰器
for name, method in cls.__dict__.items():
if callable(method) and not name.startswith('__'):
setattr(cls, name, self._log_method(method, class_logger))
def _log_method(self, method, logger):
"""创建带日志的方法装饰器"""
def wrapped(*args, **kwargs):
# 记录方法调用
logger.debug(f"Entering {method.__name__}")
try:
result = method(*args, **kwargs)
logger.debug(f"Exiting {method.__name__}")
return result
except Exception as e:
logger.error(
f"Error in {method.__name__}: {str(e)}",
exc_info=True
)
raise
return wrapped
def configure_logging():
"""配置基础日志系统"""
# 创建根日志记录器
root_logger = logging.getLogger()
root_logger.setLevel(logging.DEBUG)
# 创建控制台处理器
console_handler = logging.StreamHandler()
console_handler.setLevel(logging.INFO)
# 创建格式化器
formatter = logging.Formatter(
'%(asctime)s | %(name)-25s | %(levelname)-8s | %(message)s',
datefmt='%Y-%m-%d %H:%M:%S'
)
console_handler.setFormatter(formatter)
# 添加处理器
root_logger.addHandler(console_handler)
return root_logger
def initialize_logging_system():
"""初始化全局日志系统"""
# 配置基础日志
configure_logging()
# 初始化项目日志系统
ProjectLogger()
# 初始化日志注入器
injector = LoggerInjector()
# 重新加载已导入的模块以注入日志
reload_imported_modules()
return injector
def reload_imported_modules():
"""重新加载所有已导入的模块以注入日志"""
loaded_modules = list(sys.modules.keys())
project_modules = [m for m in loaded_modules if m.startswith('yolov10_mss_project.')]
info("Reloading modules for logger injection")
for module_name in project_modules:
# 跳过不需要重新加载的模块
if 'logger' in module_name or 'config_loader' in module_name:
continue
try:
info(f"Reloading module: {module_name}")
module = sys.modules[module_name]
importlib.reload(module)
except Exception as e:
error(f"Failed to reload module {module_name}: {str(e)}")
# 自动初始化日志系统
if not hasattr(sys, 'logger_system_initialized'):
initialize_logging_system()
sys.logger_system_initialized = True
```
## 项目启动集成
### 修改 `src/main.py`
```python
#!/usr/bin/env python3
# src/main.py
# 确保这是第一个导入
from src.utils.logger_integration import initialize_logging_system
def main():
# 初始化日志系统(自动完成)
# 注意:实际初始化在导入时已完成
# 获取日志记录器
from src.utils.logger import info
info("Application starting")
# 导入其他模块(此时会自动注入日志)
from src.core.capture import ScreenCapture
from src.core.detection import ObjectDetector
from src.core.action import ActionExecutor
info("Initializing components")
capture = ScreenCapture()
detector = ObjectDetector()
executor = ActionExecutor()
# 主循环
info("Entering main loop")
try:
while True:
frame = capture.get_frame()
detections = detector.detect(frame)
executor.execute_actions(detections)
except KeyboardInterrupt:
info("Application terminated by user")
except Exception as e:
error(f"Critical error: {str(e)}", exc_info=True)
raise
if __name__ == "__main__":
main()
```
## 使用示例
### 在任何模块中直接使用日志(无需导入)
```python
# src/core/capture.py
class ScreenCapture:
def __init__(self):
# 直接使用日志记录器(自动注入)
self.logger.info("Initializing screen capture")
self.setup_capture()
def setup_capture(self):
try:
# 直接使用快捷日志函数
info("Setting up screen capture device")
# ... 初始化代码 ...
debug(f"Capture resolution: {self.resolution}")
except Exception as e:
error(f"Capture setup failed: {str(e)}")
raise
def get_frame(self):
# 自动添加的方法级日志
# (在LoggerInjector中自动添加)
# ... 获取帧的代码 ...
return frame
```
### 在函数中使用日志
```python
# src/core/detection.py
def load_model(model_path):
"""加载YOLO模型"""
# 直接使用全局日志函数
info(f"Loading model from {model_path}")
try:
# ... 模型加载代码 ...
debug(f"Model input size: {model.input_size}")
return model
except FileNotFoundError:
error(f"Model file not found: {model_path}")
raise
except Exception as e:
exception("Unexpected error loading model")
raise
```
## 技术实现详解
### 1. 导入钩子机制
我们通过自定义 `MetaPathFinder` 和 `Loader` 实现导入钩子:
```python
class LoggerInjector(importlib.abc.MetaPathFinder, importlib.abc.Loader):
def find_spec(self, fullname, path, target=None):
# 只处理项目内模块
if fullname.startswith("yolov10_mss_project."):
return importlib.util.spec_from_loader(fullname, self)
def exec_module(self, module):
# 原始模块加载
super().exec_module(module)
# 注入日志功能
self._inject_logger(module)
```
### 2. 自动注入内容
注入器会为每个模块添加以下内容:
1. 模块级 `logger` 对象
2. 全局日志函数 (`debug`, `info`, `warning`, `error`, `critical`, `exception`)
3. 类级别的 `logger` 对象
4. 所有方法的自动日志装饰器
### 3. 方法级自动日志
通过装饰器自动添加方法入口/出口日志:
```python
def _log_method(self, method, logger):
def wrapped(*args, **kwargs):
logger.debug(f"Entering {method.__name__}")
try:
result = method(*args, **kwargs)
logger.debug(f"Exiting {method.__name__}")
return result
except Exception as e:
logger.error(f"Error in {method.__name__}: {str(e)}", exc_info=True)
raise
return wrapped
```
### 4. 已加载模块处理
对于项目启动前已加载的模块,我们重新加载它们以注入日志:
```python
def reload_imported_modules():
loaded_modules = list(sys.modules.keys())
project_modules = [m for m in loaded_modules if m.startswith('yolov10_mss_project.')]
for module_name in project_modules:
importlib.reload(sys.modules[module_name])
```
## 高级配置选项
### 自定义注入规则
在 `configs/default.yaml` 中添加日志注入配置:
```yaml
logging:
injection:
enabled: true
# 包含的模块(支持通配符)
include:
- "yolov10_mss_project.src.core.*"
- "yolov10_mss_project.src.utils.*"
# 排除的模块
exclude:
- "yolov10_mss_project.src.core.config_loader"
# 自动日志级别配置
auto_logging:
methods: true # 为方法添加自动日志
level: "DEBUG" # 自动日志级别
# 排除的方法
exclude_methods:
- "__init__"
- "__getattr__"
- "__setattr__"
```
### 修改注入器以支持配置
```python
# 在 LoggerInjector 中添加
def __init__(self, config_path="configs/default.yaml"):
self.config = ConfigLoader(config_path).get('logging.injection', {})
# ...其他初始化...
def should_inject(self, module_name):
"""检查是否应该注入日志"""
if not self.config.get('enabled', True):
return False
# 检查包含规则
included = any(fnmatch.fnmatch(module_name, pattern)
for pattern in self.config.get('include', []))
# 检查排除规则
excluded = any(fnmatch.fnmatch(module_name, pattern)
for pattern in self.config.get('exclude', []))
return included and not excluded
# 在 exec_module 中
def exec_module(self, module):
if not self.should_inject(module.__name__):
super().exec_module(module)
return
# ...注入逻辑...
```
## 性能优化措施
### 1. 延迟注入
对于大型项目,使用按需注入代替全量注入:
```python
class LazyLoggerInjector:
def __init__(self):
self.injected_modules = set()
def __call__(self, frame, event, arg):
if event != 'call':
return
# 获取调用模块
module_name = frame.f_globals.get('__name__', '')
# 检查是否需要注入
if (module_name not in self.injected_modules and
self.should_inject(module_name)):
module = sys.modules.get(module_name)
if module:
self._inject_logger(module)
self.injected_modules.add(module_name)
# 在 initialize_logging_system 中
sys.setprofile(LazyLoggerInjector())
```
### 2. 选择性方法装饰
只对符合条件的方法添加自动日志:
```python
def should_wrap_method(self, method_name):
"""决定是否包装方法"""
auto_config = self.config.get('auto_logging', {})
if not auto_config.get('methods', True):
return False
# 检查排除列表
if method_name in auto_config.get('exclude_methods', []):
return False
# 只包装公共方法
if method_name.startswith('_'):
return False
return True
```
### 3. 编译优化
使用 `compile()` 预编译日志注入代码:
```python
# 预编译常用日志代码
LOG_TEMPLATE = """
def {name}(msg, *args, **kwargs):
logger.{level}(msg, *args, **kwargs)
"""
compiled_log = compile(LOG_TEMPLATE, '<string>', 'exec')
# 在注入时使用
def _inject_log_functions(module):
# 创建新的命名空间
namespace = {'logger': module.logger}
# 编译并执行
for level in ['debug', 'info', 'warning', 'error', 'critical']:
exec(compiled_log, {'name': level, 'level': level}, namespace)
# 添加到模块
for name in ['debug', 'info', 'warning', 'error', 'critical']:
setattr(module, name, namespace[name])
```
## 特殊场景处理
### 1. 多进程支持
```python
def handle_multiprocessing():
"""处理多进程中的日志"""
from multiprocessing import util
from src.utils.logger import ProjectLogger
# 覆盖多进程日志初始化
original_logger = util.get_logger
def patched_logger():
logger = original_logger()
# 添加项目日志处理器
project_logger = ProjectLogger().get_logger()
for handler in project_logger.handlers:
logger.addHandler(handler)
return logger
util.get_logger = patched_logger
```
### 2. 异步协程支持
```python
def async_log_wrapper(func):
"""包装异步函数以添加日志"""
async def wrapped(*args, **kwargs):
logger = logging.getLogger(f"{func.__module__}.{func.__name__}")
logger.debug(f"Entering async {func.__name__}")
try:
result = await func(*args, **kwargs)
logger.debug(f"Exiting async {func.__name__}")
return result
except Exception as e:
logger.error(f"Error in async {func.__name__}: {str(e)}", exc_info=True)
raise
return wrapped
# 在注入器中对异步方法特殊处理
if inspect.iscoroutinefunction(method):
setattr(cls, name, self._async_log_method(method, class_logger))
```
### 3. 第三方库日志集成
```python
def configure_third_party_logging():
"""配置第三方库的日志"""
# 设置常见库的日志级别
for lib in ['urllib3', 'requests', 'PIL', 'matplotlib']:
logging.getLogger(lib).setLevel(logging.WARNING)
# 将第三方库日志路由到项目日志系统
root_logger = logging.getLogger()
project_logger = ProjectLogger().get_logger()
for handler in project_logger.handlers:
root_logger.addHandler(handler)
```
## 验证日志系统
创建测试脚本 `tests/test_logger_injection.py`:
```python
import sys
import os
import logging
# 将项目根目录添加到路径
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
# 初始化日志系统
from src.utils.logger_integration import initialize_logging_system
initialize_logging_system()
# 导入测试模块
from src.core.capture import ScreenCapture
from src.core.detection import load_model
def test_module_level_logging():
# 直接使用全局日志函数
info("This is a module-level info message")
debug("This is a module-level debug message")
try:
# 触发错误
1 / 0
except Exception:
exception("Module-level exception")
def test_class_logging():
capture = ScreenCapture()
capture.get_frame() # 触发自动日志
def test_function_logging():
load_model("models/yolov10n.pt")
def test_third_party_logging():
# 测试第三方库日志集成
import requests
requests.get("https://invalid.url") # 应生成警告日志
if __name__ == "__main__":
test_module_level_logging()
test_class_logging()
test_function_logging()
test_third_party_logging()
info("All logger injection tests completed")
```
## 部署注意事项
1. **启动顺序**:
- 确保 `logger_integration.py` 是最先导入的模块
- 在 `main.py` 开头显式导入
2. **性能影响**:
- 在开发环境启用完整日志注入
- 在生产环境禁用方法级自动日志:
```yaml
logging:
injection:
auto_logging:
methods: false
```
3. **依赖管理**:
- 在 `requirements.txt` 中添加:
```
import