mirror of
https://github.com/spalencsar/bearhub.git
synced 2026-07-07 02:24:16 +02:00
AUR bearhub: fix stale source cache collision (pkgrel=10)
Use tag-specific tarball name bearhub-0.10.7-bearhub.6.tar.gz so makepkg and AUR helpers do not reuse an outdated bearhub-0.10.7.tar.gz from cache. Ignore makepkg artifacts under packaging/aur/.
This commit is contained in:
Sebastian Palencsar
@@ -1,702 +0,0 @@
|
||||
import shutil
|
||||
import time
|
||||
import traceback
|
||||
from subprocess import Popen, STDOUT
|
||||
from threading import Thread
|
||||
from typing import List, Set, Type, Tuple, Dict, Optional, Generator, Callable
|
||||
|
||||
from bauh.api.abstract.controller import SoftwareManager, SearchResult, ApplicationContext, UpgradeRequirements, \
|
||||
UpgradeRequirement, TransactionResult, SoftwareAction, SettingsView, SettingsController
|
||||
from bauh.api.abstract.disk import DiskCacheLoader
|
||||
from bauh.api.abstract.handler import ProcessWatcher, TaskManager
|
||||
from bauh.api.abstract.model import SoftwarePackage, PackageUpdate, PackageHistory, PackageSuggestion, \
|
||||
CustomSoftwareAction
|
||||
from bauh.api.abstract.view import TabGroupComponent, MessageType
|
||||
from bauh.api.exception import NoInternetException
|
||||
from bauh.commons.boot import CreateConfigFile
|
||||
from bauh.commons.html import bold
|
||||
from bauh.commons.regex import RE_URL
|
||||
from bauh.commons.util import sanitize_command_input
|
||||
from bauh.view.core.config import CoreConfigManager
|
||||
from bauh.view.core.settings import GenericSettingsManager
|
||||
from bauh.view.core.update import check_for_update
|
||||
from bauh.view.util import resource
|
||||
from bauh.view.util.resource import get_path
|
||||
from bauh.view.util.util import clean_app_files, restart_app
|
||||
|
||||
|
||||
class GenericUpgradeRequirements(UpgradeRequirements):
|
||||
|
||||
def __init__(self, to_install: List[UpgradeRequirement], to_remove: List[UpgradeRequirement],
|
||||
to_upgrade: List[UpgradeRequirement], cannot_upgrade: List[SoftwarePackage],
|
||||
sub_requirements: Dict[SoftwareManager, UpgradeRequirements]):
|
||||
super(GenericUpgradeRequirements, self).__init__(to_install=to_install, to_upgrade=to_upgrade,
|
||||
to_remove=to_remove, cannot_upgrade=cannot_upgrade)
|
||||
self.sub_requirements = sub_requirements
|
||||
|
||||
|
||||
class GenericSoftwareManager(SoftwareManager, SettingsController):
|
||||
|
||||
def __init__(self, managers: List[SoftwareManager], context: ApplicationContext, config: dict,
|
||||
force_suggestions: bool = False):
|
||||
|
||||
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
|
||||
self.settings_manager: Optional[GenericSettingsManager] = None
|
||||
self.http_client = context.http_client
|
||||
self.configman = CoreConfigManager()
|
||||
self._action_reset: Optional[CustomSoftwareAction] = None
|
||||
self._dynamic_extra_actions: Optional[Dict[CustomSoftwareAction, Callable[[dict], bool]]] = None
|
||||
self.force_suggestions = force_suggestions
|
||||
|
||||
@property
|
||||
def dynamic_extra_actions(self) -> Dict[CustomSoftwareAction, Callable[[dict], bool]]:
|
||||
if self._dynamic_extra_actions is None:
|
||||
self._dynamic_extra_actions = {
|
||||
CustomSoftwareAction(i18n_label_key='action.backups',
|
||||
i18n_status_key='action.backups.status',
|
||||
i18n_description_key='action.backups.desc',
|
||||
manager_method='launch_timeshift',
|
||||
manager=self,
|
||||
icon_path='timeshift',
|
||||
requires_root=False,
|
||||
refresh=False): self.is_backups_action_available
|
||||
}
|
||||
|
||||
return self._dynamic_extra_actions
|
||||
|
||||
@property
|
||||
def action_reset(self) -> CustomSoftwareAction:
|
||||
if self._action_reset is None:
|
||||
self._action_reset = CustomSoftwareAction(i18n_label_key='action.reset',
|
||||
i18n_status_key='action.reset.status',
|
||||
i18n_description_key='action.reset.desc',
|
||||
manager_method='reset',
|
||||
icon_path=resource.get_path('img/logo.svg'),
|
||||
requires_root=False,
|
||||
manager=self,
|
||||
refresh=False)
|
||||
|
||||
return self._action_reset
|
||||
|
||||
def _is_timeshift_launcher_available(self) -> bool:
|
||||
return bool(shutil.which('timeshift-launcher'))
|
||||
|
||||
def is_backups_action_available(self, app_config: dict) -> bool:
|
||||
return bool(app_config['backup']['enabled']) and self._is_timeshift_launcher_available()
|
||||
|
||||
def reset_cache(self):
|
||||
if self._available_cache is not None:
|
||||
self._available_cache = {}
|
||||
self.working_managers.clear()
|
||||
|
||||
def launch_timeshift(self, root_password: Optional[str], watcher: ProcessWatcher):
|
||||
if self._is_timeshift_launcher_available():
|
||||
try:
|
||||
Popen(['timeshift-launcher'], stderr=STDOUT)
|
||||
return True
|
||||
except Exception:
|
||||
traceback.print_exc()
|
||||
watcher.show_message(title=self.i18n["error"].capitalize(),
|
||||
body=self.i18n['action.backups.tool_error'].format(bold('Timeshift')),
|
||||
type_=MessageType.ERROR)
|
||||
return False
|
||||
else:
|
||||
watcher.show_message(title=self.i18n["error"].capitalize(),
|
||||
body=self.i18n['action.backups.tool_error'].format(bold('Timeshift')),
|
||||
type_=MessageType.ERROR)
|
||||
return False
|
||||
|
||||
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()[0]
|
||||
self._available_cache[t] = available
|
||||
|
||||
if available:
|
||||
available = True
|
||||
else:
|
||||
available = man.is_enabled() and man.can_work()[0]
|
||||
|
||||
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):
|
||||
mti = time.time()
|
||||
apps_found = man.search(words=word, disk_loader=disk_loader, is_url=is_url, limit=-1)
|
||||
mtf = time.time()
|
||||
self.logger.info(f'{man.__class__.__name__} took {mtf - mti:.8f} seconds')
|
||||
|
||||
res.installed.extend(apps_found.installed)
|
||||
res.new.extend(apps_found.new)
|
||||
|
||||
def search(self, words: str, disk_loader: DiskCacheLoader = None, limit: int = -1, is_url: bool = False) -> SearchResult:
|
||||
ti = time.time()
|
||||
self._wait_to_be_ready()
|
||||
|
||||
res = SearchResult.empty()
|
||||
|
||||
if self.context.is_internet_available():
|
||||
norm_query = sanitize_command_input(words).lower()
|
||||
self.logger.info(f"Search query: {norm_query}")
|
||||
|
||||
if norm_query:
|
||||
is_url = bool(RE_URL.match(norm_query))
|
||||
disk_loader = self.disk_loader_factory.new()
|
||||
disk_loader.start()
|
||||
|
||||
threads = []
|
||||
|
||||
for man in self.managers:
|
||||
if self._can_work(man):
|
||||
t = Thread(target=self._search, args=(norm_query, is_url, man, disk_loader, res), daemon=True)
|
||||
t.start()
|
||||
threads.append(t)
|
||||
|
||||
for t in threads:
|
||||
t.join()
|
||||
|
||||
if disk_loader:
|
||||
disk_loader.stop_working()
|
||||
disk_loader.join()
|
||||
else:
|
||||
raise NoInternetException()
|
||||
|
||||
res.update_total()
|
||||
tf = time.time()
|
||||
self.logger.info(f'Took {tf - ti:.8f} seconds')
|
||||
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) -> Tuple[bool, Optional[str]]:
|
||||
return True, None
|
||||
|
||||
def _get_package_lower_name(self, pkg: SoftwarePackage):
|
||||
return pkg.name.lower()
|
||||
|
||||
def _fill_read_installed(self, man: SoftwareManager, disk_loader: DiskCacheLoader, internet_available: bool,
|
||||
output: List[SearchResult]):
|
||||
mti = time.time()
|
||||
man_res = man.read_installed(disk_loader=disk_loader, pkg_types=None, internet_available=internet_available,
|
||||
limit=-1, only_apps=False)
|
||||
mtf = time.time()
|
||||
self.logger.info(f'{man.__class__.__name__} took {mtf - mti:.4f} seconds')
|
||||
output.append(man_res)
|
||||
|
||||
def read_installed(self, disk_loader: DiskCacheLoader = None, limit: int = -1, only_apps: bool = False, pkg_types: Set[Type[SoftwarePackage]] = None, internet_available: bool = None) -> SearchResult:
|
||||
ti = time.time()
|
||||
self._wait_to_be_ready()
|
||||
|
||||
res = SearchResult([], None, 0)
|
||||
|
||||
disk_loader = None
|
||||
|
||||
net_available = self.context.is_internet_available()
|
||||
read_threads = list()
|
||||
results = list()
|
||||
|
||||
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()
|
||||
|
||||
t = Thread(target=self._fill_read_installed,
|
||||
args=(man, disk_loader, net_available, results),
|
||||
daemon=True)
|
||||
t.start()
|
||||
read_threads.append(t)
|
||||
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()
|
||||
|
||||
t = Thread(target=self._fill_read_installed,
|
||||
args=(man, disk_loader, net_available, results),
|
||||
daemon=True)
|
||||
t.start()
|
||||
read_threads.append(t)
|
||||
|
||||
for t in read_threads:
|
||||
t.join()
|
||||
|
||||
if disk_loader:
|
||||
disk_loader.stop_working()
|
||||
disk_loader.join()
|
||||
|
||||
for result in results:
|
||||
if result.installed:
|
||||
res.installed.extend(result.installed)
|
||||
res.total += result.total
|
||||
|
||||
if res.installed:
|
||||
for p in res.installed:
|
||||
if p.is_update_ignored():
|
||||
if p.categories is None:
|
||||
p.categories = ['updates_ignored']
|
||||
elif 'updates_ignored' not in p.categories:
|
||||
self._add_category(p, 'updates_ignored')
|
||||
|
||||
res.installed.sort(key=self._get_package_lower_name)
|
||||
|
||||
tf = time.time()
|
||||
self.logger.info(f'Took {tf - ti:.2f} seconds')
|
||||
return res
|
||||
|
||||
def _add_category(self, pkg: SoftwarePackage, category: str):
|
||||
if isinstance(pkg.categories, tuple):
|
||||
pkg.categories = tuple((*pkg.categories, category))
|
||||
elif isinstance(pkg.categories, list):
|
||||
pkg.categories.append(category)
|
||||
elif isinstance(pkg.categories, set):
|
||||
pkg.categories.add(category)
|
||||
|
||||
def downgrade(self, app: SoftwarePackage, root_password: Optional[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(f'Took {mtf - mti:.2f} seconds')
|
||||
return res
|
||||
else:
|
||||
raise Exception(f"Downgrading is not possible for {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 upgrade(self, requirements: GenericUpgradeRequirements, root_password: Optional[str], handler: ProcessWatcher) -> bool:
|
||||
for man, man_reqs in requirements.sub_requirements.items():
|
||||
res = man.upgrade(man_reqs, root_password, handler)
|
||||
|
||||
if not res:
|
||||
return False
|
||||
|
||||
return True
|
||||
|
||||
def _fill_post_transaction_status(self, pkg: SoftwarePackage, installed: bool):
|
||||
pkg.installed = installed
|
||||
pkg.update = False
|
||||
|
||||
if pkg.latest_version:
|
||||
pkg.version = pkg.latest_version
|
||||
|
||||
def _update_post_transaction_status(self, res: TransactionResult):
|
||||
if res.success:
|
||||
if res.installed:
|
||||
for p in res.installed:
|
||||
self._fill_post_transaction_status(p, True)
|
||||
if res.removed:
|
||||
for p in res.removed:
|
||||
self._fill_post_transaction_status(p, False)
|
||||
|
||||
def uninstall(self, pkg: SoftwarePackage, root_password: Optional[str], handler: ProcessWatcher, disk_loader: DiskCacheLoader = None) -> TransactionResult:
|
||||
man = self._get_manager_for(pkg)
|
||||
|
||||
if man:
|
||||
ti = time.time()
|
||||
disk_loader = self.disk_loader_factory.new()
|
||||
disk_loader.start()
|
||||
self.logger.info(f"Uninstalling {pkg.name}")
|
||||
try:
|
||||
res = man.uninstall(pkg, root_password, handler, disk_loader)
|
||||
disk_loader.stop_working()
|
||||
disk_loader.join()
|
||||
self._update_post_transaction_status(res)
|
||||
return res
|
||||
except Exception:
|
||||
traceback.print_exc()
|
||||
return TransactionResult(success=False, installed=[], removed=[])
|
||||
finally:
|
||||
tf = time.time()
|
||||
self.logger.info(f'Uninstallation of {pkg} took {(tf - ti) / 60:.2f} minutes')
|
||||
|
||||
def install(self, app: SoftwarePackage, root_password: Optional[str], disk_loader: DiskCacheLoader, handler: ProcessWatcher) -> TransactionResult:
|
||||
man = self._get_manager_for(app)
|
||||
|
||||
if man:
|
||||
ti = time.time()
|
||||
disk_loader = self.disk_loader_factory.new()
|
||||
disk_loader.start()
|
||||
try:
|
||||
self.logger.info(f'Installing {app}')
|
||||
res = man.install(app, root_password, disk_loader, handler)
|
||||
disk_loader.stop_working()
|
||||
disk_loader.join()
|
||||
self._update_post_transaction_status(res)
|
||||
return res
|
||||
except Exception:
|
||||
traceback.print_exc()
|
||||
return TransactionResult(success=False, installed=[], removed=[])
|
||||
finally:
|
||||
tf = time.time()
|
||||
self.logger.info(f'Installation of {app} took {(tf - ti) / 60:.2f} minutes')
|
||||
|
||||
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(f'{man.__class__.__name__} took {mtf - mti:.2f} seconds')
|
||||
return history
|
||||
|
||||
def get_managed_types(self) -> Set[Type[SoftwarePackage]]:
|
||||
available_types = set()
|
||||
|
||||
for man in self.get_working_managers():
|
||||
available_types.update(man.get_managed_types())
|
||||
|
||||
return available_types
|
||||
|
||||
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 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: SoftwareAction, app: SoftwarePackage) -> bool:
|
||||
if app is None:
|
||||
if self.managers:
|
||||
for man in self.managers:
|
||||
if self._can_work(man):
|
||||
if man.requires_root(action, app):
|
||||
return True
|
||||
return False
|
||||
else:
|
||||
man = self._get_manager_for(app)
|
||||
|
||||
if man:
|
||||
return man.requires_root(action, app)
|
||||
|
||||
def prepare(self, task_manager: TaskManager, root_password: Optional[str], internet_available: bool):
|
||||
ti = time.time()
|
||||
self.logger.info("Initializing")
|
||||
taskman = task_manager if task_manager else TaskManager() # empty task manager to prevent null pointers
|
||||
|
||||
create_config = CreateConfigFile(taskman=taskman, configman=self.configman, i18n=self.i18n,
|
||||
task_icon_path=get_path('img/logo.svg'), logger=self.logger)
|
||||
create_config.start()
|
||||
|
||||
if self.managers:
|
||||
internet_on = self.context.is_internet_available()
|
||||
prepare_tasks = []
|
||||
for man in self.managers:
|
||||
if man not in self._already_prepared and self._can_work(man):
|
||||
t = Thread(target=man.prepare, args=(taskman, root_password, internet_on), daemon=True)
|
||||
t.start()
|
||||
prepare_tasks.append(t)
|
||||
self._already_prepared.append(man)
|
||||
|
||||
for t in prepare_tasks:
|
||||
t.join()
|
||||
|
||||
create_config.join()
|
||||
tf = time.time()
|
||||
self.logger.info(f'Finished ({tf - ti:.2f} seconds)')
|
||||
|
||||
def cache_available_managers(self):
|
||||
if self.managers:
|
||||
for man in self.managers:
|
||||
self._can_work(man)
|
||||
|
||||
def list_updates(self, internet_available: bool = None) -> List[PackageUpdate]:
|
||||
self._wait_to_be_ready()
|
||||
|
||||
updates = []
|
||||
|
||||
if self.managers:
|
||||
net_available = self.context.is_internet_available()
|
||||
|
||||
for man in self.managers:
|
||||
if self._can_work(man):
|
||||
man_updates = man.list_updates(internet_available=net_available)
|
||||
if man_updates:
|
||||
updates.extend(man_updates)
|
||||
|
||||
return updates
|
||||
|
||||
def list_warnings(self, internet_available: bool = None) -> List[str]:
|
||||
warnings = []
|
||||
|
||||
int_available = self.context.is_internet_available()
|
||||
|
||||
if int_available:
|
||||
updates_msg = check_for_update(self.logger, self.http_client, self.i18n)
|
||||
|
||||
if updates_msg:
|
||||
warnings.append(updates_msg)
|
||||
|
||||
if self.managers:
|
||||
for man in self.managers:
|
||||
if self._can_work(man):
|
||||
man_warnings = man.list_warnings(internet_available=int_available)
|
||||
|
||||
if man_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(f'{man.__class__.__name__} took {mtf - mti:.5f} seconds')
|
||||
|
||||
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 self.force_suggestions or bool(self.config['suggestions']['enabled']):
|
||||
if self.managers and self.context.is_internet_available():
|
||||
suggestions, threads = [], []
|
||||
for man in self.managers:
|
||||
t = Thread(target=self._fill_suggestions,
|
||||
args=(suggestions, man, int(self.config['suggestions']['by_type']), filter_installed),
|
||||
daemon=True)
|
||||
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: CustomSoftwareAction, pkg: SoftwarePackage, root_password: Optional[str], watcher: ProcessWatcher):
|
||||
if action.requires_internet and not self.context.is_internet_available():
|
||||
raise NoInternetException()
|
||||
|
||||
man = action.manager if action.manager else self._get_manager_for(pkg)
|
||||
|
||||
if man:
|
||||
return eval(f"man.{action.manager_method}({'pkg=pkg, ' if pkg else ''}root_password=root_password, watcher=watcher)")
|
||||
|
||||
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(f'Launching {pkg}')
|
||||
man.launch(pkg)
|
||||
|
||||
def get_screenshots(self, pkg: SoftwarePackage) -> Generator[str, None, None]:
|
||||
man = self._get_manager_for(pkg)
|
||||
|
||||
if man:
|
||||
yield from man.get_screenshots(pkg)
|
||||
|
||||
def get_working_managers(self):
|
||||
return [m for m in self.managers if self._can_work(m)]
|
||||
|
||||
def get_settings(self) -> Optional[Generator[SettingsView, None, None]]:
|
||||
if self.settings_manager is None:
|
||||
self.settings_manager = GenericSettingsManager(managers=self.managers,
|
||||
working_managers=self.working_managers,
|
||||
configman=self.configman,
|
||||
context=self.context)
|
||||
else:
|
||||
self.settings_manager.managers = self.managers
|
||||
self.settings_manager.working_managers = self.working_managers
|
||||
|
||||
yield SettingsView(self, self.settings_manager.get_settings())
|
||||
|
||||
def save_settings(self, component: TabGroupComponent) -> Tuple[bool, Optional[List[str]]]:
|
||||
return self.settings_manager.save_settings(component)
|
||||
|
||||
def _map_pkgs_by_manager(self, pkgs: List[SoftwarePackage], pkg_filters: list = None) -> Dict[SoftwareManager, List[SoftwarePackage]]:
|
||||
by_manager = {}
|
||||
for pkg in pkgs:
|
||||
if pkg_filters and not all((1 for f in pkg_filters if f(pkg))):
|
||||
continue
|
||||
|
||||
man = self._get_manager_for(pkg)
|
||||
|
||||
if man:
|
||||
man_pkgs = by_manager.get(man)
|
||||
|
||||
if man_pkgs is None:
|
||||
man_pkgs = []
|
||||
by_manager[man] = man_pkgs
|
||||
|
||||
man_pkgs.append(pkg)
|
||||
|
||||
return by_manager
|
||||
|
||||
def get_upgrade_requirements(self, pkgs: List[SoftwarePackage], root_password: Optional[str], watcher: ProcessWatcher) -> UpgradeRequirements:
|
||||
by_manager = self._map_pkgs_by_manager(pkgs)
|
||||
res = GenericUpgradeRequirements([], [], [], [], {})
|
||||
|
||||
if by_manager:
|
||||
for man, pkgs in by_manager.items():
|
||||
ti = time.time()
|
||||
man_reqs = man.get_upgrade_requirements(pkgs, root_password, watcher)
|
||||
tf = time.time()
|
||||
self.logger.info(f'{man.__class__.__name__} took {tf - ti:.2f} seconds')
|
||||
|
||||
if not man_reqs:
|
||||
return # it means the process should be stopped
|
||||
|
||||
if man_reqs:
|
||||
res.sub_requirements[man] = man_reqs
|
||||
if man_reqs.to_install:
|
||||
res.to_install.extend(man_reqs.to_install)
|
||||
|
||||
if man_reqs.to_remove:
|
||||
res.to_remove.extend(man_reqs.to_remove)
|
||||
|
||||
if man_reqs.to_upgrade:
|
||||
res.to_upgrade.extend(man_reqs.to_upgrade)
|
||||
|
||||
if man_reqs.cannot_upgrade:
|
||||
res.cannot_upgrade.extend(man_reqs.cannot_upgrade)
|
||||
|
||||
return res
|
||||
|
||||
def reset(self, root_password: Optional[str], watcher: ProcessWatcher) -> bool:
|
||||
body = f"<p>{self.i18n['action.reset.body_1'].format(bold(self.context.app_name))}</p>" \
|
||||
f"<p>{self.i18n['action.reset.body_2']}</p>"
|
||||
|
||||
if watcher.request_confirmation(title=self.i18n['action.reset'],
|
||||
body=body,
|
||||
confirmation_label=self.i18n['proceed'].capitalize(),
|
||||
deny_label=self.i18n['cancel'].capitalize()):
|
||||
|
||||
try:
|
||||
clean_app_files(managers=self.managers, logs=False)
|
||||
restart_app()
|
||||
except Exception:
|
||||
return False
|
||||
|
||||
return True
|
||||
|
||||
def gen_custom_actions(self) -> Generator[CustomSoftwareAction, None, None]:
|
||||
if self.managers:
|
||||
working_managers = []
|
||||
|
||||
for man in self.managers:
|
||||
if self._can_work(man):
|
||||
working_managers.append(man)
|
||||
|
||||
if working_managers:
|
||||
working_managers.sort(key=lambda m: m.__class__.__name__)
|
||||
|
||||
for man in working_managers:
|
||||
for action in man.gen_custom_actions():
|
||||
action.manager = man
|
||||
yield action
|
||||
|
||||
app_config = self.configman.get_config()
|
||||
|
||||
for action, available in self.dynamic_extra_actions.items():
|
||||
if available(app_config):
|
||||
yield action
|
||||
|
||||
yield self.action_reset
|
||||
|
||||
def _fill_sizes(self, man: SoftwareManager, pkgs: List[SoftwarePackage]):
|
||||
ti = time.time()
|
||||
man.fill_sizes(pkgs)
|
||||
tf = time.time()
|
||||
self.logger.info(f'{man.__class__.__name__} took {tf - ti:.2f} seconds')
|
||||
|
||||
def fill_sizes(self, pkgs: List[SoftwarePackage]):
|
||||
by_manager = self._map_pkgs_by_manager(pkgs, pkg_filters=[lambda p: p.size is None])
|
||||
|
||||
if by_manager:
|
||||
threads = []
|
||||
for man, man_pkgs in by_manager.items():
|
||||
if man_pkgs:
|
||||
t = Thread(target=self._fill_sizes, args=(man, man_pkgs), daemon=True)
|
||||
t.start()
|
||||
threads.append(t)
|
||||
|
||||
for t in threads:
|
||||
t.join()
|
||||
|
||||
def ignore_update(self, pkg: SoftwarePackage):
|
||||
manager = self._get_manager_for(pkg)
|
||||
|
||||
if manager:
|
||||
manager.ignore_update(pkg)
|
||||
|
||||
if pkg.is_update_ignored():
|
||||
if pkg.categories is None:
|
||||
pkg.categories = ['updates_ignored']
|
||||
elif 'updates_ignored' not in pkg.categories:
|
||||
self._add_category(pkg, 'updates_ignored')
|
||||
|
||||
def revert_ignored_update(self, pkg: SoftwarePackage):
|
||||
manager = self._get_manager_for(pkg)
|
||||
|
||||
if manager:
|
||||
manager.revert_ignored_update(pkg)
|
||||
|
||||
if not pkg.is_update_ignored() and pkg.categories and 'updates_ignored' in pkg.categories:
|
||||
if isinstance(pkg.categories, tuple):
|
||||
pkg.categories = tuple(c for c in pkg.categories if c != 'updates_ignored')
|
||||
else:
|
||||
pkg.categories.remove('updates_ignored')
|
||||
@@ -1,328 +0,0 @@
|
||||
import os
|
||||
import re
|
||||
import shutil
|
||||
import time
|
||||
import traceback
|
||||
from io import StringIO, BytesIO
|
||||
from logging import Logger
|
||||
from math import floor
|
||||
from pathlib import Path
|
||||
from threading import Thread
|
||||
from typing import Optional, Tuple
|
||||
|
||||
from bauh.api.abstract.download import FileDownloader
|
||||
from bauh.api.abstract.handler import ProcessWatcher
|
||||
from bauh.api.http import HttpClient
|
||||
from bauh.commons.html import bold
|
||||
from bauh.commons.system import ProcessHandler, SimpleProcess
|
||||
from bauh.commons.view_utils import get_human_size_str
|
||||
from bauh.view.util.translation import I18n
|
||||
|
||||
RE_HAS_EXTENSION = re.compile(r'.+\.\w+$')
|
||||
|
||||
|
||||
class SelfFileDownloader(FileDownloader):
|
||||
|
||||
def __init__(self, logger: Logger, i18n: I18n, http_client: HttpClient,
|
||||
check_ssl: bool):
|
||||
self._logger = logger
|
||||
self._i18n = i18n
|
||||
self._client = http_client
|
||||
self._ssl = check_ssl
|
||||
|
||||
def is_multithreaded(self) -> bool:
|
||||
return False
|
||||
|
||||
def can_work(self) -> bool:
|
||||
return True
|
||||
|
||||
def get_supported_multithreaded_clients(self) -> Tuple[str, ...]:
|
||||
return tuple()
|
||||
|
||||
def is_multithreaded_client_available(self, name: str) -> bool:
|
||||
return False
|
||||
|
||||
def list_available_multithreaded_clients(self) -> Tuple[str, ...]:
|
||||
return tuple()
|
||||
|
||||
def get_supported_clients(self) -> Tuple[str, ...]:
|
||||
return tuple()
|
||||
|
||||
def download(self, file_url: str, watcher: Optional[ProcessWatcher], output_path: str, cwd: str,
|
||||
root_password: Optional[str] = None, substatus_prefix: str = None, display_file_size: bool = True,
|
||||
max_threads: int = None, known_size: int = None) -> bool:
|
||||
try:
|
||||
res = self._client.get(url=file_url, ignore_ssl=not self._ssl, stream=True)
|
||||
except Exception:
|
||||
return False
|
||||
|
||||
try:
|
||||
content_length = int(res.headers.get("content-length", 0))
|
||||
except Exception:
|
||||
content_length = 0
|
||||
self._logger.warning(f"Could not retrieve the content-length for file '{file_url}'")
|
||||
|
||||
file_name = file_url.split("/")[-1]
|
||||
msg = StringIO()
|
||||
msg.write(f"{substatus_prefix} " if substatus_prefix else "")
|
||||
msg.write(f"{self._i18n['downloading']} {bold(file_name)}")
|
||||
base_msg = msg.getvalue()
|
||||
|
||||
byte_stream = BytesIO()
|
||||
total_downloaded = 0
|
||||
known_size = content_length and content_length > 0
|
||||
total_size_str = get_human_size_str(content_length) if known_size > 0 else "?"
|
||||
|
||||
try:
|
||||
for data in res.iter_content(chunk_size=1024):
|
||||
byte_stream.write(data)
|
||||
total_downloaded += len(data)
|
||||
perc = f"({(total_downloaded / content_length) * 100:.2f}%) " if known_size > 0 else ""
|
||||
watcher.change_substatus(f"{perc}{base_msg} ({get_human_size_str(total_downloaded)} / {total_size_str})")
|
||||
except Exception:
|
||||
self._logger.error(f"Unexpected exception while downloading file from '{file_url}'")
|
||||
traceback.print_exc()
|
||||
return False
|
||||
|
||||
self._logger.info(f"Writing downloaded file content to disk: {output_path}")
|
||||
|
||||
try:
|
||||
with open(output_path, "wb+") as f:
|
||||
f.write(byte_stream.getvalue())
|
||||
except Exception:
|
||||
self._logger.error(f"Unexpected exception when saving downloaded content to disk: {output_path}")
|
||||
traceback.print_exc()
|
||||
return False
|
||||
|
||||
return True
|
||||
|
||||
|
||||
class AdaptableFileDownloader(FileDownloader):
|
||||
|
||||
def __init__(self, logger: Logger, multithread_enabled: bool, i18n: I18n, http_client: HttpClient,
|
||||
multithread_client: str, check_ssl: bool):
|
||||
self.logger = logger
|
||||
self.multithread_enabled = multithread_enabled
|
||||
self.i18n = i18n
|
||||
self.http_client = http_client
|
||||
self.supported_multithread_clients = ("aria2", "axel")
|
||||
self.multithread_client = multithread_client
|
||||
self.check_ssl = check_ssl
|
||||
self._self_downloader = SelfFileDownloader(logger=logger,
|
||||
i18n=i18n,
|
||||
http_client=http_client,
|
||||
check_ssl=check_ssl)
|
||||
|
||||
@staticmethod
|
||||
def is_aria2c_available() -> bool:
|
||||
return bool(shutil.which('aria2c'))
|
||||
|
||||
@staticmethod
|
||||
def is_axel_available() -> bool:
|
||||
return bool(shutil.which('axel'))
|
||||
|
||||
def _get_aria2c_process(self, url: str, output_path: str, cwd: str, root_password: Optional[str], threads: int) -> SimpleProcess:
|
||||
cmd = ['aria2c', url,
|
||||
'--no-conf',
|
||||
'-x', '16',
|
||||
'--enable-color=false',
|
||||
'--stderr=true',
|
||||
'--summary-interval=0',
|
||||
'--disable-ipv6',
|
||||
'-k', '1M',
|
||||
'--allow-overwrite=true',
|
||||
'-c',
|
||||
'-t', '5',
|
||||
'--max-file-not-found=3',
|
||||
'--file-allocation=none',
|
||||
'--console-log-level=error']
|
||||
|
||||
if threads > 1:
|
||||
cmd.append('-s')
|
||||
cmd.append(str(threads))
|
||||
|
||||
if output_path:
|
||||
output_split = output_path.split('/')
|
||||
cmd.append('-d')
|
||||
cmd.append('/'.join(output_split[:-1]))
|
||||
cmd.append('-o')
|
||||
cmd.append(output_split[-1])
|
||||
|
||||
return SimpleProcess(cmd=cmd, root_password=root_password, cwd=cwd)
|
||||
|
||||
def _get_axel_process(self, url: str, output_path: str, cwd: str, root_password: Optional[str], threads: int) -> SimpleProcess:
|
||||
cmd = ['axel', url, '-n', str(threads), '-4', '-c', '-T', '5']
|
||||
|
||||
if not self.check_ssl:
|
||||
cmd.append('-k')
|
||||
|
||||
if output_path:
|
||||
cmd.append(f'--output={output_path}')
|
||||
|
||||
return SimpleProcess(cmd=cmd, cwd=cwd, root_password=root_password)
|
||||
|
||||
def _rm_bad_file(self, file_name: str, output_path: str, cwd, handler: ProcessHandler, root_password: Optional[str]):
|
||||
to_delete = output_path if output_path else f'{cwd}/{file_name}'
|
||||
|
||||
if to_delete and os.path.exists(to_delete):
|
||||
self.logger.info(f'Removing downloaded file {to_delete}')
|
||||
success, _ = handler.handle_simple(SimpleProcess(['rm', '-rf', to_delete], root_password=root_password))
|
||||
return success
|
||||
|
||||
def _concat_file_size(self, file_url: str, base_substatus: StringIO, watcher: ProcessWatcher):
|
||||
watcher.change_substatus(f'{base_substatus.getvalue()} ( ? Mb )')
|
||||
|
||||
try:
|
||||
size = self.http_client.get_content_length(file_url)
|
||||
|
||||
if size:
|
||||
base_substatus.write(f' ( {size} )')
|
||||
watcher.change_substatus(base_substatus.getvalue())
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
def _get_appropriate_threads_number(self, max_threads: int, known_size: int) -> int:
|
||||
if max_threads and max_threads > 0:
|
||||
threads = max_threads
|
||||
elif known_size:
|
||||
threads = 16 if known_size >= 16000000 else floor(known_size / 1000000)
|
||||
|
||||
if threads <= 0:
|
||||
threads = 1
|
||||
else:
|
||||
threads = 16
|
||||
|
||||
return threads
|
||||
|
||||
def _download_with_threads(self, client: str, file_url: str, output_path: str, cwd: str,
|
||||
max_threads: int, known_size: int, display_file_size: bool, handler: ProcessHandler,
|
||||
root_password: Optional[str] = None, substatus_prefix: Optional[str] = None) \
|
||||
-> Tuple[float, bool]:
|
||||
|
||||
threads = self._get_appropriate_threads_number(max_threads, known_size)
|
||||
|
||||
if client == 'aria2':
|
||||
start_time = time.time()
|
||||
process = self._get_aria2c_process(file_url, output_path, cwd, root_password, threads)
|
||||
downloader = 'aria2'
|
||||
else:
|
||||
start_time = time.time()
|
||||
process = self._get_axel_process(file_url, output_path, cwd, root_password, threads)
|
||||
downloader = 'axel'
|
||||
|
||||
name = file_url.split('/')[-1]
|
||||
|
||||
if output_path and not RE_HAS_EXTENSION.match(name) and RE_HAS_EXTENSION.match(output_path):
|
||||
name = output_path.split('/')[-1]
|
||||
|
||||
if handler.watcher:
|
||||
msg = StringIO()
|
||||
msg.write(f'{substatus_prefix} ' if substatus_prefix else '')
|
||||
msg.write(f"{bold('[{}]'.format(downloader))} {self.i18n['downloading']} {bold(name)}")
|
||||
|
||||
if display_file_size:
|
||||
if known_size:
|
||||
msg.write(f' ( {get_human_size_str(known_size)} )')
|
||||
handler.watcher.change_substatus(msg.getvalue())
|
||||
else:
|
||||
Thread(target=self._concat_file_size, args=(file_url, msg, handler.watcher), daemon=True).start()
|
||||
else:
|
||||
msg.write(' ( ? Mb )')
|
||||
handler.watcher.change_substatus(msg.getvalue())
|
||||
|
||||
success, _ = handler.handle_simple(process)
|
||||
return start_time, success
|
||||
|
||||
def download(self, file_url: str, watcher: ProcessWatcher, output_path: str = None, cwd: str = None, root_password: Optional[str] = None, substatus_prefix: str = None, display_file_size: bool = True, max_threads: int = None, known_size: int = None) -> bool:
|
||||
self.logger.info(f'Downloading {file_url}')
|
||||
handler = ProcessHandler(watcher)
|
||||
file_name = file_url.split('/')[-1]
|
||||
|
||||
final_cwd = cwd if cwd else '.'
|
||||
|
||||
success = False
|
||||
start_time = time.time()
|
||||
try:
|
||||
if output_path:
|
||||
if os.path.exists(output_path):
|
||||
self.logger.info(f'Removing old file found before downloading: {output_path}')
|
||||
os.remove(output_path)
|
||||
self.logger.info(f'Old file {output_path} removed')
|
||||
else:
|
||||
output_dir = os.path.dirname(output_path)
|
||||
|
||||
try:
|
||||
Path(output_dir).mkdir(exist_ok=True, parents=True)
|
||||
except OSError:
|
||||
self.logger.error(f"Could not make download directory '{output_dir}'")
|
||||
watcher.print(self.i18n['error.mkdir'].format(dir=output_dir))
|
||||
return False
|
||||
|
||||
threaded_client = self.get_available_multithreaded_tool()
|
||||
if threaded_client:
|
||||
start_time, success = self._download_with_threads(client=threaded_client, file_url=file_url,
|
||||
output_path=output_path,
|
||||
cwd=final_cwd, max_threads=max_threads,
|
||||
known_size=known_size, handler=handler,
|
||||
display_file_size=display_file_size,
|
||||
root_password=root_password)
|
||||
else:
|
||||
start_time = time.time()
|
||||
success = self._self_downloader.download(file_url=file_url, watcher=watcher, output_path=output_path,
|
||||
cwd=cwd, root_password=root_password,
|
||||
substatus_prefix=substatus_prefix,
|
||||
display_file_size=display_file_size, max_threads=max_threads,
|
||||
known_size=known_size)
|
||||
except Exception:
|
||||
traceback.print_exc()
|
||||
self._rm_bad_file(file_name, output_path, final_cwd, handler, root_password)
|
||||
|
||||
final_time = time.time()
|
||||
self.logger.info(f'{file_name} download took {(final_time - start_time) / 60:.4f} minutes')
|
||||
|
||||
if not success:
|
||||
self.logger.error(f"Could not download '{file_name}'")
|
||||
self._rm_bad_file(file_name, output_path, final_cwd, handler, root_password)
|
||||
|
||||
return success
|
||||
|
||||
def is_multithreaded(self) -> bool:
|
||||
return bool(self.get_available_multithreaded_tool())
|
||||
|
||||
def get_available_multithreaded_tool(self) -> str:
|
||||
if self.multithread_enabled:
|
||||
if self.multithread_client is None or self.multithread_client not in self.supported_multithread_clients:
|
||||
for client in self.supported_multithread_clients:
|
||||
if self.is_multithreaded_client_available(client):
|
||||
return client
|
||||
else:
|
||||
possible_clients = {*self.supported_multithread_clients}
|
||||
|
||||
if self.is_multithreaded_client_available(self.multithread_client):
|
||||
return self.multithread_client
|
||||
else:
|
||||
possible_clients.remove(self.multithread_client)
|
||||
|
||||
for client in possible_clients:
|
||||
if self.is_multithreaded_client_available(client):
|
||||
return client
|
||||
|
||||
def can_work(self) -> bool:
|
||||
return True
|
||||
|
||||
def get_supported_multithreaded_clients(self) -> Tuple[str, ...]:
|
||||
return self.supported_multithread_clients
|
||||
|
||||
def is_multithreaded_client_available(self, name: str) -> bool:
|
||||
if name == 'aria2':
|
||||
return self.is_aria2c_available()
|
||||
elif name == 'axel':
|
||||
return self.is_axel_available()
|
||||
else:
|
||||
return False
|
||||
|
||||
def list_available_multithreaded_clients(self) -> Tuple[str, ...]:
|
||||
return tuple(c for c in self.supported_multithread_clients if self.is_multithreaded_client_available(c))
|
||||
|
||||
def get_supported_clients(self) -> Tuple[str, ...]:
|
||||
return "self", "aria2", "axel"
|
||||
@@ -1,628 +0,0 @@
|
||||
import os
|
||||
import time
|
||||
import traceback
|
||||
from math import floor
|
||||
from threading import Thread
|
||||
from typing import List, Tuple, Optional, Dict, Type, Iterable
|
||||
|
||||
from PyQt5.QtWidgets import QApplication, QStyleFactory
|
||||
|
||||
from bauh import ROOT_DIR, __app_name__
|
||||
from bauh.api.abstract.context import ApplicationContext
|
||||
from bauh.api.abstract.controller import SoftwareManager, SettingsController, SettingsView
|
||||
from bauh.api.abstract.view import TabComponent, InputOption, TextComponent, MultipleSelectComponent, \
|
||||
PanelComponent, FormComponent, TabGroupComponent, SingleSelectComponent, SelectViewType, TextInputComponent, \
|
||||
FileChooserComponent, RangeInputComponent, ViewComponentAlignment
|
||||
from bauh.commons.view_utils import new_select
|
||||
from bauh.view.core import timeshift
|
||||
from bauh.view.core.config import CoreConfigManager, BACKUP_REMOVE_METHODS, BACKUP_DEFAULT_REMOVE_METHOD
|
||||
from bauh.view.core.downloader import AdaptableFileDownloader
|
||||
from bauh.view.util import translation
|
||||
|
||||
|
||||
class GenericSettingsManager(SettingsController):
|
||||
|
||||
def __init__(self, context: ApplicationContext, managers: List[SoftwareManager],
|
||||
working_managers: List[SoftwareManager], configman: CoreConfigManager):
|
||||
self.context = context
|
||||
self.i18n = context.i18n
|
||||
self.managers = managers
|
||||
self.working_managers = working_managers
|
||||
self.logger = context.logger
|
||||
self.file_downloader = self.context.file_downloader
|
||||
self.configman = configman
|
||||
self._settings_views: Optional[Dict[Type, List[SettingsView]]] = None
|
||||
|
||||
def get_settings(self) -> TabGroupComponent:
|
||||
tabs = list()
|
||||
|
||||
gem_opts, def_gem_opts, gem_tabs = [], set(), []
|
||||
|
||||
self._settings_views = dict()
|
||||
|
||||
for man in self.managers:
|
||||
can_work, reason_not_work = man.can_work()
|
||||
modname = man.__module__.split('.')[-2]
|
||||
|
||||
man_settings = man.get_settings() if can_work else None
|
||||
if man_settings:
|
||||
for view in man_settings:
|
||||
icon_path = view.icon_path
|
||||
|
||||
if not icon_path:
|
||||
icon_path = f"{ROOT_DIR}/gems/{modname}/resources/img/{modname}.svg"
|
||||
|
||||
tab_name = view.label if view.label else self.i18n.get(f'gem.{modname}.label', modname.capitalize())
|
||||
gem_tabs.append(TabComponent(label=tab_name, content=view.component, icon_path=icon_path))
|
||||
|
||||
views = self._settings_views.get(man.__class__, list())
|
||||
self._settings_views[man.__class__] = views
|
||||
views.append(view)
|
||||
|
||||
help_tip = reason_not_work if not can_work and reason_not_work else self.i18n.get(f'gem.{modname}.info')
|
||||
opt = InputOption(label=self.i18n.get(f'gem.{modname}.label', modname.capitalize()),
|
||||
tooltip=help_tip,
|
||||
value=modname,
|
||||
icon_path=f'{ROOT_DIR}/gems/{modname}/resources/img/{modname}.svg',
|
||||
read_only=not can_work,
|
||||
extra_properties={'warning': 'true'} if not can_work else None)
|
||||
gem_opts.append(opt)
|
||||
|
||||
if man.is_enabled() and man in self.working_managers:
|
||||
def_gem_opts.add(opt)
|
||||
|
||||
core_config = self.configman.get_config()
|
||||
|
||||
if gem_opts:
|
||||
type_help = TextComponent(html=self.i18n['core.config.types.tip'])
|
||||
gem_opts.sort(key=lambda o: o.value)
|
||||
gem_selector = MultipleSelectComponent(label=None,
|
||||
tooltip=None,
|
||||
options=gem_opts,
|
||||
max_width=floor(self.context.screen_width * 0.22),
|
||||
default_options=def_gem_opts,
|
||||
id_="gems")
|
||||
tabs.append(TabComponent(label=self.i18n['core.config.tab.types'],
|
||||
content=PanelComponent([type_help, FormComponent([gem_selector], spaces=False)]),
|
||||
id_='core.types'))
|
||||
|
||||
tabs.append(self._gen_general_settings(core_config))
|
||||
tabs.append(self._gen_interface_settings(core_config))
|
||||
tabs.append(self._gen_tray_settings(core_config))
|
||||
tabs.append(self._gen_adv_settings(core_config))
|
||||
|
||||
bkp_settings = self._gen_backup_settings(core_config)
|
||||
|
||||
if bkp_settings:
|
||||
tabs.append(bkp_settings)
|
||||
|
||||
for tab in gem_tabs:
|
||||
tabs.append(tab)
|
||||
|
||||
return TabGroupComponent(tabs)
|
||||
|
||||
def _gen_adv_settings(self, core_config: dict) -> TabComponent:
|
||||
|
||||
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_trim = new_select(label=self.i18n['core.config.trim.after_upgrade'],
|
||||
tip=self.i18n['core.config.trim.after_upgrade.tip'],
|
||||
value=core_config['disk']['trim']['after_upgrade'],
|
||||
opts=[(self.i18n['yes'].capitalize(), True, None),
|
||||
(self.i18n['no'].capitalize(), False, None),
|
||||
(self.i18n['ask'].capitalize(), None, None)],
|
||||
id_='trim_after_upgrade')
|
||||
|
||||
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_check_ssl = self._gen_bool_component(label=self.i18n['core.config.download.check_ssl'],
|
||||
tooltip=self.i18n['core.config.download.check_ssl.tip'],
|
||||
value=core_config['download']['check_ssl'],
|
||||
id_='download.check_ssl')
|
||||
|
||||
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'])
|
||||
|
||||
select_mthread_client = self._gen_multithread_client_select(core_config)
|
||||
|
||||
inputs = [select_dmthread, select_mthread_client, select_check_ssl, select_trim, select_dep_check,
|
||||
input_data_exp, input_icon_exp]
|
||||
panel = PanelComponent([FormComponent(inputs, spaces=False)], id_='advanced')
|
||||
return TabComponent(self.i18n['core.config.tab.advanced'].capitalize(), panel, None, 'core.adv')
|
||||
|
||||
def _gen_multithread_client_select(self, core_config: dict) -> SingleSelectComponent:
|
||||
available_mthread_clients = [*self.file_downloader.list_available_multithreaded_clients()]
|
||||
available_mthread_clients.sort()
|
||||
|
||||
default_i18n_key = 'default' if available_mthread_clients else 'core.config.download.multithreaded_client.none'
|
||||
mthread_client_opts = [(self.i18n[default_i18n_key].capitalize(), None, None)]
|
||||
|
||||
for client in available_mthread_clients:
|
||||
mthread_client_opts.append((client, client, None))
|
||||
|
||||
current_mthread_client = core_config['download']['multithreaded_client']
|
||||
|
||||
if current_mthread_client not in available_mthread_clients:
|
||||
current_mthread_client = None
|
||||
|
||||
return new_select(label=self.i18n['core.config.download.multithreaded_client'],
|
||||
tip=self.i18n['core.config.download.multithreaded_client.tip'],
|
||||
id_="mthread_client",
|
||||
opts=mthread_client_opts,
|
||||
value=current_mthread_client)
|
||||
|
||||
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'}
|
||||
de_path = str(core_config['ui']['tray']['default_icon']) if core_config['ui']['tray']['default_icon'] else None
|
||||
select_def_icon = FileChooserComponent(id_='def_icon',
|
||||
label=self.i18n["core.config.ui.tray.default_icon"],
|
||||
tooltip=self.i18n["core.config.ui.tray.default_icon.tip"],
|
||||
file_path=de_path,
|
||||
allowed_extensions=allowed_exts)
|
||||
|
||||
up_path = str(core_config['ui']['tray']['updates_icon']) if core_config['ui']['tray']['updates_icon'] else None
|
||||
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=up_path,
|
||||
allowed_extensions=allowed_exts)
|
||||
|
||||
sub_comps = [FormComponent([select_def_icon, select_up_icon, input_update_interval], spaces=False)]
|
||||
return TabComponent(self.i18n['core.config.tab.tray'].capitalize(),
|
||||
PanelComponent(sub_comps, id_='tray'), None, 'core.tray')
|
||||
|
||||
def _gen_interface_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'],
|
||||
value=bool(core_config['ui']['hdpi']),
|
||||
id_='hdpi')
|
||||
|
||||
scale_tip = self.i18n['core.config.ui.auto_scale.tip'].format('QT_AUTO_SCREEN_SCALE_FACTOR')
|
||||
select_ascale = self._gen_bool_component(label=self.i18n['core.config.ui.auto_scale'],
|
||||
tooltip=scale_tip,
|
||||
value=bool(core_config['ui']['auto_scale']),
|
||||
id_='auto_scale')
|
||||
|
||||
try:
|
||||
scale = float(core_config['ui']['scale_factor'])
|
||||
|
||||
if scale < 1.0:
|
||||
scale = 1.0
|
||||
except ValueError:
|
||||
scale = 1.0
|
||||
|
||||
select_scale = RangeInputComponent(id_="scalef", label=self.i18n['core.config.ui.scale_factor'] + ' (%)',
|
||||
tooltip=self.i18n['core.config.ui.scale_factor.tip'],
|
||||
min_value=100, max_value=400, step_value=5, value=int(scale * 100))
|
||||
|
||||
if not core_config['ui']['qt_style']:
|
||||
cur_style = QApplication.instance().property('qt_style')
|
||||
else:
|
||||
cur_style = core_config['ui']['qt_style']
|
||||
|
||||
style_opts = [InputOption(label=s.capitalize(), value=s.lower()) for s in QStyleFactory.keys()]
|
||||
|
||||
default_style = [o for o in style_opts if o.value == cur_style]
|
||||
|
||||
if not default_style:
|
||||
if cur_style:
|
||||
default_style = InputOption(label=cur_style, value=cur_style)
|
||||
style_opts.append(default_style)
|
||||
else:
|
||||
default_style = style_opts[0]
|
||||
else:
|
||||
default_style = default_style[0]
|
||||
|
||||
select_style = SingleSelectComponent(label=self.i18n['style'].capitalize(),
|
||||
tooltip=self.i18n['core.config.ui.qt_style.tooltip'],
|
||||
options=style_opts,
|
||||
default_option=default_style,
|
||||
type_=SelectViewType.COMBO,
|
||||
alignment=ViewComponentAlignment.CENTER,
|
||||
id_="style")
|
||||
|
||||
systheme_tip = self.i18n['core.config.ui.system_theme.tip'].format(app=__app_name__)
|
||||
select_system_theme = self._gen_bool_component(label=self.i18n['core.config.ui.system_theme'],
|
||||
tooltip=systheme_tip,
|
||||
value=bool(core_config['ui']['system_theme']),
|
||||
id_='system_theme')
|
||||
|
||||
input_maxd = TextInputComponent(label=self.i18n['core.config.ui.max_displayed'],
|
||||
tooltip=self.i18n['core.config.ui.max_displayed.tip'],
|
||||
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_ascale, select_scale,
|
||||
select_dicons, select_system_theme,
|
||||
select_style, input_maxd], spaces=False)]
|
||||
|
||||
return TabComponent(self.i18n['core.config.tab.ui'].capitalize(),
|
||||
PanelComponent(sub_comps, id_='interface'), None, 'core.ui')
|
||||
|
||||
def _gen_general_settings(self, core_config: dict) -> TabComponent:
|
||||
locale_keys = translation.get_available_keys()
|
||||
locale_opts = [InputOption(label=self.i18n[f'locale.{k}'].capitalize(), value=k) for k in locale_keys]
|
||||
|
||||
current_locale = None
|
||||
|
||||
if core_config['locale']:
|
||||
current_locale = [loc for loc in locale_opts if loc.value == core_config['locale']]
|
||||
|
||||
if not current_locale:
|
||||
if self.i18n.current_key:
|
||||
current_locale = [loc for loc in locale_opts if loc.value == self.i18n.current_key]
|
||||
|
||||
if not current_locale:
|
||||
current_locale = [loc for loc in locale_opts if loc.value == self.i18n.default_key]
|
||||
|
||||
current_locale = current_locale[0] if current_locale else None
|
||||
|
||||
sel_locale = SingleSelectComponent(label=self.i18n['core.config.locale.label'],
|
||||
options=locale_opts,
|
||||
default_option=current_locale,
|
||||
type_=SelectViewType.COMBO,
|
||||
alignment=ViewComponentAlignment.CENTER,
|
||||
id_='locale')
|
||||
|
||||
sel_store_pwd = self._gen_bool_component(label=self.i18n['core.config.store_password'].capitalize(),
|
||||
tooltip=self.i18n['core.config.store_password.tip'].capitalize(),
|
||||
id_="store_pwd",
|
||||
value=bool(core_config['store_root_password']))
|
||||
|
||||
notify_tip = self.i18n['core.config.system.notifications.tip'].capitalize()
|
||||
sel_sys_notify = self._gen_bool_component(label=self.i18n['core.config.system.notifications'].capitalize(),
|
||||
tooltip=notify_tip,
|
||||
value=bool(core_config['system']['notifications']),
|
||||
id_="sys_notify")
|
||||
|
||||
sel_load_apps = self._gen_bool_component(label=self.i18n['core.config.boot.load_apps'],
|
||||
tooltip=self.i18n['core.config.boot.load_apps.tip'],
|
||||
value=bool(core_config['boot']['load_apps']),
|
||||
id_='boot.load_apps')
|
||||
|
||||
sel_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")
|
||||
|
||||
inp_reboot = new_select(label=self.i18n['core.config.updates.reboot'],
|
||||
tip=self.i18n['core.config.updates.reboot.tip'],
|
||||
id_='ask_for_reboot',
|
||||
max_width=None,
|
||||
value=bool(core_config['updates']['ask_for_reboot']),
|
||||
opts=[(self.i18n['ask'].capitalize(), True, None),
|
||||
(self.i18n['no'].capitalize(), False, None)])
|
||||
|
||||
inputs = [sel_locale, sel_store_pwd, sel_sys_notify, sel_load_apps, inp_reboot, sel_sugs, inp_sugs]
|
||||
panel = PanelComponent([FormComponent(inputs, spaces=False)], id_='general')
|
||||
return TabComponent(self.i18n['core.config.tab.general'].capitalize(), panel, None, 'core.gen')
|
||||
|
||||
def _gen_bool_component(self, label: str, tooltip: Optional[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 _save_settings(self, general: PanelComponent,
|
||||
advanced: PanelComponent,
|
||||
backup: PanelComponent,
|
||||
ui: PanelComponent,
|
||||
tray: PanelComponent,
|
||||
gems_panel: PanelComponent) -> Tuple[bool, Optional[List[str]]]:
|
||||
core_config = self.configman.get_config()
|
||||
|
||||
# general
|
||||
gen_form = general.get_component_by_idx(0, FormComponent)
|
||||
|
||||
locale = gen_form.get_component('locale', SingleSelectComponent).get_selected()
|
||||
|
||||
if locale != self.i18n.current_key:
|
||||
core_config['locale'] = locale
|
||||
|
||||
notifications = gen_form.get_component('sys_notify', SingleSelectComponent).get_selected()
|
||||
core_config['system']['notifications'] = notifications
|
||||
|
||||
suggestions = gen_form.get_component('sugs_enabled', SingleSelectComponent).get_selected()
|
||||
core_config['suggestions']['enabled'] = suggestions
|
||||
|
||||
store_root_pwd = gen_form.get_component('store_pwd', SingleSelectComponent).get_selected()
|
||||
core_config['store_root_password'] = store_root_pwd
|
||||
|
||||
sugs_by_type = gen_form.get_component('sugs_by_type', TextInputComponent).get_int_value()
|
||||
core_config['suggestions']['by_type'] = sugs_by_type
|
||||
|
||||
ask_reboot = gen_form.get_component('ask_for_reboot', SingleSelectComponent).get_selected()
|
||||
core_config['updates']['ask_for_reboot'] = ask_reboot
|
||||
|
||||
load_apps = gen_form.get_component('boot.load_apps', SingleSelectComponent).get_selected()
|
||||
core_config['boot']['load_apps'] = load_apps
|
||||
|
||||
# advanced
|
||||
adv_form = advanced.get_component_by_idx(0, FormComponent)
|
||||
|
||||
download_mthreaded = adv_form.get_component('down_mthread', SingleSelectComponent).get_selected()
|
||||
core_config['download']['multithreaded'] = download_mthreaded
|
||||
|
||||
mthread_client = adv_form.get_component('mthread_client', SingleSelectComponent).get_selected()
|
||||
core_config['download']['multithreaded_client'] = mthread_client
|
||||
|
||||
check_ssl = adv_form.get_component('download.check_ssl', SingleSelectComponent).get_selected()
|
||||
core_config['download']['check_ssl'] = check_ssl
|
||||
|
||||
if isinstance(self.file_downloader, AdaptableFileDownloader):
|
||||
self.file_downloader.multithread_client = mthread_client
|
||||
self.file_downloader.multithread_enabled = download_mthreaded
|
||||
self.file_downloader.check_ssl = check_ssl
|
||||
|
||||
single_dep_check = adv_form.get_component('dep_check', SingleSelectComponent).get_selected()
|
||||
core_config['system']['single_dependency_checking'] = single_dep_check
|
||||
|
||||
data_exp = adv_form.get_component('data_exp', TextInputComponent).get_int_value()
|
||||
core_config['memory_cache']['data_expiration'] = data_exp
|
||||
|
||||
icon_exp = adv_form.get_component('icon_exp', TextInputComponent).get_int_value()
|
||||
core_config['memory_cache']['icon_expiration'] = icon_exp
|
||||
|
||||
trim_after_upgrade = adv_form.get_component('trim_after_upgrade', SingleSelectComponent).get_selected()
|
||||
core_config['disk']['trim']['after_upgrade'] = trim_after_upgrade
|
||||
|
||||
# backup
|
||||
if backup:
|
||||
bkp_form = backup.get_component_by_idx(0, FormComponent)
|
||||
|
||||
core_config['backup']['enabled'] = bkp_form.get_component('enabled', SingleSelectComponent).get_selected()
|
||||
core_config['backup']['mode'] = bkp_form.get_component('mode', SingleSelectComponent).get_selected()
|
||||
core_config['backup']['type'] = bkp_form.get_component('type', SingleSelectComponent).get_selected()
|
||||
core_config['backup']['install'] = bkp_form.get_component('install', SingleSelectComponent).get_selected()
|
||||
core_config['backup']['upgrade'] = bkp_form.get_component('upgrade', SingleSelectComponent).get_selected()
|
||||
|
||||
bkp_remove_method = bkp_form.get_component('remove_method', SingleSelectComponent).get_selected()
|
||||
core_config['backup']['remove_method'] = bkp_remove_method
|
||||
|
||||
bkp_uninstall = bkp_form.get_component('uninstall', SingleSelectComponent).get_selected()
|
||||
core_config['backup']['uninstall'] = bkp_uninstall
|
||||
|
||||
bkp_downgrade = bkp_form.get_component('downgrade', SingleSelectComponent).get_selected()
|
||||
core_config['backup']['downgrade'] = bkp_downgrade
|
||||
|
||||
# tray
|
||||
tray_form = tray.get_component_by_idx(0, FormComponent)
|
||||
|
||||
updates_interval = tray_form.get_component('updates_interval', TextInputComponent).get_int_value()
|
||||
core_config['updates']['check_interval'] = updates_interval
|
||||
|
||||
def_icon_path = tray_form.get_component('def_icon', FileChooserComponent).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', FileChooserComponent).file_path
|
||||
core_config['ui']['tray']['updates_icon'] = up_icon_path if up_icon_path else None
|
||||
|
||||
# ui
|
||||
ui_form = ui.get_component_by_idx(0, FormComponent)
|
||||
|
||||
core_config['download']['icons'] = ui_form.get_component('down_icons', SingleSelectComponent).get_selected()
|
||||
core_config['ui']['hdpi'] = ui_form.get_component('hdpi', SingleSelectComponent).get_selected()
|
||||
|
||||
previous_autoscale = core_config['ui']['auto_scale']
|
||||
|
||||
core_config['ui']['auto_scale'] = ui_form.get_component('auto_scale', SingleSelectComponent).get_selected()
|
||||
|
||||
if previous_autoscale and not core_config['ui']['auto_scale']:
|
||||
self.logger.info("Deleting environment variable QT_AUTO_SCREEN_SCALE_FACTOR")
|
||||
del os.environ['QT_AUTO_SCREEN_SCALE_FACTOR']
|
||||
|
||||
core_config['ui']['scale_factor'] = ui_form.get_component('scalef').value / 100
|
||||
|
||||
table_max = ui_form.get_component('table_max', TextInputComponent).get_int_value()
|
||||
core_config['ui']['table']['max_displayed'] = table_max
|
||||
|
||||
style = ui_form.get_component('style', SingleSelectComponent).get_selected()
|
||||
|
||||
if core_config['ui']['qt_style']:
|
||||
cur_style = core_config['ui']['qt_style']
|
||||
else:
|
||||
cur_style = QApplication.instance().property('qt_style')
|
||||
|
||||
if style != cur_style:
|
||||
core_config['ui']['qt_style'] = style
|
||||
QApplication.instance().setProperty('qt_style', style)
|
||||
|
||||
core_config['ui']['system_theme'] = ui_form.get_component('system_theme', SingleSelectComponent).get_selected()
|
||||
|
||||
# gems
|
||||
checked_gems = gems_panel.components[1].get_component('gems', MultipleSelectComponent).get_selected_values()
|
||||
|
||||
for man in self.managers:
|
||||
modname = man.__module__.split('.')[-2]
|
||||
enabled = modname in checked_gems
|
||||
man.set_enabled(enabled)
|
||||
|
||||
if core_config['gems'] is None and len(checked_gems) == len(self.managers):
|
||||
sel_gems = None
|
||||
else:
|
||||
sel_gems = checked_gems
|
||||
|
||||
core_config['gems'] = sel_gems
|
||||
|
||||
try:
|
||||
self.configman.save_config(core_config)
|
||||
return True, None
|
||||
except Exception:
|
||||
return False, [traceback.format_exc()]
|
||||
|
||||
def _save_views(self, views: Iterable[SettingsView], success_list: List[bool], warnings: List[str]):
|
||||
success = False
|
||||
|
||||
for view in views:
|
||||
try:
|
||||
res = view.save()
|
||||
|
||||
if res:
|
||||
success, errors = res[0], res[1]
|
||||
|
||||
if errors:
|
||||
warnings.extend(errors)
|
||||
except Exception:
|
||||
self.logger.error(f"An exception happened while {view.controller.__class__.__name__}"
|
||||
f" was trying to save settings")
|
||||
traceback.print_exc()
|
||||
finally:
|
||||
success_list.append(success)
|
||||
|
||||
def _save_core_settings(self, tabs: TabGroupComponent, success_list: List[bool], warnings: List[str]):
|
||||
success = False
|
||||
|
||||
try:
|
||||
bkp = tabs.get_tab('core.bkp')
|
||||
success, errors = self._save_settings(general=tabs.get_tab('core.gen').get_content(PanelComponent),
|
||||
advanced=tabs.get_tab('core.adv').get_content(PanelComponent),
|
||||
tray=tabs.get_tab('core.tray').get_content(PanelComponent),
|
||||
backup=bkp.get_content(PanelComponent) if bkp else None,
|
||||
ui=tabs.get_tab('core.ui').get_content(PanelComponent),
|
||||
gems_panel=tabs.get_tab('core.types').get_content(PanelComponent))
|
||||
if errors:
|
||||
warnings.extend(errors)
|
||||
|
||||
except Exception:
|
||||
self.logger.error("An exception happened while saving the core settings")
|
||||
traceback.print_exc()
|
||||
finally:
|
||||
success_list.append(success)
|
||||
|
||||
def save_settings(self, component: TabGroupComponent) -> Tuple[bool, Optional[List[str]]]:
|
||||
ti = time.time()
|
||||
save_threads, warnings, success_list = [], [], []
|
||||
|
||||
save_core = Thread(target=self._save_core_settings, args=(component, success_list, warnings), daemon=True)
|
||||
save_core.start()
|
||||
save_threads.append(save_core)
|
||||
|
||||
if self._settings_views:
|
||||
|
||||
for views in self._settings_views.values():
|
||||
save_view = Thread(target=self._save_views, args=(views, success_list, warnings), daemon=True)
|
||||
save_view.start()
|
||||
save_threads.append(save_view)
|
||||
|
||||
for t in save_threads:
|
||||
t.join()
|
||||
|
||||
success = all(success_list)
|
||||
tf = time.time()
|
||||
self.logger.info(f"Saving all settings took {tf - ti:.8f} seconds")
|
||||
return success, warnings
|
||||
|
||||
def _gen_backup_settings(self, core_config: dict) -> Optional[TabComponent]:
|
||||
if timeshift.is_available():
|
||||
enabled_opt = self._gen_bool_component(label=self.i18n['core.config.backup'],
|
||||
tooltip=None,
|
||||
value=bool(core_config['backup']['enabled']),
|
||||
id_='enabled')
|
||||
|
||||
ops_opts = [(self.i18n['yes'].capitalize(), True, None),
|
||||
(self.i18n['no'].capitalize(), False, None),
|
||||
(self.i18n['ask'].capitalize(), None, None)]
|
||||
|
||||
install_mode = new_select(label=self.i18n['core.config.backup.install'],
|
||||
tip=None,
|
||||
value=core_config['backup']['install'],
|
||||
opts=ops_opts,
|
||||
id_='install')
|
||||
|
||||
uninstall_mode = new_select(label=self.i18n['core.config.backup.uninstall'],
|
||||
tip=None,
|
||||
value=core_config['backup']['uninstall'],
|
||||
opts=ops_opts,
|
||||
id_='uninstall')
|
||||
|
||||
upgrade_mode = new_select(label=self.i18n['core.config.backup.upgrade'],
|
||||
tip=None,
|
||||
value=core_config['backup']['upgrade'],
|
||||
opts=ops_opts,
|
||||
id_='upgrade')
|
||||
|
||||
downgrade_mode = new_select(label=self.i18n['core.config.backup.downgrade'],
|
||||
tip=None,
|
||||
value=core_config['backup']['downgrade'],
|
||||
opts=ops_opts,
|
||||
id_='downgrade')
|
||||
|
||||
mode = new_select(
|
||||
label=self.i18n['core.config.backup.mode'],
|
||||
tip=None,
|
||||
value=core_config['backup']['mode'],
|
||||
opts=[
|
||||
(self.i18n['core.config.backup.mode.incremental'], 'incremental',
|
||||
self.i18n['core.config.backup.mode.incremental.tip']),
|
||||
(self.i18n['core.config.backup.mode.only_one'], 'only_one',
|
||||
self.i18n['core.config.backup.mode.only_one.tip'])
|
||||
],
|
||||
id_='mode')
|
||||
type_ = new_select(label=self.i18n['type'].capitalize(),
|
||||
tip=None,
|
||||
value=core_config['backup']['type'],
|
||||
opts=[('rsync', 'rsync', None), ('btrfs', 'btrfs', None)],
|
||||
id_='type')
|
||||
|
||||
remove_method = core_config['backup']['remove_method']
|
||||
|
||||
if not remove_method or remove_method not in BACKUP_REMOVE_METHODS:
|
||||
remove_method = BACKUP_DEFAULT_REMOVE_METHOD
|
||||
|
||||
remove_i18n = 'core.config.backup.remove_method'
|
||||
remove_opts = ((self.i18n[f'{remove_i18n}.{m}'], m, self.i18n[f'{remove_i18n}.{m}.tip'])
|
||||
for m in sorted(BACKUP_REMOVE_METHODS))
|
||||
|
||||
remove_label = f'{self.i18n[remove_i18n]} ({self.i18n["core.config.backup.mode"]} ' \
|
||||
f'"{self.i18n["core.config.backup.mode.only_one"].capitalize()}")'
|
||||
|
||||
sel_remove = new_select(label=remove_label,
|
||||
tip=None,
|
||||
value=remove_method,
|
||||
opts=remove_opts,
|
||||
capitalize_label=False,
|
||||
id_='remove_method')
|
||||
|
||||
inputs = [enabled_opt, type_, mode, sel_remove, install_mode, uninstall_mode, upgrade_mode, downgrade_mode]
|
||||
panel = PanelComponent([FormComponent(inputs, spaces=False)], id_='backup')
|
||||
return TabComponent(self.i18n['core.config.tab.backup'].capitalize(), panel, None, 'core.bkp')
|
||||
@@ -1,32 +0,0 @@
|
||||
import os
|
||||
from typing import Optional, Dict
|
||||
from bauh import __app_name__
|
||||
|
||||
|
||||
def read_suggestions_mapping() -> Optional[Dict[str, str]]:
|
||||
file_path = f'/etc/{__app_name__}/suggestions.conf'
|
||||
|
||||
if os.path.isfile(file_path):
|
||||
try:
|
||||
with open(file_path) as f:
|
||||
file_content = f.read()
|
||||
except FileNotFoundError:
|
||||
return
|
||||
|
||||
if not file_content:
|
||||
return
|
||||
|
||||
mapping = {}
|
||||
for line in file_content.split('\n'):
|
||||
line_strip = line.strip()
|
||||
|
||||
if not line_strip.startswith('#'):
|
||||
gem_file = line_strip.split('=')
|
||||
|
||||
if len(gem_file) == 2:
|
||||
gem_name, file_url = gem_file[0].strip(), gem_file[1].strip()
|
||||
|
||||
if gem_name and file_url:
|
||||
mapping[gem_name] = file_url
|
||||
|
||||
return mapping if mapping else None
|
||||
@@ -1,38 +0,0 @@
|
||||
import re
|
||||
import shutil
|
||||
from typing import Optional, Generator
|
||||
|
||||
from bauh import __app_name__
|
||||
from bauh.commons.system import SimpleProcess, new_root_subprocess
|
||||
|
||||
RE_SNAPSHOTS = re.compile(r'\d+\s+>\s+([\w\-_]+)\s+.+<{}>'.format(__app_name__))
|
||||
|
||||
|
||||
def is_available() -> bool:
|
||||
return bool(shutil.which('timeshift'))
|
||||
|
||||
|
||||
def delete_all_snapshots(root_password: Optional[str]) -> SimpleProcess:
|
||||
return SimpleProcess(('timeshift', '--delete-all', '--scripted'), root_password=root_password)
|
||||
|
||||
|
||||
def delete(snapshot_name: str, root_password: Optional[str]) -> SimpleProcess:
|
||||
return SimpleProcess(('timeshift', '--delete', '--snapshot', snapshot_name),
|
||||
shell=True, root_password=root_password)
|
||||
|
||||
|
||||
def create_snapshot(root_password: Optional[str], mode: str) -> SimpleProcess:
|
||||
return SimpleProcess(('timeshift', '--create', '--scripted', f'--{mode}', '--comments', f'<{__app_name__}>'),
|
||||
root_password=root_password)
|
||||
|
||||
|
||||
def read_created_snapshots(root_password: Optional[str]) -> Generator[str, None, None]:
|
||||
proc = new_root_subprocess(cmd=('timeshift', '--list'), root_password=root_password, shell=True)
|
||||
proc.wait()
|
||||
|
||||
if proc.returncode == 0:
|
||||
output = '\n'.join((o.decode() for o in proc.stdout))
|
||||
|
||||
if output:
|
||||
for name in RE_SNAPSHOTS.findall(output):
|
||||
yield name
|
||||
@@ -1,12 +0,0 @@
|
||||
from pathlib import Path
|
||||
|
||||
from bauh.api.paths import CACHE_DIR
|
||||
|
||||
TRAY_CHECK_FILE = f'{CACHE_DIR}/notify_tray' # it is a file that signals to the tray icon it should recheck for updates
|
||||
|
||||
|
||||
def notify_tray():
|
||||
Path(CACHE_DIR).mkdir(exist_ok=True, parents=True)
|
||||
|
||||
with open(TRAY_CHECK_FILE, 'w+') as f:
|
||||
f.write('')
|
||||
@@ -1,58 +0,0 @@
|
||||
import logging
|
||||
import os
|
||||
from pathlib import Path
|
||||
|
||||
from bauh import __app_name__, __version__
|
||||
from bauh.api.http import HttpClient
|
||||
from bauh.api.paths import CACHE_DIR
|
||||
from bauh.commons.html import bold, link
|
||||
from bauh.commons.version_util import normalize_version
|
||||
from bauh.view.util.translation import I18n
|
||||
|
||||
|
||||
def check_for_update(logger: logging.Logger, http_client: HttpClient, i18n: I18n, tray: bool = False) -> str:
|
||||
"""
|
||||
:param logger:
|
||||
:param http_client:
|
||||
:param i18n:
|
||||
:param file_prefix: notification file prefix
|
||||
:return: bauh update warning string or 'None' if no update is available
|
||||
"""
|
||||
logger.info("Checking for updates")
|
||||
|
||||
try:
|
||||
releases = http_client.get_json('https://api.github.com/repos/spalencsar/bearhub/releases')
|
||||
|
||||
if releases:
|
||||
latest = None
|
||||
|
||||
for r in releases:
|
||||
if not r['draft']:
|
||||
latest = r
|
||||
break
|
||||
|
||||
if latest and latest.get('tag_name'):
|
||||
notifications_dir = f'{CACHE_DIR}/updates'
|
||||
release_file = '{}/{}{}'.format(notifications_dir, '' if not tray else 'tray_', latest['tag_name'])
|
||||
if os.path.exists(release_file):
|
||||
logger.info("Release {} already notified".format(latest['tag_name']))
|
||||
elif normalize_version(latest['tag_name']) > normalize_version(__version__):
|
||||
try:
|
||||
Path(notifications_dir).mkdir(parents=True, exist_ok=True)
|
||||
with open(release_file, 'w+') as f:
|
||||
f.write('')
|
||||
except Exception:
|
||||
logger.error("An error occurred while trying to create the update notification file: {}".format(release_file))
|
||||
|
||||
if tray:
|
||||
return i18n['tray.warning.update_available'].format(__app_name__, latest['tag_name'])
|
||||
else:
|
||||
return i18n['warning.update_available'].format(bold(__app_name__), bold(latest['tag_name']), link(latest.get('html_url', '?')))
|
||||
else:
|
||||
logger.info("No updates available")
|
||||
else:
|
||||
logger.warning("No official release found")
|
||||
else:
|
||||
logger.warning("No releases returned from the GitHub API")
|
||||
except Exception:
|
||||
logger.error("An error occurred while trying to retrieve the current releases")
|
||||
Reference in New Issue
Block a user