From 50893bff99543aff0cdeb58783109a38749b5037 Mon Sep 17 00:00:00 2001 From: Vinicius Moreira Date: Wed, 26 Aug 2020 12:54:56 -0300 Subject: [PATCH] [snap] improvement -> full support refactored to use the Snapd socket instead of the Ubuntu's old Snap API --- CHANGELOG.md | 3 + bauh/api/abstract/controller.py | 4 +- bauh/gems/snap/constants.py | 1 - bauh/gems/snap/controller.py | 252 +++++++++++++++-------------- bauh/gems/snap/model.py | 47 ++++-- bauh/gems/snap/resources/locale/ca | 5 +- bauh/gems/snap/resources/locale/de | 5 +- bauh/gems/snap/resources/locale/en | 5 +- bauh/gems/snap/resources/locale/es | 5 +- bauh/gems/snap/resources/locale/it | 5 +- bauh/gems/snap/resources/locale/pt | 5 +- bauh/gems/snap/resources/locale/ru | 5 +- bauh/gems/snap/resources/locale/tr | 5 +- bauh/gems/snap/snap.py | 213 +----------------------- bauh/gems/snap/snapd.py | 136 ++++++++++++++++ bauh/gems/snap/worker.py | 70 -------- 16 files changed, 347 insertions(+), 419 deletions(-) delete mode 100644 bauh/gems/snap/constants.py create mode 100644 bauh/gems/snap/snapd.py delete mode 100644 bauh/gems/snap/worker.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 34afa926..7ecab792 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -48,6 +48,8 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/). - Flatpak - creating the exports path **~/.local/share/flatpak/exports/share** (if it does not exist) and adding it to install/upgrade/downgrade/remove commands path to prevent warning messages. [#128](https://github.com/vinifmor/bauh/issues/128) - downgrade function refactored +- Snap + - full support refactored to use the Snapd socket instead of the Ubuntu's old Snap API (which was recently disabled). Now the 'read' operations are faster, a only the icon is cached to the disk. - minor UI improvements @@ -80,6 +82,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/). - some environment variables are not available during the common operations (install, upgrade, downgrade, uninstall, launch) - minor fixes - Snap + - not able to install classic Snaps due to Ubuntu's old Snaps API shutdown - some environment variables are not available during the common operations (install, upgrade, downgrade, uninstall, launch) - Web - some environment variable are not available during the launch process diff --git a/bauh/api/abstract/controller.py b/bauh/api/abstract/controller.py index 3b7d5774..8ce3f477 100644 --- a/bauh/api/abstract/controller.py +++ b/bauh/api/abstract/controller.py @@ -3,7 +3,7 @@ import os import shutil from abc import ABC, abstractmethod from pathlib import Path -from typing import List, Set, Type, Tuple +from typing import List, Set, Type, Tuple, Optional import yaml @@ -17,7 +17,7 @@ from bauh.api.abstract.view import ViewComponent class SearchResult: - def __init__(self, installed: List[SoftwarePackage], new: List[SoftwarePackage], total: int): + def __init__(self, installed: Optional[List[SoftwarePackage]], new: Optional[List[SoftwarePackage]], total: int): """ :param installed: already installed packages :param new: new packages found diff --git a/bauh/gems/snap/constants.py b/bauh/gems/snap/constants.py deleted file mode 100644 index 099a232e..00000000 --- a/bauh/gems/snap/constants.py +++ /dev/null @@ -1 +0,0 @@ -SNAP_API_URL = 'https://search.apps.ubuntu.com/api/v1' diff --git a/bauh/gems/snap/controller.py b/bauh/gems/snap/controller.py index 6ad82683..048380b8 100644 --- a/bauh/gems/snap/controller.py +++ b/bauh/gems/snap/controller.py @@ -1,25 +1,23 @@ import re import time -from datetime import datetime from threading import Thread -from typing import List, Set, Type +from typing import List, Set, Type, Optional from bauh.api.abstract.controller import SoftwareManager, SearchResult, ApplicationContext, UpgradeRequirements, \ TransactionResult from bauh.api.abstract.disk import DiskCacheLoader from bauh.api.abstract.handler import ProcessWatcher, TaskManager from bauh.api.abstract.model import SoftwarePackage, PackageHistory, PackageUpdate, PackageSuggestion, \ - SuggestionPriority, CustomSoftwareAction + SuggestionPriority, CustomSoftwareAction, PackageStatus from bauh.api.abstract.view import SingleSelectComponent, SelectViewType, InputOption from bauh.commons import resource, internet from bauh.commons.category import CategoriesDownloader from bauh.commons.html import bold -from bauh.commons.system import SystemProcess, ProcessHandler, new_root_subprocess +from bauh.commons.system import SystemProcess, ProcessHandler, new_root_subprocess, get_human_size_str from bauh.gems.snap import snap, URL_CATEGORIES_FILE, SNAP_CACHE_PATH, CATEGORIES_FILE_PATH, SUGGESTIONS_FILE, \ - get_icon_path -from bauh.gems.snap.constants import SNAP_API_URL + get_icon_path, snapd from bauh.gems.snap.model import SnapApplication -from bauh.gems.snap.worker import SnapAsyncDataLoader +from bauh.gems.snap.snapd import SnapdClient RE_AVAILABLE_CHANNELS = re.compile(re.compile(r'(\w+)\s+(snap install.+)')) @@ -47,36 +45,12 @@ class SnapManager(SoftwareManager): i18n_confirm_key='snap.action.refresh.confirm') ] - def get_info_path(self) -> str: - if self.info_path is None: - self.info_path = snap.get_app_info_path() - - return self.info_path - - def map_json(self, app_json: dict, installed: bool, disk_loader: DiskCacheLoader, internet: bool = True) -> SnapApplication: - app = SnapApplication(publisher=app_json.get('publisher'), - rev=app_json.get('rev'), - notes=app_json.get('notes'), - has_apps_field=app_json.get('apps_field', False), - id=app_json.get('name'), - name=app_json.get('name'), - version=app_json.get('version'), - latest_version=app_json.get('version'), - description=app_json.get('description', app_json.get('summary')), - verified_publisher=app_json.get('developer_validation', '') == 'verified', - extra_actions=self.custom_actions) - - if app.publisher and app.publisher.endswith('*'): - app.verified_publisher = True - app.publisher = app.publisher.replace('*', '') - + def _fill_categories(self, app: SnapApplication): categories = self.categories.get(app.name.lower()) if categories: app.categories = categories - app.installed = installed - if not app.is_application(): categories = app.categories @@ -87,70 +61,67 @@ class SnapManager(SoftwareManager): if 'runtime' not in categories: categories.append('runtime') - api_data = self.api_cache.get(app_json['name']) - expired_data = api_data and api_data.get('expires_at') and api_data['expires_at'] <= datetime.utcnow() - - if (not api_data or expired_data) and app.is_application(): - if disk_loader and app.installed: - disk_loader.fill(app) - - if internet: - SnapAsyncDataLoader(app=app, api_cache=self.api_cache, manager=self, context=self.context).start() - else: - app.fill_cached_data(api_data) - - return app - def search(self, words: str, disk_loader: DiskCacheLoader, limit: int = -1, is_url: bool = False) -> SearchResult: - if is_url: + if is_url or (not snap.is_installed() and not snapd.is_running()): return SearchResult([], [], 0) - if snap.is_snapd_running(): + snapd_client = SnapdClient(self.logger) + apps_found = snapd_client.query(words) + + res = SearchResult([], [], 0) + + if apps_found: installed = self.read_installed(disk_loader).installed - res = SearchResult([], [], 0) - - for app_json in snap.search(words): - + for app_json in apps_found: already_installed = None if installed: - already_installed = [i for i in installed if i.id == app_json.get('name')] + already_installed = [i for i in installed if i.id == app_json.get('id')] already_installed = already_installed[0] if already_installed else None if already_installed: res.installed.append(already_installed) else: - res.new.append(self.map_json(app_json, installed=False, disk_loader=disk_loader)) + res.new.append(self._map_to_app(app_json, installed=False)) - res.total = len(res.installed) + len(res.new) - return res - else: - return SearchResult([], [], 0) + res.total = len(res.installed) + len(res.new) + return res def read_installed(self, disk_loader: DiskCacheLoader, limit: int = -1, only_apps: bool = False, pkg_types: Set[Type[SoftwarePackage]] = None, internet_available: bool = None) -> SearchResult: - info_path = self.get_info_path() - - if snap.is_snapd_running() and info_path: - installed = [self.map_json(app_json, installed=True, disk_loader=disk_loader, internet=internet_available) for app_json in snap.read_installed(info_path)] + if snap.is_installed() and snapd.is_running(): + snapd_client = SnapdClient(self.logger) + app_names = {a['snap'] for a in snapd_client.list_only_apps()} + installed = [self._map_to_app(app_json=appjson, + installed=True, + disk_loader=disk_loader, + is_application=app_names and appjson['name'] in app_names) for appjson in snapd_client.list_all_snaps()] return SearchResult(installed, None, len(installed)) else: return SearchResult([], None, 0) def downgrade(self, pkg: SnapApplication, root_password: str, watcher: ProcessWatcher) -> bool: + if not snap.is_installed(): + watcher.print("'snap' seems not to be installed") + return False + if not snapd.is_running(): + watcher.print("'snapd' seems not to be running") + return False + return ProcessHandler(watcher).handle_simple(snap.downgrade_and_stream(pkg.name, root_password))[0] def upgrade(self, requirements: UpgradeRequirements, root_password: str, watcher: ProcessWatcher) -> SystemProcess: raise Exception("'upgrade' is not supported by {}".format(SnapManager.__class__.__name__)) def uninstall(self, pkg: SnapApplication, root_password: str, watcher: ProcessWatcher, disk_loader: DiskCacheLoader) -> TransactionResult: - uninstalled = ProcessHandler(watcher).handle_simple(snap.uninstall_and_stream(pkg.name, root_password))[0] + if snap.is_installed() and snapd.is_running(): + uninstalled = ProcessHandler(watcher).handle_simple(snap.uninstall_and_stream(pkg.name, root_password))[0] - if uninstalled: - if self.suggestions_cache: - self.suggestions_cache.delete(pkg.name) + if uninstalled: + if self.suggestions_cache: + self.suggestions_cache.delete(pkg.name) - return TransactionResult(success=True, installed=None, removed=[pkg]) + return TransactionResult(success=True, installed=None, removed=[pkg]) return TransactionResult.fail() @@ -162,17 +133,29 @@ class SnapManager(SoftwareManager): self.api_cache.delete(pkg.id) def get_info(self, pkg: SnapApplication) -> dict: - info = snap.get_info(pkg.name, attrs=('license', 'contact', 'commands', 'snap-id', 'tracking', 'installed')) - info['description'] = pkg.description - info['publisher'] = pkg.publisher - info['revision'] = pkg.rev - info['name'] = pkg.name + info = { + 'description': pkg.description, + 'developer': pkg.developer, + 'license': pkg.license, + 'contact': pkg.contact, + 'snap-id': pkg.id, + 'name': pkg.name, + 'publisher': pkg.publisher, + 'revision': pkg.rev, + 'tracking': pkg.tracking, + 'channel': pkg.channel, + 'type': pkg.type + } - if info.get('commands'): - info['commands'] = ' '.join(info['commands']) + if pkg.installed: + commands = [*{c['name'] for c in SnapdClient(self.logger).list_commands(pkg.name)}] + commands.sort() + info['commands'] = commands - if info.get('license') and info['license'] == 'unset': - del info['license'] + if pkg.installed_size: + info['installed_size']: get_human_size_str(pkg.installed_size) + elif pkg.download_size: + info['download_size'] = get_human_size_str(pkg.download_size) return info @@ -180,13 +163,16 @@ class SnapManager(SoftwareManager): raise Exception("'get_history' is not supported by {}".format(pkg.__class__.__name__)) def install(self, pkg: SnapApplication, root_password: str, disk_loader: DiskCacheLoader, watcher: ProcessWatcher) -> TransactionResult: - info_path = self.get_info_path() - - if not info_path: - self.logger.warning('Information directory was not found. It will not be possible to determine if the installed application can be launched') - # retrieving all installed so it will be possible to know the additional installed runtimes after the operation succeeds - installed_names = snap.list_installed_names() + if not snap.is_installed(): + watcher.print("'snap' seems not to be installed") + return TransactionResult.fail() + + if not snapd.is_running(): + watcher.print("'snapd' seems not to be running") + return TransactionResult.fail() + + installed_names = {s['name'] for s in SnapdClient(self.logger).list_all_snaps()} res, output = ProcessHandler(watcher).handle_simple(snap.install_and_stream(pkg.name, pkg.confinement, root_password)) @@ -208,17 +194,10 @@ class SnapManager(SoftwareManager): deny_label=self.i18n['cancel']): self.logger.info("Installing '{}' with the custom command '{}'".format(pkg.name, channel_select.value)) res = ProcessHandler(watcher).handle(SystemProcess(new_root_subprocess(channel_select.value.value.split(' '), root_password=root_password))) - - if res and info_path: - pkg.has_apps_field = snap.has_apps_field(pkg.name, info_path) - return self._gen_installation_response(success=res, pkg=pkg, installed=installed_names, disk_loader=disk_loader) else: self.logger.error("Could not find available channels in the installation output: {}".format(output)) - else: - if info_path: - pkg.has_apps_field = snap.has_apps_field(pkg.name, info_path) return self._gen_installation_response(success=res, pkg=pkg, installed=installed_names, disk_loader=disk_loader) @@ -278,7 +257,7 @@ class SnapManager(SoftwareManager): def list_warnings(self, internet_available: bool) -> List[str]: if snap.is_installed(): - if not snap.is_snapd_running(): + if not snapd.is_running(): snap_bold = bold('Snap') return [self.i18n['snap.notification.snapd_unavailable'].format(bold('snapd'), snap_bold), self.i18n['snap.notification.snap.disable'].format(snap_bold, bold('{} > {}'.format(self.i18n['settings'].capitalize(), @@ -291,24 +270,57 @@ class SnapManager(SoftwareManager): self.logger.warning('It seems Snap API is not available. Search output: {}'.format(output)) return [self.i18n['snap.notifications.api.unavailable'].format(bold('Snaps'), bold('Snap'))] - def _fill_suggestion(self, pkg_name: str, priority: SuggestionPriority, out: List[PackageSuggestion]): - res = self.http_client.get_json(SNAP_API_URL + '/search?q=package_name:{}'.format(pkg_name)) + def _fill_suggestion(self, name: str, priority: SuggestionPriority, snapd_client: SnapdClient, out: List[PackageSuggestion]): + res = snapd_client.find_by_name(name) - if res and res['_embedded']['clickindex:package']: - pkg = res['_embedded']['clickindex:package'][0] - pkg['rev'] = pkg['revision'] - pkg['name'] = pkg_name + if res: + if len(res) == 1: + app_json = res[0] + else: + jsons_found = [p for p in res if p['name'] == name] + app_json = jsons_found[0] if jsons_found else None - sug = PackageSuggestion(self.map_json(pkg, installed=False, disk_loader=None), priority) - self.suggestions_cache.add(pkg_name, sug) - out.append(sug) - else: - self.logger.warning("Could not retrieve suggestion '{}'".format(pkg_name)) + if app_json: + sug = PackageSuggestion(self._map_to_app(app_json, False), priority) + self.suggestions_cache.add(name, sug) + out.append(sug) + return + + self.logger.warning("Could not retrieve suggestion '{}'".format(name)) + + def _map_to_app(self, app_json: dict, installed: bool, disk_loader: Optional[DiskCacheLoader] = None, is_application: bool = False) -> SnapApplication: + app = SnapApplication(id=app_json.get('id'), + name=app_json.get('name'), + license=app_json.get('license'), + version=app_json.get('version'), + latest_version=app_json.get('version'), + description=app_json.get('description', app_json.get('summary')), + installed=installed, + rev=app_json.get('revision'), + publisher=app_json['publisher'].get('display-name', app_json['publisher'].get('username')), + verified_publisher=app_json['publisher'].get('validation') == 'verified', + icon_url=app_json.get('icon'), + screenshots={m['url'] for m in app_json.get('media', []) if m['type'] == 'screenshot'}, + download_size=app_json.get('download-size'), + channel=app_json.get('channel'), + confinement=app_json.get('confinement'), + app_type=app_json.get('type'), + app=is_application, + installed_size=app_json.get('installed-size'), + extra_actions=self.custom_actions) + + if disk_loader and app.installed: + disk_loader.fill(app) + + self._fill_categories(app) + + app.status = PackageStatus.READY + return app def list_suggestions(self, limit: int, filter_installed: bool) -> List[PackageSuggestion]: res = [] - if snap.is_snapd_running(): + if snapd.is_running(): self.logger.info('Downloading suggestions file {}'.format(SUGGESTIONS_FILE)) file = self.http_client.get(SUGGESTIONS_FILE) @@ -319,7 +331,8 @@ class SnapManager(SoftwareManager): self.logger.info('Mapping suggestions') suggestions, threads = [], [] - installed = {i.name.lower() for i in self.read_installed(disk_loader=None).installed} if filter_installed else None + snapd_client = SnapdClient(self.logger) + installed = {s['name'].lower() for s in snapd_client.list_all_snaps()} for l in file.text.split('\n'): if l: @@ -333,7 +346,7 @@ class SnapManager(SoftwareManager): if cached_sug: res.append(cached_sug) else: - t = Thread(target=self._fill_suggestion, args=(name, SuggestionPriority(int(sug[0])), res)) + t = Thread(target=self._fill_suggestion, args=(name, SuggestionPriority(int(sug[0])), snapd_client, res)) t.start() threads.append(t) time.sleep(0.001) # to avoid being blocked @@ -350,22 +363,21 @@ class SnapManager(SoftwareManager): return True def launch(self, pkg: SnapApplication): - snap.run(pkg, self.context.logger) + commands = SnapdClient(self.logger).list_commands(pkg.name) - def get_screenshots(self, pkg: SoftwarePackage) -> List[str]: - res = self.http_client.get_json('{}/search?q={}'.format(SNAP_API_URL, pkg.name)) - - if res: - if res.get('_embedded') and res['_embedded'].get('clickindex:package'): - snap_data = res['_embedded']['clickindex:package'][0] - - if snap_data.get('screenshot_urls'): - return snap_data['screenshot_urls'] - else: - self.logger.warning("No 'screenshots_urls' defined for {}".format(pkg)) + if commands: + if len(commands) == 1: + cmd = commands[0]['name'] else: - self.logger.error('It seems the API is returning a different response: {}'.format(res)) - else: - self.logger.warning('Could not retrieve data for {}'.format(pkg)) + desktop_cmd = [c for c in commands if 'desktop-file' in c] - return [] + if desktop_cmd: + cmd = desktop_cmd[0]['name'] + else: + cmd = commands[0]['name'] + + self.logger.info("Running '{}': {}".format(pkg.name, cmd)) + snap.run(cmd) + + def get_screenshots(self, pkg: SnapApplication) -> List[str]: + return pkg.screenshots if pkg.has_screenshots() else [] diff --git a/bauh/gems/snap/model.py b/bauh/gems/snap/model.py index e3ec5984..58e787af 100644 --- a/bauh/gems/snap/model.py +++ b/bauh/gems/snap/model.py @@ -1,26 +1,46 @@ -from typing import List +from typing import List, Optional, Set from bauh.api.abstract.model import SoftwarePackage, CustomSoftwareAction from bauh.commons import resource from bauh.gems.snap import ROOT_DIR -KNOWN_RUNTIME_NAMES = {'snapd', 'snapcraft', 'multipass'} - class SnapApplication(SoftwarePackage): def __init__(self, id: str = None, name: str = None, version: str = None, latest_version: str = None, description: str = None, publisher: str = None, rev: str = None, notes: str = None, - confinement: str = None, has_apps_field: bool = None, verified_publisher: bool = False, extra_actions: List[CustomSoftwareAction] = None): + confinement: str = None, verified_publisher: bool = False, + extra_actions: List[CustomSoftwareAction] = None, + screenshots: Optional[Set[str]] = None, + license: Optional[str] = None, + installed: bool = False, + icon_url: Optional[str] = None, + download_size: Optional[int] = None, + developer: Optional[str] = None, + contact: Optional[str] = None, + tracking: Optional[str] = None, + app_type: Optional[str] = None, + channel: Optional[str] = None, + app: bool = False, + installed_size: Optional[int] = None): super(SnapApplication, self).__init__(id=id, name=name, version=version, - latest_version=latest_version, description=description) + latest_version=latest_version, description=description, + license=license, installed=installed, icon_url=icon_url) self.publisher = publisher self.rev = rev self.notes = notes self.confinement = confinement - self.has_apps_field = has_apps_field self.verified_publisher = verified_publisher self.extra_actions = extra_actions + self.screenshots = screenshots + self.download_size = download_size + self.developer = developer + self.contact = contact + self.tracking = tracking + self.type = app_type + self.channel = channel + self.app = app + self.installed_size = installed_size def supports_disk_cache(self): return self.installed @@ -44,7 +64,10 @@ class SnapApplication(SoftwarePackage): return self.get_default_icon_path() def is_application(self) -> bool: - return not self.installed or ((self.has_apps_field is None or self.has_apps_field) and self.name.lower() not in KNOWN_RUNTIME_NAMES) + if self.installed: + return self.app + else: + return self.type == 'app' def get_disk_cache_path(self): return super(SnapApplication, self).get_disk_cache_path() + '/installed/' + self.name @@ -54,20 +77,17 @@ class SnapApplication(SoftwarePackage): def get_data_to_cache(self): return { - "icon_url": self.icon_url, - 'confinement': self.confinement, - 'description': self.description, 'categories': self.categories } def fill_cached_data(self, data: dict): if data: - for base_attr in ('icon_url', 'description', 'confinement', 'categories'): + for base_attr in self.get_data_to_cache().keys(): if data.get(base_attr): setattr(self, base_attr, data[base_attr]) def can_be_run(self) -> bool: - return self.installed and self.is_application() + return bool(self.installed and self.is_application()) def get_publisher(self): return self.publisher @@ -79,6 +99,9 @@ class SnapApplication(SoftwarePackage): def supports_backup(self) -> bool: return True + def has_screenshots(self) -> bool: + return not self.installed and self.screenshots + def __eq__(self, other): if isinstance(other, SnapApplication): return self.name == other.name diff --git a/bauh/gems/snap/resources/locale/ca b/bauh/gems/snap/resources/locale/ca index 04da1593..935f9ae1 100644 --- a/bauh/gems/snap/resources/locale/ca +++ b/bauh/gems/snap/resources/locale/ca @@ -2,15 +2,18 @@ gem.snap.info=Aplicacions publicades a https://snapcraft.io/store snap.action.refresh.confirm=Actualitza {} ? snap.action.refresh.label=Actualitza snap.action.refresh.status=S’està actualitzant +snap.info.channel=channel snap.info.commands=ordres snap.info.contact=contacte snap.info.description=descripció +snap.info.download_size=Size (Download) snap.info.installed=instal·lada +snap.info.installed_size=Size (Instalattion) snap.info.license=llicència snap.info.publisher=publicador snap.info.revision=revisió -snap.info.size=mida snap.info.tracking=seguiment +snap.info.type=type snap.info.version=versió snap.install.available_channels.help=Seleccioneu-ne un si voleu continuar snap.install.available_channels.message=No hi ha un canal {} (stable) disponible per a {}. Però més avall n’hi ha els següents diff --git a/bauh/gems/snap/resources/locale/de b/bauh/gems/snap/resources/locale/de index da013a37..7a672965 100644 --- a/bauh/gems/snap/resources/locale/de +++ b/bauh/gems/snap/resources/locale/de @@ -2,15 +2,18 @@ gem.snap.info=Anwendungen von https://snapcraft.io/store snap.action.refresh.confirm=Erneuern {} ? snap.action.refresh.label=Erneuern snap.action.refresh.status=Erneuern +snap.info.channel=channel snap.info.commands=Kommandos snap.info.contact=Kontakt snap.info.description=Beschreibung +snap.info.download_size=Size (Download) snap.info.installed=installiert +snap.info.installed_size=Size (Instalattion) snap.info.license=Lizenz snap.info.publisher=Herausgeber snap.info.revision=Revision -snap.info.size=Größe snap.info.tracking=Tracking +snap.info.type=type snap.info.version=Version snap.install.available_channels.help=Wähle einen aus um fortzufahren snap.install.available_channels.message=Es ist kein {} Channel verfügbar für {}. Es gibt jedoch folgende diff --git a/bauh/gems/snap/resources/locale/en b/bauh/gems/snap/resources/locale/en index 5dc82da9..46ed39d0 100644 --- a/bauh/gems/snap/resources/locale/en +++ b/bauh/gems/snap/resources/locale/en @@ -2,15 +2,18 @@ gem.snap.info=Applications published at https://snapcraft.io/store snap.action.refresh.confirm=Refresh {} ? snap.action.refresh.label=Refresh snap.action.refresh.status=Refreshing +snap.info.channel=channel snap.info.commands=commands snap.info.contact=contact snap.info.description=description +snap.info.download_size=Size (Download) snap.info.installed=installed +snap.info.installed_size=Size (Instalattion) snap.info.license=license snap.info.publisher=publisher snap.info.revision=revision -snap.info.size=size snap.info.tracking=tracking +snap.info.type=type snap.info.version=version snap.install.available_channels.help=Select one if you want to continue snap.install.available_channels.message=There is no {} channel available for {}. But there are these below diff --git a/bauh/gems/snap/resources/locale/es b/bauh/gems/snap/resources/locale/es index f6494044..d2d612f3 100644 --- a/bauh/gems/snap/resources/locale/es +++ b/bauh/gems/snap/resources/locale/es @@ -2,15 +2,18 @@ gem.snap.info=Aplicativos publicados en https://snapcraft.io/store snap.action.refresh.confirm=Actualizar {} ? snap.action.refresh.label=Actualizar snap.action.refresh.status=Actualizando +snap.info.channel=canal snap.info.commands=comandos snap.info.contact=contacto snap.info.description=descripción +snap.info.download_size=Tamaño (Descarga) snap.info.installed=instalado +snap.info.installed_size=Tamaño (Instalación) snap.info.license=licencia snap.info.publisher=publicador snap.info.revision=revisión -snap.info.size=tamaño snap.info.tracking=tracking +snap.info.type=tipo snap.info.version=versión snap.install.available_channels.help=Seleccione uno si desea continuar snap.install.available_channels.message=No hay un canal {} ( stable ) disponible para {}. Pero hay estos otros abajo diff --git a/bauh/gems/snap/resources/locale/it b/bauh/gems/snap/resources/locale/it index 6d66aee1..254cfadc 100644 --- a/bauh/gems/snap/resources/locale/it +++ b/bauh/gems/snap/resources/locale/it @@ -2,15 +2,18 @@ gem.snap.info=Applicazioni pubblicate su https://snapcraft.io/store snap.action.refresh.confirm=Ripristina {} ? snap.action.refresh.label=Ripristina snap.action.refresh.status=Ripristinare +snap.info.channel=channel snap.info.commands=commands snap.info.contact=contact snap.info.description=description +snap.info.download_size=Size (Download) snap.info.installed=installed +snap.info.installed_size=Size (Instalattion) snap.info.license=license snap.info.publisher=publisher snap.info.revision=revision -snap.info.size=size snap.info.tracking=tracking +snap.info.type=type snap.info.version=version snap.install.available_channels.help=Selezionane uno se vuoi continuare snap.install.available_channels.message=Non esiste un {} canale disponibile per {}. Ma ci sono questi sotto diff --git a/bauh/gems/snap/resources/locale/pt b/bauh/gems/snap/resources/locale/pt index 35223efa..54fafa34 100644 --- a/bauh/gems/snap/resources/locale/pt +++ b/bauh/gems/snap/resources/locale/pt @@ -2,15 +2,18 @@ gem.snap.info=Aplicativos publicados em https://snapcraft.io/store snap.action.refresh.confirm=Atualizar {} ? snap.action.refresh.label=Atualizar snap.action.refresh.status=Atualizando +snap.info.channel=canal snap.info.commands=comandos snap.info.contact=contato snap.info.description=descrição +snap.info.download_size=Tamanho (Download) snap.info.installed=instalado +snap.info.installed_size=Tamanho (Instalação) snap.info.license=licença snap.info.publisher=publicador snap.info.revision=revisão -snap.info.size=tamanho snap.info.tracking=tracking +snap.info.type=tipo snap.info.version=versão snap.install.available_channels.help=Selecione algum se quiser continuar snap.install.available_channels.message=Não há um canal {} ( stable ) disponível para {}. Porém existem estes outros abaixo diff --git a/bauh/gems/snap/resources/locale/ru b/bauh/gems/snap/resources/locale/ru index a7d60891..76fc3d96 100644 --- a/bauh/gems/snap/resources/locale/ru +++ b/bauh/gems/snap/resources/locale/ru @@ -2,15 +2,18 @@ gem.snap.info=Приложения, опубликованные на https://sn snap.action.refresh.confirm=Обновить {} ? snap.action.refresh.label=Обновить snap.action.refresh.status=Обновляется +snap.info.channel=channel snap.info.commands=Команды snap.info.contact=Контакт snap.info.description=Описание +snap.info.download_size=Size (Download) snap.info.installed=Размер установки +snap.info.installed_size=Size (Instalattion) snap.info.license=Лицензия snap.info.publisher=Издатель snap.info.revision=Ревизия -snap.info.size=Размер snap.info.tracking=Отслеживание +snap.info.type=type snap.info.version=Версия snap.install.available_channels.help=Выберите один, если вы хотите продолжить snap.install.available_channels.message=Нет одного канала {}, доступного для {}. Но есть такие ниже diff --git a/bauh/gems/snap/resources/locale/tr b/bauh/gems/snap/resources/locale/tr index 9347c7ce..9799d921 100644 --- a/bauh/gems/snap/resources/locale/tr +++ b/bauh/gems/snap/resources/locale/tr @@ -2,15 +2,18 @@ gem.snap.info=https://snapcraft.io/store adresinde yayınlanan uygulamalar snap.action.refresh.confirm=Yenile {} ? snap.action.refresh.label=Yenile snap.action.refresh.status=Yenileniyor +snap.info.channel=channel snap.info.commands=komutlar snap.info.contact=iletişim snap.info.description=açıklama +snap.info.download_size=Size (Download) snap.info.installed=yüklü +snap.info.installed_size=Size (Instalattion) snap.info.license=lisans snap.info.publisher=yayıncı snap.info.revision=revizyon -snap.info.size=boyut snap.info.tracking=izleme +snap.info.type=type snap.info.version=sürüm snap.install.available_channels.help=Devam etmek istiyorsanız birini seçin snap.install.available_channels.message={} için hiçbir {} kanalı yok. Ama aşağıda bunlar var diff --git a/bauh/gems/snap/snap.py b/bauh/gems/snap/snap.py index d114b5fa..74e0e890 100644 --- a/bauh/gems/snap/snap.py +++ b/bauh/gems/snap/snap.py @@ -1,174 +1,15 @@ -import logging import os -import re import subprocess from io import StringIO -from typing import List, Tuple, Set +from typing import Tuple -from bauh.commons.system import run_cmd, new_subprocess, SimpleProcess -from bauh.gems.snap.model import SnapApplication +from bauh.commons.system import run_cmd, SimpleProcess BASE_CMD = 'snap' -RE_SNAPD_STATUS = re.compile('\s+') -RE_SNAPD_SERVICES = re.compile(r'snapd\.\w+.+') -def is_installed(): - res = run_cmd('which snap', print_error=False) - return res and not res.strip().startswith('which ') - - -def is_snapd_running() -> bool: - output = run_cmd('systemctl list-units') - - snapd_services = RE_SNAPD_SERVICES.findall(output) - - socket, socket_running, service, service_running = False, False, False, False - if snapd_services: - for service_line in snapd_services: - line_split = RE_SNAPD_STATUS.split(service_line) - - running = line_split[3] in {'listening', 'running'} - - if line_split[0] == 'snapd.service': - service = True - service_running = running - elif line_split[0] == 'snapd.socket': - socket = True - socket_running = running - - return socket and socket_running and (not service or service_running) - - -def app_str_to_json(app: str) -> dict: - app_data = [word for word in app.split(' ') if word] - app_json = { - 'name': app_data[0], - 'version': app_data[1], - 'rev': app_data[2], - 'tracking': app_data[3], - 'publisher': app_data[4] if len(app_data) >= 5 else None, - 'notes': app_data[5] if len(app_data) >= 6 else None - } - - return app_json - - -def get_info(app_name: str, attrs: tuple = None): - full_info_lines = run_cmd('{} info {}'.format(BASE_CMD, app_name)) - - data = {} - - if full_info_lines: - re_attrs = r'\w+' if not attrs else '|'.join(attrs) - info_map = re.findall(r'({}):\s+(.+)'.format(re_attrs), full_info_lines) - - for info in info_map: - val = info[1].strip() - - if info[0] == 'installed': - val_split = [s for s in val.split(' ') if s] - data['version'] = val_split[0] - - if len(val_split) > 2: - data['size'] = val_split[2] - else: - data[info[0]] = val - - if not attrs or 'description' in attrs: - desc = re.findall(r'\|\n+((\s+.+\n+)+)', full_info_lines) - data['description'] = ''.join([w.strip() for w in desc[0][0].strip().split('\n')]).replace('.', '.\n') if desc else None - - if not attrs or 'commands' in attrs: - commands = re.findall(r'commands:\s*\n*((\s+-\s.+\s*\n)+)', full_info_lines) - data['commands'] = commands[0][0].strip().replace('- ', '').split('\n') if commands else None - - return data - - -def read_installed(info_path: str) -> List[dict]: - res = run_cmd('{} list'.format(BASE_CMD), print_error=False) - - apps = [] - - if res and len(res) > 0: - lines = res.split('\n') - - if not lines[0].startswith('error'): - for idx, app_str in enumerate(lines): - if idx > 0 and app_str: - apps.append(app_str_to_json(app_str)) - - info_out = new_subprocess(['cat', *[info_path.format(a['name']) for a in apps]]).stdout - - idx = -1 - for o in new_subprocess(['grep', '-E', '(summary|apps)', '--colour=never'], stdin=info_out).stdout: - if o: - line = o.decode() - - if line.startswith('summary:'): - idx += 1 - apps[idx]['summary'] = line.split(':')[1].strip() - else: - apps[idx]['apps_field'] = True - - return apps - - -def get_app_info_path() -> str: - if os.path.exists('/snap'): - return '/snap/{}/current/meta/snap.yaml' - elif os.path.exists('/var/lib/snapd/snap'): - return '/var/lib/snapd/snap/{}/current/meta/snap.yaml' - else: - return None - - -def has_apps_field(name: str, info_path: str) -> bool: - info_out = new_subprocess(['cat', info_path.format(name)]).stdout - - res = False - for o in new_subprocess(['grep', '-E', 'apps', '--colour=never'], stdin=info_out).stdout: - if o: - line = o.decode() - - if line.startswith('apps:'): - res = True - - return res - - -def search(word: str, exact_name: bool = False) -> List[dict]: - apps = [] - - res = run_cmd('{} find "{}"'.format(BASE_CMD, word), print_error=False) - - if res: - res = res.split('\n') - - if not res[0].startswith('No matching'): - for idx, app_str in enumerate(res): - if idx > 0 and app_str: - app_data = [word for word in app_str.split(' ') if word] - - if exact_name and app_data[0] != word: - continue - - apps.append({ - 'name': app_data[0], - 'version': app_data[1], - 'publisher': app_data[2], - 'notes': app_data[3] if app_data[3] != '-' else None, - 'summary': app_data[4] if len(app_data) == 5 else '', - 'rev': None, - 'tracking': None, - 'type': None - }) - - if exact_name and len(apps) > 0: - break - - return apps +def is_installed() -> bool: + return bool(run_cmd('which {}'.format(BASE_CMD), print_error=False)) def uninstall_and_stream(app_name: str, root_password: str) -> SimpleProcess: @@ -199,56 +40,16 @@ def refresh_and_stream(app_name: str, root_password: str) -> SimpleProcess: shell=True) -def run(app: SnapApplication, logger: logging.Logger): - info = get_info(app.name, 'commands') - app_name = app.name.lower() - - if info.get('commands'): - - logger.info('Available commands found for {}: {}'.format(app_name, info['commands'])) - - commands = [c.strip() for c in info['commands']] - - # trying to find an exact match command: - command = None - - for c in commands: - if c.lower() == app_name: - command = c - logger.info("Found exact match command for '{}'".format(app_name)) - break - - if not command: - for c in commands: - if not c.endswith('.apm'): - command = c - - if command: - logger.info("Running '{}'".format(command)) - subprocess.Popen(['{} run {}'.format(BASE_CMD, command)], shell=True, env={**os.environ}) - return - - logger.error("No valid command found for '{}'".format(app_name)) - else: - logger.error("No command found for '{}'".format(app_name)) +def run(cmd: str): + subprocess.Popen(['snap run {}'.format(cmd)], shell=True, env={**os.environ}) def is_api_available() -> Tuple[bool, str]: output = StringIO() - for o in SimpleProcess(['snap', 'search']).instance.stdout: + for o in SimpleProcess([BASE_CMD, 'search']).instance.stdout: if o: output.write(o.decode()) output.seek(0) output = output.read() return 'error:' not in output, output - - -def list_installed_names() -> Set[str]: - res = run_cmd('{} list'.format(BASE_CMD), print_error=False) - - if res: - lines = res.split('\n') - - if not lines[0].startswith('error'): - return {l.split(' ')[0].strip() for l in lines[1:] if l} diff --git a/bauh/gems/snap/snapd.py b/bauh/gems/snap/snapd.py new file mode 100644 index 00000000..091d8fc0 --- /dev/null +++ b/bauh/gems/snap/snapd.py @@ -0,0 +1,136 @@ +import re +import socket +import traceback +from logging import Logger +from typing import Optional, List + +from requests import Session +from requests.adapters import HTTPAdapter +from urllib3.connection import HTTPConnection +from urllib3.connectionpool import HTTPConnectionPool + +from bauh.commons.system import run_cmd + +URL_BASE = 'http://snapd/v2' +RE_SNAPD_STATUS = re.compile('\s+') +RE_SNAPD_SERVICES = re.compile(r'snapd\.\w+.+') + + +class SnapdConnection(HTTPConnection): + def __init__(self): + super(SnapdConnection, self).__init__('localhost') + + def connect(self): + self.sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + self.sock.connect("/run/snapd.socket") + + +class SnapdConnectionPool(HTTPConnectionPool): + def __init__(self): + super(SnapdConnectionPool, self).__init__('localhost') + + def _new_conn(self): + return SnapdConnection() + + +class SnapdAdapter(HTTPAdapter): + + def get_connection(self, url, proxies=None): + return SnapdConnectionPool() + + +class SnapdClient: + + def __init__(self, logger: Logger): + self.logger = logger + self.session = self._new_session() + + def _new_session(self) -> Optional[Session]: + try: + session = Session() + session.mount("http://snapd/", SnapdAdapter()) + return session + except: + self.logger.error("Could not establish a connection to 'snapd.socker'") + traceback.print_exc() + + def query(self, query: str) -> Optional[List[dict]]: + final_query = query.strip() + + if final_query and self.session: + res = self.session.get(url='{}/find'.format(URL_BASE), params={'q': final_query}) + + if res.status_code == 200: + json_res = res.json() + + if json_res['status-code'] == 200: + return json_res['result'] + + def find_by_name(self, name: str) -> Optional[List[dict]]: + if name and self.session: + res = self.session.get('{}/find?name={}'.format(URL_BASE, name)) + + if res.status_code == 200: + json_res = res.json() + + if json_res['status-code'] == 200: + return json_res['result'] + + def list_all_snaps(self) -> List[dict]: + if self.session: + res = self.session.get('{}/snaps'.format(URL_BASE)) + + if res.status_code == 200: + json_res = res.json() + + if json_res['status-code'] == 200: + return json_res['result'] + + return [] + + def list_only_apps(self) -> List[dict]: + if self.session: + res = self.session.get('{}/apps'.format(URL_BASE)) + + if res.status_code == 200: + json_res = res.json() + + if json_res['status-code'] == 200: + return json_res['result'] + return [] + + def list_commands(self, name: str) -> List[dict]: + if self.session: + res = self.session.get('{}/apps?names={}'.format(URL_BASE, name)) + + if res.status_code == 200: + json_res = res.json() + + if json_res['status-code'] == 200: + return [r for r in json_res['result'] if r['snap'] == name] + return [] + + +def is_running() -> bool: + output = run_cmd('systemctl list-units', print_error=False) + + if not output: + return False + + snapd_services = RE_SNAPD_SERVICES.findall(output) + + snap_socket, socket_running, service, service_running = False, False, False, False + if snapd_services: + for service_line in snapd_services: + line_split = RE_SNAPD_STATUS.split(service_line) + + running = line_split[3] in {'listening', 'running'} + + if line_split[0] == 'snapd.service': + service = True + service_running = running + elif line_split[0] == 'snapd.socket': + snap_socket = True + socket_running = running + + return snap_socket and socket_running and (not service or service_running) diff --git a/bauh/gems/snap/worker.py b/bauh/gems/snap/worker.py deleted file mode 100644 index 72dd76c2..00000000 --- a/bauh/gems/snap/worker.py +++ /dev/null @@ -1,70 +0,0 @@ -import traceback -from threading import Thread - -from bauh.api.abstract.cache import MemoryCache -from bauh.api.abstract.context import ApplicationContext -from bauh.api.abstract.controller import SoftwareManager -from bauh.api.abstract.model import PackageStatus -from bauh.gems.snap import snap -from bauh.gems.snap.constants import SNAP_API_URL -from bauh.gems.snap.model import SnapApplication - - -class SnapAsyncDataLoader(Thread): - - def __init__(self, app: SnapApplication, manager: SoftwareManager, api_cache: MemoryCache, - context: ApplicationContext): - super(SnapAsyncDataLoader, self).__init__(daemon=True) - self.app = app - self.id_ = '{}#{}'.format(self.__class__.__name__, id(self)) - self.manager = manager - self.http_client = context.http_client - self.api_cache = api_cache - self.persist = False - self.download_icons = context.download_icons - self.logger = context.logger - - def run(self): - if self.app: - self.app.status = PackageStatus.LOADING_DATA - - try: - res = self.http_client.session.get('{}/search?q={}'.format(SNAP_API_URL, self.app.name)) - - if res: - try: - snap_list = res.json()['_embedded']['clickindex:package'] - except: - self.logger.warning('Snap API response responded differently from expected for app: {}'.format(self.app.name)) - return - - if not snap_list: - self.logger.warning("Could not retrieve app data for id '{}'. Server response: {}. Body: {}".format(self.app.id, res.status_code, res.content.decode())) - else: - snap_data = snap_list[0] - - api_data = { - 'confinement': snap_data.get('confinement'), - 'description': snap_data.get('description'), - 'icon_url': snap_data.get('icon_url') if self.download_icons else None - } - - self.api_cache.add(self.app.id, api_data) - self.app.confinement = api_data['confinement'] - self.app.icon_url = api_data['icon_url'] - - if not api_data.get('description'): - api_data['description'] = snap.get_info(self.app.name, ('description',)).get('description') - - self.app.description = api_data['description'] - self.persist = self.app.supports_disk_cache() - else: - self.logger.warning("Could not retrieve app data for id '{}'. Server response: {}. Body: {}".format(self.app.id, res.status_code, res.content.decode())) - except: - self.logger.error("Could not retrieve app data for id '{}'".format(self.app.id)) - traceback.print_exc() - - self.app.status = PackageStatus.READY - - if self.persist: - self.manager.cache_to_disk(pkg=self.app, icon_bytes=None, only_icon=False)