""" 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 logging import lib.common.utils as utils import lib.common.exceptions as exceptions from lib.common.decorators import getrequest from lib.common.decorators import postrequest from lib.web.pages.templates import web_templates from lib.db.db_plugins import DBPlugins from lib.plugins.plugin_manager.plugin_manager import PluginManager @getrequest.route('/api/pluginsform') def get_plugins_form_html(_webserver, _namespace=None, _sort_col=None, _sort_dir=None, filter_dict=None): plugins_form = PluginsFormHTML(_webserver.config) _area = _webserver.query_data.get('area') _plugin = _webserver.query_data.get('plugin') _repo = _webserver.query_data.get('repo') if _area is None and _plugin is None and _repo is None: _webserver.do_mime_response( 404, 'text/html', web_templates['htmlError'] .format('404 - Badly formed request')) elif _area: try: form = plugins_form.get(_area) _webserver.do_mime_response(200, 'text/html', form) except exceptions.CabernetException as ex: _webserver.do_mime_response( 404, 'text/html', web_templates['htmlError'] .format('404 - Badly formed area request')) elif _plugin and _repo: try: form = plugins_form.get_plugin(_repo, _plugin) _webserver.do_mime_response(200, 'text/html', form) except exceptions.CabernetException as ex: _webserver.do_mime_response( 404, 'text/html', web_templates['htmlError'] .format('404 - Badly formed plugin request')) else: # case where plugin and repo are not provided together _webserver.do_mime_response( 404, 'text/html', web_templates['htmlError'] .format('404 - Badly formed plugin/repo request')) @postrequest.route('/api/pluginsform') def post_plugins_html(_webserver): action = _webserver.query_data.get('action') pluginid = _webserver.query_data.get('pluginId') repoid = _webserver.query_data.get('repoId') if action and pluginid and repoid: action = action[0] pluginid = pluginid[0] repoid = repoid[0] if action == "deletePlugin": pm = PluginManager(_webserver.plugins) results = pm.delete_plugin(repoid, pluginid, _webserver.sched_queue) _webserver.do_mime_response(200, 'text/html', 'STATUS: Deleting plugin: {}:{}
'.format(repoid, pluginid) + str(results)) if action == "addInstance": pm = PluginManager(_webserver.plugins) results = pm.add_instance(repoid, pluginid, _webserver.sched_queue) _webserver.do_mime_response(200, 'text/html', 'STATUS: Adding Instance plugin: {}:{}
'.format(repoid, pluginid) + str(results)) elif action == "installPlugin": pm = PluginManager(_webserver.plugins) results = pm.install_plugin(repoid, pluginid, _webserver.sched_queue) _webserver.do_mime_response(200, 'text/html', 'STATUS: Installing plugin: {}:{}
'.format(repoid, pluginid) + str(results)) elif action == "upgradePlugin": pm = PluginManager(_webserver.plugins) results = pm.upgrade_plugin(repoid, pluginid, _webserver.sched_queue) _webserver.do_mime_response(200, 'text/html', 'STATUS: Installing plugin: {}:{}
'.format(repoid, pluginid) + str(results)) else: _webserver.do_mime_response(200, 'text/html', "doing something else"+str(action[0])) else: _webserver.do_mime_response( 404, 'text/html', web_templates['htmlError'] .format('404 - Badly formed request')) class PluginsFormHTML: def __init__(self, _config): self.logger = logging.getLogger(__name__) self.config = _config self.plugin_db = DBPlugins(self.config) self.active_tab_name = None self.num_of_plugins = 0 self.plugin_data = None self.area = None def get(self, _area): self.area = _area return ''.join([self.header, self.body]) def get_plugin(self, _repo_id, _plugin_id): plugin_defn = self.plugin_db.get_plugins( _installed=None, _repo_id=_repo_id, _plugin_id=_plugin_id) if not plugin_defn: self.logger.warning( 'HTTP request: Unknown plugin: {}' .format(_plugin_id)) raise exceptions.CabernetException( 'Unknown Plugin: {}' .format(_plugin_id)) plugin_defn = plugin_defn[0] return ''.join([self.get_plugin_header(plugin_defn), self.get_menu_section(plugin_defn), self.get_plugin_section(plugin_defn)]) def get_menu_top_section(self, _plugin_defn): return ''.join([ '' ]) def get_menu_items(self, _plugin_defn): if not _plugin_defn['external']: menu_list='' elif _plugin_defn['version']['installed']: # delete and possible upgrade... menu_list = '' if _plugin_defn['version']['latest'] != _plugin_defn['version']['current']: menu_list = ''.join([ '' ]) menu_list += ''.join([ '' ]) else: # install menu_list = ''.join([ '' ]) return menu_list def get_menu_section(self, _plugin_defn): pluginid = _plugin_defn['id'] repoid = _plugin_defn['repoid'] return ''.join([ '', '
' '', '', '', '
']) def get_plugin_header(self, _plugin_defn): instances = self.plugin_db.get_instances(_namespace=_plugin_defn['name']) if instances: # array of instance names instances = instances[_plugin_defn['name']] else: instances = None if not _plugin_defn['version'].get('latest'): _plugin_defn['version']['latest'] = None html = ''.join([ '
', '
', str(_plugin_defn['name']), '
', '
', str(_plugin_defn['summary']), '
', '
', '',
                _plugin_defn['name'],'', '
', '
', str(_plugin_defn['description']), '
', '
' ]) return html def get_plugin_section(self, _plugin_defn): pluginid = _plugin_defn['id'] repoid = _plugin_defn['repoid'] instances = self.plugin_db.get_instances(_namespace=_plugin_defn['name']) if instances: # array of instance names instances = instances[_plugin_defn['name']] else: instances = None if not _plugin_defn['version'].get('latest'): _plugin_defn['version']['latest'] = None latest_version = _plugin_defn['version']['latest'] upgrade_available = '' if latest_version != _plugin_defn['version']['current'] and _plugin_defn['external']: upgrade_available = ''.join([ '
' '', '', '', '' \ .format(latest_version), '
' ]) if _plugin_defn['version']['installed']: version_installed_div = ''.join([ '
', '
Version Installed:
', '
', str(_plugin_defn['version']['current']), '
', upgrade_available, '
', ]) else: version_installed_div = '' if _plugin_defn.get('changelog'): changelog_div = ''.join([ '
', '
Change Log:
', '
', '', str(_plugin_defn['changelog']), '
', '
', ]) else: changelog_div = '' html = ''.join([ '', version_installed_div, '
', '
Latest Version:
', '
', str(_plugin_defn['version']['latest']), '
', '
', changelog_div, '
', '
Dependencies:
', '
', str(_plugin_defn['dependencies']), '
', '
', '
', '
Source:
', '
', str(_plugin_defn['source']), '
', '
', '
', '
License:
', '
', str(_plugin_defn['license']), '
', '
', '
', '
Author:
', '
', str(_plugin_defn['provider-name']), '
', '
', '
', '
Origin:
', '
', 'Cabernet Plugin Repository', '
', '
', '
', '
Category:
', '
', str(_plugin_defn['category']), '
', '
', '
', '
Related Website:
', '
', str(_plugin_defn['website']), '
', '
', '
', '
Instances:
', '
', str(instances), '
', '
', '
', '
' ]) return html def form_plugins(self, _is_installed): plugin_defns = self.plugin_db.get_plugins( _is_installed, None, None) if not plugin_defns: if self.area == 'My_Plugins': return ''.join([ 'No plugins are installed. Go to Catalog and select a plugin to install.' ]) elif self.area == 'Catalog': return ''.join([ 'All available plugins are installed' ]) plugins_list = '' for plugin_defn in sorted(plugin_defns, key=lambda p: p['id']): repo_id = plugin_defn['repoid'] plugin_id = plugin_defn['id'] plugin_name = plugin_defn['name'] img_size = self.lookup_config_size() latest_version = plugin_defn['version']['latest'] upgrade_available = '' if _is_installed and plugin_defn['external']: if latest_version != plugin_defn['version']['current']: upgrade_available = '
Upgrade to {}
' \ .format(latest_version) current_version = plugin_defn['version']['current'] elif not _is_installed: current_version = plugin_defn['version']['latest'] else: current_version = 'Internal' plugins_list += ''.join([ '' ]) return plugins_list @property def header(self): return ''.join([ '', '' ]) @property def form(self): if self.area == 'My_Plugins': forms_html = ''.join([ '
', self.form_plugins(True), '
']) elif self.area == 'Catalog': forms_html = ''.join([ 'Plugins Available To Install:' '
', self.form_plugins(_is_installed=False), '
']) else: self.logger.warning('HTTP request: unknown area: {}'.format(self.area)) raise exceptions.CabernetException('Unknown Tab: {}'.format(self.area)) return forms_html @property def body(self): return ''.join([ '', self.form, '']) def lookup_config_size(self): size_text = self.config['channels']['thumbnail_size'] if size_text == 'None': return 0 elif size_text == 'Tiny(16)': return 16 elif size_text == 'Small(48)': return 48 elif size_text == 'Medium(128)': return 128 elif size_text == 'Large(180)': return 180 elif size_text == 'X-Large(270)': return 270 elif size_text == 'Full-Size': return None else: return None