""" MIT License Copyright (C) 2023 ROCKY4546 https://github.com/rocky4546 This file is part of Cabernet Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions: The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software. """ import configparser import logging import json import importlib import importlib.resources import os import pathlib import lib.common.exceptions as exceptions import lib.common.utils as utils from .plugin import Plugin from .repo_handler import RepoHandler from lib.db.db_plugins import DBPlugins from lib.db.db_channels import DBChannels from lib.db.db_config_defn import DBConfigDefn PLUGIN_DEFN_FILE = 'plugin_defn.json' class PluginHandler: logger = None cls_plugins = None def __init__(self, _config_obj): self.plugins = {} self.config_obj = _config_obj if PluginHandler.logger is None: PluginHandler.logger = logging.getLogger(__name__) self.plugin_defn = self.load_plugin_defn() self.check_external_plugin_folder() self.repos = RepoHandler(self.config_obj) self.repos.load_cabernet_repo() self.collect_plugins(self.config_obj.data['paths']['internal_plugins_pkg'], False) self.collect_plugins(self.config_obj.data['paths']['external_plugins_pkg'], True) self.cleanup_config_missing_plugins() if PluginHandler.cls_plugins is not None: del PluginHandler.cls_plugins PluginHandler.cls_plugins = self.plugins def terminate(self, _plugin_name): """ calls terminate to the plugin requested """ self.plugins[_plugin_name].terminate() del self.plugins[_plugin_name] def check_external_plugin_folder(self): """ If the folder does not exists, then create it and place the __init__.py file in it. """ ext_folder = pathlib.Path(self.config_obj.data['paths']['main_dir']) \ .joinpath(self.config_obj.data['paths']['external_plugins_pkg']) init_file = ext_folder.joinpath('__init__.py') if not init_file.exists(): self.logger.notice('Creating external plugin folder for use by Cabernet') try: if not ext_folder.exists(): os.makedirs(ext_folder) f = open(init_file, 'wb') f.close() except PermissionError as e: self.logger.warning('ERROR: {} unable to create {}'.format(str(e), init_file)) def collect_plugins(self, _plugins_pkg, _is_external): pkg = importlib.util.find_spec(_plugins_pkg) if not pkg: # module folder does not exist, do nothing self.logger.notice( 'plugin folder {} does not exist with a __init__.py empty file in it.' .format(_plugins_pkg)) return for folder in importlib.resources.contents(_plugins_pkg): self.collect_plugin(_plugins_pkg, _is_external, folder) self.del_missing_plugins() def cleanup_config_missing_plugins(self): """ Case where the plugin is deleted from folder, but database and config still have data. """ ch_db = DBChannels(self.config_obj.data) ns_inst_list = ch_db.get_channel_instances() ns_list = ch_db.get_channel_names() for ns in ns_list: ns = ns['namespace'] if not self.plugins.get(ns) and self.config_obj.data.get(ns.lower()): for nv in self.config_obj.data.get(ns.lower()).items(): new_value = self.set_value_type(nv[1]) self.config_obj.data[ns.lower()][nv[0]] = new_value for ns_inst in ns_inst_list: if not self.plugins.get(ns_inst['namespace']): inst_name = utils.instance_config_section(ns_inst['namespace'], ns_inst['instance']) if self.config_obj.data.get(inst_name): for nv in self.config_obj.data.get(inst_name).items(): new_value = self.set_value_type(nv[1]) self.config_obj.data[inst_name][nv[0]] = new_value db_configdefn = DBConfigDefn(self.config_obj.data) db_configdefn.add_config(self.config_obj.data) def set_value_type(self, _value): if not isinstance(_value, str): return _value if _value == 'True': return True elif _value == 'False': return False elif _value.isdigit(): return int(_value) else: return _value def collect_plugin(self, _plugins_pkg, _is_external, _folder): if _folder.startswith('__'): return try: importlib.resources.read_text(_plugins_pkg, _folder) except (IsADirectoryError, PermissionError): try: plugin = Plugin(self.config_obj, self.plugin_defn, _plugins_pkg, _folder, _is_external) self.plugins[plugin.name] = plugin except (exceptions.CabernetException, AttributeError): pass except UnicodeDecodeError: pass except Exception: pass return def del_missing_plugins(self): """ updates to uninstalled the plugins from the db that are no longer present """ plugin_db = DBPlugins(self.config_obj.data) plugin_dblist = plugin_db.get_plugins(_installed=True) if plugin_dblist: for p_dict in plugin_dblist: if (p_dict['name'] not in self.plugins) and (p_dict['name'] != utils.CABERNET_ID): p_dict['version']['installed'] = False plugin_db.save_plugin(p_dict) def load_plugin_defn(self): try: defn_file = importlib.resources.read_text(self.config_obj.data['paths']['resources_pkg'], PLUGIN_DEFN_FILE) self.logger.debug('Plugin Defn file loaded') defn = json.loads(defn_file) except FileNotFoundError: self.logger.warning('PLUGIN DEFN FILE NOT FOUND AT {} {}'.format( self.config_obj.data['paths']['resources_dir'], PLUGIN_DEFN_FILE)) defn = {} return defn def initialize_plugins(self): for name, plugin in self.plugins.items(): if not plugin.enabled or not self.config_obj.data[plugin.name.lower()]['enabled']: self.logger.info('Plugin {} is disabled in config.ini'.format(plugin.name)) plugin.enabled = False else: try: plugin.plugin_obj = plugin.init_func(plugin, self.plugins) except exceptions.CabernetException: self.logger.debug('Setting plugin {} to disabled'.format(plugin.name)) self.config_obj.data[plugin.name.lower()]['enabled'] = False plugin.enabled = False