mirror of
https://github.com/spalencsar/bearhub.git
synced 2026-07-06 21:44:16 +02:00
init
This commit is contained in:
Vinicius Moreira
@@ -4,6 +4,10 @@ All notable changes to this project will be documented in this file.
|
||||
|
||||
The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/).
|
||||
|
||||
## [0.5.0]
|
||||
### Comments
|
||||
- Env variables / arguments FPAKMAN_SNAP / FPAKMAN_FLATPAK were removed. To disable a supported type, just uninstall its module.
|
||||
|
||||
## [0.4.2]
|
||||
### Fixes:
|
||||
- [Snaps read index error](https://github.com/vinifmor/fpakman/issues/30)
|
||||
|
||||
@@ -49,14 +49,9 @@ You can change some application settings via environment variables or arguments
|
||||
- **FPAKMAN_ICON_EXPIRATION**: define a custom expiration time in SECONDS for cached icons. Default: 300 (5 minutes).
|
||||
- **FPAKMAN_DISK_CACHE**: enables / disables disk cache. When disk cache is enabled, the installed applications data are loaded faster. Use **0** (disable) or **1** (enable, default).
|
||||
- **FPAKMAN_DOWNLOAD_ICONS**: Enables / disables app icons download. It may improve the application speed depending on how applications data are being retrieved. Use **0** (disable) or **1** (enable, default).
|
||||
- **FPAKMAN_FLATPAK**: enables / disables flatpak usage. Use **0** (disable) or **1** (enabled, default)
|
||||
- **FPAKMAN_SNAP**: enables / disables snap usage. Use **0** (disable) or **1** (enabled, default)
|
||||
- **FPAKMAN_CHECK_PACKAGING_ONCE**: If the available supported packaging types should be checked ONLY once. It improves the application speed if enabled, but can generate errors if you uninstall any packaging technology while using it, and every time a supported packaging type is installed it will only be available after a restart. Use **0** (disable, default) or **1** (enable).
|
||||
|
||||
### How to improve the application performance
|
||||
- If you don't care about a specific packaging technology and don't want **fpakman** to deal with it, just disable it via the specific argument or environment variable. For instance, if I don't care
|
||||
about **snaps**, I can initialize the application setting "snap=0" (**fpakman --snap=0**). This will improve the application response time, since it won't need to do any verifications associated
|
||||
with the technology that I don't care every time an action is executed.
|
||||
- If you don't care about restarting **fpakman** every time a new supported packaging technology is installed, set "check-packaging-once=1" (**fpakman --check-packaging-once=1**). This can reduce the application response time up to 80% in some scenarios, since it won't need to recheck if the packaging type is available for every action you request.
|
||||
- If you don't mind to see the applications icons, you can set "download-icons=0" (**fpakman --download-icons=0**). The application may have a slight response improvement, since it will reduce the parallelism within it.
|
||||
|
||||
|
||||
@@ -11,13 +11,10 @@ from colorama import Fore
|
||||
from fpakman import __version__, __app_name__
|
||||
from fpakman.core import resource
|
||||
from fpakman.core.controller import GenericApplicationManager
|
||||
from fpakman.core.disk import DiskCacheLoaderFactory
|
||||
from fpakman_api.util.disk import DiskCacheLoaderFactory
|
||||
from fpakman.core.flatpak.constants import FLATPAK_CACHE_PATH
|
||||
from fpakman.core.flatpak.controller import FlatpakManager
|
||||
from fpakman.core.flatpak.model import FlatpakApplication
|
||||
from fpakman.core.snap.constants import SNAP_CACHE_PATH
|
||||
from fpakman.core.snap.controller import SnapManager
|
||||
from fpakman.core.snap.model import SnapApplication
|
||||
from fpakman.util import util
|
||||
from fpakman.util.cache import Cache
|
||||
from fpakman.util.memory import CacheCleaner
|
||||
@@ -42,7 +39,6 @@ parser.add_argument('-n', '--update-notification', action="store", choices=[0, 1
|
||||
parser.add_argument('-dc', '--disk-cache', action="store", choices=[0, 1], default=os.getenv('FPAKMAN_DISK_CACHE', 1), type=int, help='Enables / disables disk cache. When disk cache is enabled, the installed applications data are loaded faster. Default: %(default)s')
|
||||
parser.add_argument('-di', '--download-icons', action="store", choices=[0, 1], default=os.getenv('FPAKMAN_DOWNLOAD_ICONS', 1), type=int, help='Enables / disables app icons download. It may improve the application speed, depending of how applications data are retrieved by their extensions.')
|
||||
parser.add_argument('--flatpak', action="store", default=os.getenv('FPAKMAN_FLATPAK', 1), choices=[0, 1], type=int, help='Enables / disables flatpak usage. Default: %(default)s')
|
||||
parser.add_argument('--snap', action="store", default=os.getenv('FPAKMAN_SNAP', 1), choices=[0, 1], type=int, help='Enables / disables snap usage. Default: %(default)s')
|
||||
parser.add_argument('-co', '--check-packaging-once', action="store", default=os.getenv('FPAKMAN_CHECK_PACKAGING_ONCE', 0), choices=[0, 1], type=int, help='If the available supported packaging types should be checked ONLY once. It improves the application speed if enabled, but can generate errors if you uninstall any packaging technology while using it, and every time a supported packaging type is installed it will only be available after a restart. Default: %(default)s')
|
||||
args = parser.parse_args()
|
||||
|
||||
@@ -92,13 +88,14 @@ if args.flatpak:
|
||||
Path(FLATPAK_CACHE_PATH).mkdir(parents=True, exist_ok=True)
|
||||
|
||||
if args.snap:
|
||||
snap_api_cache = Cache(expiration_time=args.cache_exp)
|
||||
cache_map[SnapApplication] = snap_api_cache
|
||||
managers.append(SnapManager(app_args=args, disk_cache=args.disk_cache, api_cache=snap_api_cache, http_session=http_session))
|
||||
caches.append(snap_api_cache)
|
||||
|
||||
if args.disk_cache:
|
||||
Path(SNAP_CACHE_PATH).mkdir(parents=True, exist_ok=True)
|
||||
pass
|
||||
# snap_api_cache = Cache(expiration_time=args.cache_exp)
|
||||
# cache_map[SnapApplication] = snap_api_cache
|
||||
# managers.append(SnapManager(app_args=args, disk_cache=args.disk_cache, api_cache=snap_api_cache, http_session=http_session))
|
||||
# caches.append(snap_api_cache)
|
||||
#
|
||||
# if args.disk_cache:
|
||||
# Path(SNAP_CACHE_PATH).mkdir(parents=True, exist_ok=True)
|
||||
|
||||
icon_cache = Cache(expiration_time=args.icon_exp)
|
||||
caches.append(icon_cache)
|
||||
|
||||
@@ -1,6 +0,0 @@
|
||||
from pathlib import Path
|
||||
|
||||
from fpakman import __app_name__
|
||||
|
||||
HOME_PATH = Path.home()
|
||||
CACHE_PATH = '{}/.cache/{}'.format(HOME_PATH, __app_name__)
|
||||
@@ -1,88 +1,11 @@
|
||||
from abc import ABC, abstractmethod
|
||||
from argparse import Namespace
|
||||
from typing import List, Dict
|
||||
|
||||
from fpakman.core.disk import DiskCacheLoader, DiskCacheLoaderFactory
|
||||
from fpakman.core.model import Application, ApplicationUpdate
|
||||
from fpakman.core.system import FpakmanProcess
|
||||
|
||||
|
||||
class ApplicationManager(ABC):
|
||||
|
||||
def __init__(self, app_args):
|
||||
self.app_args = app_args
|
||||
|
||||
@abstractmethod
|
||||
def search(self, word: str, disk_loader: DiskCacheLoader) -> Dict[str, List[Application]]:
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def read_installed(self, disk_loader: DiskCacheLoader) -> List[Application]:
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def downgrade_app(self, app: Application, root_password: str) -> FpakmanProcess:
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def clean_cache_for(self, app: Application):
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def can_downgrade(self):
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def update_and_stream(self, app: Application) -> FpakmanProcess:
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def uninstall_and_stream(self, app: Application, root_password: str) -> FpakmanProcess:
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def get_app_type(self):
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def get_info(self, app: Application) -> dict:
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def get_history(self, app: Application) -> List[dict]:
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def install_and_stream(self, app: Application, root_password: str) -> FpakmanProcess:
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def is_enabled(self) -> bool:
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def cache_to_disk(self, app: Application, icon_bytes: bytes, only_icon: bool):
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def requires_root(self, action: str, app: Application):
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def refresh(self, app: Application, root_password: str) -> FpakmanProcess:
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def prepare(self):
|
||||
"""
|
||||
Callback executed before the ApplicationManager starts to work.
|
||||
:return:
|
||||
"""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def list_updates(self) -> List[ApplicationUpdate]:
|
||||
pass
|
||||
from fpakman_api.abstract.controller import ApplicationManager
|
||||
from fpakman_api.abstract.model import Application, ApplicationUpdate
|
||||
from fpakman_api.util.disk import DiskCacheLoader
|
||||
from fpakman_api.util.disk import DiskCacheLoaderFactory
|
||||
from fpakman_api.util.system import FpakmanProcess
|
||||
|
||||
|
||||
class GenericApplicationManager(ApplicationManager):
|
||||
@@ -143,7 +66,7 @@ class GenericApplicationManager(ApplicationManager):
|
||||
disk_loader = self.disk_loader_factory.new()
|
||||
disk_loader.start()
|
||||
|
||||
apps_found = man.search(word=norm_word, disk_loader=disk_loader)
|
||||
apps_found = man.search(words=norm_word, disk_loader=disk_loader)
|
||||
res['installed'].extend(apps_found['installed'])
|
||||
res['new'].extend(apps_found['new'])
|
||||
|
||||
|
||||
@@ -1,75 +0,0 @@
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
from threading import Thread, Lock
|
||||
from typing import List, Dict
|
||||
|
||||
from fpakman.core.model import Application
|
||||
from fpakman.util.cache import Cache
|
||||
|
||||
|
||||
class DiskCacheLoader(Thread):
|
||||
|
||||
def __init__(self, enabled: bool, cache_map: Dict[type, Cache], apps: List[Application] = []):
|
||||
super(DiskCacheLoader, self).__init__(daemon=True)
|
||||
self.apps = apps
|
||||
self.stop = False
|
||||
self.lock = Lock()
|
||||
self.cache_map = cache_map
|
||||
self.enabled = enabled
|
||||
|
||||
def run(self):
|
||||
|
||||
if self.enabled:
|
||||
while True:
|
||||
if self.apps:
|
||||
self.lock.acquire()
|
||||
app = self.apps[0]
|
||||
del self.apps[0]
|
||||
self.lock.release()
|
||||
self.fill_cached_data(app)
|
||||
elif self.stop:
|
||||
break
|
||||
|
||||
def add(self, app: Application):
|
||||
if self.enabled:
|
||||
if app and app.supports_disk_cache():
|
||||
self.lock.acquire()
|
||||
self.apps.append(app)
|
||||
self.lock.release()
|
||||
|
||||
def fill_cached_data(self, app: Application):
|
||||
if self.enabled:
|
||||
if os.path.exists(app.get_disk_data_path()):
|
||||
with open(app.get_disk_data_path()) as f:
|
||||
cached_data = json.loads(f.read())
|
||||
app.fill_cached_data(cached_data)
|
||||
self.cache_map.get(app.__class__).add_non_existing(app.base_data.id, cached_data)
|
||||
|
||||
|
||||
class DiskCacheLoaderFactory:
|
||||
|
||||
def __init__(self, disk_cache: bool, cache_map: Dict[type, Cache]):
|
||||
self.disk_cache = disk_cache
|
||||
self.cache_map = cache_map
|
||||
|
||||
def new(self):
|
||||
return DiskCacheLoader(enabled=self.disk_cache, cache_map=self.cache_map)
|
||||
|
||||
|
||||
def save(app: Application, icon_bytes: bytes = None, only_icon: bool = False):
|
||||
|
||||
if app.supports_disk_cache():
|
||||
|
||||
if not only_icon:
|
||||
Path(app.get_disk_cache_path()).mkdir(parents=True, exist_ok=True)
|
||||
data = app.get_data_to_cache()
|
||||
|
||||
with open(app.get_disk_data_path(), 'w+') as f:
|
||||
f.write(json.dumps(data))
|
||||
|
||||
if icon_bytes:
|
||||
Path(app.get_disk_cache_path()).mkdir(parents=True, exist_ok=True)
|
||||
|
||||
with open(app.get_disk_icon_path(), 'wb+') as f:
|
||||
f.write(icon_bytes)
|
||||
@@ -1,3 +0,0 @@
|
||||
|
||||
class NoInternetException(Exception):
|
||||
pass
|
||||
@@ -1,5 +0,0 @@
|
||||
from fpakman.core.constants import CACHE_PATH
|
||||
|
||||
FLATHUB_URL = 'https://flathub.org'
|
||||
FLATHUB_API_URL = FLATHUB_URL + '/api/v1'
|
||||
FLATPAK_CACHE_PATH = '{}/flatpak/installed'.format(CACHE_PATH)
|
||||
@@ -1,99 +0,0 @@
|
||||
from abc import ABC, abstractmethod
|
||||
from enum import Enum
|
||||
|
||||
from fpakman.core import resource
|
||||
|
||||
|
||||
class ApplicationStatus(Enum):
|
||||
READY = 1
|
||||
LOADING_DATA = 2
|
||||
|
||||
|
||||
class ApplicationData:
|
||||
|
||||
def __init__(self, id: str, version: str, name: str = None, description: str = None, latest_version: str = None, icon_url: str = None):
|
||||
self.id = id
|
||||
self.name = name
|
||||
self.version = version
|
||||
self.description = description
|
||||
self.latest_version = latest_version
|
||||
self.icon_url = icon_url
|
||||
|
||||
|
||||
class Application(ABC):
|
||||
|
||||
def __init__(self, base_data: ApplicationData, status: ApplicationStatus = ApplicationStatus.READY, installed: bool = False, update: bool = False):
|
||||
self.base_data = base_data
|
||||
self.status = status
|
||||
self.installed = installed
|
||||
self.update = update
|
||||
|
||||
@abstractmethod
|
||||
def has_history(self):
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def has_info(self):
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def can_be_downgraded(self):
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def can_be_uninstalled(self):
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def can_be_installed(self):
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def can_be_refreshed(self):
|
||||
return self.installed
|
||||
|
||||
@abstractmethod
|
||||
def get_type(self):
|
||||
pass
|
||||
|
||||
def get_default_icon_path(self):
|
||||
return resource.get_path('img/logo.svg')
|
||||
|
||||
@abstractmethod
|
||||
def is_library(self):
|
||||
pass
|
||||
|
||||
def supports_disk_cache(self):
|
||||
return self.installed and not self.is_library()
|
||||
|
||||
@abstractmethod
|
||||
def get_disk_cache_path(self):
|
||||
pass
|
||||
|
||||
def get_disk_icon_path(self):
|
||||
return '{}/icon.png'.format(self.get_disk_cache_path())
|
||||
|
||||
def get_disk_data_path(self):
|
||||
return '{}/data.json'.format(self.get_disk_cache_path())
|
||||
|
||||
@abstractmethod
|
||||
def get_data_to_cache(self):
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def fill_cached_data(self, data: dict):
|
||||
pass
|
||||
|
||||
def __str__(self):
|
||||
return '{} (id={}, name={})'.format(self.__class__.__name__, self.base_data.id, self.base_data.name)
|
||||
|
||||
|
||||
class ApplicationUpdate:
|
||||
|
||||
def __init__(self, app_id: str, version: str, app_type: str):
|
||||
self.id = app_id
|
||||
self.version = version
|
||||
self.type = app_type
|
||||
|
||||
def __str__(self):
|
||||
return '{} (id={}, type={}, new_version={})'.format(self.__class__.__name__, self.id, self.type, self.type)
|
||||
@@ -1,4 +1,3 @@
|
||||
|
||||
from fpakman import ROOT_DIR
|
||||
|
||||
|
||||
|
||||
@@ -1,4 +0,0 @@
|
||||
from fpakman.core.constants import CACHE_PATH
|
||||
|
||||
SNAP_API_URL = 'https://search.apps.ubuntu.com/api/v1'
|
||||
SNAP_CACHE_PATH = '{}/snap/installed'.format(CACHE_PATH)
|
||||
@@ -1,135 +0,0 @@
|
||||
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, ApplicationUpdate
|
||||
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]:
|
||||
return [self.map_json(app_json, installed=True, disk_loader=disk_loader) for app_json in snap.read_installed()]
|
||||
|
||||
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.confinement, 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
|
||||
|
||||
def list_updates(self) -> List[ApplicationUpdate]:
|
||||
return []
|
||||
@@ -1,60 +0,0 @@
|
||||
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, confinement: str = None):
|
||||
super(SnapApplication, self).__init__(base_data=base_data)
|
||||
self.publisher = publisher
|
||||
self.rev = rev
|
||||
self.notes = notes
|
||||
self.type = app_type
|
||||
self.confinement = confinement
|
||||
|
||||
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,
|
||||
'confinement': self.confinement,
|
||||
'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('confinement'):
|
||||
self.confinement = data['confinement']
|
||||
@@ -1,134 +0,0 @@
|
||||
import re
|
||||
import subprocess
|
||||
from typing import List
|
||||
|
||||
from fpakman.core import system
|
||||
|
||||
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] if len(app_data) >= 5 else None,
|
||||
'notes': app_data[5] if len(app_data) >= 6 else None
|
||||
}
|
||||
|
||||
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, confinement: str, root_password: str) -> subprocess.Popen:
|
||||
|
||||
install_cmd = [BASE_CMD, 'install', app_name] # default
|
||||
|
||||
if confinement == 'classic':
|
||||
install_cmd.append('--classic')
|
||||
|
||||
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)
|
||||
@@ -1,85 +0,0 @@
|
||||
import time
|
||||
import traceback
|
||||
|
||||
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_API_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.persist = False
|
||||
self.download_icons = download_icons
|
||||
|
||||
def run(self):
|
||||
if self.app:
|
||||
self.app.status = ApplicationStatus.LOADING_DATA
|
||||
|
||||
for _ in range(0, self.attempts):
|
||||
try:
|
||||
res = self.http_session.get('{}/search?q={}'.format(SNAP_API_URL, self.app.base_data.name), timeout=self.timeout)
|
||||
|
||||
if res.status_code == 200 and res.text:
|
||||
|
||||
try:
|
||||
snap_list = res.json()['_embedded']['clickindex:package']
|
||||
except:
|
||||
self.log_msg('Snap API response responded differently from expected for app: {}'.format(self.app.base_data.name))
|
||||
break
|
||||
|
||||
if not snap_list:
|
||||
break
|
||||
|
||||
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.base_data.id, api_data)
|
||||
self.app.confinement = api_data['confinement']
|
||||
self.app.base_data.icon_url = api_data['icon_url']
|
||||
|
||||
if not api_data.get('description'):
|
||||
api_data['description'] = snap.get_info(self.app.base_data.name, ('description',)).get('description')
|
||||
|
||||
self.app.base_data.description = api_data['description']
|
||||
|
||||
self.app.status = ApplicationStatus.READY
|
||||
self.persist = self.app.supports_disk_cache()
|
||||
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(self.app.base_data.id), Fore.YELLOW)
|
||||
traceback.print_exc()
|
||||
time.sleep(0.5)
|
||||
|
||||
self.app.status = ApplicationStatus.READY
|
||||
|
||||
if self.persist:
|
||||
self.manager.cache_to_disk(app=self.app, icon_bytes=None, only_icon=False)
|
||||
|
||||
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)
|
||||
|
||||
@@ -1,51 +0,0 @@
|
||||
import os
|
||||
import subprocess
|
||||
from typing import List
|
||||
|
||||
from fpakman import __app_name__
|
||||
from fpakman.core import resource
|
||||
|
||||
|
||||
class FpakmanProcess:
|
||||
|
||||
def __init__(self, subproc: subprocess.Popen, success_phrase: str = None, wrong_error_phrase: str = '[sudo] password for'):
|
||||
self.subproc = subproc
|
||||
self.success_pgrase = success_phrase
|
||||
self.wrong_error_phrase = wrong_error_phrase
|
||||
|
||||
|
||||
def run_cmd(cmd: str, expected_code: int = 0, ignore_return_code: bool = False, print_error: bool = True) -> str:
|
||||
args = {
|
||||
"shell": True,
|
||||
"stdout": subprocess.PIPE,
|
||||
"env": {'LANG': 'en'}
|
||||
}
|
||||
|
||||
if not print_error:
|
||||
args["stderr"] = subprocess.DEVNULL
|
||||
|
||||
res = subprocess.run(cmd, **args)
|
||||
return res.stdout.decode() if ignore_return_code or res.returncode == expected_code else None
|
||||
|
||||
|
||||
def stream_cmd(cmd: List[str]):
|
||||
return cmd_to_subprocess(cmd).stdout
|
||||
|
||||
|
||||
def cmd_to_subprocess(cmd: List[str]):
|
||||
return subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, env={'LANG': 'en'})
|
||||
|
||||
|
||||
def notify_user(msg: str, icon_path: str = resource.get_path('img/logo.svg')):
|
||||
os.system("notify-send -a {} {} '{}'".format(__app_name__, "-i {}".format(icon_path) if icon_path else '', msg))
|
||||
|
||||
|
||||
def cmd_as_root(cmd: List[str], root_password: str) -> subprocess.Popen:
|
||||
pwdin, final_cmd = None, []
|
||||
|
||||
if root_password is not None:
|
||||
final_cmd.extend(['sudo', '-S'])
|
||||
pwdin = stream_cmd(['echo', root_password])
|
||||
|
||||
final_cmd.extend(cmd)
|
||||
return subprocess.Popen(final_cmd, stdin=pwdin, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
|
||||
@@ -1,31 +0,0 @@
|
||||
from io import StringIO
|
||||
from threading import Thread
|
||||
|
||||
from colorama import Fore
|
||||
|
||||
from fpakman.core.model import Application
|
||||
|
||||
|
||||
class AsyncDataLoader(Thread):
|
||||
|
||||
def __init__(self, app: Application):
|
||||
super(AsyncDataLoader, self).__init__(daemon=True)
|
||||
self.id_ = '{}#{}'.format(self.__class__.__name__, id(self))
|
||||
self.app = app
|
||||
|
||||
def log_msg(self, msg: str, color: int = None):
|
||||
final_msg = StringIO()
|
||||
|
||||
if color:
|
||||
final_msg.write(str(color))
|
||||
|
||||
final_msg.write('[{}] '.format(self.id_))
|
||||
|
||||
final_msg.write(msg)
|
||||
|
||||
if color:
|
||||
final_msg.write(Fore.RESET)
|
||||
|
||||
final_msg.seek(0)
|
||||
|
||||
print(final_msg.read())
|
||||
Binary file not shown.
|
Before Width: | Height: | Size: 1.1 KiB |
@@ -1,3 +1,4 @@
|
||||
pyqt5>=5.12
|
||||
requests>=2.22
|
||||
colorama>=0.4.1
|
||||
fpakman_api==0.1.0
|
||||
Reference in New Issue
Block a user