Source code for AutoArchive._application.archiving.actions.list_action

# _list_action.py
#
# Project: AutoArchive
# License: GNU GPLv3
#
# Copyright (C) 2003 - 2023 Róbert Čerňanský



from datetime import date

from AutoArchive._infrastructure.configuration import Options, OptionsUtils
from AutoArchive._infrastructure.ui import VerbosityLevels, DisplayField, FieldStretchiness, MultiFieldLine
from .action_utils import ActionUtils
from .._archive_info import _BackupLevelRestartReasons
from ..._action.action_result import ActionResult
from ..._action.action import Action



[docs] class ListAction(Action): def __init__(self, actionUtils: ActionUtils, selectedArchiveSpecs): super().__init__(actionUtils.actionResult) self.__actionUtils = actionUtils self.__selectedArchiveSpecs = selectedArchiveSpecs self.__actionResult = actionUtils.actionResult self.__archiving = actionUtils.archiving self.__componentUi = actionUtils.componentUi
[docs] def executeAction_(self) -> bool: archiveSpecsForProcessing = self.__actionUtils.getArchiveSpecsForProcessing(self.__selectedArchiveSpecs) if archiveSpecsForProcessing is None: return False orphanedArchives = frozenset(self.__actionUtils.getOrphanedArchives(archiveSpecsForProcessing)) for archiveSpecInfo in archiveSpecsForProcessing: processingArchSpec = archiveSpecInfo.name or archiveSpecInfo.path self.__componentUi.setProcessingArchSpec(processingArchSpec) # if the archive is orphaned do not even try to get information about it because it most certainly will # fail and an error will be shown which is undesirable if processingArchSpec in orphanedArchives: continue archiveInfo = self.__archiving.getArchiveInfo(archiveSpecInfo.path) if archiveInfo is None: self.__actionResult.set(ActionResult.ActionResults.Failed) continue if self.__componentUi.verbosity == VerbosityLevels.Verbose: self.__showVerboseArchiveInfo(archiveInfo) else: self.__showStandardArchiveInfo(archiveInfo) self.__componentUi.setProcessingArchSpec(None) archiveSpecsNames = {spec.name for spec in archiveSpecsForProcessing} for orphanedArchiveName in orphanedArchives: self.__componentUi.setProcessingArchSpec(orphanedArchiveName) # if user passed some arguments (selected some archives) and option ALL is not enabled then we will going # to list only those orphaned archives which were specified (as arguments) if (len(self.__selectedArchiveSpecs) > 0 and not self.__actionUtils.configuration[Options.ALL]) and \ (orphanedArchiveName not in archiveSpecsNames): continue archiveInfo = self.__archiving.getStoredArchiveInfo(orphanedArchiveName) if self.__componentUi.verbosity == VerbosityLevels.Verbose: self.__showVerboseArchiveInfo(archiveInfo, True) else: self.__showStandardArchiveInfo(archiveInfo, True) self.__componentUi.setProcessingArchSpec(None) return True
def __showVerboseArchiveInfo(self, archiveInfo, orphaned = False): archiveInfoMsg = str.format("Name: {}\n", self.__bracket(archiveInfo.name, orphaned)) archiveInfoMsg += str.format("Root: {}\n", self.__question(archiveInfo.path, True)) archiveInfoMsg += str.format("Archiver type: {}\n", OptionsUtils.archiverTypeToStr(archiveInfo.archiverType)) archiveInfoMsg += str.format("Destination directory: {}\n", archiveInfo.destDir) archiveInfoMsg += str.format("Current backup level/next/max.: {}\n", self.__formatLevelsString(archiveInfo)) archiveInfoMsg += str.format( "Target backup level for non-full restart: {}\n", self.__bracket(self.__question(archiveInfo.restartLevel, archiveInfo.incremental is not None), not archiveInfo.incremental or not archiveInfo.restarting)) archiveInfoMsg += str.format( "Upcoming restart reason: {}\n", self.__bracket(self.__question(self.__reasonToStr(archiveInfo.restartReason), archiveInfo.incremental is not None), not archiveInfo.incremental or not archiveInfo.restarting)) archiveInfoMsg += str.format("Restart count/max.: {}/{}\n", self.__bracket(self.__dash(archiveInfo.restartCount), not archiveInfo.incremental or not archiveInfo.restarting), self.__bracket(self.__dash(archiveInfo.fullRestartAfterCount), not archiveInfo.incremental or not archiveInfo.restarting)) age = (date.today() - archiveInfo.lastRestart).days if archiveInfo.lastRestart is not None else None archiveInfoMsg += str.format("Days since last restart/max.: {}/{}\n", self.__bracket(self.__dash(age), not archiveInfo.incremental or not archiveInfo.restarting), self.__bracket(self.__dash(archiveInfo.restartAfterAge), not archiveInfo.incremental or not archiveInfo.restarting)) age = (date.today() - archiveInfo.lastFullRestart).days if archiveInfo.lastFullRestart is not None else None archiveInfoMsg += str.format("Days since last full restart/max.: {}/{}\n", self.__bracket(self.__dash(age), not archiveInfo.incremental or not archiveInfo.restarting), self.__bracket(self.__dash(archiveInfo.fullRestartAfterAge), not archiveInfo.incremental or not archiveInfo.restarting)) self.__componentUi.showVerbose(archiveInfoMsg) def __showStandardArchiveInfo(self, archiveInfo, orphaned = False): name = DisplayField(self.__bracket(archiveInfo.name, orphaned), 0.087, FieldStretchiness.Medium) path = DisplayField(self.__question(archiveInfo.path, True), 0.447, FieldStretchiness.Normal) destDir = DisplayField(archiveInfo.destDir, 0.448, FieldStretchiness.Normal) levels = DisplayField(self.__formatLevelsString(archiveInfo), 0.015, FieldStretchiness.Low) self.__componentUi.presentMultiFieldLine(MultiFieldLine([name, path, destDir, levels], 1000)) def __formatLevelsString(self, archiveInfo): levels = str.format( "{}/{}/{}", self.__bracket(self.__dash(archiveInfo.backupLevel), not archiveInfo.incremental), self.__bracket(self.__question(archiveInfo.nextBackupLevel, archiveInfo.incremental is not None and archiveInfo.backupLevel is not None), not archiveInfo.incremental), self.__bracket(self.__dash(archiveInfo.restartAfterLevel), not archiveInfo.incremental or not archiveInfo.restarting)) return levels @staticmethod def __bracket(token, condition): leftBracket = "" rightBracket = "" if condition and token is not None: leftBracket = "[" rightBracket = "]" return str.format("{leftBracket}{token}{rightBracket}", leftBracket = leftBracket, token = token, rightBracket = rightBracket) @staticmethod def __dash(value): if value is None: return "-" else: return value @classmethod def __question(cls, value, condition): if condition and value is None: return "?" else: return cls.__dash(value) @classmethod def __reasonToStr(cls, reason): if reason is _BackupLevelRestartReasons.NoRestart: reasonStr = "No restart scheduled for the next backup." elif reason is _BackupLevelRestartReasons.RestartCountLimitReached: reasonStr = "Maximal restart count reached." elif reason is _BackupLevelRestartReasons.LastFullRestartAgeLimitReached: reasonStr = "Maximal age without full restart reached." elif reason is _BackupLevelRestartReasons.BackupLevelLimitReached: reasonStr = "Maximal backup level reached." elif reason is _BackupLevelRestartReasons.LastRestartAgeLimitReached: reasonStr = "Maximal age without a restart reached." elif reason is None: reasonStr = None else: reasonStr = "Unknown reason." return reasonStr