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
|
import gc, glob, imp, 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 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
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 = {}
self.waiting_requirement = {}
def list_modules(self) -> typing.List[str]:
return sorted(glob.glob(os.path.join(self.directory, "*.py")))
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, "%s.py" % 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):
path = self._module_path(name)
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")
module = imp.load_source(self._import_name(name), path)
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"])
module_object._context = context
module_object._import_name = name
if name in self.modules:
raise ModuleNameCollisionException("Module name '%s' "
"attempted to be used twice")
return module_object
def load_module(self, bot: "IRCBot.Bot", name: str):
try:
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[module._import_name] = module
if name in self.waiting_requirement:
for requirement_name in self.waiting_requirement:
self.load_module(bot, requirement_name)
self.log.debug("Module '%s' loaded", [name])
def load_modules(self, bot: "IRCBot.Bot", whitelist: typing.List[str]=[],
blacklist: typing.List[str]=[]):
for 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()
module = self.modules[name]
if hasattr(module, "unload"):
try:
module.unload()
except:
pass
del self.modules[name]
context = module._context
self.events.purge_context(context)
self.exports.purge_context(context)
self.timers.purge_context(context)
del sys.modules[self._import_name(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)",
[name, references, "" if references == 1 else "s"])
if references > 0:
self.log.debug("References left for '%s': %s",
[name, ", ".join([str(referrer) for referrer in referrers])])
|
