Skip to content

engine.plugins

from engine.plugins import *

engine.plugins

PLUGIN_DIR = 'plugins' module-attribute

PluginManager

Source code in engine/plugins.py 
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
class PluginManager:
    def __init__(self, plugin_dir: str = PLUGIN_DIR) -> None:
        self.plugin_dir = plugin_dir
        self._before_hooks: list[Callable] = []
        self._after_hooks: list[Callable] = []
        self._before_validate_hooks: list[Callable] = []
        self._after_failure_hooks: list[Callable] = []
        self._after_retry_hooks: list[Callable] = []
        self._before_archive_hooks: list[Callable] = []
        self._plugins: dict[str, Any] = {}
        self.load_all()

    def load_all(self) -> None:
        if not os.path.isdir(self.plugin_dir):
            os.makedirs(self.plugin_dir, exist_ok=True)
            init_file = os.path.join(self.plugin_dir, "__init__.py")
            if not os.path.exists(init_file):
                with open(init_file, "w") as f:
                    f.write("")
            return

        sys.path.insert(0, os.path.dirname(self.plugin_dir))

        for fname in sorted(os.listdir(self.plugin_dir)):
            if not fname.endswith(".py") or fname.startswith("_"):
                continue
            mod_name = fname[:-3]
            try:
                mod = importlib.import_module(f"{os.path.basename(self.plugin_dir)}.{mod_name}")
                self._register_module(mod, mod_name)
            except Exception as e:
                AppLogger.warn(f"Plugin load failed: {mod_name} -> {e}")

        sys.path.pop(0)

    def _register_module(self, mod: Any, name: str) -> None:
        hooks = 0
        for attr_name in dir(mod):
            attr = getattr(mod, attr_name)
            if not callable(attr) or attr_name.startswith("_"):
                continue

            if hasattr(attr, "_before_send"):
                self._before_hooks.append(attr)
                hooks += 1
            if hasattr(attr, "_after_send"):
                self._after_hooks.append(attr)
                hooks += 1
            if hasattr(attr, "_before_validate"):
                self._before_validate_hooks.append(attr)
                hooks += 1
            if hasattr(attr, "_after_failure"):
                self._after_failure_hooks.append(attr)
                hooks += 1
            if hasattr(attr, "_after_retry"):
                self._after_retry_hooks.append(attr)
                hooks += 1
            if hasattr(attr, "_before_archive"):
                self._before_archive_hooks.append(attr)
                hooks += 1

        if hooks:
            self._plugins[name] = mod
            AppLogger.info(f"Plugin loaded: {name} ({hooks} hook(s))")

    def run_before_send(self, context: dict[str, Any]) -> dict[str, Any]:
        for hook in self._before_hooks:
            try:
                result = hook(context)
                if isinstance(result, dict):
                    context.update(result)
            except Exception as e:
                AppLogger.warn(f"before_send plugin error: {e}")
        return context

    def run_after_send(self, context: dict[str, Any], result: dict[str, Any]) -> None:
        for hook in self._after_hooks:
            try:
                hook(context, result)
            except Exception as e:
                AppLogger.warn(f"after_send plugin error: {e}")

    def run_before_validate(self, context: dict[str, Any]) -> dict[str, Any]:
        for hook in self._before_validate_hooks:
            try:
                result = hook(context)
                if isinstance(result, dict):
                    context.update(result)
            except Exception as e:
                AppLogger.warn(f"before_validate plugin error: {e}")
        return context

    def run_after_failure(self, context: dict[str, Any]) -> None:
        for hook in self._after_failure_hooks:
            try:
                hook(context)
            except Exception as e:
                AppLogger.warn(f"after_failure plugin error: {e}")

    def run_after_retry(self, context: dict[str, Any], attempt: int) -> None:
        for hook in self._after_retry_hooks:
            try:
                hook(context, attempt)
            except Exception as e:
                AppLogger.warn(f"after_retry plugin error: {e}")

    def run_before_archive(self, context: dict[str, Any]) -> dict[str, Any]:
        for hook in self._before_archive_hooks:
            try:
                result = hook(context)
                if isinstance(result, dict):
                    context.update(result)
            except Exception as e:
                AppLogger.warn(f"before_archive plugin error: {e}")
        return context

    @property
    def plugin_count(self) -> int:
        return len(self._plugins)

plugin_count property

plugin_dir = plugin_dir instance-attribute

__init__(plugin_dir=PLUGIN_DIR)

