📚 课程目录第3章插件式日志系统

3. 插件式日志系统

任务说明

构建一个插件式日志系统，核心目标是通过插件机制动态注册日志处理器（Handler），支持文件、网络、数据库三种内置插件，并允许在运行时加载新插件而无需修改核心代码。系统需提供统一的日志接口，将日志消息分发给所有已注册的处理器。

系统架构

核心实现

系统由三个核心类组成：LogHandler（插件抽象基类）、PluginManager（插件注册与动态加载）、Logger（日志分发入口）。插件通过register()方法注册，或通过load_plugin()从外部Python文件动态加载。

# plugin_logger.py
import importlib.util
import inspect
import logging
import os
import socket
import sqlite3
from datetime import datetime
from pathlib import Path

class LogHandler:
    """插件抽象基类：所有日志处理器必须继承并实现 emit()"""
    plugin_id = "base"

    def emit(self, level: str, message: str) -> None:
        raise NotImplementedError

class FileHandler(LogHandler):
    plugin_id = "file"

    def __init__(self, filename="app.log"):
        self.filename = filename

    def emit(self, level: str, message: str) -> None:
        with open(self.filename, "a", encoding="utf-8") as f:
            f.write(f"[{datetime.now().isoformat()}] [{level}] {message}\n")

class NetworkHandler(LogHandler):
    plugin_id = "network"

    def __init__(self, host="127.0.0.1", port=9999):
        self.host = host
        self.port = port

    def emit(self, level: str, message: str) -> None:
        sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        data = f"[{datetime.now().isoformat()}] [{level}] {message}".encode()
        sock.sendto(data, (self.host, self.port))
        sock.close()

class DatabaseHandler(LogHandler):
    plugin_id = "database"

    def __init__(self, db_path="logs.db"):
        self.db_path = db_path
        self._init_db()

    def _init_db(self):
        conn = sqlite3.connect(self.db_path)
        conn.execute(
            "CREATE TABLE IF NOT EXISTS logs "
            "(id INTEGER PRIMARY KEY AUTOINCREMENT, "
            "timestamp TEXT, level TEXT, message TEXT)"
        )
        conn.commit()
        conn.close()

    def emit(self, level: str, message: str) -> None:
        conn = sqlite3.connect(self.db_path)
        conn.execute(
            "INSERT INTO logs (timestamp, level, message) VALUES (?, ?, ?)",
            (datetime.now().isoformat(), level, message),
        )
        conn.commit()
        conn.close()

class PluginManager:
    def __init__(self):
        self._handlers = {}

    def register(self, handler: LogHandler) -> None:
        """注册插件实例到注册表"""
        self._handlers[handler.plugin_id] = handler

    def load_plugin(self, file_path: str) -> None:
        """从外部 .py 文件动态加载插件类并注册"""
        file_path = Path(file_path).resolve()
        module_name = f"plugin_{file_path.stem}_{id(file_path)}"
        spec = importlib.util.spec_from_file_location(module_name, file_path)
        module = importlib.util.module_from_spec(spec)
        spec.loader.exec_module(module)

        # 查找模块中所有 LogHandler 子类（排除基类自身）
        for _, obj in inspect.getmembers(module, inspect.isclass):
            if issubclass(obj, LogHandler) and obj is not LogHandler:
                self.register(obj())
                print(f"已加载插件: {obj.plugin_id}")

    def get_handler(self, plugin_id: str) -> LogHandler:
        return self._handlers.get(plugin_id)

    def list_plugins(self) -> list:
        return list(self._handlers.keys())

class Logger:
    def __init__(self):
        self.manager = PluginManager()

    def log(self, level: str, message: str) -> None:
        """将日志消息分发给所有已注册的处理器"""
        for handler in self.manager._handlers.values():
            handler.emit(level, message)

    # 便捷方法
    def info(self, msg: str):
        self.log("INFO", msg)

    def error(self, msg: str):
        self.log("ERROR", msg)

外部插件示例

以下代码保存为 console_plugin.py，演示动态加载机制——无需修改核心代码即可新增控制台输出能力：

# console_plugin.py
from plugin_logger import LogHandler

class ConsoleHandler(LogHandler):
    plugin_id = "console"

    def emit(self, level: str, message: str) -> None:
        print(f"[{level}] {message}")

使用与验证

将上述代码保存为 plugin_logger.py，在相同目录下创建 console_plugin.py，然后运行以下测试脚本：

# test_plugin_logger.py
from plugin_logger import Logger, FileHandler, NetworkHandler, DatabaseHandler

# 1. 创建日志器并注册三个内置插件
logger = Logger()
logger.manager.register(FileHandler("app.log"))
logger.manager.register(NetworkHandler("127.0.0.1", 9999))
logger.manager.register(DatabaseHandler("logs.db"))

# 2. 动态加载外部插件
logger.manager.load_plugin("console_plugin.py")

# 3. 确认插件列表
print("已注册插件:", logger.manager.list_plugins())

# 4. 发送测试日志
logger.info("系统启动")
logger.error("数据库连接失败")

# 5. 验证文件输出
print("\n--- app.log 内容 ---")
with open("app.log", encoding="utf-8") as f:
    print(f.read())

# 6. 验证数据库输出
import sqlite3
conn = sqlite3.connect("logs.db")
rows = conn.execute("SELECT level, message FROM logs").fetchall()
print("--- logs.db 内容 ---")
for level, msg in rows:
    print(f"[{level}] {msg}")
conn.close()

验证方法：运行 python test_plugin_logger.py，观察以下判定点：

控制台输出 已注册插件: ['file', 'network', 'database', 'console']，证明动态加载成功；
app.log 文件中出现两条带时间戳的日志记录；
logs.db 的 logs 表中存在对应两条记录；
若在另一终端用 nc -u -l 9999 监听UDP端口，可收到网络日志报文（此步骤可选）。

判定标准：上述1～3项全部满足，且文件与数据库内容包含 INFO 系统启动 和 ERROR 数据库连接失败 两条消息，即判定系统功能正常。若需扩展新处理器，只需编写继承 LogHandler 的类并调用 load_plugin() 即可，核心代码无需改动。