supporting snaps

fpakman/core/snap/constants.py

@@ -0,0 +1,4 @@
+from fpakman.core.constants import CACHE_PATH
+
+SNAP_STORE_URL = 'https://snapcraft.io'
+SNAP_CACHE_PATH = '{}/snap/installed'.format(CACHE_PATH)

fpakman/core/snap/controller.py

@@ -0,0 +1,133 @@
+import os
+import shutil
+from argparse import Namespace
+from datetime import datetime
+from typing import Dict, List
+
+from fpakman.core import disk
+from fpakman.core.controller import ApplicationManager
+from fpakman.core.disk import DiskCacheLoader
+from fpakman.core.model import ApplicationData, Application
+from fpakman.core.snap import snap
+from fpakman.core.snap.model import SnapApplication
+from fpakman.core.snap.worker import SnapAsyncDataLoader
+from fpakman.core.system import FpakmanProcess
+from fpakman.util.cache import Cache
+
+
+class SnapManager(ApplicationManager):
+
+    def __init__(self, app_args: Namespace,  api_cache: Cache, disk_cache: bool, http_session):
+        super(SnapManager, self).__init__(app_args=app_args)
+        self.api_cache = api_cache
+        self.http_session = http_session
+        self.disk_cache = disk_cache
+
+    def map_json(self, app_json: dict, installed: bool,  disk_loader: DiskCacheLoader) -> SnapApplication:
+        app = SnapApplication(publisher=app_json.get('publisher'),
+                              rev=app_json.get('rev'),
+                              notes=app_json.get('notes'),
+                              app_type=app_json.get('type'),
+                              base_data=ApplicationData(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')
+                                                        ))
+
+        if app.publisher:
+            app.publisher = app.publisher.replace('*', '')
+
+        app.installed = installed
+
+        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 not app.is_library():
+            if disk_loader and app.installed:
+                disk_loader.add(app)
+
+            SnapAsyncDataLoader(app=app, api_cache=self.api_cache, manager=self, http_session=self.http_session, download_icons=self.app_args.download_icons).start()
+        else:
+            app.fill_cached_data(api_data)
+
+        return app
+
+    def search(self, word: str, disk_loader: DiskCacheLoader) -> Dict[str, List[SnapApplication]]:
+        installed = self.read_installed(disk_loader)
+
+        res = {'installed': [], 'new': []}
+
+        for app_json in snap.search(word):
+
+            already_installed = None
+
+            if installed:
+                already_installed = [i for i in installed if i.base_data.id == app_json.get('name')]
+                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))
+
+        return res
+
+    def read_installed(self, disk_loader: DiskCacheLoader) -> List[SnapApplication]:
+        res = [self.map_json(app_json, installed=True, disk_loader=disk_loader) for app_json in snap.read_installed()]
+        return res
+
+    def downgrade_app(self, app: Application, root_password: str) -> FpakmanProcess:
+        return FpakmanProcess(subproc=snap.downgrade_and_stream(app.base_data.name, root_password), wrong_error_phrase=None)
+
+    def clean_cache_for(self, app: SnapApplication):
+        self.api_cache.delete(app.base_data.name)
+
+        if app.supports_disk_cache() and os.path.exists(app.get_disk_cache_path()):
+            shutil.rmtree(app.get_disk_cache_path())
+
+    def can_downgrade(self):
+        return True
+
+    def update_and_stream(self, app: SnapApplication) -> FpakmanProcess:
+        pass
+
+    def uninstall_and_stream(self, app: SnapApplication, root_password: str) -> FpakmanProcess:
+        return FpakmanProcess(subproc=snap.uninstall_and_stream(app.base_data.name, root_password))
+
+    def get_app_type(self):
+        return SnapApplication
+
+    def get_info(self, app: SnapApplication) -> dict:
+        info = snap.get_info(app.base_data.name, attrs=('license', 'contact', 'commands', 'snap-id', 'tracking', 'installed'))
+        info['description'] = app.base_data.description
+        info['publisher'] = app.publisher
+        info['revision'] = app.rev
+        info['name'] = app.base_data.name
+
+        if info.get('commands'):
+            info['commands'] = ' '.join(info['commands'])
+
+        return info
+
+    def get_history(self, app: Application) -> List[dict]:
+        return []
+
+    def install_and_stream(self, app: SnapApplication, root_password: str) -> FpakmanProcess:
+        return FpakmanProcess(subproc=snap.install_and_stream(app.base_data.name, app.install_cmd, root_password))
+
+    def is_enabled(self) -> bool:
+        return snap.is_installed()
+
+    def cache_to_disk(self, app: Application, icon_bytes: bytes, only_icon: bool):
+        if self.disk_cache and app.supports_disk_cache():
+            disk.save(app, icon_bytes, only_icon)
+
+    def requires_root(self, action: str, app: SnapApplication):
+        return action != 'search'
+
+    def refresh(self, app: SnapApplication, root_password: str) -> FpakmanProcess:
+        return FpakmanProcess(subproc=snap.refresh_and_stream(app.base_data.name, root_password))
+
+    def prepare(self):
+        pass

