Gnarwhal/sshare
1
0
Fork 0
Code Issues 2 Pull requests Projects Releases 3 Packages Wiki Activity Actions

Refactoring

Browse source 
- Flatten plugin module into Plugin class
- Move logger to its own class and rework it slightly
- Rename source and name to get_source and get_name since name() now clashes with the plugin name
This commit is contained in:
Gnarwhal 2024-09-03 19:47:49 +00:00
parent 00f3321230
commit a9fa15d643
Signed by: Gnarwhal
GPG key ID: 0989A73D8C421174
7 changed files with 82 additions and 68 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
examples

@ -1 +1 @@
Subproject commit 49bd55b9951270f233683f7bd51f70f8eab43985
Subproject commit f8b077f3764c16935462ffb818bdb5aeda75222b

43
src/sshare/logger.py Normal file
View file

@ -0,0 +1,43 @@
# This file is part of SSHare.
#
# SSHare is free software: you can redistribute it and/or modify it under the terms of
# the GNU General Public License as published by the Free Software Foundation,
# either version 3 of the License, or (at your option) any later version.
#
# SSHare is distributed in the hope that it will be useful, but WITHOUT ANY
# WARRANTY; without even the implied warranty of MERCHANTABILITY
# or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
# more details.
#
# You should have received a copy of the GNU General Public License along with
# SSHare. If not, see <https://www.gnu.org/licenses/>.
from sshare.plugin import Plugin
class Logger:
def __init__(self, *args, **kwargs):
if kwargs.get("preload_command_line") == True:
self.loggers = [ Plugin.internal("command_line") ]
else:
self.loggers = []
self.add(*args)
def add(self, *args, **kwargs):
for logger in args:
self.loggers.append()
def info(self, message):
for logger in self.loggers:
logger.info(message)
def warn(self, message):
for logger in self.loggers:
logger.warn(message)
def error(self, message):
for logger in self.loggers:
logger.error(message)
def fatal(self, message, error_code=1):
self.error(message)
sys.exit(error_code)

95
src/sshare/main.py
View file

