mirror of
https://github.com/spalencsar/bearhub.git
synced 2026-07-07 10:34:15 +02:00
317 lines
9.9 KiB
Python
Executable File
317 lines
9.9 KiB
Python
Executable File
import os
|
|
import shutil
|
|
from abc import ABC, abstractmethod
|
|
from datetime import datetime
|
|
from threading import Lock
|
|
from typing import List
|
|
|
|
import requests
|
|
|
|
from fpakman.core import flatpak, disk
|
|
from fpakman.core.disk import DiskCacheLoader, DiskCacheLoaderFactory
|
|
from fpakman.core.model import FlatpakApplication, ApplicationData, ApplicationStatus, Application
|
|
from fpakman.util.cache import Cache
|
|
|
|
|
|
class ApplicationManager(ABC):
|
|
|
|
@abstractmethod
|
|
def search(self, word: str, disk_loader: DiskCacheLoader) -> List[Application]:
|
|
pass
|
|
|
|
@abstractmethod
|
|
def read_installed(self, disk_loader: DiskCacheLoader, keep_workers: bool) -> List[Application]:
|
|
pass
|
|
|
|
@abstractmethod
|
|
def downgrade_app(self, app: Application, root_password: str):
|
|
pass
|
|
|
|
@abstractmethod
|
|
def clean_cache_for(self, app: Application):
|
|
pass
|
|
|
|
@abstractmethod
|
|
def can_downgrade(self):
|
|
pass
|
|
|
|
@abstractmethod
|
|
def update_and_stream(self, app: Application):
|
|
pass
|
|
|
|
@abstractmethod
|
|
def uninstall_and_stream(self, app: Application):
|
|
pass
|
|
|
|
@abstractmethod
|
|
def get_app_type(self) -> str:
|
|
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):
|
|
pass
|
|
|
|
@abstractmethod
|
|
def is_enabled(self) -> bool:
|
|
pass
|
|
|
|
@abstractmethod
|
|
def cache_to_disk(self, app: Application, icon_bytes: bytes, only_icon: bool):
|
|
pass
|
|
|
|
|
|
from fpakman.core.worker import FlatpakAsyncDataLoaderManager
|
|
|
|
|
|
class FlatpakManager(ApplicationManager):
|
|
|
|
def __init__(self, api_cache: Cache, disk_cache: bool):
|
|
self.api_cache = api_cache
|
|
self.http_session = requests.Session()
|
|
self.lock_read = Lock()
|
|
self.disk_cache = disk_cache
|
|
self.async_data_loader = FlatpakAsyncDataLoaderManager(api_cache=self.api_cache, manager=self)
|
|
flatpak.set_default_remotes()
|
|
|
|
def get_app_type(self):
|
|
return FlatpakApplication
|
|
|
|
def _map_to_model(self, app: dict, installed: bool, disk_loader: DiskCacheLoader) -> FlatpakApplication:
|
|
|
|
model = FlatpakApplication(arch=app.get('arch'),
|
|
branch=app.get('branch'),
|
|
origin=app.get('origin'),
|
|
runtime=app.get('runtime'),
|
|
ref=app.get('ref'),
|
|
commit=app.get('commit'),
|
|
base_data=ApplicationData(id=app.get('id'),
|
|
name=app.get('name'),
|
|
version=app.get('version'),
|
|
latest_version=app.get('latest_version')))
|
|
model.installed = installed
|
|
|
|
api_data = self.api_cache.get(app['id'])
|
|
|
|
expired_data = api_data and api_data.get('expires_at') and api_data['expires_at'] <= datetime.utcnow()
|
|
|
|
if not api_data or expired_data:
|
|
if not app['runtime']:
|
|
disk_loader.add(model) # preloading cached disk data
|
|
model.status = ApplicationStatus.LOADING_DATA
|
|
self.async_data_loader.load(model)
|
|
|
|
else:
|
|
model.fill_cached_data(api_data)
|
|
|
|
return model
|
|
|
|
def search(self, word: str, disk_loader: DiskCacheLoader) -> List[FlatpakApplication]:
|
|
|
|
res = []
|
|
apps_found = flatpak.search(word)
|
|
|
|
if apps_found:
|
|
already_read = set()
|
|
installed_apps = self.read_installed(disk_loader=disk_loader, keep_workers=True)
|
|
|
|
if installed_apps:
|
|
for app_found in apps_found:
|
|
for installed_app in installed_apps:
|
|
if app_found['id'] == installed_app.base_data.id:
|
|
res.append(installed_app)
|
|
already_read.add(app_found['id'])
|
|
|
|
for app_found in apps_found:
|
|
if app_found['id'] not in already_read:
|
|
res.append(self._map_to_model(app_found, False, disk_loader))
|
|
|
|
disk_loader.stop = True
|
|
disk_loader.join()
|
|
self.async_data_loader.stop_current_workers()
|
|
|
|
return res
|
|
|
|
def read_installed(self, disk_loader: DiskCacheLoader, keep_workers: bool = False) -> List[FlatpakApplication]:
|
|
|
|
self.lock_read.acquire()
|
|
|
|
try:
|
|
installed = flatpak.list_installed()
|
|
|
|
if installed:
|
|
installed.sort(key=lambda p: p['name'].lower())
|
|
|
|
available_updates = flatpak.list_updates_as_str()
|
|
|
|
models = []
|
|
|
|
for app in installed:
|
|
model = self._map_to_model(app, True, disk_loader)
|
|
model.update = app['id'] in available_updates
|
|
models.append(model)
|
|
|
|
if not keep_workers:
|
|
self.async_data_loader.stop_current_workers()
|
|
|
|
return models
|
|
|
|
return []
|
|
|
|
finally:
|
|
self.lock_read.release()
|
|
|
|
def can_downgrade(self):
|
|
return True
|
|
|
|
def downgrade_app(self, app: FlatpakApplication, root_password: str):
|
|
|
|
commits = flatpak.get_app_commits(app.ref, app.origin)
|
|
|
|
commit_idx = commits.index(app.commit)
|
|
|
|
# downgrade is not possible if the app current commit in the first one:
|
|
if commit_idx == len(commits) - 1:
|
|
return None
|
|
|
|
return flatpak.downgrade_and_stream(app.ref, commits[commit_idx + 1], root_password)
|
|
|
|
def clean_cache_for(self, app: FlatpakApplication):
|
|
self.api_cache.delete(app.base_data.id)
|
|
|
|
if app.supports_disk_cache() and os.path.exists(app.get_disk_cache_path()):
|
|
shutil.rmtree(app.get_disk_cache_path())
|
|
|
|
def update_and_stream(self, app: FlatpakApplication):
|
|
return flatpak.update_and_stream(app.ref)
|
|
|
|
def uninstall_and_stream(self, app: FlatpakApplication):
|
|
return flatpak.uninstall_and_stream(app.ref)
|
|
|
|
def get_info(self, app: FlatpakApplication) -> dict:
|
|
app_info = flatpak.get_app_info_fields(app.base_data.id, app.branch)
|
|
app_info['name'] = app.base_data.name
|
|
app_info['type'] = 'runtime' if app.runtime else 'app'
|
|
app_info['description'] = app.base_data.description
|
|
return app_info
|
|
|
|
def get_history(self, app: FlatpakApplication) -> List[dict]:
|
|
return flatpak.get_app_commits_data(app.ref, app.origin)
|
|
|
|
def install_and_stream(self, app: FlatpakApplication):
|
|
return flatpak.install_and_stream(app.base_data.id, app.origin)
|
|
|
|
def is_enabled(self):
|
|
return flatpak.is_installed()
|
|
|
|
def cache_to_disk(self, app: FlatpakApplication, icon_bytes: bytes, only_icon: bool):
|
|
if self.disk_cache and app.supports_disk_cache():
|
|
disk.save(app, icon_bytes, only_icon)
|
|
|
|
|
|
class GenericApplicationManager(ApplicationManager):
|
|
|
|
def __init__(self, managers: List[ApplicationManager], disk_loader_factory: DiskCacheLoaderFactory):
|
|
self.managers = managers
|
|
self.map = {m.get_app_type(): m for m in self.managers}
|
|
self.disk_loader_factory = disk_loader_factory
|
|
|
|
def search(self, word: str, disk_loader: DiskCacheLoader = None) -> List[Application]:
|
|
apps = []
|
|
disk_loader = self.disk_loader_factory.new()
|
|
disk_loader.start()
|
|
|
|
for man in self.managers:
|
|
if man.is_enabled():
|
|
apps.extend(man.search(word, disk_loader))
|
|
|
|
disk_loader.stop = True
|
|
disk_loader.join()
|
|
return apps
|
|
|
|
def read_installed(self, disk_loader: DiskCacheLoader = None, keep_workers: bool = False) -> List[Application]:
|
|
installed = []
|
|
|
|
disk_loader = self.disk_loader_factory.new()
|
|
disk_loader.start()
|
|
|
|
for man in self.managers:
|
|
if man.is_enabled():
|
|
installed.extend(man.read_installed(disk_loader=disk_loader, keep_workers=keep_workers))
|
|
|
|
disk_loader.stop = True
|
|
disk_loader.join()
|
|
|
|
return installed
|
|
|
|
def can_downgrade(self):
|
|
return True
|
|
|
|
def downgrade_app(self, app: Application, root_password: str):
|
|
man = self._get_manager_for(app)
|
|
|
|
if man and man.can_downgrade():
|
|
return man.downgrade_app(app, root_password)
|
|
else:
|
|
raise Exception("downgrade is not possible for {}".format(app.__class__.__name__))
|
|
|
|
def clean_cache_for(self, app: Application):
|
|
man = self._get_manager_for(app)
|
|
|
|
if man:
|
|
return man.clean_cache_for(app)
|
|
|
|
def update_and_stream(self, app: Application):
|
|
man = self._get_manager_for(app)
|
|
|
|
if man:
|
|
return man.update_and_stream(app)
|
|
|
|
def uninstall_and_stream(self, app: Application):
|
|
man = self._get_manager_for(app)
|
|
|
|
if man:
|
|
return man.uninstall_and_stream(app)
|
|
|
|
def install_and_stream(self, app: Application):
|
|
man = self._get_manager_for(app)
|
|
|
|
if man:
|
|
return man.install_and_stream(app)
|
|
|
|
def get_info(self, app: Application):
|
|
man = self._get_manager_for(app)
|
|
|
|
if man:
|
|
return man.get_info(app)
|
|
|
|
def get_history(self, app: Application):
|
|
man = self._get_manager_for(app)
|
|
|
|
if man:
|
|
return man.get_history(app)
|
|
|
|
def get_app_type(self):
|
|
return None
|
|
|
|
def is_enabled(self):
|
|
return True
|
|
|
|
def _get_manager_for(self, app: Application) -> ApplicationManager:
|
|
man = self.map[app.__class__]
|
|
return man if man and man.is_enabled() else None
|
|
|
|
def cache_to_disk(self, app: Application, icon_bytes: bytes, only_icon: bool):
|
|
if self.disk_loader_factory.disk_cache and app.supports_disk_cache():
|
|
man = self._get_manager_for(app)
|
|
|
|
if man:
|
|
return man.cache_to_disk(app, icon_bytes=icon_bytes, only_icon=only_icon)
|