fpakman/core/snap/model.py

@@ -0,0 +1,60 @@
+from fpakman.core import resource
+from fpakman.core.model import Application, ApplicationData
+from fpakman.core.snap.constants import SNAP_CACHE_PATH
+
+
+class SnapApplication(Application):
+
+    def __init__(self, base_data: ApplicationData, publisher: str, rev: str, notes: str, app_type: str, install_cmd: str = None):
+        super(SnapApplication, self).__init__(base_data=base_data)
+        self.publisher = publisher
+        self.rev = rev
+        self.notes = notes
+        self.type = app_type
+        self.install_cmd = install_cmd
+
+    def has_history(self):
+        return False
+
+    def has_info(self):
+        return True
+
+    def can_be_downgraded(self):
+        return self.installed
+
+    def can_be_uninstalled(self):
+        return self.installed
+
+    def can_be_installed(self):
+        return not self.installed
+
+    def can_be_refreshed(self):
+        return self.installed
+
+    def get_type(self):
+        return 'snap'
+
+    def get_default_icon_path(self):
+        return resource.get_path('img/snapcraft.png')
+
+    def is_library(self):
+        return self.type in ('core', 'base', 'snapd') or self.base_data.name.startswith('gtk-') or self.base_data.name.startswith('gnome-')
+
+    def get_disk_cache_path(self):
+        return '{}/{}'.format(SNAP_CACHE_PATH, self.base_data.name)
+
+    def get_data_to_cache(self):
+        return {
+            "icon_url": self.base_data.icon_url,
+            'install_cmd': self.install_cmd,
+            'description': self.base_data.description
+        }
+
+    def fill_cached_data(self, data: dict):
+        if data:
+            for base_attr in ('icon_url', 'description'):
+                if data.get(base_attr):
+                    setattr(self.base_data, base_attr, data[base_attr])
+
+            if data.get('install_cmd'):
+                self.install_cmd = data['install_cmd']

fpakman/core/snap/snap.py

@@ -0,0 +1,145 @@
+import re
+import subprocess
+from typing import List
+
+import requests
+from bs4 import BeautifulSoup
+
+from fpakman.core import system
+from fpakman.core.snap.constants import SNAP_STORE_URL
+
+BASE_CMD = 'snap'
+
+
+def is_installed():
+    version = get_snapd_version()
+    return False if version is None else True
+
+
+def get_version():
+    res = system.run_cmd('{} --version'.format(BASE_CMD), print_error=False)
+    return res.split('\n')[0].split(' ')[-1].strip() if res else None
+
+
+def get_snapd_version():
+    res = system.run_cmd('{} --version'.format(BASE_CMD), print_error=False)
+
+    if not res:
+        return None
+    else:
+        lines = res.split('\n')
+
+        if lines and len(lines) >= 2:
+            version = lines[1].split(' ')[-1].strip()
+            return version if version and version.lower() != 'unavailable' else None
+        else:
+            return None
+
+
+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],
+        'notes': app_data[5]
+    }
+
+    app_json.update(get_info(app_json['name'], ('summary', 'type', 'description')))
+    return app_json
+
+
+def get_info(app_name: str, attrs: tuple = None):
+    full_info_lines = system.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:
+            data[info[0]] = info[1].strip()
+
+        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].replace('-', '').strip().split('\n') if commands else None
+
+    return data
+
+
+def read_installed() -> List[dict]:
+    res = system.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))
+
+    return apps
+
+
+def search(word: str) -> List[dict]:
+    apps = []
+
+    res = system.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]
+                    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
+                    })
+
+    return apps
+
+
+def uninstall_and_stream(app_name: str, root_password: str):
+    return system.cmd_as_root([BASE_CMD, 'remove', app_name], root_password)
+
+
+def install_and_stream(app_name: str, custom_install_cmd: str, root_password: str) -> subprocess.Popen:
+
+    install_cmd = [BASE_CMD, 'install', app_name]  # default
+
+    if custom_install_cmd:
+        install_cmd = custom_install_cmd.split(' ')
+    else:   # tries to retrieve the snapstore proper installation command
+        res = requests.get('{}/{}'.format(SNAP_STORE_URL, app_name))
+
+        if res.status_code == 200:
+            soup = BeautifulSoup(res.text, 'html.parser')
+            input_install_cmd = soup.find("input", {"id": "snap-install"})
+            install_cmd = input_install_cmd.get("value").strip().split(' ')
+
+    return system.cmd_as_root(install_cmd, root_password)
+
+
+def downgrade_and_stream(app_name: str, root_password: str) -> subprocess.Popen:
+    return system.cmd_as_root([BASE_CMD, 'revert', app_name], root_password)
+
+
+def refresh_and_stream(app_name: str, root_password: str) -> subprocess.Popen:
+    return system.cmd_as_root([BASE_CMD, 'refresh', app_name], root_password)

