import re import time import traceback from threading import Thread from typing import List, Set, Type, Tuple from PyQt5.QtWidgets import QStyleFactory, QApplication from bauh import ROOT_DIR from bauh.api.abstract.controller import SoftwareManager, SearchResult, ApplicationContext from bauh.api.abstract.disk import DiskCacheLoader from bauh.api.abstract.handler import ProcessWatcher from bauh.api.abstract.model import SoftwarePackage, PackageUpdate, PackageHistory, PackageSuggestion, PackageAction from bauh.api.abstract.view import FormComponent, ViewComponent, TabGroupComponent, TabComponent, SingleSelectComponent, \ InputOption, PanelComponent, SelectViewType, TextInputComponent, FileChooserComponent, MultipleSelectComponent, \ TextComponent from bauh.api.exception import NoInternetException from bauh.commons import internet from bauh.commons.html import bold from bauh.view.core import config from bauh.view.core.config import read_config from bauh.view.util import translation RE_IS_URL = re.compile(r'^https?://.+') class GenericSoftwareManager(SoftwareManager): def __init__(self, managers: List[SoftwareManager], context: ApplicationContext, config: dict): super(GenericSoftwareManager, self).__init__(context=context) self.managers = managers self.map = {t: m for m in self.managers for t in m.get_managed_types()} self._available_cache = {} if config['system']['single_dependency_checking'] else None self.thread_prepare = None self.i18n = context.i18n self.disk_loader_factory = context.disk_loader_factory self.logger = context.logger self._already_prepared = [] self.working_managers = [] self.config = config def reset_cache(self): if self._available_cache is not None: self._available_cache = {} self.working_managers.clear() def _sort(self, apps: List[SoftwarePackage], word: str) -> List[SoftwarePackage]: exact_name_matches, contains_name_matches, others = [], [], [] for app in apps: lower_name = app.name.lower() if word == lower_name: exact_name_matches.append(app) elif word in lower_name: contains_name_matches.append(app) else: others.append(app) res = [] for app_list in (exact_name_matches, contains_name_matches, others): app_list.sort(key=lambda a: a.name.lower()) res.extend(app_list) return res def _can_work(self, man: SoftwareManager): if self._available_cache is not None: available = False for t in man.get_managed_types(): available = self._available_cache.get(t) if available is None: available = man.is_enabled() and man.can_work() self._available_cache[t] = available if available: available = True else: available = man.is_enabled() and man.can_work() if available: if man not in self.working_managers: self.working_managers.append(man) else: if man in self.working_managers: self.working_managers.remove(man) return available def _search(self, word: str, is_url: bool, man: SoftwareManager, disk_loader, res: SearchResult): if self._can_work(man): mti = time.time() apps_found = man.search(words=word, disk_loader=disk_loader, is_url=is_url) mtf = time.time() self.logger.info(man.__class__.__name__ + " took {0:.2f} seconds".format(mtf - mti)) res.installed.extend(apps_found.installed) res.new.extend(apps_found.new) def search(self, word: str, disk_loader: DiskCacheLoader = None, limit: int = -1, is_url: bool = False) -> SearchResult: ti = time.time() self._wait_to_be_ready() res = SearchResult([], [], 0) if internet.is_available(self.context.http_client, self.context.logger): norm_word = word.strip().lower() url_words = RE_IS_URL.match(norm_word) disk_loader = self.disk_loader_factory.new() disk_loader.start() threads = [] for man in self.managers: t = Thread(target=self._search, args=(norm_word, url_words, man, disk_loader, res)) t.start() threads.append(t) for t in threads: t.join() if disk_loader: disk_loader.stop_working() disk_loader.join() res.installed = self._sort(res.installed, norm_word) res.new = self._sort(res.new, norm_word) res.total = len(res.installed) + len(res.new) else: raise NoInternetException() tf = time.time() self.logger.info('Took {0:.2f} seconds'.format(tf - ti)) return res def _wait_to_be_ready(self): if self.thread_prepare: self.thread_prepare.join() self.thread_prepare = None def set_enabled(self, enabled: bool): pass def can_work(self) -> bool: return True def _is_internet_available(self, res: dict): res['available'] = internet.is_available(self.context.http_client, self.context.logger) def _get_internet_check(self, res: dict) -> Thread: t = Thread(target=self._is_internet_available, args=(res,)) t.start() return t def read_installed(self, disk_loader: DiskCacheLoader = None, limit: int = -1, only_apps: bool = False, pkg_types: Set[Type[SoftwarePackage]] = None, net_check: bool = None) -> SearchResult: ti = time.time() self._wait_to_be_ready() net_check = {} thread_internet_check = self._get_internet_check(net_check) res = SearchResult([], None, 0) disk_loader = None internet_available = None if not pkg_types: # any type for man in self.managers: if self._can_work(man): if not disk_loader: disk_loader = self.disk_loader_factory.new() disk_loader.start() if internet_available is None: thread_internet_check.join() internet_available = net_check.get('available', True) mti = time.time() man_res = man.read_installed(disk_loader=disk_loader, pkg_types=None, internet_available=internet_available) mtf = time.time() self.logger.info(man.__class__.__name__ + " took {0:.2f} seconds".format(mtf - mti)) res.installed.extend(man_res.installed) res.total += man_res.total else: man_already_used = [] for t in pkg_types: man = self.map.get(t) if man and (man not in man_already_used) and self._can_work(man): if not disk_loader: disk_loader = self.disk_loader_factory.new() disk_loader.start() if internet_available is None: thread_internet_check.join() internet_available = net_check.get('available', True) mti = time.time() man_res = man.read_installed(disk_loader=disk_loader, pkg_types=None, internet_available=internet_available) mtf = time.time() self.logger.info(man.__class__.__name__ + " took {0:.2f} seconds".format(mtf - mti)) res.installed.extend(man_res.installed) res.total += man_res.total if disk_loader: disk_loader.stop_working() disk_loader.join() tf = time.time() self.logger.info('Took {0:.2f} seconds'.format(tf - ti)) return res def downgrade(self, app: SoftwarePackage, root_password: str, handler: ProcessWatcher) -> bool: man = self._get_manager_for(app) if man and app.can_be_downgraded(): mti = time.time() res = man.downgrade(app, root_password, handler) mtf = time.time() self.logger.info('Took {0:.2f} seconds'.format(mtf - mti)) return res else: raise Exception("downgrade is not possible for {}".format(app.__class__.__name__)) def clean_cache_for(self, app: SoftwarePackage): man = self._get_manager_for(app) if man: return man.clean_cache_for(app) def update(self, app: SoftwarePackage, root_password: str, handler: ProcessWatcher) -> bool: man = self._get_manager_for(app) if man: return man.update(app, root_password, handler) def uninstall(self, app: SoftwarePackage, root_password: str, handler: ProcessWatcher) -> bool: man = self._get_manager_for(app) if man: return man.uninstall(app, root_password, handler) def install(self, app: SoftwarePackage, root_password: str, handler: ProcessWatcher) -> bool: man = self._get_manager_for(app) if man: ti = time.time() try: self.logger.info('Installing {}'.format(app)) return man.install(app, root_password, handler) except: traceback.print_exc() return False finally: tf = time.time() self.logger.info('Installation of {}'.format(app) + 'took {0:.2f} minutes'.format((tf - ti)/60)) def get_info(self, app: SoftwarePackage): man = self._get_manager_for(app) if man: return man.get_info(app) def get_history(self, app: SoftwarePackage) -> PackageHistory: man = self._get_manager_for(app) if man: mti = time.time() history = man.get_history(app) mtf = time.time() self.logger.info(man.__class__.__name__ + " took {0:.2f} seconds".format(mtf - mti)) return history def get_managed_types(self) -> Set[Type[SoftwarePackage]]: pass def is_enabled(self): return True def _get_manager_for(self, app: SoftwarePackage) -> SoftwareManager: man = self.map[app.__class__] return man if man and self._can_work(man) else None def cache_to_disk(self, pkg: SoftwarePackage, icon_bytes: bytes, only_icon: bool): if self.context.disk_cache and pkg.supports_disk_cache(): man = self._get_manager_for(pkg) if man: return man.cache_to_disk(pkg, icon_bytes=icon_bytes, only_icon=only_icon) def requires_root(self, action: str, app: SoftwarePackage): man = self._get_manager_for(app) if man: return man.requires_root(action, app) def _prepare(self): if self.managers: for man in self.managers: if man not in self._already_prepared and self._can_work(man): man.prepare() self._already_prepared.append(man) def prepare(self): self.thread_prepare = Thread(target=self._prepare) self.thread_prepare.start() def list_updates(self, net_check: bool = None) -> List[PackageUpdate]: self._wait_to_be_ready() updates = [] if self.managers: net_check = {} thread_internet_check = self._get_internet_check(net_check) for man in self.managers: if self._can_work(man): if thread_internet_check.is_alive(): thread_internet_check.join() man_updates = man.list_updates(internet_available=net_check['available']) if man_updates: updates.extend(man_updates) return updates def list_warnings(self, internet_available: bool = None) -> List[str]: warnings = [] if self.managers: int_available = internet.is_available(self.context.http_client, self.context.logger) for man in self.managers: if man.is_enabled(): man_warnings = man.list_warnings(internet_available=int_available) if man_warnings: if warnings is None: warnings = [] warnings.extend(man_warnings) return warnings def _fill_suggestions(self, suggestions: list, man: SoftwareManager, limit: int, filter_installed: bool): if self._can_work(man): mti = time.time() man_sugs = man.list_suggestions(limit=limit, filter_installed=filter_installed) mtf = time.time() self.logger.info(man.__class__.__name__ + ' took {0:.2f} seconds'.format(mtf - mti)) if man_sugs: if 0 < limit < len(man_sugs): man_sugs = man_sugs[0:limit] suggestions.extend(man_sugs) def list_suggestions(self, limit: int, filter_installed: bool) -> List[PackageSuggestion]: if bool(self.config['suggestions']['enabled']): if self.managers and internet.is_available(self.context.http_client, self.context.logger): suggestions, threads = [], [] for man in self.managers: t = Thread(target=self._fill_suggestions, args=(suggestions, man, int(self.config['suggestions']['by_type']), filter_installed)) t.start() threads.append(t) for t in threads: t.join() if suggestions: suggestions.sort(key=lambda s: s.priority.value, reverse=True) return suggestions return [] def execute_custom_action(self, action: PackageAction, pkg: SoftwarePackage, root_password: str, watcher: ProcessWatcher): man = self._get_manager_for(pkg) if man: return exec('man.{}(pkg=pkg, root_password=root_password, watcher=watcher)'.format(action.manager_method)) def is_default_enabled(self) -> bool: return True def launch(self, pkg: SoftwarePackage): self._wait_to_be_ready() man = self._get_manager_for(pkg) if man: self.logger.info('Launching {}'.format(pkg)) man.launch(pkg) def get_screenshots(self, pkg: SoftwarePackage): man = self._get_manager_for(pkg) if man: return man.get_screenshots(pkg) def get_working_managers(self): return [m for m in self.managers if self._can_work(m)] def _gen_bool_component(self, label: str, tooltip: str, value: bool, id_: str) -> SingleSelectComponent: opts = [InputOption(label=self.i18n['yes'].capitalize(), value=True), InputOption(label=self.i18n['no'].capitalize(), value=False)] return SingleSelectComponent(label=label, options=opts, default_option=[o for o in opts if o.value == value][0], type_=SelectViewType.RADIO, tooltip=tooltip, max_per_line=len(opts), id_=id_) def _gen_general_settings(self, core_config: dict) -> TabComponent: locale_opts = [InputOption(label=self.i18n['locale.{}'.format(k)].capitalize(), value=k) for k in translation.get_available_keys()] select_locale = SingleSelectComponent(label=self.i18n['core.config.locale.label'], options=locale_opts, default_option=[l for l in locale_opts if l.value == core_config['locale']][0], type_=SelectViewType.COMBO, id_='locale') select_sysnotify = self._gen_bool_component(label=self.i18n['core.config.system.notifications'].capitalize(), tooltip=self.i18n['core.config.system.notifications.tip'].capitalize(), value=bool(core_config['system']['notifications']), id_="sys_notify") select_sugs = self._gen_bool_component(label=self.i18n['core.config.suggestions.activated'].capitalize(), tooltip=self.i18n['core.config.suggestions.activated.tip'].capitalize(), id_="sugs_enabled", value=bool(core_config['suggestions']['enabled'])) inp_sugs = TextInputComponent(label=self.i18n['core.config.suggestions.by_type'], tooltip=self.i18n['core.config.suggestions.by_type.tip'], value=str(core_config['suggestions']['by_type']), only_int=True, id_="sugs_by_type") sub_comps = [FormComponent([select_locale, select_sysnotify, select_sugs, inp_sugs], spaces=False)] return TabComponent(self.i18n['core.config.tab.general'].capitalize(), PanelComponent(sub_comps), None, 'core.gen') def _gen_adv_settings(self, core_config: dict) -> TabComponent: select_dcache = self._gen_bool_component(label=self.i18n['core.config.disk_cache'], tooltip=self.i18n['core.config.disk_cache.tip'].format( app=self.context.app_name), value=core_config['disk_cache']['enabled'], id_='dcache') input_data_exp = TextInputComponent(label=self.i18n['core.config.mem_cache.data_exp'], tooltip=self.i18n['core.config.mem_cache.data_exp.tip'], value=str(core_config['memory_cache']['data_expiration']), only_int=True, id_="data_exp") input_icon_exp = TextInputComponent(label=self.i18n['core.config.mem_cache.icon_exp'], tooltip=self.i18n['core.config.mem_cache.icon_exp.tip'], value=str(core_config['memory_cache']['icon_expiration']), only_int=True, id_="icon_exp") select_dep_check = self._gen_bool_component(label=self.i18n['core.config.system.dep_checking'], tooltip=self.i18n['core.config.system.dep_checking.tip'], value=core_config['system']['single_dependency_checking'], id_='dep_check') select_dmthread = self._gen_bool_component(label=self.i18n['core.config.download.multithreaded'], tooltip=self.i18n['core.config.download.multithreaded.tip'], id_="down_mthread", value=core_config['download']['multithreaded']) sub_comps = [FormComponent([select_dcache, select_dmthread, select_dep_check, input_data_exp, input_icon_exp], spaces=False)] return TabComponent(self.i18n['core.config.tab.advanced'].capitalize(), PanelComponent(sub_comps), None, 'core.adv') def _gen_tray_settings(self, core_config: dict) -> TabComponent: input_update_interval = TextInputComponent(label=self.i18n['core.config.updates.interval'].capitalize(), tooltip=self.i18n['core.config.updates.interval.tip'], only_int=True, value=str(core_config['updates']['check_interval']), id_="updates_interval") allowed_exts = {'png', 'svg', 'jpg', 'jpeg', 'ico', 'xpm'} select_def_icon = FileChooserComponent(id_='def_icon', label=self.i18n["core.config.ui.tray.default_icon"].capitalize(), tooltip=self.i18n["core.config.ui.tray.default_icon.tip"].capitalize(), file_path=str(core_config['ui']['tray']['default_icon']) if core_config['ui']['tray']['default_icon'] else None, allowed_extensions=allowed_exts) select_up_icon = FileChooserComponent(id_='up_icon', label=self.i18n["core.config.ui.tray.updates_icon"].capitalize(), tooltip=self.i18n["core.config.ui.tray.updates_icon.tip"].capitalize(), file_path=str(core_config['ui']['tray']['updates_icon']) if core_config['ui']['tray']['updates_icon'] else None, allowed_extensions=allowed_exts) sub_comps = [FormComponent([input_update_interval, select_def_icon, select_up_icon], spaces=False)] return TabComponent(self.i18n['core.config.tab.tray'].capitalize(), PanelComponent(sub_comps), None, 'core.tray') def _gen_ui_settings(self, core_config: dict) -> TabComponent: select_hdpi = self._gen_bool_component(label=self.i18n['core.config.ui.hdpi'], tooltip=self.i18n['core.config.ui.hdpi.tip'].format( app=self.context.app_name), value=bool(core_config['ui']['hdpi']), id_='hdpi') cur_style = QApplication.instance().style().objectName().lower() if not core_config['ui']['style'] else core_config['ui']['style'] style_opts = [InputOption(label=s.capitalize(), value=s.lower()) for s in QStyleFactory.keys()] select_style = SingleSelectComponent(label=self.i18n['style'].capitalize(), options=style_opts, default_option=[o for o in style_opts if o.value == cur_style][0], type_=SelectViewType.COMBO, id_="style") input_maxd = TextInputComponent(label=self.i18n['core.config.ui.max_displayed'].capitalize(), tooltip=self.i18n['core.config.ui.max_displayed.tip'].capitalize(), only_int=True, id_="table_max", value=str(core_config['ui']['table']['max_displayed'])) select_dicons = self._gen_bool_component(label=self.i18n['core.config.download.icons'], tooltip=self.i18n['core.config.download.icons.tip'], id_="down_icons", value=core_config['download']['icons']) sub_comps = [FormComponent([select_hdpi, select_dicons, select_style, input_maxd], spaces=False)] return TabComponent(self.i18n['core.config.tab.ui'].capitalize(), PanelComponent(sub_comps), None, 'core.ui') def _save_settings(self, general: PanelComponent, advanced: PanelComponent, ui: PanelComponent, tray: PanelComponent, gems_panel: PanelComponent) -> Tuple[bool, List[str]]: core_config = config.read_config() # general general_form = general.components[0] locale = general_form.get_component('locale').get_selected() if locale != self.i18n.current_key: core_config['locale'] = locale core_config['system']['notifications'] = general_form.get_component('sys_notify').get_selected() core_config['suggestions']['enabled'] = general_form.get_component('sugs_enabled').get_selected() sugs_by_type = general_form.get_component('sugs_by_type').get_int_value() core_config['suggestions']['by_type'] = sugs_by_type # advanced adv_form = advanced.components[0] core_config['disk_cache']['enabled'] = adv_form.get_component('dcache').get_selected() download_mthreaded = adv_form.get_component('down_mthread').get_selected() core_config['download']['multithreaded'] = download_mthreaded single_dep_check = adv_form.get_component('dep_check').get_selected() core_config['system']['single_dependency_checking'] = single_dep_check data_exp = adv_form.get_component('data_exp').get_int_value() core_config['memory_cache']['data_expiration'] = data_exp icon_exp = adv_form.get_component('icon_exp').get_int_value() core_config['memory_cache']['icon_expiration'] = icon_exp # tray tray_form = tray.components[0] core_config['updates']['check_interval'] = tray_form.get_component('updates_interval').get_int_value() def_icon_path = tray_form.get_component('def_icon').file_path core_config['ui']['tray']['default_icon'] = def_icon_path if def_icon_path else None up_icon_path = tray_form.get_component('up_icon').file_path core_config['ui']['tray']['updates_icon'] = up_icon_path if up_icon_path else None # ui ui_form = ui.components[0] core_config['download']['icons'] = ui_form.get_component('down_icons').get_selected() core_config['ui']['hdpi'] = ui_form.get_component('hdpi').get_selected() core_config['ui']['table']['max_displayed'] = ui_form.get_component('table_max').get_int_value() style = ui_form.get_component('style').get_selected() cur_style = core_config['ui']['style'] if core_config['ui']['style'] else QApplication.instance().style().objectName().lower() if style != cur_style: core_config['ui']['style'] = style # gems checked_gems = gems_panel.get_component('gems').get_selected_values() for man in self.managers: modname = man.__module__.split('.')[-2] enabled = modname in checked_gems man.set_enabled(enabled) core_config['gems'] = None if core_config['gems'] is None and len(checked_gems) == len(self.managers) else checked_gems try: config.save(core_config) return True, None except: return False, [traceback.format_exc()] def get_settings(self) -> ViewComponent: tabs = list() gem_opts, def_gem_opts, gem_tabs = [], set(), [] for man in self.managers: if self._can_work(man): man_comp = man.get_settings() modname = man.__module__.split('.')[-2] icon_path = "{r}/gems/{n}/resources/img/{n}.svg".format(r=ROOT_DIR, n=modname) if man_comp: gem_tabs.append(TabComponent(label=modname.capitalize(), content=man_comp, icon_path=icon_path, id_=modname)) opt = InputOption(label=self.i18n.get('gem.{}.label'.format(modname), modname.capitalize()), tooltip=self.i18n.get('gem.{}.info'.format(modname)), value=modname, icon_path='{r}/gems/{n}/resources/img/{n}.svg'.format(r=ROOT_DIR, n=modname)) gem_opts.append(opt) if man.is_enabled() and man in self.working_managers: def_gem_opts.add(opt) core_config = read_config() tabs.append(self._gen_general_settings(core_config)) if gem_opts: type_help = TextComponent(html=self.i18n['core.config.types.tip']) gem_selector = MultipleSelectComponent(label=None, tooltip=None, options=gem_opts, default_options=def_gem_opts, id_="gems") tabs.append(TabComponent(label=self.i18n['core.config.tab.types'], content=PanelComponent([type_help, gem_selector]), id_='core.types')) tabs.append(self._gen_ui_settings(core_config)) tabs.append(self._gen_tray_settings(core_config)) tabs.append(self._gen_adv_settings(core_config)) for tab in gem_tabs: tabs.append(tab) return TabGroupComponent(tabs) def save_settings(self, component: TabGroupComponent) -> Tuple[bool, List[str]]: saved, warnings = True, [] success, errors = self._save_settings(general=component.get_tab('core.gen').content, advanced=component.get_tab('core.adv').content, tray=component.get_tab('core.tray').content, ui=component.get_tab('core.ui').content, gems_panel=component.get_tab('core.types').content) if not success: saved = False if errors: warnings.extend(errors) for man in self.managers: if man: modname = man.__module__.split('.')[-2] tab = component.get_tab(modname) if not tab: self.logger.warning("Tab for {} was not found".format(man.__class__.__name__)) else: res = man.save_settings(tab.content) if res: success, errors = res[0], res[1] if not success: saved = False if errors: warnings.extend(errors) return saved, warnings