borgmatic/borgmatic/borg/list.py

import argparse
import copy
import logging
import re

from borgmatic.borg import environment, feature, flags, rlist
from borgmatic.execute import execute_command

logger = logging.getLogger(__name__)


MAKE_FLAGS_EXCLUDES = ('repository', 'archive', 'successful', 'paths', 'find_paths')


def make_list_command(
    repository,
    storage_config,
    local_borg_version,
    list_arguments,
    local_path='borg',
    remote_path=None,
):
    '''
    Given a local or remote repository path, a storage config dict, the arguments to the list
    action, and local and remote Borg paths, return a command as a tuple to list archives or paths
    within an archive.
    '''
    lock_wait = storage_config.get('lock_wait', None)

    return (
        (local_path, 'list')
        + (
            ('--info',)
            if logger.getEffectiveLevel() == logging.INFO and not list_arguments.json
            else ()
        )
        + (
            ('--debug', '--show-rc')
            if logger.isEnabledFor(logging.DEBUG) and not list_arguments.json
            else ()
        )
        + flags.make_flags('remote-path', remote_path)
        + flags.make_flags('lock-wait', lock_wait)
        + flags.make_flags_from_arguments(list_arguments, excludes=MAKE_FLAGS_EXCLUDES,)
        + (
            flags.make_repository_archive_flags(
                repository, list_arguments.archive, local_borg_version
            )
            if list_arguments.archive
            else flags.make_repository_flags(repository, local_borg_version)
        )
        + (tuple(list_arguments.paths) if list_arguments.paths else ())
    )


def make_find_paths(find_paths):
    '''
    Given a sequence of path fragments or patterns as passed to `--find`, transform all path
    fragments into glob patterns. Pass through existing patterns untouched.

    For example, given find_paths of:

      ['foo.txt', 'pp:root/somedir']

    ... transform that into:

      ['sh:**/*foo.txt*/**', 'pp:root/somedir']
    '''
    if not find_paths:
        return ()

    return tuple(
        find_path
        if re.compile(r'([-!+RrPp] )|(\w\w:)').match(find_path)
        else f'sh:**/*{find_path}*/**'
        for find_path in find_paths
    )


def list_archive(
    repository,
    storage_config,
    local_borg_version,
    list_arguments,
    local_path='borg',
    remote_path=None,
):
    '''
    Given a local or remote repository path, a storage config dict, the local Borg version, the
    arguments to the list action, and local and remote Borg paths, display the output of listing
    the files of a Borg archive (or return JSON output). If list_arguments.find_paths are given,
    list the files by searching across multiple archives. If neither find_paths nor archive name
    are given, instead list the archives in the given repository.
    '''
    if not list_arguments.archive and not list_arguments.find_paths:
        if feature.available(feature.Feature.RLIST, local_borg_version):
            logger.warning(
                'Omitting the --archive flag on the list action is deprecated when using Borg 2.x. Use the rlist action instead.'
            )

        rlist_arguments = argparse.Namespace(
            repository=repository,
            short=list_arguments.short,
            format=list_arguments.format,
            json=list_arguments.json,
            prefix=list_arguments.prefix,
            glob_archives=list_arguments.glob_archives,
            sort_by=list_arguments.sort_by,
            first=list_arguments.first,
            last=list_arguments.last,
        )
        return rlist.list_repository(
            repository, storage_config, local_borg_version, rlist_arguments, local_path, remote_path
        )

    if feature.available(feature.Feature.RLIST, local_borg_version):
        for flag_name in ('prefix', 'glob-archives', 'sort-by', 'first', 'last'):
            if getattr(list_arguments, flag_name.replace('-', '_'), None):
                raise ValueError(
                    f'The --{flag_name} flag on the list action is not supported when using the --archive flag and Borg 2.x.'
                )

    borg_environment = environment.make_environment(storage_config)

    # If there are any paths to find (and there's not a single archive already selected), start by
    # getting a list of archives to search.
    if list_arguments.find_paths and not list_arguments.archive:
        rlist_arguments = argparse.Namespace(
            repository=repository,
            short=True,
            format=None,
            json=None,
            prefix=list_arguments.prefix,
            glob_archives=list_arguments.glob_archives,
            sort_by=list_arguments.sort_by,
            first=list_arguments.first,
            last=list_arguments.last,
        )

        # Ask Borg to list archives. Capture its output for use below.
        archive_lines = tuple(
            execute_command(
                rlist.make_rlist_command(
                    repository,
                    storage_config,
                    local_borg_version,
                    rlist_arguments,
                    local_path,
                    remote_path,
                ),
                output_log_level=None,
                borg_local_path=local_path,
                extra_environment=borg_environment,
            )
            .strip('\n')
            .split('\n')
        )
    else:
        archive_lines = (list_arguments.archive,)

    # For each archive listed by Borg, run list on the contents of that archive.
    for archive in archive_lines:
        logger.warning(f'{repository}: Listing archive {archive}')

        archive_arguments = copy.copy(list_arguments)
        archive_arguments.archive = archive
        main_command = make_list_command(
            repository,
            storage_config,
            local_borg_version,
            archive_arguments,
            local_path,
            remote_path,
        ) + make_find_paths(list_arguments.find_paths)

        output = execute_command(
            main_command,
            output_log_level=None if list_arguments.json else logging.WARNING,
            borg_local_path=local_path,
            extra_environment=borg_environment,
        )

        if list_arguments.json:
            return output