import glob
import json
import os
import re
import shutil
import sqlite3
import subprocess
import traceback
from datetime import datetime
from pathlib import Path
from typing import Set, Type, List, Tuple, Optional, Iterable, Generator
from colorama import Fore
from bauh import __app_name__
from bauh.api.abstract.context import ApplicationContext
from bauh.api.abstract.controller import SoftwareManager, SearchResult, UpgradeRequirements, UpgradeRequirement, \
TransactionResult, SoftwareAction, SettingsView, SettingsController
from bauh.api.abstract.disk import DiskCacheLoader
from bauh.api.abstract.handler import ProcessWatcher, TaskManager
from bauh.api.abstract.model import SoftwarePackage, PackageHistory, PackageUpdate, PackageSuggestion, \
SuggestionPriority, CustomSoftwareAction
from bauh.api.abstract.view import MessageType, FormComponent, InputOption, SingleSelectComponent, \
SelectViewType, TextInputComponent, PanelComponent, FileChooserComponent, ViewObserver
from bauh.api.paths import DESKTOP_ENTRIES_DIR
from bauh.commons import resource
from bauh.commons.boot import CreateConfigFile
from bauh.commons.html import bold
from bauh.commons.system import SystemProcess, new_subprocess, ProcessHandler, SimpleProcess
from bauh.commons.version_util import normalize_version
from bauh.gems.appimage import query, INSTALLATION_DIR, APPIMAGE_SHARED_DIR, ROOT_DIR, \
APPIMAGE_CONFIG_DIR, UPDATES_IGNORED_FILE, util, get_default_manual_installation_file_dir, DATABASE_APPS_FILE, \
DATABASE_RELEASES_FILE, APPIMAGE_CACHE_DIR, get_icon_path, DOWNLOAD_DIR
from bauh.gems.appimage.config import AppImageConfigManager
from bauh.gems.appimage.model import AppImage
from bauh.gems.appimage.util import replace_desktop_entry_exec_command
from bauh.gems.appimage.worker import DatabaseUpdater, SymlinksVerifier, AppImageSuggestionsDownloader
RE_DESKTOP_ICON = re.compile(r'Icon\s*=\s*.+\n')
RE_ICON_ENDS_WITH = re.compile(r'.+\.(png|svg)$')
RE_APPIMAGE_NAME = re.compile(r'(.+)\.appimage', flags=re.IGNORECASE)
class ManualInstallationFileObserver(ViewObserver):
def __init__(self, name: Optional[TextInputComponent], version: TextInputComponent):
self.name = name
self.version = version
def on_change(self, file_path: str):
if file_path:
name_found = RE_APPIMAGE_NAME.findall(file_path.split('/')[-1])
if name_found:
name_split = name_found[0].split('-')
if self.name:
self.name.set_value(name_split[0].strip())
if len(name_split) > 1:
self.version.set_value(name_split[1].strip())
else:
if self.name:
self.name.set_value(None)
self.version.set_value(None)
class AppImageManager(SoftwareManager, SettingsController):
def __init__(self, context: ApplicationContext):
super(AppImageManager, self).__init__(context=context)
self.i18n = context.i18n
self.api_cache = context.cache_factory.new()
context.disk_loader_factory.map(AppImage, self.api_cache)
self.enabled = True
self.http_client = context.http_client
self.logger = context.logger
self.file_downloader = context.file_downloader
self.configman = AppImageConfigManager()
self._custom_actions: Optional[Iterable[CustomSoftwareAction]] = None
self._action_self_install: Optional[CustomSoftwareAction] = None
self._app_repository: Optional[str] = None
self._search_unfilled_attrs: Optional[Tuple[str, ...]] = None
self._suggestions_downloader: Optional[AppImageSuggestionsDownloader] = None
def install_file(self, root_password: Optional[str], watcher: ProcessWatcher) -> bool:
max_width = 350
file_chooser = FileChooserComponent(label=self.i18n['file'].capitalize(),
allowed_extensions={'AppImage', '*'},
search_path=get_default_manual_installation_file_dir(),
max_width=max_width)
input_name = TextInputComponent(label=self.i18n['name'].capitalize(), max_width=max_width)
input_version = TextInputComponent(label=self.i18n['version'].capitalize(), max_width=max_width)
file_chooser.observers.append(ManualInstallationFileObserver(input_name, input_version))
input_description = TextInputComponent(label=self.i18n['description'].capitalize(), max_width=max_width)
cat_ops = [InputOption(label=self.i18n['category.none'].capitalize(), value=0)]
cat_ops.extend([InputOption(label=self.i18n.get(f'category.{c.lower()}', c.lower()).capitalize(), value=c) for c in self.context.default_categories])
inp_cat = SingleSelectComponent(label=self.i18n['category'], type_=SelectViewType.COMBO, options=cat_ops,
default_option=cat_ops[0], max_width=max_width)
form = FormComponent(label='', components=[file_chooser, input_name, input_version, input_description, inp_cat],
spaces=False)
while True:
if watcher.request_confirmation(title=self.i18n['appimage.custom_action.install_file.details'], body=None,
components=[form],
confirmation_label=self.i18n['proceed'].capitalize(),
deny_label=self.i18n['cancel'].capitalize(),
min_height=100, max_width=max_width + 150):
if not file_chooser.file_path or not os.path.isfile(file_chooser.file_path) or not file_chooser.file_path.lower().strip().endswith('.appimage'):
watcher.request_confirmation(title=self.i18n['error'].capitalize(),
body=self.i18n['appimage.custom_action.install_file.invalid_file'],
deny_button=False)
elif not input_name.get_value() or not input_name.get_value().strip():
watcher.request_confirmation(title=self.i18n['error'].capitalize(),
body=self.i18n['appimage.custom_action.install_file.invalid_name'],
deny_button=False)
else:
break
else:
return False
appim = AppImage(i18n=self.i18n, imported=True, manual_update=True)
appim.name = input_name.get_value().strip()
appim.local_file_path = file_chooser.file_path
appim.version = input_version.get_value()
appim.latest_version = input_version.get_value()
appim.description = input_description.get_value()
appim.categories = ['Imported']
if inp_cat.get_selected() != cat_ops[0].value:
appim.categories.append(inp_cat.get_selected())
res = self.install(root_password=root_password, pkg=appim, disk_loader=None, watcher=watcher).success
if res:
appim.installed = True
self.cache_to_disk(appim, None, False)
return res
def update_file(self, pkg: AppImage, root_password: Optional[str], watcher: ProcessWatcher):
max_width = 350
file_chooser = FileChooserComponent(label=self.i18n['file'].capitalize(),
allowed_extensions={'AppImage', '*'},
search_path=get_default_manual_installation_file_dir(),
max_width=max_width)
input_version = TextInputComponent(label=self.i18n['version'].capitalize(), max_width=max_width)
file_chooser.observers.append(ManualInstallationFileObserver(None, input_version))
while True:
if watcher.request_confirmation(title=self.i18n['appimage.custom_action.manual_update.details'], body=None,
components=[FormComponent(label='', components=[file_chooser, input_version], spaces=False)],
confirmation_label=self.i18n['proceed'].capitalize(),
deny_label=self.i18n['cancel'].capitalize(),
min_height=100, max_width=max_width + 150):
if not file_chooser.file_path or not os.path.isfile(file_chooser.file_path) or not file_chooser.file_path.lower().strip().endswith('.appimage'):
watcher.request_confirmation(title=self.i18n['error'].capitalize(),
body=self.i18n['appimage.custom_action.install_file.invalid_file'],
deny_button=False)
else:
break
else:
return False
pkg.local_file_path = file_chooser.file_path
pkg.version = input_version.get_value()
pkg.manual_update = True
reqs = UpgradeRequirements(to_install=None, to_remove=None, to_upgrade=[UpgradeRequirement(pkg=pkg)], cannot_upgrade=None)
return self.upgrade(reqs, root_password=root_password, watcher=watcher)
def _get_db_connection(self, db_path: str) -> sqlite3.Connection:
if os.path.exists(db_path):
try:
return sqlite3.connect(db_path)
except Exception:
self.logger.error(f"Could not connect to database file '{db_path}'")
traceback.print_exc()
else:
self.logger.warning(f"Could not get a connection for database '{db_path}'")
def _gen_app_key(self, app: AppImage):
return f"{app.name.lower()}{app.repository.lower() if app.repository else ''}"
def search(self, words: str, disk_loader: DiskCacheLoader, limit: int = -1, is_url: bool = False) -> SearchResult:
if is_url:
return SearchResult.empty()
apps_conn = self._get_db_connection(DATABASE_APPS_FILE)
if not apps_conn:
return SearchResult.empty()
not_installed, found_map = [], {}
try:
cursor = apps_conn.cursor()
cursor.execute(query.SEARCH_APPS_BY_NAME_OR_DESCRIPTION.format(words, words))
idx = 0
for r in cursor.fetchall():
app = AppImage(*r, i18n=self.i18n)
not_installed.append(app)
found_map[self._gen_app_key(app)] = {'app': app, 'idx': idx}
idx += 1
except Exception:
self.logger.error("An exception happened while querying the 'apps' database")
traceback.print_exc()
try:
installed = self.read_installed(connection=apps_conn, disk_loader=disk_loader, limit=limit, only_apps=False, pkg_types=None, internet_available=True).installed
except Exception:
installed = None
installed_found = []
if installed:
lower_words = words.lower()
for appim in installed:
found = False
if not_installed and found_map:
key = self._gen_app_key(appim)
new_found = found_map.get(key)
if new_found:
if not appim.imported:
for attr in self.search_unfilled_attrs:
if getattr(appim, attr) is None:
setattr(appim, attr, getattr(new_found['app'], attr))
del not_installed[new_found['idx']]
installed_found.append(appim)
found = True
if not found and (lower_words in appim.name.lower() or (appim.description and lower_words in appim.description.lower())):
installed_found.append(appim)
try:
apps_conn.close()
except Exception:
self.logger.error(f"An exception happened when trying to close the connection to database file '{DATABASE_APPS_FILE}'")
traceback.print_exc()
return SearchResult(new=not_installed, installed=installed_found, total=len(not_installed) + len(installed_found))
def read_installed(self, disk_loader: Optional[DiskCacheLoader], limit: int = -1, only_apps: bool = False,
pkg_types: Optional[Set[Type[SoftwarePackage]]] = None, internet_available: bool = None, connection: sqlite3.Connection = None) -> SearchResult:
installed_apps = []
res = SearchResult(installed_apps, [], 0)
if os.path.exists(INSTALLATION_DIR):
installed = glob.glob(f'{INSTALLATION_DIR}/*/data.json')
if installed:
names = set()
for path in installed:
if path:
with open(path) as f:
app = AppImage(installed=True, i18n=self.i18n, **json.loads(f.read()))
app.icon_url = app.icon_path
installed_apps.append(app)
names.add(f"'{app.name.lower()}'")
if installed_apps:
apps_con = self._get_db_connection(DATABASE_APPS_FILE) if not connection else connection
if apps_con:
try:
cursor = apps_con.cursor()
cursor.execute(query.FIND_APPS_BY_NAME.format(','.join(names)))
for tup in cursor.fetchall():
for app in installed_apps:
if app.name.lower() == tup[0].lower() and (not app.repository or app.repository.lower() == tup[1].lower()):
continuous_version = app.version == 'continuous'
continuous_update = tup[2] == 'continuous'
if tup[3]:
if continuous_version and not continuous_update:
app.update = True
elif continuous_update and not continuous_version:
app.update = False
else:
if tup[2]:
try:
latest_version = normalize_version(tup[2])
installed_version = normalize_version(app.version)
app.update = latest_version > installed_version
except Exception:
app.update = False
traceback.print_exc()
if app.update:
app.latest_version = tup[2]
app.url_download_latest_version = tup[3]
break
except Exception:
self.logger.error(f"An exception happened while querying the database file '{DATABASE_APPS_FILE}'")
traceback.print_exc()
finally:
if not connection: # the connection can only be closed if it was opened within this method
apps_con.close()
ignored_updates = self._read_ignored_updates()
if ignored_updates:
for app in installed_apps:
if app.supports_ignored_updates() and app.name in ignored_updates:
app.updates_ignored = True
res.total = len(res.installed)
return res
def downgrade(self, pkg: AppImage, root_password: Optional[str], watcher: ProcessWatcher) -> bool:
versions = self.get_history(pkg)
if len(versions.history) == 1:
watcher.show_message(title=self.i18n['appimage.downgrade.impossible.title'],
body=self.i18n['appimage.downgrade.impossible.body'].format(bold(pkg.name)),
type_=MessageType.ERROR)
return False
elif versions.pkg_status_idx == -1:
watcher.show_message(title=self.i18n['appimage.downgrade.impossible.title'],
body=self.i18n['appimage.downgrade.unknown_version.body'].format(bold(pkg.name)),
type_=MessageType.ERROR)
return False
elif versions.pkg_status_idx == len(versions.history) - 1:
watcher.show_message(title=self.i18n['appimage.downgrade.impossible.title'],
body=self.i18n['appimage.downgrade.first_version'].format(bold(pkg.name)),
type_=MessageType.ERROR)
return False
else:
old_release = versions.history[versions.pkg_status_idx + 1]
pkg.version = old_release['0_version']
pkg.latest_version = pkg.version
pkg.url_download = old_release['2_url_download']
download_data = self._download(pkg=pkg, watcher=watcher)
if not download_data:
return False
if self.uninstall(pkg, root_password, watcher).success:
if self._install(pkg=pkg, watcher=watcher, pre_downloaded_file=download_data).success:
self.cache_to_disk(pkg, None, False)
return True
else:
watcher.show_message(title=self.i18n['error'],
body=self.i18n['appimage.downgrade.install_version'].format(bold(pkg.version), bold(pkg.url_download)),
type_=MessageType.ERROR)
else:
watcher.show_message(title=self.i18n['error'],
body=self.i18n['appimage.error.uninstall_current_version'].format(bold(pkg.name)),
type_=MessageType.ERROR)
return False
def upgrade(self, requirements: UpgradeRequirements, root_password: Optional[str], watcher: ProcessWatcher) -> bool:
not_upgraded = []
for req in requirements.to_upgrade:
watcher.change_status(f"{self.i18n['manage_window.status.upgrading']} {req.pkg.name} ({req.pkg.version})...")
download_data = None
# always changing the 'imported' field based on the 'manual_update' flag that means
# "the user is manually installing/updating the AppImage"
if not req.pkg.manual_update:
req.pkg.imported = False
if req.pkg.categories and "Imported" in req.pkg.categories:
# removing the imported category in case the file is not considered imported anymore
req.pkg.categories.remove("Imported")
# manual file updates do not required download it
if not req.pkg.imported:
download_data = self._download(req.pkg, watcher)
if not download_data:
not_upgraded.append(req.pkg)
watcher.change_substatus('')
continue
if not self.uninstall(req.pkg, root_password, watcher).success:
not_upgraded.append(req.pkg)
watcher.change_substatus('')
continue
if not self._install(pkg=req.pkg, watcher=watcher, pre_downloaded_file=download_data).success:
not_upgraded.append(req.pkg)
watcher.change_substatus('')
continue
self.cache_to_disk(req.pkg, None, False)
all_failed = len(not_upgraded) == len(requirements.to_upgrade)
if not_upgraded:
pkgs_str = ''.join((f'{app.name}' for app in not_upgraded))
watcher.show_message(title=self.i18n['error' if all_failed else 'warning'].capitalize(),
body=self.i18n['appimage.upgrade.failed'].format(apps=f''),
type_=MessageType.ERROR if all_failed else MessageType.WARNING)
watcher.change_substatus('')
return not all_failed
def uninstall(self, pkg: AppImage, root_password: Optional[str], watcher: ProcessWatcher, disk_loader: DiskCacheLoader = None) -> TransactionResult:
if os.path.exists(pkg.get_disk_cache_path()):
handler = ProcessHandler(watcher)
if not handler.handle(SystemProcess(new_subprocess(['rm', '-rf', pkg.get_disk_cache_path()]))):
watcher.show_message(title=self.i18n['error'], body=self.i18n['appimage.uninstall.error.remove_folder'].format(bold(pkg.get_disk_cache_path())))
return TransactionResult.fail()
de_path = self._gen_desktop_entry_path(pkg)
if os.path.exists(de_path):
os.remove(de_path)
self.revert_ignored_update(pkg)
if pkg.symlink and os.path.islink(pkg.symlink):
self.logger.info(f"Removing symlink '{pkg.symlink}'")
try:
os.remove(pkg.symlink)
self.logger.info(f"symlink '{pkg.symlink}' successfully removed")
except Exception:
msg = f"could not remove symlink '{pkg.symlink}'"
self.logger.error(msg)
if watcher:
watcher.print(f"[error] {msg}")
self._add_self_latest_version(pkg) # only for self installation
return TransactionResult(success=True, installed=None, removed=[pkg])
def _add_self_latest_version(self, app: AppImage):
if app.name == self.context.app_name and app.repository == self.app_repository and not app.url_download_latest_version:
history = self.get_history(app)
if not history or not history.history:
self.logger.warning(f"Could not retrieve '{app.name}' versions. "
f"It will not be possible to determine the current latest version")
else:
app.version = history.history[0]['0_version']
app.latest_version = app.version
app.url_download = history.history[0]['2_url_download']
app.url_download_latest_version = app.url_download
def get_managed_types(self) -> Set[Type[SoftwarePackage]]:
return {AppImage}
def clean_cache_for(self, pkg: AppImage):
pass
def get_info(self, pkg: AppImage) -> dict:
data = pkg.get_data_to_cache()
if not pkg.installed:
for key in ('install_dir', 'symlink', 'icon_path'):
if key in data:
del data[key]
if data.get('url_download'):
size = self.http_client.get_content_length(data['url_download'])
if size:
data['size'] = size
categories = data.get('categories')
if categories:
data['categories'] = [self.i18n.get(f'category.{c.lower()}', self.i18n.get(c, c)).capitalize() for c in data['categories']]
if data.get('symlink') and not os.path.islink(data['symlink']):
del data['symlink']
if not data.get("license"):
data["license"] = self.i18n["unknown"].lower()
return data
def _sort_release(self, rel: tuple):
return rel[0]
def get_history(self, pkg: AppImage) -> PackageHistory:
history = []
res = PackageHistory(pkg, history, -1)
app_con = self._get_db_connection(DATABASE_APPS_FILE)
if not app_con:
return res
try:
cursor = app_con.cursor()
cursor.execute(query.FIND_APP_ID_BY_REPO_AND_NAME.format(pkg.repository.lower() if pkg.repository else '', pkg.name.lower()))
app_tuple = cursor.fetchone()
if not app_tuple:
self.logger.warning(f"Could not retrieve {pkg} from the database '{DATABASE_APPS_FILE}'")
return res
except Exception:
self.logger.error(f"An exception happened while querying the database file '{DATABASE_APPS_FILE}'")
traceback.print_exc()
app_con.close()
return res
app_con.close()
releases_con = self._get_db_connection(DATABASE_RELEASES_FILE)
if not releases_con:
return res
try:
cursor = releases_con.cursor()
releases = cursor.execute(query.FIND_RELEASES_BY_APP_ID.format(app_tuple[0]))
if releases:
treated_releases = [(normalize_version(r[0]), r[0], *r[1:]) for r in releases]
treated_releases.sort(key=self._sort_release, reverse=True)
for idx, tup in enumerate(treated_releases):
ver = tup[1]
date_ = datetime.strptime(tup[3], '%Y-%m-%dT%H:%M:%SZ') if tup[3] else ''
history.append({'0_version': ver, '1_published_at': date_, '2_url_download': tup[2]})
if res.pkg_status_idx == -1 and pkg.version == ver:
res.pkg_status_idx = idx
return res
except Exception:
self.logger.error(f"An exception happened while querying the database file '{DATABASE_RELEASES_FILE}'")
traceback.print_exc()
finally:
releases_con.close()
def _find_desktop_file(self, folder: str) -> Optional[str]:
for r, d, files in os.walk(folder):
for f in files:
if f.endswith('.desktop'):
return f
def _find_icon_file(self, folder: str) -> Optional[str]:
for f in glob.glob(folder + ('/**' if not folder.endswith('/') else '**'), recursive=True):
if RE_ICON_ENDS_WITH.match(f):
return f
def _download(self, pkg: AppImage, watcher: ProcessWatcher) -> Optional[Tuple[str, str]]:
appimage_url = pkg.url_download_latest_version if pkg.update else pkg.url_download
if not appimage_url:
watcher.show_message(title=self.i18n['error'],
body=self.i18n['appimage.download.no_url'].format(app=bold(pkg.name)),
type_=MessageType.ERROR)
return
file_name = appimage_url.split('/')[-1]
pkg.version = pkg.latest_version
pkg.url_download = appimage_url
try:
Path(DOWNLOAD_DIR).mkdir(exist_ok=True, parents=True)
except OSError:
watcher.show_message(title=self.i18n['error'],
body=self.i18n['error.mkdir'].format(dir=bold(DOWNLOAD_DIR)),
type_=MessageType.ERROR)
return
file_path = f'{DOWNLOAD_DIR}/{file_name}'
downloaded = self.file_downloader.download(file_url=pkg.url_download, watcher=watcher,
output_path=file_path, cwd=str(Path.home()))
if not downloaded:
watcher.show_message(title=self.i18n['error'],
body=self.i18n['appimage.download.error'].format(bold(pkg.url_download)),
type_=MessageType.ERROR)
return
return file_name, file_path
def install(self, pkg: AppImage, root_password: Optional[str], disk_loader: Optional[DiskCacheLoader], watcher: ProcessWatcher) -> TransactionResult:
return self._install(pkg=pkg, watcher=watcher)
def _install(self, pkg: AppImage, watcher: ProcessWatcher, pre_downloaded_file: Optional[Tuple[str, str]] = None) \
-> TransactionResult:
handler = ProcessHandler(watcher)
out_dir = f'{INSTALLATION_DIR}/{pkg.get_clean_name()}'
counter = 0
while True:
if os.path.exists(out_dir):
self.logger.info(f"Installation dir '{out_dir}' already exists. Generating a different one")
out_dir += f'-{counter}'
counter += 1
else:
break
Path(out_dir).mkdir(parents=True, exist_ok=True)
pkg.install_dir = out_dir
# when the package is being imported/upgraded there is no need to download it
if pkg.manual_update:
downloaded, file_name = True, pkg.local_file_path.split('/')[-1]
install_file_path = out_dir + '/' + file_name
try:
moved, output = handler.handle_simple(SimpleProcess(['mv', pkg.local_file_path, install_file_path]))
except Exception:
output = ''
self.logger.error(f"Could not rename file '{pkg.local_file_path}' as '{install_file_path}'")
moved = False
if not moved:
watcher.show_message(title=self.i18n['error'].capitalize(),
body=self.i18n['appimage.install.imported.rename_error'].format(bold(pkg.local_file_path.split('/')[-1]),
bold(output)),
type_=MessageType.ERROR)
return TransactionResult.fail()
else:
download_data = pre_downloaded_file if pre_downloaded_file else self._download(pkg, watcher)
if not download_data:
return TransactionResult.fail()
file_name, download_path = download_data[0], download_data[1]
install_file_path = f'{out_dir}/{file_name}'
try:
shutil.move(download_path, install_file_path)
except OSError:
watcher.show_message(title=self.i18n['error'],
body=self.i18n['error.mvfile'].formmat(src=bold(download_path),
dest=bold(install_file_path)))
return TransactionResult.fail()
watcher.change_substatus(self.i18n['appimage.install.permission'].format(bold(file_name)))
permission_given = handler.handle(SystemProcess(new_subprocess(['chmod', 'a+x', install_file_path])))
if permission_given:
watcher.change_substatus(self.i18n['appimage.install.extract'].format(bold(file_name)))
try:
res, output = handler.handle_simple(
SimpleProcess([install_file_path, '--appimage-extract'], cwd=out_dir))
if 'Error: Failed to register AppImage in AppImageLauncherFS' in output:
watcher.show_message(title=self.i18n['error'],
body=self.i18n['appimage.install.appimagelauncher.error'].format(
appimgl=bold('AppImageLauncher'), app=bold(pkg.name)),
type_=MessageType.ERROR)
handler.handle(SystemProcess(new_subprocess(['rm', '-rf', out_dir])))
return TransactionResult.fail()
except Exception:
watcher.show_message(title=self.i18n['error'],
body=traceback.format_exc(),
type_=MessageType.ERROR)
traceback.print_exc()
handler.handle(SystemProcess(new_subprocess(['rm', '-rf', out_dir])))
return TransactionResult.fail()
watcher.change_substatus(self.i18n['appimage.install.desktop_entry'])
extracted_folder = f'{out_dir}/squashfs-root'
if os.path.exists(extracted_folder):
desktop_entry = self._find_desktop_file(extracted_folder)
with open(f'{extracted_folder}/{desktop_entry}') as f:
de_content = f.read()
if de_content:
de_content = replace_desktop_entry_exec_command(desktop_entry=de_content,
appname=pkg.name,
file_path=install_file_path)
extracted_icon = self._find_icon_file(extracted_folder)
if extracted_icon:
icon_path = out_dir + '/logo.' + extracted_icon.split('/')[-1].split('.')[-1]
shutil.copy(extracted_icon, icon_path)
if de_content:
de_content = RE_DESKTOP_ICON.sub(f'Icon={icon_path}\n', de_content)
pkg.icon_path = icon_path
if not de_content:
de_content = pkg.to_desktop_entry()
Path(DESKTOP_ENTRIES_DIR).mkdir(parents=True, exist_ok=True)
with open(self._gen_desktop_entry_path(pkg), 'w+') as f:
f.write(de_content)
try:
shutil.rmtree(extracted_folder)
except Exception:
traceback.print_exc()
SymlinksVerifier.create_symlink(app=pkg, file_path=install_file_path, logger=self.logger,
watcher=watcher)
return TransactionResult(success=True, installed=[pkg], removed=[])
else:
watcher.show_message(title=self.i18n['error'],
body=f'Could extract content from {bold(file_name)}',
type_=MessageType.ERROR)
handler.handle(SystemProcess(new_subprocess(['rm', '-rf', out_dir])))
return TransactionResult.fail()
def _gen_desktop_entry_path(self, app: AppImage) -> str:
return f'{DESKTOP_ENTRIES_DIR}/{__app_name__}_appimage_{app.get_clean_name()}.desktop'
def is_enabled(self) -> bool:
return self.enabled
def set_enabled(self, enabled: bool):
self.enabled = enabled
def _is_sqlite3_available(self) -> bool:
return bool(shutil.which('sqlite3'))
def can_work(self) -> Tuple[bool, Optional[str]]:
if not self.context.is_system_x86_64():
return False, self.i18n['message.requires_architecture'].format(arch=bold('x86_64'))
if not self._is_sqlite3_available():
return False, self.i18n['missing_dep'].format(dep=bold('sqlite3'))
return True, None
def requires_root(self, action: SoftwareAction, pkg: AppImage) -> bool:
return False
def prepare(self, task_manager: TaskManager, root_password: Optional[str], internet_available: bool):
create_config = CreateConfigFile(taskman=task_manager, configman=self.configman, i18n=self.i18n,
task_icon_path=get_icon_path(), logger=self.logger)
create_config.start()
symlink_check = SymlinksVerifier(taskman=task_manager, i18n=self.i18n, logger=self.logger)
symlink_check.start()
if internet_available:
DatabaseUpdater(taskman=task_manager, i18n=self.context.i18n,
create_config=create_config, http_client=self.context.http_client,
logger=self.context.logger).start()
if not self.suggestions_downloader.is_custom_local_file_mapped():
self.suggestions_downloader.taskman = task_manager
self.suggestions_downloader.create_config = create_config
self.suggestions_downloader.register_task()
self.suggestions_downloader.start()
def list_updates(self, internet_available: bool) -> List[PackageUpdate]:
res = self.read_installed(disk_loader=None, internet_available=internet_available)
updates = []
if res.installed:
for app in res.installed:
if app.update and not app.is_update_ignored():
updates.append(PackageUpdate(pkg_id=app.name, pkg_type='AppImage', version=app.latest_version, name=app.name))
return updates
def list_warnings(self, internet_available: bool) -> List[str]:
dbfiles = glob.glob(f'{APPIMAGE_CACHE_DIR}/*.db')
if not dbfiles or len({f for f in (DATABASE_APPS_FILE, DATABASE_RELEASES_FILE) if f in dbfiles}) != 2:
return [self.i18n['appimage.warning.missing_db_files'].format(appimage=bold('AppImage'))]
def list_suggestions(self, limit: int, filter_installed: bool) -> Optional[List[PackageSuggestion]]:
if limit == 0:
return
connection = self._get_db_connection(DATABASE_APPS_FILE)
if connection:
self.suggestions_downloader.taskman = TaskManager()
suggestions = tuple(self.suggestions_downloader.read())
if not suggestions:
self.logger.warning("Could not read AppImage suggestions")
return
else:
self.logger.info("Mapping AppImage suggestions")
try:
if filter_installed:
installed = {i.name.lower() for i in self.read_installed(disk_loader=None,
connection=connection).installed}
else:
installed = None
sugs_map = {}
for s in suggestions:
lsplit = s.split('=')
name = lsplit[1].strip()
if limit < 0 or len(sugs_map) < limit:
if not installed or not name.lower() in installed:
sugs_map[name] = SuggestionPriority(int(lsplit[0]))
else:
break
cursor = connection.cursor()
cursor.execute(query.FIND_APPS_BY_NAME_FULL.format(','.join([f"'{s}'" for s in sugs_map.keys()])))
res = []
for t in cursor.fetchall():
app = AppImage(*t, i18n=self.i18n)
res.append(PackageSuggestion(app, sugs_map[app.name.lower()]))
self.logger.info(f"Mapped {len(res)} AppImage suggestions")
return res
except Exception:
traceback.print_exc()
finally:
connection.close()
def is_default_enabled(self) -> bool:
return True
def launch(self, pkg: AppImage):
installation_dir = pkg.get_disk_cache_path()
if os.path.exists(installation_dir):
appimag_path = util.find_appimage_file(installation_dir)
if appimag_path:
subprocess.Popen(args=[appimag_path], shell=True, env={**os.environ},
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, stdin=subprocess.DEVNULL)
else:
self.logger.error(f"Could not find the AppImage file of '{pkg.name}' in '{installation_dir}'")
def cache_to_disk(self, pkg: SoftwarePackage, icon_bytes: Optional[bytes], only_icon: bool):
self.serialize_to_disk(pkg, icon_bytes, only_icon)
def get_screenshots(self, pkg: AppImage) -> Generator[str, None, None]:
if pkg.url_screenshot:
yield pkg.url_screenshot
def clear_data(self, logs: bool = True):
for f in glob.glob(f'{APPIMAGE_SHARED_DIR}/*.db'):
try:
if logs:
print(f'[{__app_name__}][appimage] Deleting {f}')
os.remove(f)
if logs:
print(f'{Fore.YELLOW}[{__app_name__}][appimage] {f} deleted{Fore.RESET}')
except Exception:
if logs:
print(f'{Fore.RED}[{__app_name__}][appimage] An exception has happened when deleting {f}{Fore.RESET}')
traceback.print_exc()
def get_settings(self) -> Optional[Generator[SettingsView, None, None]]:
config_ = self.configman.get_config()
comps = [
TextInputComponent(label=self.i18n['appimage.config.database.expiration'],
value=int(config_['database']['expiration']) if isinstance(
config_['database']['expiration'], int) else '',
tooltip=self.i18n['appimage.config.database.expiration.tip'],
only_int=True,
id_='appim_db_exp'),
TextInputComponent(label=self.i18n['appimage.config.suggestions.expiration'],
value=int(config_['suggestions']['expiration']) if isinstance(
config_['suggestions']['expiration'], int) else '',
tooltip=self.i18n['appimage.config.suggestions.expiration.tip'],
only_int=True,
id_='appim_sugs_exp')
]
yield SettingsView(self, PanelComponent([FormComponent(components=comps)]))
def save_settings(self, component: PanelComponent) -> Tuple[bool, Optional[List[str]]]:
config_ = self.configman.get_config()
form = component.get_component_by_idx(0, FormComponent)
config_['database']['expiration'] = form.get_component('appim_db_exp', TextInputComponent).get_int_value()
config_['suggestions']['expiration'] = form.get_component('appim_sugs_exp', TextInputComponent).get_int_value()
try:
self.configman.save_config(config_)
return True, None
except Exception:
return False, [traceback.format_exc()]
def gen_custom_actions(self) -> Generator[CustomSoftwareAction, None, None]:
if self._custom_actions is None:
self._custom_actions = (CustomSoftwareAction(i18n_label_key='appimage.custom_action.install_file',
i18n_status_key='appimage.custom_action.install_file.status',
i18n_description_key='appimage.custom_action.install_file.desc',
manager=self,
manager_method='install_file',
icon_path=resource.get_path('img/appimage.svg', ROOT_DIR),
requires_root=False,
requires_confirmation=False),
CustomSoftwareAction(i18n_label_key='appimage.custom_action.update_db',
i18n_status_key='appimage.custom_action.update_db.status',
i18n_description_key='appimage.custom_action.update_db.desc',
manager=self,
manager_method='update_database',
icon_path=resource.get_path('img/appimage.svg', ROOT_DIR),
requires_root=False,
requires_internet=True))
if self._get_self_appimage_running() and not self._is_self_installed():
yield self.action_self_install
yield from self._custom_actions
def get_upgrade_requirements(self, pkgs: List[AppImage], root_password: Optional[str], watcher: ProcessWatcher) -> UpgradeRequirements:
to_update = []
for pkg in pkgs:
requirement = UpgradeRequirement(pkg)
installed_size = self.http_client.get_content_length_in_bytes(pkg.url_download)
upgrade_size = self.http_client.get_content_length_in_bytes(pkg.url_download_latest_version)
requirement.required_size = upgrade_size
if upgrade_size and installed_size:
requirement.extra_size = upgrade_size - installed_size
to_update.append(requirement)
return UpgradeRequirements([], [], to_update, [])
def _read_ignored_updates(self) -> Set[str]:
ignored = set()
if os.path.exists(UPDATES_IGNORED_FILE):
with open(UPDATES_IGNORED_FILE) as f:
ignored_txt = f.read()
for l in ignored_txt.split('\n'):
if l:
line_clean = l.strip()
if line_clean:
ignored.add(line_clean)
return ignored
def ignore_update(self, pkg: AppImage):
current_ignored = self._read_ignored_updates()
if pkg.name not in current_ignored:
current_ignored.add(pkg.name)
self._write_ignored_updates(current_ignored)
pkg.updates_ignored = True
def _write_ignored_updates(self, names: Set[str]):
Path(APPIMAGE_CONFIG_DIR).mkdir(parents=True, exist_ok=True)
ignored_list = [*names]
ignored_list.sort()
with open(UPDATES_IGNORED_FILE, 'w+') as f:
if ignored_list:
for ignored in ignored_list:
f.write(f'{ignored}\n')
else:
f.write('')
def revert_ignored_update(self, pkg: AppImage):
current_ignored = self._read_ignored_updates()
if current_ignored and pkg.name in current_ignored:
current_ignored.remove(pkg.name)
self._write_ignored_updates(current_ignored)
pkg.updates_ignored = False
def update_database(self, root_password: Optional[str], watcher: ProcessWatcher) -> bool:
db_updater = DatabaseUpdater(i18n=self.i18n, http_client=self.context.http_client,
logger=self.context.logger, watcher=watcher, taskman=TaskManager())
res = db_updater.download_databases()
return res
@property
def search_unfilled_attrs(self) -> Tuple[str, ...]:
if self._search_unfilled_attrs is None:
self._search_unfilled_attrs = ('icon_url', 'url_download_latest_version', 'author', 'license', 'repository',
'source', 'url_screenshot')
return self._search_unfilled_attrs
def self_install(self, root_password: Optional[str], watcher: ProcessWatcher) -> bool:
file_path = self._get_self_appimage_running()
if not file_path:
return False
if self._is_self_installed():
return False
app = AppImage(name=self.context.app_name, version=self.context.app_version,
categories=['system'], author=self.context.app_name, repository=self.app_repository,
license='zlib/libpng')
res = self._install(pkg=app, watcher=watcher,
pre_downloaded_file=(os.path.basename(file_path), file_path))
if res.success:
app.installed = True
de_path = self._gen_desktop_entry_path(app)
if de_path and os.path.exists(de_path):
with open(de_path) as f:
bauh_entry = f.read()
if bauh_entry:
comments = re.compile(r'Comment(\[\w+])?\s*=\s*(.+)').findall(bauh_entry)
if comments:
locale = f'{self.i18n.current_key}' if self.i18n.current_key != self.i18n.default_key else None
for key, desc in comments:
if desc:
if not key:
app.description = desc # default description
if not locale:
break
elif key == locale:
app.description = desc # localized description
break
else:
self.context.logger.warning(f"Could not find the 'Comment' fields from {self.context.app_name}'s desktop entry")
else:
self.context.logger.warning(f"{self.context.app_name} desktop entry is empty. Is is not possible to determine the 'description' field")
else:
self.context.logger.warning(f"{self.context.app_name} desktop file not found ({de_path}). It is not possible to determine the 'description' field")
self.cache_to_disk(app, None, False)
return res.success
def _is_self_installed(self) -> bool:
return os.path.exists(f'{INSTALLATION_DIR}/{self.context.app_name}/data.json')
def _get_self_appimage_running(self) -> Optional[str]:
file = os.getenv('APPIMAGE')
if not file:
return
app_exec = os.getenv('APPRUN_STARTUP_EXEC_ARGS')
if not app_exec:
return
if os.path.basename(app_exec).split(' ')[0] != self.context.app_name:
return
if not os.path.exists(file):
return
return file
@property
def action_self_install(self) -> CustomSoftwareAction:
if self._action_self_install is None:
self._action_self_install = CustomSoftwareAction(i18n_label_key='appimage.custom_action.self_install',
i18n_status_key='appimage.custom_action.self_install.status',
i18n_description_key='appimage.custom_action.self_install.desc',
manager=self,
manager_method='self_install',
icon_path=resource.get_path('img/appimage.svg', ROOT_DIR),
requires_root=False,
refresh=True,
requires_internet=False)
return self._action_self_install
@property
def app_repository(self) -> str:
if self._app_repository is None:
self._app_repository = 'https://github.com/spalencsar/bearhub'
return self._app_repository
@property
def suggestions_downloader(self) -> AppImageSuggestionsDownloader:
if not self._suggestions_downloader:
file_url = self.context.get_suggestion_url(self.__module__)
self._suggestions_downloader = AppImageSuggestionsDownloader(taskman=TaskManager(),
i18n=self.context.i18n,
http_client=self.context.http_client,
logger=self.context.logger,
file_url=file_url)
if self._suggestions_downloader.is_custom_local_file_mapped():
self.logger.info(f"Local AppImage suggestions file mapped: {file_url}")
return self._suggestions_downloader