1
2
3
4
5
6
7
8
9
10
11
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
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
|
import enum, gc, glob, importlib, io, inspect, os, sys, typing, uuid
from src import Config, EventManager, Exports, IRCBot, Logging, Timers, utils
class ModuleException(Exception):
pass
class ModuleWarning(Exception):
pass
class ModuleNotFoundException(ModuleException):
pass
class ModuleNameCollisionException(ModuleException):
pass
class ModuleLoadException(ModuleException):
pass
class ModuleUnloadException(ModuleException):
pass
class ModuleNotLoadedWarning(ModuleWarning):
pass
class ModuleType(enum.Enum):
FILE = 0
DIRECTORY = 1
class BaseModule(object):
def __init__(self,
bot: "IRCBot.Bot",
events: EventManager.EventHook,
exports: Exports.Exports,
timers: Timers.Timers,
log: Logging.Log):
self.bot = bot
self.events = events
self.exports = exports
self.timers = timers
self.log = log
self.on_load()
def on_load(self):
pass
def unload(self):
pass
class LoadedModule(object):
def __init__(self,
name: str,
module: BaseModule,
context: str,
import_name: str):
self.name = name
self.module = module
self.context = context
self.import_name = import_name
class ModuleManager(object):
def __init__(self,
events: EventManager.EventHook,
exports: Exports.Exports,
timers: Timers.Timers,
config: Config.Config,
log: Logging.Log,
directory: str):
self.events = events
self.exports = exports
self.config = config
self.timers = timers
self.log = log
self.directory = directory
self.modules = {} # type: typing.Dict[str, LoadedModule]
self.waiting_requirement = {} # type: typing.Dict[str, typing.Set[str]]
def list_modules(self) -> typing.List[typing.Tuple[ModuleType, str]]:
modules = []
for file_module in glob.glob(os.path.join(self.directory, "*.py")):
modules.append((ModuleType.FILE, file_module))
for directory_module in glob.glob(os.path.join(
self.directory, "*", "__init__.py")):
directory = os.path.dirname(directory_module)
modules.append((ModuleType.DIRECTORY, directory))
return sorted(modules, key=lambda module: module[1])
def _module_name(self, path: str) -> str:
return os.path.basename(path).rsplit(".py", 1)[0].lower()
def _module_path(self, name: str) -> str:
return os.path.join(self.directory, name)
def _import_name(self, name: str) -> str:
return "bitbot_%s" % name
def _get_magic(self, obj: typing.Any, magic: str, default: typing.Any
) -> typing.Any:
return getattr(obj, magic) if hasattr(obj, magic) else default
def _load_module(self, bot: "IRCBot.Bot", name: str) -> LoadedModule:
path = self._module_path(name)
if os.path.isdir(path) and os.path.isfile(os.path.join(
path, "__init__.py")):
path = os.path.join(path, "__init__.py")
else:
path = "%s.py" % path
for hashflag, value in utils.parse.hashflags(path):
if hashflag == "ignore":
# nope, ignore this module.
raise ModuleNotLoadedWarning("module ignored")
elif hashflag == "require-config" and value:
if not self.config.get(value.lower(), None):
# nope, required config option not present.
raise ModuleNotLoadedWarning("required config not present")
elif hashflag == "require-module" and value:
requirement = value.lower()
if not requirement in self.modules:
if not requirement in self.waiting_requirement:
self.waiting_requirement[requirement] = set([])
self.waiting_requirement[requirement].add(path)
raise ModuleNotLoadedWarning("waiting for requirement")
import_name = self._import_name(name)
import_spec = importlib.util.spec_from_file_location(import_name, path)
module = importlib.util.module_from_spec(import_spec)
sys.modules[import_name] = module
loader = typing.cast(importlib.abc.Loader, import_spec.loader)
loader.exec_module(module)
module_object_pointer = getattr(module, "Module", None)
if not module_object_pointer:
raise ModuleLoadException("module '%s' doesn't have a "
"'Module' class." % name)
if not inspect.isclass(module_object_pointer):
raise ModuleLoadException("module '%s' has a 'Module' attribute "
"but it is not a class." % name)
context = str(uuid.uuid4())
context_events = self.events.new_context(context)
context_exports = self.exports.new_context(context)
context_timers = self.timers.new_context(context)
module_object = module_object_pointer(bot, context_events,
context_exports, context_timers, self.log)
if not hasattr(module_object, "_name"):
module_object._name = name.title()
for attribute_name in dir(module_object):
attribute = getattr(module_object, attribute_name)
for hook in self._get_magic(attribute,
utils.consts.BITBOT_HOOKS_MAGIC, []):
context_events.on(hook["event"]).hook(attribute,
**hook["kwargs"])
for export in self._get_magic(module_object,
utils.consts.BITBOT_EXPORTS_MAGIC, []):
context_exports.add(export["setting"], export["value"])
if name in self.modules:
raise ModuleNameCollisionException("Module name '%s' "
"attempted to be used twice")
return LoadedModule(name, module_object, context, import_name)
def load_module(self, bot: "IRCBot.Bot", name: str):
try:
loaded_module = self._load_module(bot, name)
except ModuleWarning as warning:
self.log.warn("Module '%s' not loaded", [name])
raise
except Exception as e:
self.log.error("Failed to load module \"%s\": %s",
[name, str(e)])
raise
self.modules[loaded_module.name] = loaded_module
if loaded_module.name in self.waiting_requirement:
for requirement_name in self.waiting_requirement[
loaded_module.name]:
self.load_module(bot, requirement_name)
self.log.debug("Module '%s' loaded", [loaded_module.name])
def load_modules(self, bot: "IRCBot.Bot", whitelist: typing.List[str]=[],
blacklist: typing.List[str]=[]):
for type, path in self.list_modules():
name = self._module_name(path)
if name in whitelist or (not whitelist and not name in blacklist):
try:
self.load_module(bot, name)
except ModuleWarning:
pass
def unload_module(self, name: str):
if not name in self.modules:
raise ModuleNotFoundException()
loaded_module = self.modules[name]
if hasattr(loaded_module.module, "unload"):
try:
loaded_module.module.unload()
except:
pass
del self.modules[loaded_module.name]
context = loaded_module.context
self.events.purge_context(context)
self.exports.purge_context(context)
self.timers.purge_context(context)
module = loaded_module.module
del loaded_module.module
del sys.modules[loaded_module.import_name]
references = sys.getrefcount(module)
referrers = gc.get_referrers(module)
del module
references -= 1 # 'del module' removes one reference
references -= 1 # one of the refs is from getrefcount
self.log.debug("Module '%s' unloaded (%d reference%s)",
[loaded_module.name, references,
"" if references == 1 else "s"])
if references > 0:
self.log.debug("References left for '%s': %s",
[loaded_module.name,
", ".join([str(referrer) for referrer in referrers])])
|