@ -22,38 +22,12 @@ import subprocess
import sys
from pathlib import Path
from sshare.logger import Logger
from sshare.plugin import Plugin
from sshare.plugins.config import Flag
from sshare.plugins.config import NoArgument
from sshare.plugins.config import NoDefault
from sshare.version import version
class Congloggerate:
def __init__(self, loggers):
def info(message):
for logger in loggers:
logger.info(message)
def warn(message):
for logger in loggers:
logger.warn(message)
def error(message):
for logger in loggers:
logger.error(message)
self.info = info
self.warn = warn
self.error = error
fatalicize(self)
def fatalicize(logger):
def fatal(message):
logger.error(message)
sys.exit(1)
setattr(logger, "fatal", fatal)
from sshare.version import version
def main():
@ -65,13 +39,10 @@ def main():
# so that it can be used to report errors while loading and
# configuring plugins
# i.e. before other logging plugins have had a chance to be initialised
command_line = Plugin.internal("command_line")
logger = command_line.module
fatalicize(logger)
logger = Logger(preload_command_line=True)
# Load inbuilt plugins
plugins_flat = [
command_line,
Plugin.internal("file"),
Plugin.internal("current_time"),
Plugin.internal("append_type"),
@ -107,15 +78,15 @@ def main():
if config.get("plugins") == None:
config["plugins"] = {}
for plugin in plugins_flat:
if hasattr(plugin.module, "config"):
if hasattr(plugin, "config"):
plugin_config = config["plugins"].get(plugin.name)
if plugin_config != None:
for config_entry in plugin_config.items():
plugin.module.config[config_entry[0]] = config_entry[1]
plugin.config[config_entry[0]] = config_entry[1]
else:
setattr(plugin.module, "config", {})
if hasattr(plugin.module, "args"):
for arg_name, arg in plugin.module.args.items():
setattr(plugin, "config", {})
if hasattr(plugin, "args"):
for arg_name, arg in plugin.args.items():
if arg.is_valid():
arg.bind(plugin, arg_name)
def check_flag(flag):
@ -137,33 +108,33 @@ def main():
value = getattr(arguments, arg)
if value != NoArgument:
if value != Flag:
plugin.module.config[config] = value
plugin.config[config] = value
del argument_map[arg]
# Sort plugins by type and check activation criteria
error = False
for plugin in plugins_flat:
if isinstance(plugin.module.plugin_type, str):
plugin.module.plugin_type = [ plugin.module.plugin_type ]
for plugin_type in plugin.module.plugin_type:
if isinstance(plugin.plugin_type, str):
plugin.plugin_type = [ plugin.plugin_type ]
for plugin_type in plugin.plugin_type:
plugins_of_type = plugins.get(plugin_type)
if plugins_of_type == None:
logger.error(f"Error: Plugin '{plugin.name}' has an invalid plugin type '{plugin_type}'")
error = True
else:
active = True
if hasattr(plugin.module, "activate"):
criteria = plugin.module.activate
if isinstance(plugin.module.activate, dict):
criteria = plugin.module.activate.get(plugin_type)
if hasattr(plugin, "activate"):
criteria = plugin.activate
if isinstance(plugin.activate, dict):
criteria = plugin.activate.get(plugin_type)
if criteria != None:
for criterion in criteria:
active = not plugin.module.args[criterion].dest in argument_map
active = not plugin.args[criterion].dest in argument_map
if not active:
break
plugins_of_type["active" if active else "inactive"].append(plugin)
if active:
for config_entry in plugin.module.config.items():
for config_entry in plugin.config.items():
if config_entry[1] == NoDefault:
logger.error(f"Error: Value 'plugins.{plugin.name}.{config_entry[0]}' has no default value and must be specified explicitly")
error = True
@ -175,11 +146,11 @@ def main():
logger.error(f"No '{plugin_type}' plugins activated. Activate at least one of:")
for plugin in plugins_of_type["inactive"]:
logger.error(f"{plugin.name}:")
criteria = plugin.module.activate
if isinstance(plugin.module.activate, dict):
criteria = plugin.module.activate[plugin_type]
criteria = plugin.activate
if isinstance(plugin.activate, dict):
criteria = plugin.activate[plugin_type]
for criterion in criteria:
logger.error(f" {plugin.module.args[criterion].pretty()}")
logger.error(f" {plugin.args[criterion].pretty()}")
error = True
if error:
sys.exit(1)
@ -188,24 +159,24 @@ def main():
error = False
class PluginConfig: pass
for plugin in plugins_flat:
config = plugin.module.config
plugin.module.config = PluginConfig()
config = plugin.config
plugin.config = PluginConfig()
for config_entry in config.items():
setattr(plugin.module.config, config_entry[0], config_entry[1])
setattr(plugin.config, config_entry[0], config_entry[1])
if error:
sys.exit(1)
# Initialise plugins
for plugin in plugins_flat:
setattr(plugin.module, "logger", logger)
if hasattr(plugin.module, "init"):
error = error or plugin.module.init()
setattr(plugin, "logger", logger)
if hasattr(plugin, "init"):
error = error or plugin.init()
logger = Congloggerate([ logger.module for logger in plugins["logger"]["active"] ])
logger.add(*plugins["logger"]["active"])
sources = []
for plugin in plugins["source"]["active"]:
sources.append(plugin.module.source())
sources.append(plugin.get_source())
if len(sources) == 0:
logger.error("Error: No sources provided. Must activate at least one source plugin")
log_activations(logger, plugins["source"])
@ -213,15 +184,15 @@ def main():
for index, source in enumerate(sources):
name = ""
for plugin in plugins["name"]["active"]:
name = plugin.module.name(name, source)
name = plugin.get_name(name, source)
sources[index] = name, source
for name, source in sources:
for plugin in plugins["upload"]["active"]:
plugin.module.upload(name, source)
plugin.upload(name, source)
for name, _ in sources:
for plugin in plugins["result"]["active"]:
plugin.module.result(name)
plugin.result(name)
sys.exit(0)

4
src/sshare/plugin.py
View file

@ -17,8 +17,8 @@ import importlib.util
class Plugin:
def __init__(self, name, module):
self.name = name
self.module = module
self.__dict__ = module.__dict__
self.name = name
@staticmethod
def internal(name):

2
src/sshare/plugins/default/append_type.py
View file

@ -17,7 +17,7 @@ from ..source import Raw
plugin_type = "name"
def name(name, source):
def get_name(name, source):
if isinstance(source, File):
if source.path.is_dir():
return name

2
src/sshare/plugins/default/current_time.py
View file

@ -38,7 +38,7 @@ def init():
elif config.base > 62:
logger.fatal("Error: 'base' cannot be greater than 62")
def name(name, source):
def get_name(name, source):
return name + _rebase(config.base, time.time_ns())
def _rebase(base, number):

2
src/sshare/plugins/default/file.py
View file

@ -27,7 +27,7 @@ args = {
)
}
def source():
def get_source():
file = File(config.file)
if file.path.is_dir():
logger.info(f"Uploading directory '{config.file}'")