"""
Module containing the commands of the in-game Python system.
"""
from datetime import datetime
from django.conf import settings
from evennia.contrib.base_systems.ingame_python.utils import get_event_handler
from evennia.utils.ansi import raw
from evennia.utils.eveditor import EvEditor
from evennia.utils.evtable import EvTable
from evennia.utils.utils import class_from_module, time_format
COMMAND_DEFAULT_CLASS = class_from_module(settings.COMMAND_DEFAULT_CLASS)
# Permissions
WITH_VALIDATION = getattr(settings, "callbackS_WITH_VALIDATION", None)
WITHOUT_VALIDATION = getattr(settings, "callbackS_WITHOUT_VALIDATION", "developer")
VALIDATING = getattr(settings, "callbackS_VALIDATING", "developer")
# Split help text
BASIC_HELP = "Add, edit or delete callbacks."
BASIC_USAGES = [
"@call <object name> [= <callback name>]",
"@call/add <object name> = <callback name> [parameters]",
"@call/edit <object name> = <callback name> [callback number]",
"@call/del <object name> = <callback name> [callback number]",
"@call/tasks [object name [= <callback name>]]",
]
BASIC_SWITCHES = [
"add - add and edit a new callback",
"edit - edit an existing callback",
"del - delete an existing callback",
"tasks - show the list of differed tasks",
]
VALIDATOR_USAGES = ["@call/accept [object name = <callback name> [callback number]]"]
VALIDATOR_SWITCHES = ["accept - show callbacks to be validated or accept one"]
BASIC_TEXT = """
This command is used to manipulate callbacks. A callback can be linked to
an object, to fire at a specific moment. You can use the command without
switches to see what callbacks are active on an object:
@call self
You can also specify a callback name if you want the list of callbacks
associated with this object of this name:
@call north = can_traverse
You can also add a number after the callback name to see details on one callback:
@call here = say 2
You can also add, edit or remove callbacks using the add, edit or del switches.
Additionally, you can see the list of differed tasks created by callbacks
(chained events to be called) using the /tasks switch.
"""
VALIDATOR_TEXT = """
You can also use this command to validate callbacks. Depending on your game
setting, some users might be allowed to add new callbacks, but these callbacks
will not be fired until you accept them. To see the callbacks needing
validation, enter the /accept switch without argument:
@call/accept
A table will show you the callbacks that are not validated yet, who created
them and when. You can then accept a specific callback:
@call here = enter 1
Use the /del switch to remove callbacks that should not be connected.
"""
[docs]class CmdCallback(COMMAND_DEFAULT_CLASS):
"""
Command to edit callbacks.
"""
key = "@call"
aliases = ["@callback", "@callbacks", "@calls"]
locks = "cmd:perm({})".format(VALIDATING)
if WITH_VALIDATION:
locks += " or perm({})".format(WITH_VALIDATION)
help_category = "Building"
[docs] def get_help(self, caller, cmdset):
"""
Return the help message for this command and this caller.
The help text of this specific command will vary depending
on user permission.
Args:
caller (Object or Account): the caller asking for help on the command.
cmdset (CmdSet): the command set (if you need additional commands).
Returns:
docstring (str): the help text to provide the caller for this command.
"""
lock = "perm({}) or perm(callbacks_validating)".format(VALIDATING)
validator = caller.locks.check_lockstring(caller, lock)
text = "\n" + BASIC_HELP + "\n\nUsages:\n "
# Usages
text += "\n ".join(BASIC_USAGES)
if validator:
text += "\n " + "\n ".join(VALIDATOR_USAGES)
# Switches
text += "\n\nSwitches:\n "
text += "\n ".join(BASIC_SWITCHES)
if validator:
text += "\n " + "\n ".join(VALIDATOR_SWITCHES)
# Text
text += "\n" + BASIC_TEXT
if validator:
text += "\n" + VALIDATOR_TEXT
return text
[docs] def func(self):
"""Command body."""
caller = self.caller
lock = "perm({}) or perm(events_validating)".format(VALIDATING)
validator = caller.locks.check_lockstring(caller, lock)
lock = "perm({}) or perm(events_without_validation)".format(WITHOUT_VALIDATION)
autovalid = caller.locks.check_lockstring(caller, lock)
# First and foremost, get the callback handler and set other variables
self.handler = get_event_handler()
self.obj = None
rhs = self.rhs or ""
self.callback_name, sep, self.parameters = rhs.partition(" ")
self.callback_name = self.callback_name.lower()
self.is_validator = validator
self.autovalid = autovalid
if self.handler is None:
caller.msg("The event handler is not running, can't " "access the event system.")
return
# Before the equal sign, there is an object name or nothing
if self.lhs:
self.obj = caller.search(self.lhs)
if not self.obj:
return
# Switches are mutually exclusive
switch = self.switches and self.switches[0] or ""
if switch in ("", "add", "edit", "del") and self.obj is None:
caller.msg("Specify an object's name or #ID.")
return
if switch == "":
self.list_callbacks()
elif switch == "add":
self.add_callback()
elif switch == "edit":
self.edit_callback()
elif switch == "del":
self.del_callback()
elif switch == "accept" and validator:
self.accept_callback()
elif switch in ["tasks", "task"]:
self.list_tasks()
else:
caller.msg("Mutually exclusive or invalid switches were " "used, cannot proceed.")
[docs] def list_callbacks(self):
"""Display the list of callbacks connected to the object."""
obj = self.obj
callback_name = self.callback_name
parameters = self.parameters
callbacks = self.handler.get_callbacks(obj)
types = self.handler.get_events(obj)
if callback_name:
# Check that the callback name can be found in this object
created = callbacks.get(callback_name)
if created is None:
self.msg("No callback {} has been set on {}.".format(callback_name, obj))
return
if parameters:
# Check that the parameter points to an existing callback
try:
number = int(parameters) - 1
assert number >= 0
callback = callbacks[callback_name][number]
except (ValueError, AssertionError, IndexError):
self.msg(
"The callback {} {} cannot be found in {}.".format(
callback_name, parameters, obj
)
)
return
# Display the callback's details
author = callback.get("author")
author = author.key if author else "|gUnknown|n"
updated_by = callback.get("updated_by")
updated_by = updated_by.key if updated_by else "|gUnknown|n"
created_on = callback.get("created_on")
created_on = (
created_on.strftime("%Y-%m-%d %H:%M:%S") if created_on else "|gUnknown|n"
)
updated_on = callback.get("updated_on")
updated_on = (
updated_on.strftime("%Y-%m-%d %H:%M:%S") if updated_on else "|gUnknown|n"
)
msg = "Callback {} {} of {}:".format(callback_name, parameters, obj)
msg += "\nCreated by {} on {}.".format(author, created_on)
msg += "\nUpdated by {} on {}".format(updated_by, updated_on)
if self.is_validator:
if callback.get("valid"):
msg += "\nThis callback is |rconnected|n and active."
else:
msg += "\nThis callback |rhasn't been|n accepted yet."
msg += "\nCallback code:\n"
msg += raw(callback["code"])
self.msg(msg)
return
# No parameter has been specified, display the table of callbacks
cols = ["Number", "Author", "Updated", "Param"]
if self.is_validator:
cols.append("Valid")
table = EvTable(*cols, width=78)
table.reformat_column(0, align="r")
now = datetime.now()
for i, callback in enumerate(created):
author = callback.get("author")
author = author.key if author else "|gUnknown|n"
updated_on = callback.get("updated_on")
if updated_on is None:
updated_on = callback.get("created_on")
if updated_on:
updated_on = "{} ago".format(
time_format((now - updated_on).total_seconds(), 4).capitalize()
)
else:
updated_on = "|gUnknown|n"
parameters = callback.get("parameters", "")
row = [str(i + 1), author, updated_on, parameters]
if self.is_validator:
row.append("Yes" if callback.get("valid") else "No")
table.add_row(*row)
self.msg(str(table))
else:
names = list(set(list(types.keys()) + list(callbacks.keys())))
table = EvTable("Callback name", "Number", "Description", valign="t", width=78)
table.reformat_column(0, width=20)
table.reformat_column(1, width=10, align="r")
table.reformat_column(2, width=48)
for name in sorted(names):
number = len(callbacks.get(name, []))
lines = sum(len(e["code"].splitlines()) for e in callbacks.get(name, []))
no = "{} ({})".format(number, lines)
description = types.get(name, (None, "Chained event."))[1]
description = description.strip("\n").splitlines()[0]
table.add_row(name, no, description)
self.msg(str(table))
[docs] def add_callback(self):
"""Add a callback."""
obj = self.obj
callback_name = self.callback_name
types = self.handler.get_events(obj)
# Check that the callback exists
if not callback_name.startswith("chain_") and callback_name not in types:
self.msg(
"The callback name {} can't be found in {} of "
"typeclass {}.".format(callback_name, obj, type(obj))
)
return
definition = types.get(callback_name, (None, "Chained event."))
description = definition[1]
self.msg(raw(description.strip("\n")))
# Open the editor
callback = self.handler.add_callback(
obj, callback_name, "", self.caller, False, parameters=self.parameters
)
# Lock this callback right away
self.handler.db.locked.append((obj, callback_name, callback["number"]))
# Open the editor for this callback
self.caller.db._callback = callback
EvEditor(
self.caller,
loadfunc=_ev_load,
savefunc=_ev_save,
quitfunc=_ev_quit,
key="Callback {} of {}".format(callback_name, obj),
persistent=True,
codefunc=_ev_save,
)
[docs] def edit_callback(self):
"""Edit a callback."""
obj = self.obj
callback_name = self.callback_name
parameters = self.parameters
callbacks = self.handler.get_callbacks(obj)
types = self.handler.get_events(obj)
# If no callback name is specified, display the list of callbacks
if not callback_name:
self.list_callbacks()
return
# Check that the callback exists
if callback_name not in callbacks:
self.msg("The callback name {} can't be found in {}.".format(callback_name, obj))
return
# If there's only one callback, just edit it
if len(callbacks[callback_name]) == 1:
number = 0
callback = callbacks[callback_name][0]
else:
if not parameters:
self.msg("Which callback do you wish to edit? Specify a number.")
self.list_callbacks()
return
# Check that the parameter points to an existing callback
try:
number = int(parameters) - 1
assert number >= 0
callback = callbacks[callback_name][number]
except (ValueError, AssertionError, IndexError):
self.msg(
"The callback {} {} cannot be found in {}.".format(
callback_name, parameters, obj
)
)
return
# If caller can't edit without validation, forbid editing
# others' works
if not self.autovalid and callback["author"] is not self.caller:
self.msg("You cannot edit this callback created by someone else.")
return
# If the callback is locked (edited by someone else)
if (obj, callback_name, number) in self.handler.db.locked:
self.msg("This callback is locked, you cannot edit it.")
return
self.handler.db.locked.append((obj, callback_name, number))
# Check the definition of the callback
definition = types.get(callback_name, (None, "Chained event."))
description = definition[1]
self.msg(raw(description.strip("\n")))
# Open the editor
callback = dict(callback)
self.caller.db._callback = callback
EvEditor(
self.caller,
loadfunc=_ev_load,
savefunc=_ev_save,
quitfunc=_ev_quit,
key="Callback {} of {}".format(callback_name, obj),
persistent=True,
codefunc=_ev_save,
)
[docs] def del_callback(self):
"""Delete a callback."""
obj = self.obj
callback_name = self.callback_name
parameters = self.parameters
callbacks = self.handler.get_callbacks(obj)
types = self.handler.get_events(obj)
# If no callback name is specified, display the list of callbacks
if not callback_name:
self.list_callbacks()
return
# Check that the callback exists
if callback_name not in callbacks:
self.msg("The callback name {} can't be found in {}.".format(callback_name, obj))
return
# If there's only one callback, just delete it
if len(callbacks[callback_name]) == 1:
number = 0
callback = callbacks[callback_name][0]
else:
if not parameters:
self.msg("Which callback do you wish to delete? Specify " "a number.")
self.list_callbacks()
return
# Check that the parameter points to an existing callback
try:
number = int(parameters) - 1
assert number >= 0
callback = callbacks[callback_name][number]
except (ValueError, AssertionError, IndexError):
self.msg(
"The callback {} {} cannot be found in {}.".format(
callback_name, parameters, obj
)
)
return
# If caller can't edit without validation, forbid deleting
# others' works
if not self.autovalid and callback["author"] is not self.caller:
self.msg("You cannot delete this callback created by someone else.")
return
# If the callback is locked (edited by someone else)
if (obj, callback_name, number) in self.handler.db.locked:
self.msg("This callback is locked, you cannot delete it.")
return
# Delete the callback
self.handler.del_callback(obj, callback_name, number)
self.msg("The callback {}[{}] of {} was deleted.".format(callback_name, number + 1, obj))
[docs] def accept_callback(self):
"""Accept a callback."""
obj = self.obj
callback_name = self.callback_name
parameters = self.parameters
# If no object, display the list of callbacks to be checked
if obj is None:
table = EvTable("ID", "Type", "Object", "Name", "Updated by", "On", width=78)
table.reformat_column(0, align="r")
now = datetime.now()
for obj, name, number in self.handler.db.to_valid:
callbacks = self.handler.get_callbacks(obj).get(name)
if callbacks is None:
continue
try:
callback = callbacks[number]
except IndexError:
continue
type_name = obj.typeclass_path.split(".")[-1]
by = callback.get("updated_by")
by = by.key if by else "|gUnknown|n"
updated_on = callback.get("updated_on")
if updated_on is None:
updated_on = callback.get("created_on")
if updated_on:
updated_on = "{} ago".format(
time_format((now - updated_on).total_seconds(), 4).capitalize()
)
else:
updated_on = "|gUnknown|n"
table.add_row(obj.id, type_name, obj, name, by, updated_on)
self.msg(str(table))
return
# An object was specified
callbacks = self.handler.get_callbacks(obj)
types = self.handler.get_events(obj)
# If no callback name is specified, display the list of callbacks
if not callback_name:
self.list_callbacks()
return
# Check that the callback exists
if callback_name not in callbacks:
self.msg("The callback name {} can't be found in {}.".format(callback_name, obj))
return
if not parameters:
self.msg("Which callback do you wish to accept? Specify a number.")
self.list_callbacks()
return
# Check that the parameter points to an existing callback
try:
number = int(parameters) - 1
assert number >= 0
callback = callbacks[callback_name][number]
except (ValueError, AssertionError, IndexError):
self.msg(
"The callback {} {} cannot be found in {}.".format(callback_name, parameters, obj)
)
return
# Accept the callback
if callback["valid"]:
self.msg("This callback has already been accepted.")
else:
self.handler.accept_callback(obj, callback_name, number)
self.msg(
"The callback {} {} of {} has been accepted.".format(callback_name, parameters, obj)
)
[docs] def list_tasks(self):
"""List the active tasks."""
obj = self.obj
callback_name = self.callback_name
handler = self.handler
tasks = [(k, v[0], v[1], v[2]) for k, v in handler.db.tasks.items()]
if obj:
tasks = [task for task in tasks if task[2] is obj]
if callback_name:
tasks = [task for task in tasks if task[3] == callback_name]
tasks.sort()
table = EvTable("ID", "Object", "Callback", "In", width=78)
table.reformat_column(0, align="r")
now = datetime.now()
for task_id, future, obj, callback_name in tasks:
key = obj.get_display_name(self.caller)
delta = time_format((future - now).total_seconds(), 1)
table.add_row(task_id, key, callback_name, delta)
self.msg(str(table))
# Private functions to handle editing
def _ev_load(caller):
return caller.db._callback and caller.db._callback.get("code", "") or ""
def _ev_save(caller, buf):
"""Save and add the callback."""
lock = "perm({}) or perm(events_without_validation)".format(WITHOUT_VALIDATION)
autovalid = caller.locks.check_lockstring(caller, lock)
callback = caller.db._callback
handler = get_event_handler()
if (
not handler
or not callback
or not all(key in callback for key in ("obj", "name", "number", "valid"))
):
caller.msg("Couldn't save this callback.")
return False
if (callback["obj"], callback["name"], callback["number"]) in handler.db.locked:
handler.db.locked.remove((callback["obj"], callback["name"], callback["number"]))
handler.edit_callback(
callback["obj"], callback["name"], callback["number"], buf, caller, valid=autovalid
)
return True
def _ev_quit(caller):
callback = caller.db._callback
handler = get_event_handler()
if (
not handler
or not callback
or not all(key in callback for key in ("obj", "name", "number", "valid"))
):
caller.msg("Couldn't save this callback.")
return False
if (callback["obj"], callback["name"], callback["number"]) in handler.db.locked:
handler.db.locked.remove((callback["obj"], callback["name"], callback["number"]))
del caller.db._callback
caller.msg("Exited the code editor.")
