bearhub/bauh/gems/debian/tasks.py

import time
import traceback
from datetime import datetime, timedelta
from logging import Logger
from threading import Thread
from typing import Optional, Set

from bauh.api.abstract.handler import TaskManager, ProcessWatcher
from bauh.commons.html import bold
from bauh.commons.system import ProcessHandler
from bauh.gems.debian import DEBIAN_ICON_PATH, PACKAGE_SYNC_TIMESTAMP_FILE
from bauh.gems.debian.aptitude import Aptitude
from bauh.gems.debian.index import ApplicationIndexer, ApplicationsMapper
from bauh.gems.debian.model import DebianApplication
from bauh.view.util.translation import I18n


class MapApplications(Thread):

    def __init__(self, taskman: TaskManager, app_indexer: ApplicationIndexer, i18n: I18n, logger: Logger,
                 deb_config: dict, app_mapper: ApplicationsMapper, check_expiration: bool = True,
                 watcher: Optional[ProcessWatcher] = None):
        super(MapApplications, self).__init__()
        self._taskman = taskman
        self._i18n = i18n
        self._log = logger
        self._id = 'debian.map_apps'
        self._indexer = app_indexer
        self._config = deb_config
        self._app_mapper = app_mapper
        self._taskman.register_task(self._id, self._i18n['debian.task.map_apps.status'], DEBIAN_ICON_PATH)
        self.apps: Optional[Set[DebianApplication]] = None
        self.cached: Optional[bool] = None
        self._check_expiration = check_expiration
        self._watcher = watcher

    def _change_substatus(self, msg: str):
        if self._watcher:
            self._watcher.change_substatus(msg)

    def run(self):
        ti = time.time()
        self._log.info("Begin: mapping Debian applications")

        self._taskman.update_progress(self._id, 1, None)

        if not self._check_expiration or self._indexer.is_expired(self._config):
            status = self._i18n['debian.task.map_apps.check_files'].format(type="'.desktop")
            self._taskman.update_progress(self._id, 1, status)
            self._change_substatus(status)
            self.apps = self._app_mapper.map_executable_applications()
            self.cached = False
        else:
            status = self._i18n['debian.task.map_apps.read_cache']
            self._taskman.update_progress(self._id, 1, status)
            self._change_substatus(status)
            self.apps = {app for app in self._indexer.read_index()}
            self.cached = True

        self._log.info(f"Number of Debian applications found: {len(self.apps) if self.apps else 0}")
        self._taskman.update_progress(self._id, 100, None)
        self._taskman.finish_task(self._id)

        tf = time.time()
        self._log.info(f"Finish: mapping Debian applications ({tf - ti:.4f} seconds)")


class UpdateApplicationIndex(Thread):

    def __init__(self, taskman: TaskManager, app_indexer: ApplicationIndexer,
                 i18n: I18n, logger: Logger, mapping_apps: MapApplications,
                 watcher: Optional[ProcessWatcher] = None):
        super(UpdateApplicationIndex, self).__init__()
        self._taskman = taskman
        self._i18n = i18n
        self._log = logger
        self._id = 'debian.app_idx'
        self._indexer = app_indexer
        self._mapping_apps = mapping_apps
        self._taskman.register_task(self._id, self._i18n['debian.task.app_index.status'], DEBIAN_ICON_PATH)
        self._watcher = watcher

    def _change_substatus(self, msg: str):
        if self._watcher:
            self._watcher.change_substatus(msg)

    def run(self):
        ti = time.time()
        self._log.info("Begin: Debian applications indexation")

        self._taskman.update_progress(self._id, 1, self._i18n['task.waiting_task'].format(bold(self._i18n['debian.task.map_apps.status'])))
        self._mapping_apps.join()

        finish_msg = None
        if self._mapping_apps.cached:
            finish_msg = self._i18n['task.canceled']
        else:
            status = self._i18n['debian.task.update_apps_idx.status']
            self._taskman.update_progress(self._id, 50, status)
            self._change_substatus(status)

            try:
                self._indexer.update_index(self._mapping_apps.apps)
            except Exception:
                finish_msg = self._i18n['error']

        self._taskman.update_progress(self._id, 100, finish_msg)
        self._taskman.finish_task(self._id)

        tf = time.time()
        self._log.info(f"Finish: Debian applications indexation ({tf - ti:.4f} seconds)")


class SynchronizePackages(Thread):

    def __init__(self, taskman: TaskManager, i18n: I18n, logger: Logger, root_password: Optional[str],
                 aptitude: Aptitude, watcher: Optional[ProcessWatcher] = None):
        super(SynchronizePackages, self).__init__()
        self._id = 'debian.sync_pkgs'
        self._taskman = taskman
        self._i18n = i18n
        self._log = logger
        self._root_password = root_password
        self._watcher = watcher
        self._aptitude = aptitude
        self._taskman.register_task(self._id, self._i18n['debian.task.sync_pkgs.status'], DEBIAN_ICON_PATH)

    def _notify_output(self, output: str):
        self._taskman.update_output(self._id, output)

    @staticmethod
    def should_synchronize(deb_config: dict, logger: Logger) -> bool:
        try:
            period = int(deb_config.get('sync_pkgs.time', 0))
        except ValueError:
            logger.error(f"Invalid value for Debian configuration property 'sync_pkgs.time': "
                         f"{deb_config['sync_pkgs.time']}")
            return True

        if period <= 0:
            logger.warning("Packages synchronization will always be done ('sync_pkgs.time' <= 0 )'")
            return True

        try:
            with open(PACKAGE_SYNC_TIMESTAMP_FILE) as f:
                timestamp_str = f.read().strip()
        except FileNotFoundError:
            logger.info(f"No packages synchronization timestamp found ({PACKAGE_SYNC_TIMESTAMP_FILE})")
            return True

        try:
            last_timestamp = datetime.fromtimestamp(float(timestamp_str))
        except Exception:
            logger.error(f'Could not parse the packages synchronization timestamp: {timestamp_str} '
                         f'({PACKAGE_SYNC_TIMESTAMP_FILE})')
            traceback.print_exc()
            return True

        expired = last_timestamp + timedelta(minutes=period) <= datetime.utcnow()

        if expired:
            logger.info("Packages synchronization is outdated")
        else:
            logger.info("Packages synchronization is up-to-date")

        return expired

    def run(self) -> bool:
        ti = time.time()
        self._log.info("Begin: packages synchronization")
        self._taskman.update_progress(self._id, 1, None)

        handler = ProcessHandler(self._watcher)
        updated, _ = handler.handle_simple(self._aptitude.update(self._root_password),
                                           output_handler=self._notify_output)
        self._taskman.update_progress(self._id, 99, None)

        if updated:
            index_timestamp = datetime.utcnow().timestamp()
            finish_msg = None
            try:
                with open(PACKAGE_SYNC_TIMESTAMP_FILE, 'w+') as f:
                    f.write(str(index_timestamp))
            except OSError:
                finish_msg = self._i18n['error']
                self._log.error(f"Could not write the packages synchronization timestamp to file "
                                f"'{PACKAGE_SYNC_TIMESTAMP_FILE}'")
        else:
            finish_msg = self._i18n['error']

        self._taskman.update_progress(self._id, 100, finish_msg)
        self._taskman.finish_task(self._id)

        tf = time.time()
        self._log.info(f"Finish: packages synchronization ({tf - ti:.4f} seconds)")
        return updated