You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
669 lines
20 KiB
669 lines
20 KiB
# Imports
|
|
|
|
from commonkit import indent, parse_jinja_template, read_file, split_csv
|
|
from jinja2 import TemplateNotFound, TemplateError
|
|
import logging
|
|
import os
|
|
from ...variables import LOGGER_NAME
|
|
|
|
log = logging.getLogger(LOGGER_NAME)
|
|
|
|
# Exports
|
|
|
|
__all__ = (
|
|
"Command",
|
|
"Content",
|
|
"ItemizedCommand",
|
|
"MultipleCommands",
|
|
"Prompt",
|
|
"Sudo",
|
|
"Template",
|
|
)
|
|
|
|
# Classes
|
|
|
|
|
|
class Command(object):
|
|
|
|
def __init__(self, statement, cd=None, comment=None, condition=None, name=None, prefix=None, register=None, stop=False, sudo=None, tags=None, **kwargs):
|
|
self.cd = cd
|
|
self.comment = comment
|
|
self.condition = condition
|
|
self.name = name
|
|
self.number = None
|
|
self.prefix = prefix
|
|
self.register = register
|
|
self.statement = statement
|
|
self.stop = stop
|
|
self.tags = tags or list()
|
|
|
|
if isinstance(sudo, Sudo):
|
|
self.sudo = sudo
|
|
elif type(sudo) is str:
|
|
self.sudo = Sudo(enabled=True, user=sudo)
|
|
elif sudo is True:
|
|
self.sudo = Sudo(enabled=True)
|
|
else:
|
|
self.sudo = Sudo()
|
|
|
|
self.options = kwargs
|
|
|
|
def __getattr__(self, item):
|
|
return self.options.get(item)
|
|
|
|
def __repr__(self):
|
|
if self.comment:
|
|
return "<%s %s>" % (self.__class__.__name__, self.comment)
|
|
|
|
return "<%s>" % self.__class__.__name__
|
|
|
|
def get_statement(self, cd=True, include_comment=True, include_register=True, include_stop=True):
|
|
"""Get the full statement.
|
|
|
|
:param cd: Include the directory change, if given.
|
|
:type cd: bool
|
|
|
|
:param suppress_comment: Don't include the comment.
|
|
:type suppress_comment: bool
|
|
|
|
:rtype: str
|
|
|
|
"""
|
|
a = list()
|
|
|
|
if cd and self.cd is not None:
|
|
a.append("( cd %s &&" % self.cd)
|
|
|
|
if self.prefix is not None:
|
|
a.append("%s &&" % self.prefix)
|
|
|
|
if self.sudo:
|
|
statement = "%s %s" % (self.sudo, self._get_statement())
|
|
else:
|
|
statement = self._get_statement()
|
|
|
|
a.append("%s" % statement)
|
|
|
|
if cd and self.cd is not None:
|
|
a.append(")")
|
|
|
|
b = list()
|
|
if self.comment is not None and include_comment:
|
|
b.append("# %s" % self.comment)
|
|
|
|
if self.condition is not None:
|
|
b.append("if [[ %s ]]; then %s; fi;" % (self.condition, " ".join(a)))
|
|
else:
|
|
b.append(" ".join(a))
|
|
|
|
if self.register is not None and include_register:
|
|
b.append("%s=$?;" % self.register)
|
|
|
|
if self.stop and include_stop:
|
|
b.append("if [[ $%s -gt 0 ]]; exit 1; fi;" % self.register)
|
|
elif self.stop and include_stop:
|
|
b.append("if [[ $? -gt 0 ]]; exit 1; fi;")
|
|
else:
|
|
pass
|
|
|
|
return "\n".join(b)
|
|
|
|
@property
|
|
def is_itemized(self):
|
|
"""Always returns ``False``."""
|
|
return False
|
|
|
|
def _get_statement(self):
|
|
"""By default, get the statement passed upon command initialization.
|
|
|
|
:rtype: str
|
|
|
|
"""
|
|
return self.statement
|
|
|
|
|
|
class Content(object):
|
|
|
|
def __init__(self, content_type, caption=None, css=None, heading=None, height=None, image=None, message=None,
|
|
width=None, **kwargs):
|
|
self.caption = caption
|
|
self.css = css
|
|
self.heading = heading
|
|
self.height = height
|
|
self.image = image
|
|
self.message = message
|
|
self.width = width
|
|
self.name = "content"
|
|
self.type = content_type
|
|
|
|
self.kwargs = kwargs
|
|
|
|
def __getattr__(self, item):
|
|
return self.kwargs.get(item)
|
|
|
|
def __repr__(self):
|
|
return "<%s %s>" % (self.__class__.__name__, self.content_type)
|
|
|
|
@property
|
|
def is_itemized(self):
|
|
"""Always returns ``False``. Content cannot be itemized."""
|
|
return False
|
|
|
|
def get_output(self, output_format):
|
|
if self.content_type == "explain":
|
|
return self._get_message_output(output_format)
|
|
elif self.content_type == "screenshot":
|
|
return self._get_image_output(output_format)
|
|
else:
|
|
log.warning("Invalid content type: %s" % self.content_type)
|
|
return None
|
|
|
|
# noinspection PyUnusedLocal
|
|
def get_statement(self, cd=True, include_comment=True, include_register=True, include_stop=True):
|
|
"""Override to return only bash output format."""
|
|
return self.get_output("bash")
|
|
|
|
def _get_image_output(self, output_format):
|
|
"""Get the output of the content when the content type is an image."""
|
|
a = list()
|
|
|
|
if output_format == "bash":
|
|
if self.caption:
|
|
a.append("# %s" % self.caption)
|
|
|
|
a.append("# %s" % self.image)
|
|
elif output_format == "html":
|
|
b = list()
|
|
b.append('<img src="%s"' % self.image)
|
|
b.append('alt="%s"' % self.caption or self.comment)
|
|
|
|
if self.css is not None:
|
|
b.append('class="%s"' % self.css)
|
|
|
|
if self.height is not None:
|
|
b.append('height="%s"' % self.height)
|
|
|
|
if self.width is not None:
|
|
b.append('width="%s"' % self.width)
|
|
|
|
a.append(" ".join(b) + ">")
|
|
elif output_format == "md":
|
|
a.append("" % (self.caption or self.comment, self.image))
|
|
a.append("")
|
|
elif output_format == "rst":
|
|
a.append(".. figure:: %s" % self.image)
|
|
|
|
if self.caption is not None:
|
|
a.append(indent(":alt: %s" % self.caption, 8))
|
|
|
|
if self.height is not None:
|
|
a.append(indent(":height: %s" % self.height, 8))
|
|
|
|
if self.height is not None:
|
|
a.append(indent(":height: %s" % self.height, 8))
|
|
else:
|
|
if self.caption:
|
|
a.append("%s: %s" % (self.caption, self.image))
|
|
else:
|
|
a.append(self.image)
|
|
|
|
return "\n".join(a)
|
|
|
|
def _get_message_output(self, output_format):
|
|
"""Get the output of the content when the content type is a message."""
|
|
a = list()
|
|
|
|
if output_format == "bash":
|
|
if self.heading:
|
|
a.append("# %s: %s" % (self.heading, self.message))
|
|
else:
|
|
a.append("# %s" % self.message)
|
|
|
|
a.append("")
|
|
elif output_format == "html":
|
|
if self.heading:
|
|
a.append("<h2>%s</h2>" % self.heading)
|
|
|
|
a.append("<p>%s</p>" % self.message)
|
|
elif output_format == "md":
|
|
if self.heading:
|
|
a.append("## %s" % self.heading)
|
|
a.append("")
|
|
|
|
a.append(self.message)
|
|
a.append("")
|
|
elif output_format == "rst":
|
|
if self.heading:
|
|
a.append(self.heading)
|
|
a.append("=" * len(self.heading))
|
|
a.append("")
|
|
|
|
a.append(self.message)
|
|
a.append("")
|
|
else:
|
|
if self.heading:
|
|
a.append("***** %s *****" % self.heading)
|
|
a.append("")
|
|
|
|
a.append(self.message)
|
|
a.append("")
|
|
|
|
return "\n".join(a)
|
|
|
|
|
|
class ItemizedCommand(object):
|
|
"""An itemized command represents multiple commands of with the same statement but different parameters."""
|
|
|
|
def __init__(self, callback, items, *args, **kwargs):
|
|
"""Initialize the command.
|
|
|
|
:param callback: The function to be used to generate the command.
|
|
|
|
:param items: The command arguments.
|
|
:type items: list[str]
|
|
|
|
:param name: The name of the command from the mapping. Not used and not required for programmatic use, but
|
|
automatically assigned during factory instantiation.
|
|
:type name: str
|
|
|
|
:param args: The itemized arguments. ``$item`` should be included.
|
|
|
|
Keyword arguments are passed to the command class upon instantiation.
|
|
|
|
"""
|
|
self.args = args
|
|
self.callback = callback
|
|
self.items = items
|
|
self.name = None
|
|
|
|
self.kwargs = kwargs
|
|
|
|
# Set defaults for when ItemizedCommand is referenced directly before individual commands are instantiated. For
|
|
# example, when command filtering occurs.
|
|
self.kwargs.setdefault("tags", list())
|
|
|
|
def __getattr__(self, item):
|
|
return self.kwargs.get(item)
|
|
|
|
def __repr__(self):
|
|
return "<%s %s>" % (self.__class__.__name__, self.callback.__name__)
|
|
|
|
def get_commands(self):
|
|
"""Get the commands to be executed.
|
|
|
|
:rtype: list[BaseType(Command)]
|
|
|
|
"""
|
|
kwargs = self.kwargs.copy()
|
|
kwargs['name'] = self.name
|
|
|
|
a = list()
|
|
for item in self.items:
|
|
args = list()
|
|
for arg in self.args:
|
|
args.append(arg.replace("$item", item))
|
|
|
|
command = self.callback(*args, **kwargs)
|
|
a.append(command)
|
|
|
|
return a
|
|
|
|
def get_statement(self, cd=True, include_comment=True, include_register=True, include_stop=True):
|
|
"""Override to get multiple commands."""
|
|
kwargs = self.kwargs.copy()
|
|
comment = kwargs.pop("comment", "execute multiple commands")
|
|
|
|
a = list()
|
|
|
|
if include_comment:
|
|
a.append("# %s" % comment)
|
|
|
|
commands = self.get_commands()
|
|
for c in commands:
|
|
a.append(c.get_statement(cd=cd, include_comment=False, include_register=include_register, include_stop=include_stop))
|
|
|
|
return "\n".join(a)
|
|
|
|
@property
|
|
def is_itemized(self):
|
|
"""Always returns ``True``."""
|
|
return True
|
|
|
|
|
|
class MultipleCommands(object):
|
|
"""An interface for handling command composition where multiple statements must be generated."""
|
|
|
|
def __init__(self, commands, comment=None, tags=None, **kwargs):
|
|
self.commands = commands
|
|
self.comment = comment or "run multiple commands"
|
|
self.tags = tags or list()
|
|
|
|
self.options = kwargs
|
|
|
|
def __getattr__(self, item):
|
|
return self.options.get(item)
|
|
|
|
def __iter__(self):
|
|
return iter(self.commands)
|
|
|
|
def __repr__(self):
|
|
return "<%s %s>" % (self.__class__.__name__, self.comment)
|
|
|
|
def get_statement(self, cd=True, include_comment=True, include_register=True, include_stop=True):
|
|
a = list()
|
|
if include_comment:
|
|
a.append("# %s" % self.comment)
|
|
|
|
for command in self.commands:
|
|
statement = command.get_statement(
|
|
cd=cd,
|
|
include_comment=False,
|
|
include_register=include_register,
|
|
include_stop=include_stop
|
|
)
|
|
a.append(statement)
|
|
|
|
return "\n".join(a)
|
|
|
|
|
|
class Prompt(Command):
|
|
"""Prompt the user for input."""
|
|
|
|
def __init__(self, name, back_title="Input", choices=None, default=None, dialog=False, help_text=None, label=None,
|
|
**kwargs):
|
|
"""Initialize a prompt for user input.
|
|
|
|
:param name: The variable name.
|
|
:type name: str
|
|
|
|
:param back_title: The back title of the input. Used only when ``dialog`` is enabled.
|
|
:type back_title: str
|
|
|
|
:param choices: Valid choices for the variable. May be given as a list of strings or a comma separated string.
|
|
:type choices: list[str] | str
|
|
|
|
:param default: The default value of the variable.
|
|
|
|
:param dialog: Indicates the dialog command should be used.
|
|
:type dialog: bool
|
|
|
|
:param help_text: Additional text to display. Only use when ``fancy`` is ``True``.
|
|
:type help_text: str
|
|
|
|
:param label: The label of the prompt.
|
|
|
|
"""
|
|
self.back_title = back_title
|
|
self.default = default
|
|
self.dialog_enabled = dialog
|
|
self.help_text = help_text
|
|
self.label = label or name.replace("_", " ").title()
|
|
self.variable_name = name
|
|
|
|
if type(choices) in (list, tuple):
|
|
self.choices = choices
|
|
elif type(choices) is str:
|
|
self.choices = split_csv(choices, smart=False)
|
|
else:
|
|
self.choices = None
|
|
|
|
kwargs.setdefault("comment", "prompt user for %s input" % name)
|
|
|
|
super().__init__(name, **kwargs)
|
|
|
|
def get_statement(self, cd=True, include_comment=True, include_register=True, include_stop=True):
|
|
"""Get the statement using dialog or read."""
|
|
if self.dialog_enabled:
|
|
return self._get_dialog_statement()
|
|
|
|
return self._get_read_statement()
|
|
|
|
def _get_dialog_statement(self):
|
|
"""Get the dialog statement."""
|
|
a = list()
|
|
|
|
a.append('dialog --clear --backtitle "%s" --title "%s"' % (self.back_title, self.label))
|
|
|
|
if self.choices is not None:
|
|
a.append('--menu "%s" 15 40 %s' % (self.help_text or "Select", len(self.choices)))
|
|
count = 1
|
|
for choice in self.choices:
|
|
a.append('"%s" %s' % (choice, count))
|
|
count += 1
|
|
|
|
a.append('2>/tmp/input.txt')
|
|
else:
|
|
if self.help_text is not None:
|
|
a.append('--inputbox "%s"' % self.help_text)
|
|
else:
|
|
a.append('--inputbox ""')
|
|
|
|
a.append('8 60 2>/tmp/input.txt')
|
|
|
|
b = list()
|
|
|
|
b.append('touch /tmp/input.txt')
|
|
b.append(" ".join(a))
|
|
|
|
b.append('%s=$(</tmp/input.txt)' % self.variable_name)
|
|
b.append('clear')
|
|
b.append('rm /tmp/input.txt')
|
|
|
|
if self.default is not None:
|
|
b.append('if [[ -z "$%s" ]]; then %s="%s"; fi;' % (self.variable_name, self.variable_name, self.default))
|
|
|
|
# b.append('echo "$%s"' % self.name)
|
|
|
|
return "\n".join(b)
|
|
|
|
def _get_read_statement(self):
|
|
"""Get the standard read statement."""
|
|
a = list()
|
|
|
|
if self.choices is not None:
|
|
a.append('echo "%s "' % self.label)
|
|
|
|
options = list()
|
|
for choice in self.choices:
|
|
options.append('"%s"' % choice)
|
|
|
|
a.append('options=(%s)' % " ".join(options))
|
|
a.append('select opt in "${options[@]}"')
|
|
a.append('do')
|
|
a.append(' case $opt in')
|
|
|
|
for choice in self.choices:
|
|
a.append(' "%s") %s=$opt; break;;' % (choice, self.variable_name))
|
|
|
|
# a.append(' %s) %s=$opt;;' % ("|".join(self.choices), self.name))
|
|
a.append(' *) echo "invalid choice";;')
|
|
a.append(' esac')
|
|
a.append('done')
|
|
|
|
# a.append("read %s" % self.name)
|
|
else:
|
|
a.append('echo -n "%s "' % self.label)
|
|
a.append("read %s" % self.variable_name)
|
|
|
|
if self.default is not None:
|
|
a.append('if [[ -z "$%s" ]]; then %s="%s"; fi;' % (self.variable_name, self.variable_name, self.default))
|
|
|
|
# a.append('echo "$%s"' % self.name)
|
|
|
|
return "\n".join(a)
|
|
|
|
|
|
class Sudo(object):
|
|
"""Helper class for defining sudo options."""
|
|
|
|
def __init__(self, enabled=False, user="root"):
|
|
"""Initialize the helper.
|
|
|
|
:param enabled: Indicates sudo is enabled.
|
|
:type enabled: bool
|
|
|
|
:param user: The user to be invoked.
|
|
:type user: str
|
|
|
|
"""
|
|
self.enabled = enabled
|
|
self.user = user
|
|
|
|
def __bool__(self):
|
|
return self.enabled
|
|
|
|
def __str__(self):
|
|
if self.enabled:
|
|
return "sudo -u %s" % self.user
|
|
|
|
return ""
|
|
|
|
|
|
class Template(object):
|
|
|
|
PARSER_JINJA = "jinja2"
|
|
PARSER_PYTHON = "python"
|
|
PARSER_SIMPLE = "simple"
|
|
|
|
def __init__(self, source, target, backup=True, parser=PARSER_JINJA, **kwargs):
|
|
self.backup_enabled = backup
|
|
self.context = kwargs.pop("context", dict())
|
|
self.name = "template"
|
|
self.parser = parser
|
|
self.language = kwargs.pop("lang", None)
|
|
self.locations = kwargs.pop("locations", list())
|
|
self.source = os.path.expanduser(source)
|
|
self.target = target
|
|
|
|
sudo = kwargs.pop("sudo", None)
|
|
if isinstance(sudo, Sudo):
|
|
self.sudo = sudo
|
|
elif type(sudo) is str:
|
|
self.sudo = Sudo(enabled=True, user=sudo)
|
|
elif sudo is True:
|
|
self.sudo = Sudo(enabled=True)
|
|
else:
|
|
self.sudo = Sudo()
|
|
|
|
self.kwargs = kwargs
|
|
|
|
def __getattr__(self, item):
|
|
return self.kwargs.get(item)
|
|
|
|
# def __str__(self):
|
|
# return "template"
|
|
|
|
def __repr__(self):
|
|
return "<%s %s>" % (self.__class__.__name__, self.source)
|
|
|
|
def get_content(self):
|
|
"""Parse the template.
|
|
|
|
:rtype: str | None
|
|
|
|
"""
|
|
template = self.get_template()
|
|
|
|
if self.parser == self.PARSER_SIMPLE:
|
|
content = read_file(template)
|
|
for key, value in self.context.items():
|
|
replace = "$%s$" % key
|
|
content = content.replace(replace, str(value))
|
|
|
|
return content
|
|
|
|
if self.parser == self.PARSER_PYTHON:
|
|
content = read_file(template)
|
|
return content % self.context
|
|
|
|
try:
|
|
return parse_jinja_template(template, self.context)
|
|
except TemplateNotFound:
|
|
log.error("Template not found: %s" % template)
|
|
return None
|
|
except TemplateError as e:
|
|
log.error("Could not parse %s template: %s" % (template, e))
|
|
return None
|
|
|
|
# noinspection PyUnusedLocal
|
|
def get_statement(self, cd=True, include_comment=True, include_register=True, include_stop=True):
|
|
lines = list()
|
|
if include_comment and self.comment is not None:
|
|
lines.append("# %s" % self.comment)
|
|
|
|
# TODO: Backing up a template's target is currently specific to bash.
|
|
if self.backup_enabled:
|
|
command = "%s mv %s %s.b" % (self.sudo, self.target, self.target)
|
|
lines.append('if [[ -f "%s" ]]; then %s; fi;' % (self.target, command.lstrip()))
|
|
|
|
# Get the content; e.g. parse the template.
|
|
content = self.get_content()
|
|
if content is None:
|
|
lines.append("# NOT CONTENT AVAILABLE")
|
|
return "\n".join(lines)
|
|
|
|
# Templates that are bash scripts will fail to write because of the shebang.
|
|
if content.startswith("#!"):
|
|
_content = content.split("\n")
|
|
first_line = _content.pop(0)
|
|
command = '%s echo "%s" > %s' % (self.sudo, first_line, self.target)
|
|
lines.append(command.lstrip())
|
|
command = "%s cat > %s << EOF" % (self.sudo, self.target)
|
|
lines.append(command.lstrip())
|
|
lines.append("\n".join(_content))
|
|
lines.append("EOF")
|
|
else:
|
|
command = "%s cat > %s << EOF" % (self.sudo, self.target)
|
|
lines.append(command.lstrip())
|
|
lines.append(content)
|
|
lines.append("EOF")
|
|
|
|
if include_register and self.register is not None:
|
|
lines.append("%s=$?;" % self.register)
|
|
|
|
if include_stop and self.stop:
|
|
lines.append("if [[ $%s -gt 0 ]]; exit 1; fi;" % self.register)
|
|
elif include_stop and self.stop:
|
|
lines.append("if [[ $? -gt 0 ]]; exit 1; fi;")
|
|
else:
|
|
pass
|
|
|
|
return "\n".join(lines)
|
|
|
|
def get_target_language(self):
|
|
if self.language is not None:
|
|
return self.language
|
|
|
|
if self.target.endswith(".conf"):
|
|
return "conf"
|
|
elif self.target.endswith(".ini"):
|
|
return "ini"
|
|
elif self.target.endswith(".php"):
|
|
return "php"
|
|
elif self.target.endswith(".py"):
|
|
return "python"
|
|
elif self.target.endswith(".sh"):
|
|
return "bash"
|
|
elif self.target.endswith(".yml"):
|
|
return "yaml"
|
|
else:
|
|
return "text"
|
|
|
|
def get_template(self):
|
|
"""Get the template path.
|
|
|
|
:rtype: str
|
|
|
|
"""
|
|
source = self.source
|
|
for location in self.locations:
|
|
_source = os.path.join(location, self.source)
|
|
if os.path.exists(_source):
|
|
return _source
|
|
|
|
return source
|
|
|
|
@property
|
|
def is_itemized(self):
|
|
# return "$item" in self.target
|
|
return False
|
|
|