Source code in engine/plugins.py 
13
14
15
16
17
18
19
20
21
22
def __init__(self, plugin_dir: str = PLUGIN_DIR) -> None:
    self.plugin_dir = plugin_dir
    self._before_hooks: list[Callable] = []
    self._after_hooks: list[Callable] = []
    self._before_validate_hooks: list[Callable] = []
    self._after_failure_hooks: list[Callable] = []
    self._after_retry_hooks: list[Callable] = []
    self._before_archive_hooks: list[Callable] = []
    self._plugins: dict[str, Any] = {}
    self.load_all()

load_all()

Source code in engine/plugins.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
def load_all(self) -> None:
    if not os.path.isdir(self.plugin_dir):
        os.makedirs(self.plugin_dir, exist_ok=True)
        init_file = os.path.join(self.plugin_dir, "__init__.py")
        if not os.path.exists(init_file):
            with open(init_file, "w") as f:
                f.write("")
        return

    sys.path.insert(0, os.path.dirname(self.plugin_dir))

    for fname in sorted(os.listdir(self.plugin_dir)):
        if not fname.endswith(".py") or fname.startswith("_"):
            continue
        mod_name = fname[:-3]
        try:
            mod = importlib.import_module(f"{os.path.basename(self.plugin_dir)}.{mod_name}")
            self._register_module(mod, mod_name)
        except Exception as e:
            AppLogger.warn(f"Plugin load failed: {mod_name} -> {e}")

    sys.path.pop(0)

run_after_failure(context)

Source code in engine/plugins.py 
104
105
106
107
108
109
def run_after_failure(self, context: dict[str, Any]) -> None:
    for hook in self._after_failure_hooks:
        try:
            hook(context)
        except Exception as e:
            AppLogger.warn(f"after_failure plugin error: {e}")

run_after_retry(context, attempt)

Source code in engine/plugins.py 
111
112
113
114
115
116
def run_after_retry(self, context: dict[str, Any], attempt: int) -> None:
    for hook in self._after_retry_hooks:
        try:
            hook(context, attempt)
        except Exception as e:
            AppLogger.warn(f"after_retry plugin error: {e}")

run_after_send(context, result)

Source code in engine/plugins.py 
87
88
89
90
91
92
def run_after_send(self, context: dict[str, Any], result: dict[str, Any]) -> None:
    for hook in self._after_hooks:
        try:
            hook(context, result)
        except Exception as e:
            AppLogger.warn(f"after_send plugin error: {e}")

run_before_archive(context)

Source code in engine/plugins.py 
118
119
120
121
122
123
124
125
126
def run_before_archive(self, context: dict[str, Any]) -> dict[str, Any]:
    for hook in self._before_archive_hooks:
        try:
            result = hook(context)
            if isinstance(result, dict):
                context.update(result)
        except Exception as e:
            AppLogger.warn(f"before_archive plugin error: {e}")
    return context

run_before_send(context)

Source code in engine/plugins.py 
77
78
79
80
81
82
83
84
85
def run_before_send(self, context: dict[str, Any]) -> dict[str, Any]:
    for hook in self._before_hooks:
        try:
            result = hook(context)
            if isinstance(result, dict):
                context.update(result)
        except Exception as e:
            AppLogger.warn(f"before_send plugin error: {e}")
    return context

run_before_validate(context)

Source code in engine/plugins.py 
 94
 95
 96
 97
 98
 99
100
101
102
def run_before_validate(self, context: dict[str, Any]) -> dict[str, Any]:
    for hook in self._before_validate_hooks:
        try:
            result = hook(context)
            if isinstance(result, dict):
                context.update(result)
        except Exception as e:
            AppLogger.warn(f"before_validate plugin error: {e}")
    return context

hook_after_failure(func)

Source code in engine/plugins.py 
148
149
150
def hook_after_failure(func: Callable) -> Callable:
    setattr(func, "_after_failure", True)
    return func

hook_after_retry(func)

Source code in engine/plugins.py 
153
154
155
def hook_after_retry(func: Callable) -> Callable:
    setattr(func, "_after_retry", True)
    return func

hook_after_send(func)

Source code in engine/plugins.py 
138
139
140
def hook_after_send(func: Callable) -> Callable:
    setattr(func, "_after_send", True)
    return func

hook_before_archive(func)

Source code in engine/plugins.py 
158
159
160
def hook_before_archive(func: Callable) -> Callable:
    setattr(func, "_before_archive", True)
    return func

hook_before_send(func)

Source code in engine/plugins.py 
133
134
135
def hook_before_send(func: Callable) -> Callable:
    setattr(func, "_before_send", True)
    return func

hook_before_validate(func)

Source code in engine/plugins.py 
143
144
145
def hook_before_validate(func: Callable) -> Callable:
    setattr(func, "_before_validate", True)
    return func