fpakman/core/snap/worker.py

@@ -0,0 +1,87 @@
+import time
+import traceback
+
+from bs4 import BeautifulSoup
+from colorama import Fore
+
+from fpakman.core.controller import ApplicationManager
+from fpakman.core.model import ApplicationStatus
+from fpakman.core.snap import snap
+from fpakman.core.snap.constants import SNAP_STORE_URL
+from fpakman.core.snap.model import SnapApplication
+from fpakman.core.worker import AsyncDataLoader
+from fpakman.util.cache import Cache
+
+
+class SnapAsyncDataLoader(AsyncDataLoader):
+
+    def __init__(self, app: SnapApplication, manager: ApplicationManager, http_session, api_cache: Cache, download_icons: bool, attempts: int = 2, timeout: int = 30):
+        super(SnapAsyncDataLoader, self).__init__(app=app)
+        self.manager = manager
+        self.http_session = http_session
+        self.attempts = attempts
+        self.api_cache = api_cache
+        self.timeout = timeout
+        self.to_persist = {}  # stores all data loaded by the instance
+        self.download_icons = download_icons
+
+    def run(self):
+        if self.app:
+            self.app.status = ApplicationStatus.LOADING_DATA
+            if not self.app.base_data.description:
+                self.app.base_data.description = snap.get_info(self.app.base_data.name, ('description',)).get('description')
+
+            for _ in range(0, self.attempts):
+                try:
+                    res = self.http_session.get('{}/{}'.format(SNAP_STORE_URL, self.app.base_data.name),
+                                                timeout=self.timeout)
+
+                    if res.status_code == 200 and res.text:
+                        soup = BeautifulSoup(res.text, 'html.parser')
+                        input_install_cmd = soup.find("input", {"id": "snap-install"})
+
+                        api_data = {
+                            'install_cmd': input_install_cmd.get("value").strip(),
+                            'description': self.app.base_data.description
+                        }
+
+                        img_icon = None
+
+                        if self.download_icons:
+                            img_icon = soup.find('img', {"class": "p-snap-heading__icon"})
+
+                        if img_icon and img_icon.get("src") and 'snapcraft-missing-icon' not in img_icon.get('src'):
+                            api_data['icon_url'] = img_icon.get("src")
+
+                        self.api_cache.add(self.app.base_data.id, api_data)
+                        self.app.install_cmd = api_data['install_cmd']
+                        self.app.base_data.icon_url = api_data.get('icon_url')
+                        self.app.status = ApplicationStatus.READY
+
+                        if self.app.supports_disk_cache():
+                            self.to_persist[self.app.base_data.id] = self.app
+
+                        break
+                    else:
+                        self.log_msg("Could not retrieve app data for id '{}'. Server response: {}. Body: {}".format(self.app.base_data.id, res.status_code, res.content.decode()), Fore.RED)
+                except:
+                    self.log_msg("Could not retrieve app data for id '{}'".format(app.base_data.id), Fore.YELLOW)
+                    traceback.print_exc()
+                    time.sleep(0.5)
+
+            self.app.status = ApplicationStatus.READY
+
+    def clone(self) -> "SnapAsyncDataLoader":
+        return SnapAsyncDataLoader(manager=self.manager,
+                                   api_cache=self.api_cache,
+                                   attempts=self.attempts,
+                                   http_session=self.http_session,
+                                   timeout=self.timeout,
+                                   app=self.app)
+
+    def cache_to_disk(self):
+        if self.to_persist:
+            for app in self.to_persist.values():
+                self.manager.cache_to_disk(app=app, icon_bytes=None, only_icon=False)
+
+            self.to_persist = {}