"""Miscellaneous functions for the mudpy engine."""
-# Copyright (c) 2004-2018 Jeremy Stanley <fungi@yuggoth.org>. Permission
-# to use, copy, modify, and distribute this software is granted under
-# terms provided in the LICENSE file distributed with this software.
+# Copyright (c) 2004-2021 mudpy authors. Permission to use, copy,
+# modify, and distribute this software is granted under terms
+# provided in the LICENSE file distributed with this software.
import codecs
+import datetime
import os
import random
import re
prepend_padding
)
+ def is_restricted(self):
+ """Boolean check whether command is administrative or debugging."""
+ return bool(self.get("administrative") or self.get("debugging"))
+
+ def is_admin(self):
+ """Boolean check whether an actor is controlled by an admin owner."""
+ return self.owner and self.owner.is_admin()
+
def can_run(self, command):
"""Check if the user can run this command object."""
# has to be in the commands group
if command not in self.universe.groups["command"].values():
- result = False
+ return False
+
+ # debugging commands are not allowed outside debug mode
+ if command.get("debugging") and not self.universe.debug_mode():
+ return False
# avatars of administrators can run any command
- elif self.owner and self.owner.account.get("administrator"):
- result = True
+ if self.is_admin():
+ return True
# everyone can run non-administrative commands
- elif not command.get("administrative"):
- result = True
+ if not command.is_restricted():
+ return True
# otherwise the command cannot be run by this actor
- else:
- result = False
-
- # pass back the result
- return result
+ return False
def update_location(self):
"""Make sure the location's contents contain this element."""
element.update_location()
element.clean_contents()
+ # warn when debug mode has been engaged
+ if self.debug_mode():
+ pending_loglines.append((
+ "WARNING: Unsafe debugging mode is enabled!", 6))
+
# done loading, so disallow updating elements from read-only files
self.loading = False
"""Create a new, empty Universe (the Big Bang)."""
new_universe = Universe()
for attribute in vars(self).keys():
- exec("new_universe." + attribute + " = self." + attribute)
+ setattr(new_universe, attribute, getattr(self, attribute))
new_universe.reload_flag = False
del self
return new_universe
host = self.contents["mudpy.network"].get("host")
port = self.contents["mudpy.network"].get("port")
- # if no host was specified, bind to all local addresses (preferring
+ # if no host was specified, bind to the loopback address (preferring
# ipv6)
if not host:
if socket.has_ipv6:
- host = "::"
+ host = "::1"
else:
- host = "0.0.0.0"
+ host = "127.0.0.1"
# figure out if this is ipv4 or v6
family = socket.getaddrinfo(host, port)[0][0]
self.listening_socket.listen(1)
# note that we're now ready for user connections
- log(
- "Listening for Telnet connections on: " +
- host + ":" + str(port)
- )
+ log("Listening for Telnet connections on %s port %s" % (
+ host, str(port)))
def get_time(self):
"""Convenience method to get the elapsed time counter."""
- return self.groups["internal"]["counters"].get("elapsed")
+ try:
+ return self.groups["internal"]["counters"].get("elapsed", 0)
+ except KeyError:
+ return 0
+
+ def set_time(self, elapsed):
+ """Convenience method to set the elapsed time counter."""
+ try:
+ self.groups["internal"]["counters"].set("elapsed", elapsed)
+ except KeyError:
+ # add an element for counters if it doesn't exist
+ Element("internal.counters", universe)
+ self.groups["internal"]["counters"].set("elapsed", elapsed)
def add_group(self, group, fallback=None):
"""Set up group tracking/metadata."""
if fallback not in self.files:
mudpy.data.Data(fallback, self, flags=flags)
+ def debug_mode(self):
+ """Boolean method to indicate whether unsafe debugging is enabled."""
+ return self.groups["mudpy"]["limit"].get("debug", False)
+
class User:
self.address = ""
self.authenticated = False
self.avatar = None
+ self.choice = ""
self.columns = 79
self.connection = None
self.error = ""
self.output_queue = []
self.partial_input = b""
self.password_tries = 0
+ self.rows = 23
self.state = "telopt_negotiation"
self.telopts = {}
+ self.ttype = None
+ self.universe = universe
def quit(self):
"""Log, close the connection and remove."""
if self.account:
- name = self.account.get("name")
- else:
- name = ""
- if name:
- message = "User " + name
+ name = self.account.get("name", self)
else:
- message = "An unnamed user"
- message += " logged out."
- log(message, 2)
+ name = self
+ log("Logging out %s" % name, 2)
self.deactivate_avatar()
self.connection.close()
self.remove()
self.remove()
# get rid of the old user object
- del(self)
+ del self
# create a new user object
new_user = User()
old_user.connection = self.connection
old_user.last_address = old_user.address
old_user.address = self.address
+ old_user.telopts = self.telopts
+ old_user.adjust_echoing()
# take this one out of the list and delete
self.remove()
- del(self)
+ del self
return_value = True
break
def authenticate(self):
"""Flag the user as authenticated and disconnect duplicates."""
- if self.state is not "authenticated":
+ if self.state != "authenticated":
self.authenticated = True
+ log("User %s authenticated for account %s." % (
+ self, self.account.subkey), 2)
if ("mudpy.limit" in universe.contents and self.account.subkey in
universe.contents["mudpy.limit"].get("admins")):
self.account.set("administrator", True)
- log("Administrator %s authenticated." %
- self.account.get("name"), 2)
- else:
- # log("User %s authenticated." % self.account.get("name"), 2)
- log("User %s authenticated." % self.account.subkey, 2)
+ log("Account %s is an administrator." % (
+ self.account.subkey), 2)
def show_menu(self):
"""Send the user their current menu."""
self.error = False
self.adjust_echoing()
+ def prompt(self):
+ """"Generate and return an input prompt."""
+
+ # Start with the user's preference, if one was provided
+ prompt = self.account.get("prompt")
+
+ # If the user has not set a prompt, then immediately return the default
+ # provided for the current state
+ if not prompt:
+ return get_menu_prompt(self.state)
+
+ # Allow including the World clock state
+ if "$_(time)" in prompt:
+ prompt = prompt.replace(
+ "$_(time)",
+ str(universe.get_time()))
+
+ # Append a single space for clear separation from user input
+ if prompt[-1] != " ":
+ prompt = "%s " % prompt
+
+ # Return the cooked prompt
+ return prompt
+
def adjust_echoing(self):
"""Adjust echoing to match state menu requirements."""
if mudpy.telnet.is_enabled(self, mudpy.telnet.TELOPT_ECHO,
def remove(self):
"""Remove a user from the list of connected users."""
+ log("Disconnecting account %s." % self, 0)
universe.userlist.remove(self)
def send(
if not just_prompt:
output += "$(eol)"
if add_prompt:
- output += "> "
+ output += self.prompt()
mode = self.avatar.get("mode")
if mode:
output += "(" + mode + ") "
else:
self.check_idle()
+ # ask the client for their current terminal type (RFC 1091); it's None
+ # if it's not been initialized, the empty string if it has but the
+ # output was indeterminate, "UNKNOWN" if the client specified it has no
+ # terminal types to supply
+ if self.ttype is None:
+ mudpy.telnet.request_ttype(self)
+
# if output is paused, decrement the counter
if self.state == "telopt_negotiation":
if self.negotiation_pause:
# check for some input
try:
raw_input = self.connection.recv(1024)
- except (BlockingIOError, OSError):
+ except OSError:
raw_input = b""
# we got something
universe)
self.avatar.append("inherit", "archetype.avatar")
self.account.append("avatars", self.avatar.key)
+ log("Created new avatar %s for user %s." % (
+ self.avatar.key, self.account.get("name")), 0)
def delete_avatar(self, avatar):
"""Remove an avatar from the world and from the user's list."""
if self.avatar is universe.contents[avatar]:
self.avatar = None
+ log("Deleting avatar %s for user %s." % (
+ avatar, self.account.get("name")), 0)
universe.contents[avatar].destroy()
avatars = self.account.get("avatars")
avatars.remove(avatar)
self.account.get("avatars")[index]]
self.avatar.owner = self
self.state = "active"
+ log("Activated avatar %s (%s)." % (
+ self.avatar.get("name"), self.avatar.key), 0)
self.avatar.go_home()
def deactivate_avatar(self):
"""Have the active avatar leave the world."""
if self.avatar:
+ log("Deactivating avatar %s (%s) for user %s." % (
+ self.avatar.get("name"), self.avatar.key,
+ self.account.get("name")), 0)
current = self.avatar.get("location")
if current:
self.avatar.set("default_location", current)
"""Destroy the user and associated avatars."""
for avatar in self.account.get("avatars"):
self.delete_avatar(avatar)
+ log("Destroying account %s for user %s." % (
+ self.account.get("name"), self), 0)
self.account.destroy()
def list_avatar_names(self):
avatar, 6)
return avatars
+ def is_admin(self):
+ """Boolean check whether user's account is an admin."""
+ return self.account.get("administrator", False)
+
def broadcast(message, add_prompt=True):
"""Send a message to all connected users."""
file_name = ""
max_log_lines = 0
syslog_name = ""
- timestamp = time.asctime()[4:19]
+ timestamp = datetime.datetime.now().isoformat(' ')
# turn the message into a list of nonempty lines
lines = [x for x in [(x.rstrip()) for x in message.split("\n")] if x != ""]
if file_name:
if not os.path.isabs(file_name):
file_name = os.path.join(universe.startdir, file_name)
+ os.makedirs(os.path.dirname(file_name), exist_ok=True)
file_descriptor = codecs.open(file_name, "a", "utf-8")
for line in lines:
file_descriptor.write(timestamp + " " + line + "\n")
# display to connected administrators
for user in universe.userlist:
- if user.state == "active" and user.account.get(
- "administrator"
- ) and user.account.get("loglevel", 0) <= level:
+ if (
+ user.state == "active"
+ and user.is_admin()
+ and user.account.get("loglevel", 0) <= level):
# iterate over every line in the message
full_message = ""
for line in lines:
for line in lines:
while 0 < len(universe.loglines) >= max_log_lines:
del universe.loglines[0]
- universe.loglines.append((level, timestamp + " " + line))
+ universe.loglines.append((timestamp + " " + line, level))
def get_loglines(level, start, stop):
"""Return a specific range of loglines filtered by level."""
# filter the log lines
- loglines = [x for x in universe.loglines if x[0] >= level]
+ loglines = [x for x in universe.loglines if x[1] >= level]
# we need these in several places
total_count = str(len(universe.loglines))
# don't proceed if there are no lines
if filtered_count:
- # can't start before the begining or at the end
+ # can't start before the beginning or at the end
if start > filtered_count:
start = filtered_count
if start < 1:
stop = 1
# some preamble
- message = "There are " + str(total_count)
- message += " log lines in memory and " + str(filtered_count)
- message += " at or above level " + str(level) + "."
- message += " The matching lines from " + str(stop) + " to "
- message += str(start) + " are:$(eol)$(eol)"
+ message = (
+ "There are %s log lines in memory and %s at or above level %s. "
+ "The matching lines from %s to %s are:$(eol)$(eol)" %
+ (total_count, filtered_count, level, stop, start))
# add the text from the selected lines
if stop > 1:
else:
range_lines = loglines[-start:]
for line in range_lines:
- message += " (" + str(line[0]) + ") " + line[1].replace(
- "$(", "$_("
- ) + "$(eol)"
+ message += " (%s) %s$(eol)" % (
+ line[1], line[0].replace("$(", "$_("))
# there were no lines
else:
- message = "None of the " + str(total_count)
- message += " lines in memory matches your request."
+ message = "None of the %s lines in memory matches your request." % (
+ total_count)
# pass it back
return message
# characters, printable or otherwise
abs_pos = 0
- # the current text position relative to the begining of the line,
+ # the current text position relative to the beginning of the line,
# ignoring color escape sequences
rel_pos = 0
# normalize any potentially composited unicode before we count it
text = unicodedata.normalize("NFKC", text)
- # iterate over each character from the begining of the text
+ # iterate over each character from the beginning of the text
for each_character in text:
# the current character is the escape character
# characters but the space it replaced was only one
abs_pos += 1
- # now we're at the begining of a new line, plus the
+ # now we're at the beginning of a new line, plus the
# number of characters wrapped from the previous line
rel_pos -= last_rel_whitespace
last_rel_whitespace = 0
# create the expanded list of keys
for key in data.keys():
- for count in range(data[key]):
+ for _count in range(data[key]):
expanded.append(key)
# return one at random
- return random.choice(expanded)
+ # Allow the random.randrange() call in bandit since it's not used for
+ # security/cryptographic purposes
+ return random.choice(expanded) # nosec
def random_name():
name = ""
# create a name of random length from the syllables
- for syllable in range(random.randrange(2, 6)):
+ # Allow the random.randrange() call in bandit since it's not used for
+ # security/cryptographic purposes
+ for _syllable in range(random.randrange(2, 6)): # nosec
name += weighted_choice(syllables)
# strip any leading quotemark, capitalize and return the name
elif macro.startswith("inc:"):
incfile = mudpy.data.find_file(macro[4:], universe=universe)
if os.path.exists(incfile):
- incfd = codecs.open(incfile, "r", "utf-8")
replacement = ""
- for line in incfd:
- if line.endswith("\n") and not line.endswith("\r\n"):
- line = line.replace("\n", "\r\n")
- replacement += line
- # lose the trailing eol
- replacement = replacement[:-2]
+ with codecs.open(incfile, "r", "utf-8") as incfd:
+ for line in incfd:
+ if line.endswith("\n") and not line.endswith("\r\n"):
+ line = line.replace("\n", "\r\n")
+ replacement += line
+ # lose the trailing eol
+ replacement = replacement[:-2]
else:
replacement = ""
log("Couldn't read included " + incfile + " file.", 7)
def on_pulse():
"""The things which should happen on each pulse, aside from reloads."""
- # open the listening socket if it hasn't been already
- if not hasattr(universe, "listening_socket"):
- universe.initialize_server_socket()
-
- # assign a user if a new connection is waiting
- user = check_for_connection(universe.listening_socket)
- if user:
- universe.userlist.append(user)
-
- # iterate over the connected users
- for user in universe.userlist:
- user.pulse()
-
- # add an element for counters if it doesn't exist
- if "counters" not in universe.groups.get("internal", {}):
- Element("internal.counters", universe)
+ # increase the elapsed increment counter
+ universe.set_time(universe.get_time() + 1)
# update the log every now and then
if not universe.groups["internal"]["counters"].get("mark"):
) - 1
)
+ # open the listening socket if it hasn't been already
+ if not hasattr(universe, "listening_socket"):
+ universe.initialize_server_socket()
+
+ # assign a user if a new connection is waiting
+ user = check_for_connection(universe.listening_socket)
+ if user:
+ universe.userlist.append(user)
+
+ # iterate over the connected users
+ for user in universe.userlist:
+ user.pulse()
+
# pause for a configurable amount of time (decimal seconds)
time.sleep(universe.contents["mudpy.timing"].get("increment"))
- # increase the elapsed increment counter
- universe.groups["internal"]["counters"].set(
- "elapsed", universe.groups["internal"]["counters"].get(
- "elapsed", 0
- ) + 1
- )
-
def reload_data():
"""Reload all relevant objects."""
universe.save()
old_userlist = universe.userlist[:]
+ old_loglines = universe.loglines[:]
for element in list(universe.contents.values()):
element.destroy()
- universe.load()
+ pending_loglines = universe.load()
+ new_loglines = universe.loglines[:]
+ universe.loglines = old_loglines + new_loglines + pending_loglines
for user in old_userlist:
user.reload()
return None
# note that we got one
- log("Connection from " + address[0], 2)
+ log("New connection from %s." % address[0], 2)
# disable blocking so we can proceed whether or not we can send/receive
connection.setblocking(0)
# create a new user object
user = User()
+ log("Instantiated %s for %s." % (user, address[0]), 0)
# associate this connection with it
user.connection = connection
return user
+def find_command(command_name):
+ """Try to find a command by name or abbreviation."""
+
+ # lowercase the command
+ command_name = command_name.lower()
+
+ command = None
+ if command_name in universe.groups["command"]:
+ # the command matches a command word for which we have data
+ command = universe.groups["command"][command_name]
+ else:
+ for candidate in sorted(universe.groups["command"]):
+ if candidate.startswith(command_name) and not universe.groups[
+ "command"][candidate].is_restricted():
+ # the command matches the start of a command word and is not
+ # restricted to administrators
+ command = universe.groups["command"][candidate]
+ break
+ return command
+
+
def get_menu(state, error=None, choices=None):
"""Show the correct menu text to a user."""
def get_menu_choices(user):
"""Return a dict of choice:meaning."""
- menu = universe.groups["menu"][user.state]
- create_choices = menu.get("create")
+ state = universe.groups["menu"][user.state]
+ create_choices = state.get("create")
if create_choices:
- choices = eval(create_choices)
+ choices = call_hook_function(create_choices, (user,))
else:
choices = {}
ignores = []
options = {}
creates = {}
- for facet in menu.facets():
- if facet.startswith("demand_") and not eval(
- universe.groups["menu"][user.state].get(facet)
- ):
+ for facet in state.facets():
+ if facet.startswith("demand_") and not call_hook_function(
+ universe.groups["menu"][user.state].get(facet), (user,)):
ignores.append(facet.split("_", 2)[1])
elif facet.startswith("create_"):
creates[facet] = facet.split("_", 2)[1]
options[facet] = facet.split("_", 2)[1]
for facet in creates.keys():
if not creates[facet] in ignores:
- choices[creates[facet]] = eval(menu.get(facet))
+ choices[creates[facet]] = call_hook_function(
+ state.get(facet), (user,))
for facet in options.keys():
if not options[facet] in ignores:
- choices[options[facet]] = menu.get(facet)
+ choices[options[facet]] = state.get(facet)
return choices
return universe.groups["menu"][state].get("branch")
-def get_choice_branch(user, choice):
+def get_choice_branch(user):
"""Returns the new state matching the given choice."""
branches = get_menu_branches(user.state)
- if choice in branches.keys():
- return branches[choice]
- elif choice in user.menu_choices.keys():
+ if user.choice in branches.keys():
+ return branches[user.choice]
+ elif user.choice in user.menu_choices.keys():
return get_default_branch(user.state)
else:
return ""
return universe.groups["menu"][state].get("action")
-def get_choice_action(user, choice):
+def get_choice_action(user):
"""Run any indicated script for the given choice."""
actions = get_menu_actions(user.state)
- if choice in actions.keys():
- return actions[choice]
- elif choice in user.menu_choices.keys():
+ if user.choice in actions.keys():
+ return actions[user.choice]
+ elif user.choice in user.menu_choices.keys():
return get_default_action(user.state)
else:
return ""
+def call_hook_function(fname, arglist):
+ """Safely execute named function with supplied arguments, return result."""
+
+ # all functions relative to mudpy package
+ function = mudpy
+
+ for component in fname.split("."):
+ try:
+ function = getattr(function, component)
+ except AttributeError:
+ log('Could not find mudpy.%s() for arguments "%s"'
+ % (fname, arglist), 7)
+ function = None
+ break
+ if function:
+ try:
+ return function(*arglist)
+ except Exception:
+ log('Calling mudpy.%s(%s) raised an exception...\n%s'
+ % (fname, (*arglist,), traceback.format_exc()), 7)
+
+
def handle_user_input(user):
"""The main handler, branches to a state-specific handler."""
user.send("", add_prompt=False, prepend_padding=False)
# check to make sure the state is expected, then call that handler
- if "handler_" + user.state in globals():
- exec("handler_" + user.state + "(user)")
- else:
+ try:
+ globals()["handler_" + user.state](user)
+ except KeyError:
generic_menu_handler(user)
# since we got input, flag that the menu/prompt needs to be redisplayed
# get a lower-case representation of the next line of input
if user.input_queue:
- choice = user.input_queue.pop(0)
- if choice:
- choice = choice.lower()
+ user.choice = user.input_queue.pop(0)
+ if user.choice:
+ user.choice = user.choice.lower()
else:
- choice = ""
- if not choice:
- choice = get_default_menu_choice(user.state)
- if choice in user.menu_choices:
- exec(get_choice_action(user, choice))
- new_state = get_choice_branch(user, choice)
+ user.choice = ""
+ if not user.choice:
+ user.choice = get_default_menu_choice(user.state)
+ if user.choice in user.menu_choices:
+ action = get_choice_action(user)
+ if action:
+ call_hook_function(action, (user,))
+ new_state = get_choice_branch(user)
if new_state:
user.state = new_state
else:
else:
command_name, parameters = first_word(input_data)
- # lowercase the command
- command_name = command_name.lower()
-
- # the command matches a command word for which we have data
- if command_name in universe.groups["command"]:
- command = universe.groups["command"][command_name]
- else:
- command = None
+ # expand to an actual command
+ command = find_command(command_name)
# if it's allowed, do it
+ result = None
if actor.can_run(command):
- exec(command.get("action"))
+ action_fname = command.get("action", command.key)
+ if action_fname:
+ result = call_hook_function(action_fname, (actor, parameters))
- # otherwise, give an error
- elif command_name:
- command_error(actor, input_data)
+ # if the command was not run, give an error
+ if not result:
+ mudpy.command.error(actor, input_data)
# if no input, just idle back with a prompt
else:
user.send("", just_prompt=True)
-def command_halt(actor, parameters):
- """Halt the world."""
- if actor.owner:
-
- # see if there's a message or use a generic one
- if parameters:
- message = "Halting: " + parameters
- else:
- message = "User " + actor.owner.account.get(
- "name"
- ) + " halted the world."
-
- # let everyone know
- broadcast(message, add_prompt=False)
- log(message, 8)
-
- # set a flag to terminate the world
- universe.terminate_flag = True
-
-
-def command_reload(actor):
- """Reload all code modules, configs and data."""
- if actor.owner:
-
- # let the user know and log
- actor.send("Reloading all code modules, configs and data.")
- log(
- "User " +
- actor.owner.account.get("name") + " reloaded the world.",
- 6
- )
-
- # set a flag to reload
- universe.reload_flag = True
-
-
-def command_quit(actor):
- """Leave the world and go back to the main menu."""
- if actor.owner:
- actor.owner.state = "main_utility"
- actor.owner.deactivate_avatar()
-
-
-def command_help(actor, parameters):
- """List available commands and provide help for commands."""
-
- # did the user ask for help on a specific command word?
- if parameters and actor.owner:
-
- # is the command word one for which we have data?
- if parameters in universe.groups["command"]:
- command = universe.groups["command"][parameters]
- else:
- command = None
-
- # only for allowed commands
- if actor.can_run(command):
-
- # add a description if provided
- description = command.get("description")
- if not description:
- description = "(no short description provided)"
- if command.get("administrative"):
- output = "$(red)"
- else:
- output = "$(grn)"
- output += parameters + "$(nrm) - " + description + "$(eol)$(eol)"
-
- # add the help text if provided
- help_text = command.get("help")
- if not help_text:
- help_text = "No help is provided for this command."
- output += help_text
-
- # list related commands
- see_also = command.get("see_also")
- if see_also:
- really_see_also = ""
- for item in see_also:
- if item in universe.groups["command"]:
- command = universe.groups["command"][item]
- if actor.can_run(command):
- if really_see_also:
- really_see_also += ", "
- if command.get("administrative"):
- really_see_also += "$(red)"
- else:
- really_see_also += "$(grn)"
- really_see_also += item + "$(nrm)"
- if really_see_also:
- output += "$(eol)$(eol)See also: " + really_see_also
-
- # no data for the requested command word
- else:
- output = "That is not an available command."
-
- # no specific command word was indicated
- else:
-
- # give a sorted list of commands with descriptions if provided
- output = "These are the commands available to you:$(eol)$(eol)"
- sorted_commands = list(universe.groups["command"].keys())
- sorted_commands.sort()
- for item in sorted_commands:
- command = universe.groups["command"][item]
- if actor.can_run(command):
- description = command.get("description")
- if not description:
- description = "(no short description provided)"
- if command.get("administrative"):
- output += " $(red)"
- else:
- output += " $(grn)"
- output += item + "$(nrm) - " + description + "$(eol)"
- output += ('$(eol)Enter "help COMMAND" for help on a command '
- 'named "COMMAND".')
-
- # send the accumulated output to the user
- actor.send(output)
-
-
-def command_move(actor, parameters):
- """Move the avatar in a given direction."""
- if parameters in universe.contents[actor.get("location")].portals():
- actor.move_direction(parameters)
- else:
- actor.send("You cannot go that way.")
-
-
-def command_look(actor, parameters):
- """Look around."""
- if parameters:
- actor.send("You can't look at or in anything yet.")
- else:
- actor.look_at(actor.get("location"))
-
-
-def command_say(actor, parameters):
- """Speak to others in the same area."""
-
- # check for replacement macros and escape them
- parameters = escape_macros(parameters)
-
- # if the message is wrapped in quotes, remove them and leave contents
- # intact
- if parameters.startswith('"') and parameters.endswith('"'):
- message = parameters[1:-1]
- literal = True
-
- # otherwise, get rid of stray quote marks on the ends of the message
- else:
- message = parameters.strip('''"'`''')
- literal = False
-
- # the user entered a message
- if message:
-
- # match the punctuation used, if any, to an action
- if "mudpy.linguistic" in universe.contents:
- actions = universe.contents["mudpy.linguistic"].get("actions", {})
- default_punctuation = (universe.contents["mudpy.linguistic"].get(
- "default_punctuation", "."))
- else:
- actions = {}
- default_punctuation = "."
- action = ""
-
- # reverse sort punctuation options so the longest match wins
- for mark in sorted(actions.keys(), reverse=True):
- if not literal and message.endswith(mark):
- action = actions[mark]
- break
-
- # add punctuation if needed
- if not action:
- action = actions[default_punctuation]
- if message and not (
- literal or unicodedata.category(message[-1]) == "Po"
- ):
- message += default_punctuation
-
- # failsafe checks to avoid unwanted reformatting and null strings
- if message and not literal:
-
- # decapitalize the first letter to improve matching
- message = message[0].lower() + message[1:]
-
- # iterate over all words in message, replacing typos
- if "mudpy.linguistic" in universe.contents:
- typos = universe.contents["mudpy.linguistic"].get("typos", {})
- else:
- typos = {}
- words = message.split()
- for index in range(len(words)):
- word = words[index]
- while unicodedata.category(word[0]) == "Po":
- word = word[1:]
- while unicodedata.category(word[-1]) == "Po":
- word = word[:-1]
- if word in typos.keys():
- words[index] = words[index].replace(word, typos[word])
- message = " ".join(words)
-
- # capitalize the first letter
- message = message[0].upper() + message[1:]
-
- # tell the area
- if message:
- actor.echo_to_location(
- actor.get("name") + " " + action + 's, "' + message + '"'
- )
- actor.send("You " + action + ', "' + message + '"')
-
- # there was no message
- else:
- actor.send("What do you want to say?")
-
-
-def command_chat(actor):
- """Toggle chat mode."""
- mode = actor.get("mode")
- if not mode:
- actor.set("mode", "chat")
- actor.send("Entering chat mode (use $(grn)!chat$(nrm) to exit).")
- elif mode == "chat":
- actor.remove_facet("mode")
- actor.send("Exiting chat mode.")
- else:
- actor.send("Sorry, but you're already busy with something else!")
-
-
-def command_show(actor, parameters):
- """Show program data."""
- message = ""
- arguments = parameters.split()
- if not parameters:
- message = "What do you want to show?"
- elif arguments[0] == "version":
- message = repr(universe.versions)
- elif arguments[0] == "time":
- message = universe.groups["internal"]["counters"].get(
- "elapsed"
- ) + " increments elapsed since the world was created."
- elif arguments[0] == "groups":
- message = "These are the element groups:$(eol)"
- groups = list(universe.groups.keys())
- groups.sort()
- for group in groups:
- message += "$(eol) $(grn)" + group + "$(nrm)"
- elif arguments[0] == "files":
- message = "These are the current files containing the universe:$(eol)"
- filenames = sorted(universe.files)
- for filename in filenames:
- if universe.files[filename].is_writeable():
- status = "rw"
- else:
- status = "ro"
- message += ("$(eol) $(red)(%s) $(grn)%s$(nrm)" %
- (status, filename))
- if universe.files[filename].flags:
- message += (" $(yel)[%s]$(nrm)" %
- ",".join(universe.files[filename].flags))
- elif arguments[0] == "group":
- if len(arguments) != 2:
- message = "You must specify one group."
- elif arguments[1] in universe.groups:
- message = ('These are the elements in the "' + arguments[1]
- + '" group:$(eol)')
- elements = [
- (
- universe.groups[arguments[1]][x].key
- ) for x in universe.groups[arguments[1]].keys()
- ]
- elements.sort()
- for element in elements:
- message += "$(eol) $(grn)" + element + "$(nrm)"
- else:
- message = 'Group "' + arguments[1] + '" does not exist.'
- elif arguments[0] == "file":
- if len(arguments) != 2:
- message = "You must specify one file."
- elif arguments[1] in universe.files:
- message = ('These are the nodes in the "' + arguments[1]
- + '" file:$(eol)')
- elements = sorted(universe.files[arguments[1]].data)
- for element in elements:
- message += "$(eol) $(grn)" + element + "$(nrm)"
- else:
- message = 'File "%s" does not exist.' % arguments[1]
- elif arguments[0] == "element":
- if len(arguments) != 2:
- message = "You must specify one element."
- elif arguments[1].strip(".") in universe.contents:
- element = universe.contents[arguments[1].strip(".")]
- message = ('These are the properties of the "' + arguments[1]
- + '" element (in "' + element.origin.source
- + '"):$(eol)')
- facets = element.facets()
- for facet in sorted(facets):
- message += ("$(eol) $(grn)%s: $(red)%s$(nrm)" %
- (facet, str(facets[facet])))
- else:
- message = 'Element "' + arguments[1] + '" does not exist.'
- elif arguments[0] == "result":
- if len(arguments) < 2:
- message = "You need to specify an expression."
- else:
- try:
- message = repr(eval(" ".join(arguments[1:])))
- except Exception as e:
- message = ("$(red)Your expression raised an exception...$(eol)"
- "$(eol)$(bld)%s$(nrm)" % e)
- elif arguments[0] == "log":
- if len(arguments) == 4:
- if re.match(r"^\d+$", arguments[3]) and int(arguments[3]) >= 0:
- stop = int(arguments[3])
- else:
- stop = -1
- else:
- stop = 0
- if len(arguments) >= 3:
- if re.match(r"^\d+$", arguments[2]) and int(arguments[2]) > 0:
- start = int(arguments[2])
- else:
- start = -1
- else:
- start = 10
- if len(arguments) >= 2:
- if (re.match(r"^\d+$", arguments[1])
- and 0 <= int(arguments[1]) <= 9):
- level = int(arguments[1])
- else:
- level = -1
- elif 0 <= actor.owner.account.get("loglevel", 0) <= 9:
- level = actor.owner.account.get("loglevel", 0)
- else:
- level = 1
- if level > -1 and start > -1 and stop > -1:
- message = get_loglines(level, start, stop)
- else:
- message = ("When specified, level must be 0-9 (default 1), "
- "start and stop must be >=1 (default 10 and 1).")
- else:
- message = '''I don't know what "''' + parameters + '" is.'
- actor.send(message)
-
-
-def command_create(actor, parameters):
- """Create an element if it does not exist."""
- if not parameters:
- message = "You must at least specify an element to create."
- elif not actor.owner:
- message = ""
- else:
- arguments = parameters.split()
- if len(arguments) == 1:
- arguments.append("")
- if len(arguments) == 2:
- element, filename = arguments
- if element in universe.contents:
- message = 'The "' + element + '" element already exists.'
- else:
- message = ('You create "' +
- element + '" within the universe.')
- logline = actor.owner.account.get(
- "name"
- ) + " created an element: " + element
- if filename:
- logline += " in file " + filename
- if filename not in universe.files:
- message += (
- ' Warning: "' + filename + '" is not yet '
- "included in any other file and will not be read "
- "on startup unless this is remedied.")
- Element(element, universe, filename)
- log(logline, 6)
- elif len(arguments) > 2:
- message = "You can only specify an element and a filename."
- actor.send(message)
-
-
-def command_destroy(actor, parameters):
- """Destroy an element if it exists."""
- if actor.owner:
- if not parameters:
- message = "You must specify an element to destroy."
- else:
- if parameters not in universe.contents:
- message = 'The "' + parameters + '" element does not exist.'
- else:
- universe.contents[parameters].destroy()
- message = ('You destroy "' + parameters
- + '" within the universe.')
- log(
- actor.owner.account.get(
- "name"
- ) + " destroyed an element: " + parameters,
- 6
- )
- actor.send(message)
-
-
-def command_set(actor, parameters):
- """Set a facet of an element."""
- if not parameters:
- message = "You must specify an element, a facet and a value."
- else:
- arguments = parameters.split(" ", 2)
- if len(arguments) == 1:
- message = ('What facet of element "' + arguments[0]
- + '" would you like to set?')
- elif len(arguments) == 2:
- message = ('What value would you like to set for the "' +
- arguments[1] + '" facet of the "' + arguments[0]
- + '" element?')
- else:
- element, facet, value = arguments
- if element not in universe.contents:
- message = 'The "' + element + '" element does not exist.'
- else:
- try:
- universe.contents[element].set(facet, value)
- except PermissionError:
- message = ('The "%s" element is kept in read-only file '
- '"%s" and cannot be altered.' %
- (element, universe.contents[
- element].origin.source))
- except ValueError:
- message = ('Value "%s" of type "%s" cannot be coerced '
- 'to the correct datatype for facet "%s".' %
- (value, type(value), facet))
- else:
- message = ('You have successfully (re)set the "' + facet
- + '" facet of element "' + element
- + '". Try "show element ' +
- element + '" for verification.')
- actor.send(message)
-
-
-def command_delete(actor, parameters):
- """Delete a facet from an element."""
- if not parameters:
- message = "You must specify an element and a facet."
- else:
- arguments = parameters.split(" ")
- if len(arguments) == 1:
- message = ('What facet of element "' + arguments[0]
- + '" would you like to delete?')
- elif len(arguments) != 2:
- message = "You may only specify an element and a facet."
- else:
- element, facet = arguments
- if element not in universe.contents:
- message = 'The "' + element + '" element does not exist.'
- elif facet not in universe.contents[element].facets():
- message = ('The "' + element + '" element has no "' + facet
- + '" facet.')
- else:
- universe.contents[element].remove_facet(facet)
- message = ('You have successfully deleted the "' + facet
- + '" facet of element "' + element
- + '". Try "show element ' +
- element + '" for verification.')
- actor.send(message)
-
-
-def command_error(actor, input_data):
- """Generic error for an unrecognized command word."""
-
- # 90% of the time use a generic error
- if random.randrange(10):
- message = '''I'm not sure what "''' + input_data + '''" means...'''
-
- # 10% of the time use the classic diku error
- else:
- message = "Arglebargle, glop-glyf!?!"
-
- # send the error message
- actor.send(message)
-
-
def daemonize(universe):
"""Fork and disassociate from everything."""
if file_name:
if not os.path.isabs(file_name):
file_name = os.path.join(universe.startdir, file_name)
+ os.makedirs(os.path.dirname(file_name), exist_ok=True)
file_descriptor = codecs.open(file_name, "w", "utf-8")
file_descriptor.write(pid + "\n")
file_descriptor.flush()
log("Import path: %s" % ", ".join(sys.path), 1)
log("Installed dependencies: %s" % universe.versions.dependencies_text, 1)
log("Other python packages: %s" % universe.versions.environment_text, 1)
- log("Started %s with command line: %s" % (
- universe.versions.version, " ".join(sys.argv)), 1)
+ log("Running version: %s" % universe.versions.version, 1)
+ log("Initial directory: %s" % universe.startdir, 1)
+ log("Command line: %s" % " ".join(sys.argv), 1)
# pass the initialized universe back
return universe