mirror of
https://github.com/spalencsar/bearhub.git
synced 2026-07-07 03:34:15 +02:00
143 lines
3.7 KiB
Python
Executable File
143 lines
3.7 KiB
Python
Executable File
from threading import Lock, Thread
|
|
from typing import List
|
|
|
|
import requests
|
|
|
|
from fpakman.core import flatpak
|
|
|
|
__FLATHUB_URL__ = 'https://flathub.org'
|
|
__FLATHUB_API_URL__ = __FLATHUB_URL__ + '/api/v1'
|
|
|
|
|
|
class ImpossibleDowngradeException(Exception):
|
|
pass
|
|
|
|
|
|
class FlatpakManager:
|
|
|
|
def __init__(self):
|
|
self.apps = []
|
|
self.apps_db = {}
|
|
self.http_session = requests.Session()
|
|
self.lock_db_read = Lock()
|
|
|
|
def load_database_async(self):
|
|
Thread(target=self.load_database, daemon=True).start()
|
|
|
|
def load_database(self):
|
|
|
|
self.lock_db_read.acquire()
|
|
|
|
try:
|
|
res = self.http_session.get(__FLATHUB_API_URL__ + '/apps')
|
|
|
|
if res.status_code == 200:
|
|
for app in res.json():
|
|
self.apps_db[app['flatpakAppId']] = app
|
|
finally:
|
|
self.lock_db_read.release()
|
|
|
|
def get_version(self):
|
|
return flatpak.get_version()
|
|
|
|
def read_installed(self) -> List[dict]:
|
|
|
|
installed = flatpak.list_installed()
|
|
|
|
if installed:
|
|
installed.sort(key=lambda p: p['name'].lower())
|
|
|
|
available_updates = flatpak.list_updates_as_str()
|
|
|
|
for app in installed:
|
|
|
|
if not self.apps_db:
|
|
self.load_database()
|
|
|
|
if self.apps_db:
|
|
app_data = self.apps_db.get(app['id'], None)
|
|
else:
|
|
app_data = None
|
|
|
|
if not app_data:
|
|
app['latest_version'] = None
|
|
app['icon'] = None
|
|
else:
|
|
app['latest_version'] = app_data['currentReleaseVersion']
|
|
app['icon'] = app_data['iconMobileUrl']
|
|
|
|
if app['icon'].startswith('/'):
|
|
app['icon'] = __FLATHUB_URL__ + app['icon']
|
|
|
|
app['update'] = app['id'] in available_updates
|
|
|
|
self.apps = installed
|
|
return [*self.apps]
|
|
|
|
def update_apps(self, refs: List[str]) -> List[dict]:
|
|
|
|
if self.apps:
|
|
|
|
for ref in refs:
|
|
package_found = [app for app in self.apps if app['ref'] == ref]
|
|
|
|
if package_found:
|
|
package_found = package_found[0]
|
|
updated = flatpak.update(ref)
|
|
|
|
if updated:
|
|
package_found['update'] = not updated
|
|
|
|
return [*self.apps]
|
|
|
|
return []
|
|
|
|
def _find_by_ref(self, ref: str):
|
|
package_found = [app for app in self.apps if app['ref'] == ref]
|
|
|
|
if package_found:
|
|
return package_found[0]
|
|
|
|
def update_app(self, ref: str):
|
|
|
|
"""
|
|
:param ref:
|
|
:return: the update command stream
|
|
"""
|
|
|
|
if self.apps:
|
|
|
|
app_found = self._find_by_ref(ref)
|
|
|
|
if app_found:
|
|
return flatpak.update_and_stream(ref)
|
|
|
|
def remove_app(self, ref: str):
|
|
"""
|
|
:param ref:
|
|
:return: the uninstall command stream
|
|
"""
|
|
if self.apps:
|
|
|
|
app_found = self._find_by_ref(ref)
|
|
|
|
if app_found:
|
|
return flatpak.uninstall_and_stream(ref)
|
|
|
|
def downgrade_app(self, ref: str, root_password: str):
|
|
|
|
if self.apps:
|
|
|
|
app_found = self._find_by_ref(ref)
|
|
|
|
if app_found:
|
|
commits = flatpak.get_app_commits(app_found['ref'], app_found['origin'])
|
|
|
|
commit_idx = commits.index(app_found['commit'])
|
|
|
|
# downgrade is not possible if the app current commit in the first one:
|
|
if commit_idx == len(commits) - 1:
|
|
raise ImpossibleDowngradeException()
|
|
|
|
return flatpak.downgrade_and_stream(app_found['ref'], commits[commit_idx + 1], root_password)
|