src.command_line_session

  1import logging
  2import os
  3import shlex
  4import readline #type: ignore
  5from pathlib import Path
  6import src.constants as cst
  7from src.cmd_types.meta import CommandMetadata
  8from src.cmd_types.output import CommandOutput
  9from src.extra import utils
 10from src.extra.plugins_loader import PluginLoader
 11from src.extra.utils import log_error
 12from src.plugins.plugin_default import LsCommand
 13
 14HANDLED_ERRORS = tuple(cst.ERROR_HANDLERS_MESSAGES_FORMATS.keys())
 15
 16class CommandLineSession:
 17
 18    """
 19    IO stream handler.
 20
 21    :param default_wd: Default working directory.
 22    :type default_wd: str | Path
 23
 24    :param plugins_dir: Directory to look for plugins.
 25    :type plugins_dir: str | Path
 26
 27    :param plugins_prefix: Prefix that plugins file must start with
 28    :type plugins_prefix: str
 29
 30    :param strict_load: If set to False, will not raise any error while importing plugins
 31    :type strict_load: bool
 32    """
 33
 34    def __init__(self, default_wd: str | Path = cst.DEFAULT_PWD,
 35                 plugins_dir: str = cst.PLUGINS_DIR,
 36                 plugins_prefix: str = cst.PLUGINS_PREFIX,
 37                 strict_load: bool = cst.STRICT_PLUGIN_LOADING
 38                 ):
 39        self.default_wd = default_wd or "."
 40        self.cmd_map: dict[str, CommandMetadata] = {}
 41        """Map of commands names to its metadata."""
 42
 43        self.plugins_dir = plugins_dir
 44        self.plugins_prefix = plugins_prefix
 45        self.logger = logging.getLogger(__name__)
 46        self.strict_load = strict_load
 47
 48        self.posix = utils.is_posix()
 49        """Is system posix"""
 50
 51    def shlex_split(self, cmd: str) -> list[str]:
 52        """
 53        Splits a line like bash does(with passed posix param)
 54        :param cmd: Input line(command)
 55        :return: List of split: command name is the first element, the next ones are args
 56        """
 57        return shlex.split(cmd, posix=self.posix)
 58
 59    @staticmethod
 60    def fetch_name_and_args(splitted: list) -> tuple[str, list[str]]:
 61        """
 62        Gets command name and args from shlex split line
 63        :param splitted: shlex split line
 64        :return: command name, args
 65        """
 66        if not splitted:
 67            return '', []
 68        cmd_name = splitted[0]
 69        if len(splitted) < 2:
 70            return cmd_name, []
 71        return cmd_name, splitted[1:]
 72
 73    def start_session(self):
 74        """
 75        Starts an infinite loop of commandline session
 76        :return: None
 77        """
 78        self.load_modules()
 79        default = self.default_wd
 80        to_move = Path(default).expanduser()
 81        os.chdir(to_move)
 82        while True:
 83            cmd = input(f"{os.getcwd()} $ ").strip()
 84            if not cmd.strip():
 85                continue
 86            try:
 87                res = self.execute_command(cmd)
 88                if res:
 89                    res = res.strip()
 90                    if res.stderr:
 91                        errs = res.stderr.split("\n")
 92                        for err in errs:
 93                            if not err:
 94                                continue
 95                            cmd_name = self.parse_line(cmd)[0]
 96                            log_error(f"{cmd_name}: {err}", self.logger)
 97                    if res.stdout:
 98                        print(res.stdout)
 99            except ImportError:
100                raise
101
102            except Exception as e:
103                try:
104                    name = self.shlex_split(cmd)[0]
105                    cmd_meta = self.cmd_map.get(name)
106                except ValueError:
107                    name = ''
108                    cmd_meta = CommandMetadata('default', 'default', 'default', '', LsCommand)
109                if cmd_meta.plugin_author != "default":
110                    utils.log_error(
111                        f"Author of plugin '{cmd_meta.plugin_name}' of version '{cmd_meta.plugin_version}' is a debil(real name - '{cmd_meta.plugin_author}'). His command '{name}' raised an unexpected error:",
112                        self.logger
113                    )
114                    utils.log_error(e, self.logger, exc=True)
115                else:
116                    utils.log_error(e, self.logger)
117
118    def load_modules(self, outer_strict: bool = False):
119        """
120        Loads plugins
121        :param outer_strict: if method was called out of class, will be more prioritized than self.strict_load
122        :type outer_strict: bool
123
124        :return: None
125        """
126
127        if not outer_strict:
128            outer_strict = self.strict_load
129        plugins_loader = PluginLoader(self.plugins_dir, self.plugins_prefix, outer_strict)
130        plugins_loader.load_plugins()
131
132        self.cmd_map = plugins_loader.commands
133        #self.cmd_map["reload-plugins"] = CommandMetadata("reload-plugins", "default_plugin", "default", "1.0.0", ReloadPluginsCommand)
134
135    def parse_line(self, line: str) -> tuple[str, list[str]] | None:
136        """
137        Parses input line by calling self.shlex_split and self.fetch_name_and_args
138        :param line: Input line
139        :return: command name, args | None, if line was empty
140        """
141        args = self.shlex_split(line)
142        if not args:
143            return None
144        cmd_name, cmd_args = self.fetch_name_and_args(splitted=args)
145
146        return cmd_name, cmd_args
147
148    def execute_command(self, line: str):
149        """
150        Executes input command
151        :param line:
152        :return: Result of command(None, if error occurred or command was not found)
153        """
154        parsed = self.parse_line(line)
155        if not parsed:
156            return None
157
158        cmd_name, cmd_args = parsed
159        if cmd_name == "help":
160            out = ""
161            for cmd in sorted(self.cmd_map.keys()):
162                out+=f"{cmd}\n"
163            return CommandOutput(stdout=out)
164
165
166        cmd_meta = self.cmd_map.get(cmd_name)
167        if not cmd_meta:
168            utils.write_history(line)
169            utils.log_error(f"{cmd_name}: command not found", self.logger)
170            return None
171
172        self.logger.info(line)
173        cmd_obj = cmd_meta.cmd(args = cmd_args)
174        cmd_obj.history()
175
176        if "--help" in cmd_args:
177            return cmd_obj.help() #TODO: --help keys
178
179        try:
180            return cmd_obj.handled_run()
181        except HANDLED_ERRORS as e:
182            log_error(f"{cmd_name}: {str(e)}", self.logger)