mirror of
https://github.com/spalencsar/bearhub.git
synced 2026-07-08 12:04:16 +02:00
0.3.0
### Features
- Applications search
- Now when you right-click a selected application you can:
- retrieve its information
- retrieve its commit history
- downgrade
- install and uninstall it
- "About" window available when right-clicking the tray icon.
### Improvements
- Performance and memory usage
- Adding tooltips to toolbar buttons
- "Update ?" column renamed to "Upgrade ?"
- Management panel title renamed
- Showing runtime apps when no app is available
- Allowing to specify a custom app translation with the environment variable **FPAKMAN_LOCALE**
- Adding expiration time for cached app data. Default to 1 hour. The environment variable **FPAKMAN_CACHE_EXPIRATION** can change this value.
- Now the application accepts arguments related to environment variables as well. Check 'README.md'.
- Minor GUI improvements
- Notifying only new updates
- New icon
- Progress bar
This commit is contained in:
Vinícius Moreira
GitHub
@@ -1,22 +1,134 @@
|
||||
from datetime import datetime, timedelta
|
||||
from threading import Lock
|
||||
from typing import List
|
||||
|
||||
from fpakman.core.model import FlatpakManager
|
||||
import requests
|
||||
|
||||
from fpakman.core import flatpak
|
||||
|
||||
__FLATHUB_URL__ = 'https://flathub.org'
|
||||
__FLATHUB_API_URL__ = __FLATHUB_URL__ + '/api/v1'
|
||||
|
||||
|
||||
class FlatpakController:
|
||||
class FlatpakManager:
|
||||
|
||||
def __init__(self, model: FlatpakManager):
|
||||
self.model = model
|
||||
def __init__(self, cache_expire: int = 60 * 60):
|
||||
self.cache_apps = {}
|
||||
self.cache_expire = cache_expire
|
||||
self.http_session = requests.Session()
|
||||
self.lock_db_read = Lock()
|
||||
self.lock_read = Lock()
|
||||
|
||||
def refresh(self) -> List[dict]:
|
||||
return self.model.read_installed()
|
||||
def load_full_database(self):
|
||||
|
||||
def update(self, app_ref: str):
|
||||
return self.model.update_app(app_ref)
|
||||
self.lock_db_read.acquire()
|
||||
|
||||
def check_installed(self) -> bool:
|
||||
version = self.model.get_version()
|
||||
return False if version is None else version
|
||||
try:
|
||||
res = self.http_session.get(__FLATHUB_API_URL__ + '/apps', timeout=30)
|
||||
|
||||
def get_version(self) -> str:
|
||||
return self.model.get_version()
|
||||
if res.status_code == 200:
|
||||
for app in res.json():
|
||||
self.cache_apps[app['flatpakAppId']] = app
|
||||
finally:
|
||||
self.lock_db_read.release()
|
||||
|
||||
def _request_app_data(self, app_id: str):
|
||||
|
||||
try:
|
||||
res = self.http_session.get('{}/apps/{}'.format(__FLATHUB_API_URL__, app_id), timeout=30)
|
||||
|
||||
if res.status_code == 200:
|
||||
return res.json()
|
||||
else:
|
||||
print("Could not retrieve app data for id '{}'. Server response: {}".format(app_id, res.status_code))
|
||||
except:
|
||||
print("Could not retrieve app data for id '{}'. Timeout".format(app_id))
|
||||
return None
|
||||
|
||||
def _fill_api_data(self, app: dict):
|
||||
|
||||
api_data = self.cache_apps.get(app['id'])
|
||||
|
||||
if (not app['runtime'] and not api_data) or (api_data and api_data.get('expires_at') and api_data['expires_at'] <= datetime.utcnow()): # if api data is not cached or expired, tries to retrieve it
|
||||
api_data = self._request_app_data(app['id'])
|
||||
|
||||
if api_data:
|
||||
|
||||
if self.cache_expire > 0:
|
||||
api_data['expires_at'] = datetime.utcnow() + timedelta(seconds=self.cache_expire)
|
||||
|
||||
self.cache_apps[app['id']] = api_data
|
||||
|
||||
if not api_data:
|
||||
for attr in ('latest_version', 'icon', 'description'):
|
||||
if attr not in app:
|
||||
app[attr] = None
|
||||
else:
|
||||
app['latest_version'] = api_data.get('currentReleaseVersion')
|
||||
app['icon'] = api_data.get('iconMobileUrl')
|
||||
|
||||
for attr in ('name', 'description'):
|
||||
if not app.get(attr):
|
||||
app[attr] = api_data.get(attr)
|
||||
|
||||
if app['icon'].startswith('/'):
|
||||
app['icon'] = __FLATHUB_URL__ + app['icon']
|
||||
|
||||
def search(self, word: str) -> List[dict]:
|
||||
|
||||
res = []
|
||||
apps_found = flatpak.search(word)
|
||||
|
||||
if apps_found:
|
||||
|
||||
already_read = set()
|
||||
installed_apps = self.read_installed()
|
||||
|
||||
if installed_apps:
|
||||
for app in apps_found:
|
||||
for installed_app in installed_apps:
|
||||
if app['id'] == installed_app['id']:
|
||||
res.append(installed_app)
|
||||
already_read.add(app['id'])
|
||||
|
||||
for app in apps_found:
|
||||
if app['id'] not in already_read:
|
||||
app['update'] = False
|
||||
app['installed'] = False
|
||||
self._fill_api_data(app)
|
||||
res.append(app)
|
||||
|
||||
return res
|
||||
|
||||
def read_installed(self) -> List[dict]:
|
||||
|
||||
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()
|
||||
|
||||
for app in installed:
|
||||
self._fill_api_data(app)
|
||||
app['update'] = app['id'] in available_updates
|
||||
app['installed'] = True
|
||||
|
||||
return installed
|
||||
|
||||
finally:
|
||||
self.lock_read.release()
|
||||
|
||||
def downgrade_app(self, app: dict, 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)
|
||||
|
||||
@@ -1,8 +1,11 @@
|
||||
import re
|
||||
import subprocess
|
||||
from typing import List
|
||||
|
||||
from fpakman.core import system
|
||||
|
||||
BASE_CMD = 'flatpak'
|
||||
|
||||
|
||||
def app_str_to_json(line: str, version: str) -> dict:
|
||||
|
||||
@@ -32,33 +35,53 @@ def app_str_to_json(line: str, version: str) -> dict:
|
||||
else:
|
||||
raise Exception('Unsupported version')
|
||||
|
||||
info = re.findall(r'\w+:\s.+', get_app_info(app['id'], app['branch']))
|
||||
fields_to_get = ['origin', 'arch', 'ref']
|
||||
|
||||
for field in info:
|
||||
field_val = field.split(':')
|
||||
field_name = field_val[0].lower()
|
||||
|
||||
if field_name in fields_to_get:
|
||||
app[field_name] = field_val[1].strip()
|
||||
|
||||
if field_name == 'ref':
|
||||
app['runtime'] = app['ref'].startswith('runtime/')
|
||||
extra_fields = get_app_info_fields(app['id'], app['branch'], ['origin', 'arch', 'ref', 'commit'], check_runtime=True)
|
||||
app.update(extra_fields)
|
||||
|
||||
return app
|
||||
|
||||
|
||||
def get_app_info_fields(app_id: str, branch: str, fields: List[str] = [], check_runtime: bool = False):
|
||||
info = re.findall(r'\w+:\s.+', get_app_info(app_id, branch))
|
||||
data = {}
|
||||
fields_to_retrieve = len(fields) + (1 if check_runtime and 'ref' not in fields else 0)
|
||||
|
||||
for field in info:
|
||||
|
||||
if fields and fields_to_retrieve == 0:
|
||||
break
|
||||
|
||||
field_val = field.split(':')
|
||||
field_name = field_val[0].lower()
|
||||
|
||||
if not fields or field_name in fields or (check_runtime and field_name == 'ref'):
|
||||
data[field_name] = field_val[1].strip()
|
||||
|
||||
if fields:
|
||||
fields_to_retrieve -= 1
|
||||
|
||||
if check_runtime and field_name == 'ref':
|
||||
data['runtime'] = data['ref'].startswith('runtime/')
|
||||
|
||||
return data
|
||||
|
||||
|
||||
def is_installed():
|
||||
version = get_version()
|
||||
return False if version is None else True
|
||||
|
||||
|
||||
def get_version():
|
||||
res = system.run_cmd('flatpak --version')
|
||||
res = system.run_cmd('{} --version'.format(BASE_CMD))
|
||||
return res.split(' ')[1].strip() if res else None
|
||||
|
||||
|
||||
def get_app_info(app_id: str, branch: str):
|
||||
return system.run_cmd('flatpak info {} {}'.format(app_id, branch))
|
||||
return system.run_cmd('{} info {} {}'.format(BASE_CMD, app_id, branch))
|
||||
|
||||
|
||||
def list_installed() -> List[dict]:
|
||||
apps_str = system.run_cmd('flatpak list')
|
||||
apps_str = system.run_cmd('{} list'.format(BASE_CMD))
|
||||
|
||||
if apps_str:
|
||||
version = get_version()
|
||||
@@ -68,18 +91,123 @@ def list_installed() -> List[dict]:
|
||||
return []
|
||||
|
||||
|
||||
def update(app_ref: str) -> bool:
|
||||
return bool(system.run_cmd('flatpak update -y ' + app_ref))
|
||||
|
||||
|
||||
def update_and_stream(app_ref: str):
|
||||
"""
|
||||
Updates the app reference and streams Flatpak output,
|
||||
:param app_ref:
|
||||
:return:
|
||||
"""
|
||||
return system.stream_cmd(['flatpak', 'update', '-y', app_ref])
|
||||
return system.stream_cmd([BASE_CMD, 'update', '-y', app_ref])
|
||||
|
||||
|
||||
def uninstall_and_stream(app_ref: str):
|
||||
"""
|
||||
Removes the app by its reference
|
||||
:param app_ref:
|
||||
:return:
|
||||
"""
|
||||
return system.stream_cmd([BASE_CMD, 'uninstall', app_ref, '-y'])
|
||||
|
||||
|
||||
def list_updates_as_str():
|
||||
return system.run_cmd('flatpak update', ignore_return_code=True)
|
||||
return system.run_cmd('{} update'.format(BASE_CMD), ignore_return_code=True)
|
||||
|
||||
|
||||
def downgrade_and_stream(app_ref: str, commit: str, root_password: str):
|
||||
|
||||
pwdin, downgrade_cmd = None, []
|
||||
|
||||
if root_password is not None:
|
||||
downgrade_cmd.extend(['sudo', '-S'])
|
||||
pwdin = system.stream_cmd(['echo', root_password])
|
||||
|
||||
downgrade_cmd.extend([BASE_CMD, 'update', '--commit={}'.format(commit), app_ref, '-y'])
|
||||
return subprocess.Popen(downgrade_cmd, stdin=pwdin, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL).stdout
|
||||
|
||||
|
||||
def get_app_commits(app_ref: str, origin: str) -> List[str]:
|
||||
log = system.run_cmd('{} remote-info --log {} {}'.format(BASE_CMD, origin, app_ref))
|
||||
return re.findall(r'Commit+:\s(.+)', log)
|
||||
|
||||
|
||||
def get_app_commits_data(app_ref: str, origin: str) -> List[dict]:
|
||||
log = system.run_cmd('{} remote-info --log {} {}'.format(BASE_CMD, origin, app_ref))
|
||||
|
||||
res = re.findall(r'(Commit|Subject|Date):\s(.+)', log)
|
||||
|
||||
commits = []
|
||||
|
||||
commit = {}
|
||||
|
||||
for idx, data in enumerate(res):
|
||||
commit[data[0].strip().lower()] = data[1].strip()
|
||||
|
||||
if (idx + 1) % 3 == 0:
|
||||
commits.append(commit)
|
||||
commit = {}
|
||||
|
||||
return commits
|
||||
|
||||
|
||||
def search(word: str) -> List[dict]:
|
||||
cli_version = get_version()
|
||||
|
||||
res = system.run_cmd('{} search {}'.format(BASE_CMD, word))
|
||||
|
||||
found = []
|
||||
|
||||
split_res = res.split('\n')
|
||||
|
||||
if split_res and split_res[0].lower() != 'no matches found':
|
||||
for info in split_res:
|
||||
if info:
|
||||
info_list = info.split('\t')
|
||||
|
||||
if cli_version >= '1.3.0':
|
||||
version = info_list[3].strip()
|
||||
found.append({
|
||||
'name': info_list[0].strip(),
|
||||
'description': info_list[1].strip(),
|
||||
'id': info_list[2].strip(),
|
||||
'version': version,
|
||||
'latest_version': version,
|
||||
'branch': info_list[4].strip(),
|
||||
'origin': info_list[5].strip(),
|
||||
'runtime': False,
|
||||
'arch': None, # unknown at this moment,
|
||||
'ref': None # unknown at this moment
|
||||
})
|
||||
elif cli_version >= '1.2.0':
|
||||
desc = info_list[0].split('-')
|
||||
version = info_list[2].strip()
|
||||
found.append({
|
||||
'name': desc[0].strip(),
|
||||
'description': desc[1].strip(),
|
||||
'id': info_list[1].strip(),
|
||||
'version': version,
|
||||
'latest_version': version,
|
||||
'branch': info_list[3].strip(),
|
||||
'origin': info_list[4].strip(),
|
||||
'runtime': False,
|
||||
'arch': None, # unknown at this moment,
|
||||
'ref': None # unknown at this moment
|
||||
})
|
||||
else:
|
||||
version = info_list[1].strip()
|
||||
found.append({
|
||||
'name': '',
|
||||
'description': info_list[4].strip(),
|
||||
'id': info_list[0].strip(),
|
||||
'version': version,
|
||||
'latest_version': version,
|
||||
'branch': info_list[2].strip(),
|
||||
'origin': info_list[3].strip(),
|
||||
'runtime': False,
|
||||
'arch': None, # unknown at this moment,
|
||||
'ref': None # unknown at this moment
|
||||
})
|
||||
return found
|
||||
|
||||
|
||||
def install_and_stream(app_id: str, origin: str):
|
||||
return system.stream_cmd([BASE_CMD, 'install', origin, app_id, '-y'])
|
||||
|
||||
@@ -1,108 +0,0 @@
|
||||
from threading import Lock, Thread
|
||||
from typing import List
|
||||
|
||||
import requests
|
||||
from PyQt5.QtCore import QUrl
|
||||
from PyQt5.QtGui import QPixmap, QIcon
|
||||
from PyQt5.QtNetwork import QNetworkAccessManager, QNetworkRequest
|
||||
|
||||
from fpakman.core import flatpak
|
||||
|
||||
__FLATHUB_URL__ = 'https://flathub.org'
|
||||
__FLATHUB_API_URL__ = __FLATHUB_URL__ + '/api/v1'
|
||||
|
||||
|
||||
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 update_app(self, ref: str):
|
||||
|
||||
"""
|
||||
:param ref:
|
||||
:return: the update command stream
|
||||
"""
|
||||
|
||||
if self.apps:
|
||||
|
||||
package_found = [app for app in self.apps if app['ref'] == ref]
|
||||
|
||||
if package_found:
|
||||
return flatpak.update_and_stream(ref)
|
||||
|
||||
return None
|
||||
@@ -1,11 +1,18 @@
|
||||
import os
|
||||
import subprocess
|
||||
from typing import List
|
||||
|
||||
from fpakman.core import resource
|
||||
|
||||
|
||||
def run_cmd(cmd: str, expected_code: int = 0, ignore_return_code: bool = False) -> str:
|
||||
res = subprocess.run(cmd, shell=True, stdout=subprocess.PIPE)
|
||||
res = subprocess.run(cmd, shell=True, stdout=subprocess.PIPE, env={'LANG': 'en'})
|
||||
return res.stdout.decode() if ignore_return_code or res.returncode == expected_code else None
|
||||
|
||||
|
||||
def stream_cmd(cmd: List[str]):
|
||||
return subprocess.Popen(cmd, stdout=subprocess.PIPE).stdout
|
||||
return subprocess.Popen(cmd, stdout=subprocess.PIPE, env={'LANG': 'en'}).stdout
|
||||
|
||||
|
||||
def notify_user(msg: str, icon_path: str = resource.get_path('img/logo.svg')):
|
||||
os.system("notify-send {} '{}'".format("-i {}".format(icon_path) if icon_path else '', msg))
|
||||
|
||||
@@ -2,13 +2,20 @@ import locale
|
||||
|
||||
from fpakman.core import resource
|
||||
import glob
|
||||
import re
|
||||
|
||||
HTML_RE = re.compile(r'<[^>]+>')
|
||||
|
||||
|
||||
def get_locale_keys():
|
||||
def get_locale_keys(key: str = None):
|
||||
|
||||
current_locale = locale.getdefaultlocale()
|
||||
locale_path = None
|
||||
|
||||
if key is None:
|
||||
current_locale = locale.getdefaultlocale()
|
||||
else:
|
||||
current_locale = [key.strip().lower()]
|
||||
|
||||
if current_locale:
|
||||
current_locale = current_locale[0]
|
||||
|
||||
@@ -34,3 +41,7 @@ def get_locale_keys():
|
||||
locale_obj[keyval[0].strip()] = keyval[1].strip()
|
||||
|
||||
return locale_obj
|
||||
|
||||
|
||||
def strip_html(string: str):
|
||||
return HTML_RE.sub('', string)
|
||||
|
||||
Reference in New Issue
Block a user