264 lines
8.6 KiB
Python
264 lines
8.6 KiB
Python
import argparse
|
|
import copy
|
|
import logging
|
|
import re
|
|
|
|
import borgmatic.logger
|
|
from borgmatic.borg import environment, feature, flags, rlist
|
|
from borgmatic.execute import execute_command, execute_command_and_capture_output
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
ARCHIVE_FILTER_FLAGS_MOVED_TO_RLIST = ('prefix', 'match_archives', 'sort_by', 'first', 'last')
|
|
MAKE_FLAGS_EXCLUDES = (
|
|
'repository',
|
|
'archive',
|
|
'paths',
|
|
'find_paths',
|
|
) + ARCHIVE_FILTER_FLAGS_MOVED_TO_RLIST
|
|
|
|
|
|
def make_list_command(
|
|
repository_path,
|
|
config,
|
|
local_borg_version,
|
|
list_arguments,
|
|
global_arguments,
|
|
local_path='borg',
|
|
remote_path=None,
|
|
):
|
|
'''
|
|
Given a local or remote repository path, a configuration 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 = 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('log-json', global_arguments.log_json)
|
|
+ flags.make_flags('lock-wait', lock_wait)
|
|
+ flags.make_flags_from_arguments(list_arguments, excludes=MAKE_FLAGS_EXCLUDES)
|
|
+ (
|
|
flags.make_repository_archive_flags(
|
|
repository_path, list_arguments.archive, local_borg_version
|
|
)
|
|
if list_arguments.archive
|
|
else flags.make_repository_flags(repository_path, 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 capture_archive_listing(
|
|
repository_path,
|
|
archive,
|
|
config,
|
|
local_borg_version,
|
|
global_arguments,
|
|
list_paths=None,
|
|
path_format=None,
|
|
local_path='borg',
|
|
remote_path=None,
|
|
):
|
|
'''
|
|
Given a local or remote repository path, an archive name, a configuration dict, the local Borg
|
|
version, global arguments as an argparse.Namespace, the archive paths in which to list files,
|
|
the Borg path format to use for the output, and local and remote Borg paths, capture the output
|
|
of listing that archive and return it as a list of file paths.
|
|
'''
|
|
borg_environment = environment.make_environment(config)
|
|
|
|
return tuple(
|
|
execute_command_and_capture_output(
|
|
make_list_command(
|
|
repository_path,
|
|
config,
|
|
local_borg_version,
|
|
argparse.Namespace(
|
|
repository=repository_path,
|
|
archive=archive,
|
|
paths=[f'sh:{path}' for path in list_paths] if list_paths else None,
|
|
find_paths=None,
|
|
json=None,
|
|
format=path_format or '{path}{NL}', # noqa: FS003
|
|
),
|
|
global_arguments,
|
|
local_path,
|
|
remote_path,
|
|
),
|
|
extra_environment=borg_environment,
|
|
borg_local_path=local_path,
|
|
borg_exit_codes=config.get('borg_exit_codes'),
|
|
)
|
|
.strip('\n')
|
|
.split('\n')
|
|
)
|
|
|
|
|
|
def list_archive(
|
|
repository_path,
|
|
config,
|
|
local_borg_version,
|
|
list_arguments,
|
|
global_arguments,
|
|
local_path='borg',
|
|
remote_path=None,
|
|
):
|
|
'''
|
|
Given a local or remote repository path, a configuration dict, the local Borg version, global
|
|
arguments as an argparse.Namespace, the arguments to the list action as an argparse.Namespace,
|
|
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.
|
|
'''
|
|
borgmatic.logger.add_custom_log_levels()
|
|
|
|
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_path,
|
|
short=list_arguments.short,
|
|
format=list_arguments.format,
|
|
json=list_arguments.json,
|
|
prefix=list_arguments.prefix,
|
|
match_archives=list_arguments.match_archives,
|
|
sort_by=list_arguments.sort_by,
|
|
first=list_arguments.first,
|
|
last=list_arguments.last,
|
|
)
|
|
return rlist.list_repository(
|
|
repository_path,
|
|
config,
|
|
local_borg_version,
|
|
rlist_arguments,
|
|
global_arguments,
|
|
local_path,
|
|
remote_path,
|
|
)
|
|
|
|
if list_arguments.archive:
|
|
for name in ARCHIVE_FILTER_FLAGS_MOVED_TO_RLIST:
|
|
if getattr(list_arguments, name, None):
|
|
logger.warning(
|
|
f"The --{name.replace('_', '-')} flag on the list action is ignored when using the --archive flag."
|
|
)
|
|
|
|
if list_arguments.json:
|
|
raise ValueError(
|
|
'The --json flag on the list action is not supported when using the --archive/--find flags.'
|
|
)
|
|
|
|
borg_environment = environment.make_environment(config)
|
|
borg_exit_codes = config.get('borg_exit_codes')
|
|
|
|
# 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_path,
|
|
short=True,
|
|
format=None,
|
|
json=None,
|
|
prefix=list_arguments.prefix,
|
|
match_archives=list_arguments.match_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_and_capture_output(
|
|
rlist.make_rlist_command(
|
|
repository_path,
|
|
config,
|
|
local_borg_version,
|
|
rlist_arguments,
|
|
global_arguments,
|
|
local_path,
|
|
remote_path,
|
|
),
|
|
extra_environment=borg_environment,
|
|
borg_local_path=local_path,
|
|
borg_exit_codes=borg_exit_codes,
|
|
)
|
|
.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.answer(f'{repository_path}: Listing archive {archive}')
|
|
|
|
archive_arguments = copy.copy(list_arguments)
|
|
archive_arguments.archive = archive
|
|
|
|
# This list call is to show the files in a single archive, not list multiple archives. So
|
|
# blank out any archive filtering flags. They'll break anyway in Borg 2.
|
|
for name in ARCHIVE_FILTER_FLAGS_MOVED_TO_RLIST:
|
|
setattr(archive_arguments, name, None)
|
|
|
|
main_command = make_list_command(
|
|
repository_path,
|
|
config,
|
|
local_borg_version,
|
|
archive_arguments,
|
|
global_arguments,
|
|
local_path,
|
|
remote_path,
|
|
) + make_find_paths(list_arguments.find_paths)
|
|
|
|
execute_command(
|
|
main_command,
|
|
output_log_level=logging.ANSWER,
|
|
extra_environment=borg_environment,
|
|
borg_local_path=local_path,
|
|
borg_exit_codes=borg_exit_codes,
|
|
)
|