本文整理汇总了Python中system.event_manager.EventManager类的典型用法代码示例。如果您正苦于以下问题：Python EventManager类的具体用法？Python EventManager怎么用？Python EventManager使用的例子？那么恭喜您, 这里精选的类代码示例或许可以为您提供帮助。
在下文中一共展示了EventManager类的20个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: MemosPlugin
class MemosPlugin(plugin.PluginObject):
commands = None
events = None
storage = None
# data = None # SQLite for a change
def setup(self):
self.commands = CommandManager()
self.events = EventManager()
self.storage = StorageManager()
# self.data = self.storage.get_file(self, "data", SQLITE,
# "plugins/memos/memos.sqlite")
# with self.data as c:
# # Multiline strings because of an IDE bug
# c.execute("""CREATE TABLE IF NOT EXISTS memos
# (to TEXT, from TEXT, memo TEXT)""")
self.events.add_callback("PreMessageReceived", self,
self.message_received, 0)
self.commands.register_command("memo", self.memo_command, self,
"memo.send", default=True)
def save_memo(self, sender, recipient, memo):
recipient = recipient.lower()
# with self.data as c:
# c.execute("""INSERT INTO memos VALUES (?, ?, ?)""",
# (recipient, sender, memo))
def get_memos(self, recipient):
recipient = recipient.lower()
# with self.data as c:
# c.execute("""SELECT * FROM memos WHERE from=?""", (recipient,))
# d = c.fetchall()
# return d
def message_received(self, event=PreMessageReceived):
user = event.source
target = event.target if isinstance(event.target, Channel) else user
memos = self.get_memos(user.name)
if memos:
for memo in memos:
sender = memo[1]
text = memo[2]
target.respond("Memo for %s (from %s): %s" % (user.name,
sender, text))
def memo_command(self, protocol, caller, source, command, raw_args,
parsed_args):
args = raw_args.split() # Quick fix for new command handler signature
raise NotImplementedError("This isn't done yet.")
开发者ID:case，项目名称:Ultros-contrib，代码行数:53，代码来源:__init__.py
示例2: WebhooksPlugin
class WebhooksPlugin(plugin.PluginObject):
config = None
data = None
commands = None
events = None
plugins = None
storage = None
@property
def web(self):
"""
:rtype: plugins.web.WebPlugin
"""
return self.plugins.get_plugin("Web")
def setup(self):
self.logger.trace("Entered setup method.")
self.commands = CommandManager()
self.events = EventManager()
self.plugins = PluginManager()
self.storage = StorageManager()
try:
self.config = self.storage.get_file(self, "config", YAML, "plugins/webhooks.yml")
except Exception:
self.logger.exception("Error loading configuration!")
return self._disable_self()
else:
if not self.config.exists:
self.logger.error("Unable to find config/plugins/webhooks.yml")
return self._disable_self()
self._load()
self.config.add_callback(self._load)
self.events.add_callback("Web/ServerStartedEvent", self, self.add_routes, 0)
def _load(self):
pass
def add_routes(self, _=None):
self.web.add_navbar_entry("Webhooks", "/webhooks", "url")
self.web.add_handler("/webhooks", "plugins.webhooks.routes.webhooks.Route")
self.logger.info("Registered route: /webhooks")
pass # So the regions work in PyCharm
开发者ID:EionRobb，项目名称:Ultros-contrib，代码行数:49，代码来源:__init__.py
示例3: setup
def setup(self):
self.events = EventManager()
self.storage = StorageManager()
try:
self.config = self.storage.get_file(self, "config", YAML,
"plugins/blowfish.yml")
except Exception:
self.logger.exception("Error loading configuration!")
self._disable_self()
return
if not self.config.exists:
self.logger.error("Unable to find config/plugins/blowfish.yml")
self._disable_self()
return
self.events.add_callback(
"PreMessageReceived", self, self.pre_message, 10001
)
self.events.add_callback(
"MessageSent", self, self.message_sent, 10001
)
self.events.add_callback(
"ActionReceived", self, self.message_sent, 10001
)
self.events.add_callback(
"ActionSent", self, self.message_sent, 10001
)
开发者ID:EionRobb，项目名称:Ultros-contrib，代码行数:31，代码来源:__init__.py
示例4: setup
def setup(self):
self.logger.trace("Entered setup method.")
self.commands = CommandManager()
self.events = EventManager()
self.storage = StorageManager()
try:
self.config = self.storage.get_file(self, "config", YAML,
"plugins/inter.yml")
except Exception:
self.logger.exception("Error loading configuration!")
self.logger.error(_("Disabling.."))
self._disable_self()
return
if not self.config.exists:
self.logger.error("Unable to find config/plugins/inter.yml")
self.logger.error(_("Disabling.."))
self._disable_self()
return
self.config.add_callback(self.reload)
self.commands.register_command("players", self.players_command, self,
"inter.players", default=True)
self.events.add_callback("ReactorStarted", self, self.first_load, 0)
开发者ID:domainr，项目名称:Ultros-contrib，代码行数:28，代码来源:__init__.py
示例5: __init__
def __init__(self):
self.commands = CommandManager()
self.event_manager = EventManager()
self.logger = getLogger("Manager")
self.plugman = PluginManager(self)
self.yapsy_logger = getLogger("yapsy")
self.metrics = None
开发者ID:NotAFile，项目名称:Ultros，代码行数:8，代码来源:factory_manager.py
示例6: setup
def setup(self):
self.commands = CommandManager()
self.events = EventManager()
self.storage = StorageManager()
self.data = self.storage.get_file(
self,
"data",
DBAPI,
"sqlite3:data/plugins/lastseen/users.sqlite",
"data/plugins/lastseen/users.sqlite",
check_same_thread=False
)
self.data.runQuery("CREATE TABLE IF NOT EXISTS users ("
"user TEXT, "
"protocol TEXT, "
"at INTEGER)")
self.commands.register_command("seen", self.seen_command, self,
"seen.seen", default=True)
# General events
self.events.add_callback("PreMessageReceived", self,
self.event_handler, 0, cancelled=True,
extra_args=[self.event_source_caller])
self.events.add_callback("PreCommand", self,
self.event_handler, 0, cancelled=True,
extra_args=[self.event_source_caller])
self.events.add_callback("NameChanged", self,
self.event_handler, 0, cancelled=True,
extra_args=[self.event_user_caller])
self.events.add_callback("UserDisconnected", self,
self.event_handler, 0, cancelled=True,
extra_args=[self.event_user_caller])
# Mumble events
self.events.add_callback("Mumble/UserRemove", self,
self.event_handler, 0, cancelled=True,
extra_args=[self.event_user_caller])
self.events.add_callback("Mumble/UserJoined", self,
self.event_handler, 0, cancelled=True,
extra_args=[self.event_user_caller])
self.events.add_callback("Mumble/UserMoved", self,
self.event_handler, 0, cancelled=True,
extra_args=[self.event_user_caller])
self.events.add_callback("Mumble/UserSelfMuteToggle", self,
self.event_handler, 0, cancelled=True,
extra_args=[self.event_user_caller])
self.events.add_callback("Mumble/UserSelfDeafToggle", self,
self.event_handler, 0, cancelled=True,
extra_args=[self.event_user_caller])
self.events.add_callback("Mumble/UserRecordingToggle", self,
self.event_handler, 0, cancelled=True,
extra_args=[self.event_user_caller])
开发者ID:domainr，项目名称:Ultros-contrib，代码行数:56，代码来源:__init__.py
示例7: setup
def setup(self):
"""
Called when the plugin is loaded. Performs initial setup.
"""
self.logger.trace(_("Entered setup method."))
self.storage = StorageManager()
try:
self.config = self.storage.get_file(self, "config", YAML,
"plugins/urls.yml")
except Exception:
self.logger.exception(_("Error loading configuration!"))
else:
if not self.config.exists:
self.logger.warn(_("Unable to find config/plugins/urls.yml"))
else:
self.content_types = self.config["content_types"]
self.spoofing = self.config["spoofing"]
self.logger.debug(_("Spoofing: %s") % self.spoofing)
self.channels = self.storage.get_file(self, "data", YAML,
"plugins/urls/channels.yml")
self.shortened = self.storage.get_file(
self,
"data",
DBAPI,
"sqlite3:data/plugins/urls/shortened.sqlite",
"data/plugins/urls/shortened.sqlite",
check_same_thread=False
)
self.commands = CommandManager()
self.events = EventManager()
self.reload()
def message_event_filter(event=MessageReceived):
target = event.target
type_ = event.type
return type_ == "message" \
or isinstance(target, Channel) \
or isinstance(target, User)
self.add_shortener("tinyurl", self.tinyurl)
self.events.add_callback("MessageReceived", self, self.message_handler,
1, message_event_filter)
self.commands.register_command("urls", self.urls_command, self,
"urls.manage")
self.commands.register_command("shorten", self.shorten_command, self,
"urls.shorten", default=True)
开发者ID:NotAFile，项目名称:Ultros，代码行数:53，代码来源:__init__.py
示例8: AssPlugin
class AssPlugin(plugin.PluginObject):
events = None
regex = None
def setup(self):
self.events = EventManager()
self.regex = re.compile(r"(\w+)-ass (\w+)")
self.events.add_callback("MessageReceived", self, self.ass_swap, 1)
def ass_swap(self, event=MessageReceived):
source = event.source
target = event.target
message = event.message
if re.search(self.regex, message) is None:
return
result = re.sub(self.regex, r"\1 ass-\2", message)
target.respond("%s: %s" % (source.nickname, result))
开发者ID:EionRobb，项目名称:Ultros-contrib，代码行数:22，代码来源:__init__.py
示例9: __init__
def __init__(self, name, factory, config):
NoChannelsProtocol.__init__(self, name, factory, config)
self.log = getLogger(self.name)
self.event_manager = EventManager()
self.command_manager = CommandManager()
reactor.connectTCP(
self.config["connection"]["host"],
self.config["connection"]["port"],
self.factory,
120
)
开发者ID:domainr，项目名称:Ultros-contrib，代码行数:13，代码来源:protocol.py
示例10: setup
def setup(self):
self.commands = CommandManager()
self.events = EventManager()
self.storage = StorageManager()
# self.data = self.storage.get_file(self, "data", SQLITE,
# "plugins/memos/memos.sqlite")
# with self.data as c:
# # Multiline strings because of an IDE bug
# c.execute("""CREATE TABLE IF NOT EXISTS memos
# (to TEXT, from TEXT, memo TEXT)""")
self.events.add_callback("PreMessageReceived", self,
self.message_received, 0)
self.commands.register_command("memo", self.memo_command, self,
"memo.send", default=True)
开发者ID:case，项目名称:Ultros-contrib，代码行数:16，代码来源:__init__.py
示例11: setup
def setup(self):
### Grab important shit
self.commands = CommandManager()
self.events = EventManager()
self.storage = StorageManager()
### Initial config load
try:
self.config = self.storage.get_file(self, "config", YAML,
"plugins/drunkoctopus.yml")
except Exception:
self.logger.exception("Error loading configuration!")
self.logger.error("Disabling...")
self._disable_self()
return
if not self.config.exists:
self.logger.error("Unable to find config/plugins/drunkoctopus.yml")
self.logger.error("Disabling...")
self._disable_self()
return
### Create vars and stuff
self._sobering_call = None
self._drunktalk = DrunkTalk()
### Load options from config
self._load()
self.config.add_callback(self._load)
### Register events and commands
self.events.add_callback("MessageSent",
self,
self.outgoing_message_handler,
1)
self.commands.register_command("drunkenness",
self.drunkenness_command,
self,
"drunkoctopus.drunkenness",
default=True)
self.commands.register_command("drink",
self.drink_command,
self,
"drunkoctopus.drink")
开发者ID:EionRobb，项目名称:Ultros-contrib，代码行数:45，代码来源:__init__.py
示例12: __init__
def __init__(self, name, factory, config):
self.name = name
self.factory = factory
self.config = config
self.received = ""
self.log = getLogger(self.name)
self.log.info("Setting up..")
self.command_manager = CommandManager()
self.event_manager = EventManager()
self.username = config["identity"]["username"]
self.password = config["identity"]["password"]
self.networking = config["network"]
self.tokens = config["identity"]["tokens"]
self.control_chars = config["control_chars"]
audio_conf = config.get("audio", {})
self.should_mute_self = audio_conf.get("should_mute_self", True)
self.should_deafen_self = audio_conf.get("should_deafen_self", True)
event = general_events.PreConnectEvent(self, config)
self.event_manager.run_callback("PreConnect", event)
context = self._get_client_context()
if context is None:
# Could not create a context (problem loading cert file)
self.factory.manager.remove_protocol(self.name)
return
reactor.connectSSL(
self.networking["address"],
self.networking["port"],
self.factory,
context,
120
)
event = general_events.PostConnectEvent(self, config)
self.event_manager.run_callback("PostConnect", event)
开发者ID:gsingh123，项目名称:Ultros，代码行数:42，代码来源:protocol.py
示例13: setup
def setup(self):
self.logger.trace("Entered setup method.")
self.commands = CommandManager()
self.events = EventManager()
self.plugins = PluginManager()
self.storage = StorageManager()
try:
self.config = self.storage.get_file(self, "config", YAML,
"plugins/twilio.yml")
except Exception:
self.logger.exception("Error loading configuration!")
return self._disable_self()
else:
if not self.config.exists:
self.logger.error("Unable to find config/plugins/twilio.yml")
return self._disable_self()
try:
self.data = self.storage.get_file(self, "data", JSON,
"plugins/twilio/contacts.json")
except Exception:
self.logger.exception("Error loading data!")
self.logger.error("This data file is required. Shutting down...")
return self._disable_self()
self._load()
self.config.add_callback(self._load)
self.events.add_callback("Web/ServerStartedEvent", self,
self.add_routes,
0)
self.commands.register_command("sms", self.sms_command, self,
"twilio.sms")
self.commands.register_command("mms", self.mms_command, self,
"twilio.mms")
self.commands.register_command("tw", self.tw_command, self,
"twilio.tw")
开发者ID:EionRobb，项目名称:Ultros-contrib，代码行数:40，代码来源:__init__.py
示例14: setup
def setup(self):
self.logger.trace("Entered setup method.")
self.commands = CommandManager()
self.events = EventManager()
self.plugins = PluginManager()
self.storage = StorageManager()
try:
self.config = self.storage.get_file(self, "config", YAML, "plugins/webhooks.yml")
except Exception:
self.logger.exception("Error loading configuration!")
return self._disable_self()
else:
if not self.config.exists:
self.logger.error("Unable to find config/plugins/webhooks.yml")
return self._disable_self()
self._load()
self.config.add_callback(self._load)
self.events.add_callback("Web/ServerStartedEvent", self, self.add_routes, 0)
开发者ID:EionRobb，项目名称:Ultros-contrib，代码行数:22，代码来源:__init__.py
示例15: setup
def setup(self):
### Grab important shit
self.commands = CommandManager()
self.events = EventManager()
self.storage = StorageManager()
### Initial config load
try:
self._config = self.storage.get_file(self,
"config",
YAML,
"plugins/triggers.yml")
except Exception:
self.logger.exception("Error loading configuration!")
self.logger.error("Disabling...")
self._disable_self()
return
if not self._config.exists:
self.logger.error("Unable to find config/plugins/triggers.yml")
self.logger.error("Disabling...")
self._disable_self()
return
### Register event handlers
def _message_event_filter(event=MessageReceived):
return isinstance(event.target, Channel)
self.events.add_callback("MessageReceived",
self,
self.message_handler,
1,
_message_event_filter)
self.events.add_callback("ActionReceived",
self,
self.action_handler,
1,
_message_event_filter)
开发者ID:domainr，项目名称:Ultros-contrib，代码行数:38，代码来源:__init__.py
示例16: DrunkPlugin
class DrunkPlugin(plugin.PluginObject):
commands = None
config = None
storage = None
def setup(self):
### Grab important shit
self.commands = CommandManager()
self.events = EventManager()
self.storage = StorageManager()
### Initial config load
try:
self.config = self.storage.get_file(self, "config", YAML,
"plugins/drunkoctopus.yml")
except Exception:
self.logger.exception("Error loading configuration!")
self.logger.error("Disabling...")
self._disable_self()
return
if not self.config.exists:
self.logger.error("Unable to find config/plugins/drunkoctopus.yml")
self.logger.error("Disabling...")
self._disable_self()
return
### Create vars and stuff
self._sobering_call = None
self._drunktalk = DrunkTalk()
### Load options from config
self._load()
self.config.add_callback(self._load)
### Register events and commands
self.events.add_callback("MessageSent",
self,
self.outgoing_message_handler,
1)
self.commands.register_command("drunkenness",
self.drunkenness_command,
self,
"drunkoctopus.drunkenness",
default=True)
self.commands.register_command("drink",
self.drink_command,
self,
"drunkoctopus.drink")
def reload(self):
try:
self.config.reload()
except Exception:
self.logger.exception("Error reloading configuration!")
return False
return True
def _load(self):
self._drunktalk.drunkenness = self.config["drunkenness"]
self._cooldown_enabled = self.config["cooldown"]["enabled"]
self._cooldown_time = self.config["cooldown"]["time"]
self._cooldown_amount = self.config["cooldown"]["amount"]
self._drinks = self.config["drinks"]
# Sort out the sobering deferred as necessary
if self._cooldown_enabled:
if self._sobering_call is None:
self.logger.trace("Starting sobering call due to config "
"change")
self._sobering_call = reactor.callLater(self._cooldown_time,
self._sober_up)
else:
if self._sobering_call is not None:
self.logger.trace("Cancelling sobering call due to config "
"change")
self._sobering_call.cancel()
def _sober_up(self):
self.logger.trace("Sobering up")
drunk = self._drunktalk.drunkenness
drunk -= self._cooldown_amount
if drunk < 0:
drunk = 0
self._drunktalk.drunkenness = drunk
if self._cooldown_enabled:
reactor.callLater(self._cooldown_time, self._sober_up)
def drunkenness_command(self, protocol, caller, source, command, raw_args,
parsed_args):
args = raw_args.split() # Quick fix for new command handler signature
if len(args) == 0:
caller.respond("Drunkenness level: %s" %
self._drunktalk.drunkenness)
return
elif len(args) == 1:
try:
new_drunk = int(args[0])
#.........这里部分代码省略.........
开发者ID:EionRobb，项目名称:Ultros-contrib，代码行数:101，代码来源:__init__.py
示例17: TriggersPlugin
class TriggersPlugin(plugin.PluginObject):
commands = None
events = None
storage = None
_config = None
def setup(self):
### Grab important shit
self.commands = CommandManager()
self.events = EventManager()
self.storage = StorageManager()
### Initial config load
try:
self._config = self.storage.get_file(self,
"config",
YAML,
"plugins/triggers.yml")
except Exception:
self.logger.exception("Error loading configuration!")
self.logger.error("Disabling...")
self._disable_self()
return
if not self._config.exists:
self.logger.error("Unable to find config/plugins/triggers.yml")
self.logger.error("Disabling...")
self._disable_self()
return
### Register event handlers
def _message_event_filter(event=MessageReceived):
return isinstance(event.target, Channel)
self.events.add_callback("MessageReceived",
self,
self.message_handler,
1,
_message_event_filter)
self.events.add_callback("ActionReceived",
self,
self.action_handler,
1,
_message_event_filter)
def reload(self):
try:
self._config.reload()
except Exception:
self.logger.exception("Error reloading configuration!")
return False
return True
@property
def _triggers(self):
return self._config.get("triggers", {})
def message_handler(self, event=MessageReceived):
self.event_handler(
event.caller,
event.source,
event.target,
event.message,
event.type
)
def action_handler(self, event=ActionReceived):
self.event_handler(
event.caller,
event.source,
event.target,
event.message,
"action"
)
def event_handler(self, protocol, source, target, message, e_type):
"""
Event handler for general messages
"""
allowed = self.commands.perm_handler.check("triggers.trigger",
source,
target,
protocol)
if not allowed:
return
# TODO: Rewrite this when Matcher is finished
# TODO: We check the types of half of these - do the rest
global_triggers = self._triggers.get("global", [])
proto_trigger_block = self._triggers.get("protocols", {})
proto_triggers = proto_trigger_block.get(protocol.name, {})
if not isinstance(proto_triggers, dict):
self.logger.error(
"Invalid triggers for protocol '%s'" % protocol.name
#.........这里部分代码省略.........
开发者ID:domainr，项目名称:Ultros-contrib，代码行数:101，代码来源:__init__.py
示例18: LastseenPlugin
class LastseenPlugin(plugin.PluginObject):
commands = None
events = None
storage = None
data = None # SQLite for a change
def setup(self):
self.commands = CommandManager()
self.events = EventManager()
self.storage = StorageManager()
self.data = self.storage.get_file(
self,
"data",
DBAPI,
"sqlite3:data/plugins/lastseen/users.sqlite",
"data/plugins/lastseen/users.sqlite",
check_same_thread=False
)
self.data.runQuery("CREATE TABLE IF NOT EXISTS users ("
"user TEXT, "
"protocol TEXT, "
"at INTEGER)")
self.commands.register_command("seen", self.seen_command, self,
"seen.seen", default=True)
# General events
self.events.add_callback("PreMessageReceived", self,
self.event_handler, 0, cancelled=True,
extra_args=[self.event_source_caller])
self.events.add_callback("PreCommand", self,
self.event_handler, 0, cancelled=True,
extra_args=[self.event_source_caller])
self.events.add_callback("NameChanged", self,
self.event_handler, 0, cancelled=True,
extra_args=[self.event_user_caller])
self.events.add_callback("UserDisconnected", self,
self.event_handler, 0, cancelled=True,
extra_args=[self.event_user_caller])
# Mumble events
self.events.add_callback("Mumble/UserRemove", self,
self.event_handler, 0, cancelled=True,
extra_args=[self.event_user_caller])
self.events.add_callback("Mumble/UserJoined", self,
self.event_handler, 0, cancelled=True,
extra_args=[self.event_user_caller])
self.events.add_callback("Mumble/UserMoved", self,
self.event_handler, 0, cancelled=True,
extra_args=[self.event_user_caller])
self.events.add_callback("Mumble/UserSelfMuteToggle", self,
self.event_handler, 0, cancelled=True,
extra_args=[self.event_user_caller])
self.events.add_callback("Mumble/UserSelfDeafToggle", self,
self.event_handler, 0, cancelled=True,
extra_args=[self.event_user_caller])
self.events.add_callback("Mumble/UserRecordingToggle", self,
self.event_handler, 0, cancelled=True,
extra_args=[self.event_user_caller])
def _get_user_txn(self, txn, user, protocol):
user = user.lower()
txn.execute("SELECT * FROM users WHERE user=? AND protocol=?",
(user, protocol))
r = txn.fetchone()
return r
def _get_user_callback(self, result, user, protocol, source):
if result is None:
source.respond("User '%s' not found." % user)
else:
then = math.floor(result[2])
now = math.floor(time.time())
seconds = now - then
m, s = divmod(seconds, 60)
h, m = divmod(m, 60)
d, h = divmod(h, 24)
s = int(s)
m = int(m)
h = int(h)
d = int(d)
if (s + m + h + d) == 0:
source.respond("'%s' was seen just now!" % user)
else:
constructed = "'%s' was seen" % user
to_append = []
if d > 0:
to_append.append("%s days" % d)
if h > 0:
to_append.append("%s hours" % h)
if m > 0:
to_append.append("%s minutes" % m)
#.........这里部分代码省略.........
开发者ID:domainr，项目名称:Ultros-contrib，代码行数:101，代码来源:__init__.py
示例19: FactoidsPlugin
class FactoidsPlugin(plugin.PluginObject):
CHANNEL = "channel"
PROTOCOL = "protocol"
GLOBAL = "global"
PERM_ADD = "factoids.add.%s"
PERM_SET = "factoids.set.%s"
PERM_DEL = "factoids.delete.%s"
PERM_GET = "factoids.get.%s"
(RES_INVALID_LOCATION,
RES_INVALID_METHOD, # _FOR_LOCATION - i.e. CHANNEL in PM
RES_NO_PERMS,
RES_MISSING_FACTOID) = xrange(4)
def setup(self):
# ## Grab important shit
self.commands = CommandManager()
self.events = EventManager()
self.storage = StorageManager()
self.plugman = PluginManager()
# ## Set up database
self.database = self.storage.get_file(
self,
"data",
DBAPI,
"sqlite3:data/plugins/factoids.sqlite",
"data/plugins/factoids.sqlite",
check_same_thread=False
)
self.database.add_callback(self.reload)
self.reload()
# ## Register commands
# We have multiple possible permissions per command, so we have to do
# permission handling ourselves
self.commands.register_command("addfactoid",
self.factoid_add_command,
self,
None)
self.commands.register_command("setfactoid",
self.factoid_set_command,
self,
None)
self.commands.register_command("deletefactoid",
self.factoid_delete_command,
self,
None,
["delfactoid"])
self.commands.register_command("getfactoid",
self.factoid_get_command,
self,
None, default=True)
# ## Register events
self.events.add_callback("MessageReceived",
self,
self.message_handler,
1)
self.events.add_callback("Web/ServerStartedEvent",
self,
self.web_routes,
1)
def reload(self):
with self.database as db:
db.runQuery("CREATE TABLE IF NOT EXISTS factoids ("
"factoid_key TEXT, "
"location TEXT, "
"protocol TEXT, "
"channel TEXT, "
"factoid_name TEXT, "
"info TEXT, "
"UNIQUE(factoid_key, location, protocol, channel) "
"ON CONFLICT REPLACE)")
# region Util functions
def __check_perm(self, perm, caller, source, protocol):
self.logger.trace(_("Checking for permission: '%s'"), perm)
allowed = self.commands.perm_handler.check(perm,
caller,
source,
protocol)
return allowed
def _parse_args(self, raw_args):
"""
Grabs the location, factoid name, and info from a raw_args string
"""
pos = raw_args.find(" ")
if pos < 0:
raise ValueError(_("Invalid args"))
location = raw_args[:pos]
pos2 = raw_args.find(" ", pos + 1)
if pos2 < 0:
#.........这里部分代码省略.........
开发者ID:NotAFile，项目名称:Ultros，代码行数:101，代码来源:__init__.py
-
Python入门教程 Python 是一种解释型、面向对象、动态数据类型的高级程序设计语言。 P
阅读：13941|2022-01-22
-
Python wikiutil.getFrontPage函数代码示例
阅读：10292|2022-05-24
-
Python 简介 Python 是一个高层次的结合了解释性、编译性、互动性和面向对象的脚本
阅读：4177|2022-01-22
-
Python tests.group函数代码示例
阅读：4064|2022-05-27
-
Python util.check_if_user_has_permission函数代码示例
阅读：3889|2022-05-27
-
Python 练习实例98 Python 100例题目：从键盘输入一个字符串，将小写字母全部转换成大
阅读：3539|2022-01-22
-
Python 环境搭建 本章节我们将向大家介绍如何在本地搭建 Python 开发环境。 Py
阅读：3070|2022-01-22
-
Python 基础语法 Python 语言与 Perl，C 和 Java 等语言有许多相似之处。但是，也
阅读：2731|2022-01-22
-
Python output.darkgreen函数代码示例
阅读：2682|2022-05-25
-
Python 中文编码前面章节中我们已经学会了如何用 Python 输出 Hello, World!，英文没
阅读：2349|2022-01-22
|
客服电话
APP下载
官方微信
请发表评论