Save the last successful backup and add --last-backup
Write the completion time and backup directory to $XDG_STATE_HOME/simple_backup/last_backup after each successful backup (as the target user when run as root), for status bar reminders. --last-backup prints it.
This commit is contained in:
@@ -315,6 +315,12 @@ class Backup:
|
||||
if self._remote:
|
||||
self._server = f'{self.ssh_user}@{self.ssh_host}:'
|
||||
|
||||
@property
|
||||
def destination(self) -> str:
|
||||
"""The current backup directory, prefixed with 'user@host:' for remote backups"""
|
||||
|
||||
return f'{self._server}{self._output_dir}'
|
||||
|
||||
def remove_old_backups(self) -> None:
|
||||
"""Remove old backups if there are more than indicated by 'keep'"""
|
||||
|
||||
@@ -902,6 +908,8 @@ def _parse_arguments() -> argparse.Namespace:
|
||||
help='Run rsync on remote server with sudo if allowed')
|
||||
parser.add_argument('--numeric-ids', action='store_true',
|
||||
help='Use rsync \'--numeric-ids\' option (don\'t map uid/gid values by name)')
|
||||
parser.add_argument('--last-backup', action='store_true',
|
||||
help='Show when the last successful backup was made and exit')
|
||||
|
||||
args = parser.parse_args()
|
||||
|
||||
@@ -1077,6 +1085,103 @@ def _notify(text: str) -> None:
|
||||
os.seteuid(euid)
|
||||
|
||||
|
||||
def _state_owner(user: str | None) -> pwd.struct_passwd | None:
|
||||
"""The user whose state file is used when running as root on their behalf (sudo, doas, -u)"""
|
||||
|
||||
if os.geteuid() != 0 or user is None:
|
||||
return None
|
||||
|
||||
try:
|
||||
owner = pwd.getpwnam(user)
|
||||
except KeyError:
|
||||
return None
|
||||
|
||||
return owner if owner.pw_uid != 0 else None
|
||||
|
||||
|
||||
def _state_file(owner: pwd.struct_passwd | None) -> str:
|
||||
if owner is not None:
|
||||
# XDG_STATE_HOME belongs to root here
|
||||
base = os.path.join(owner.pw_dir, '.local', 'state')
|
||||
else:
|
||||
base = os.getenv('XDG_STATE_HOME', '')
|
||||
|
||||
# The XDG spec says to ignore relative paths
|
||||
if not os.path.isabs(base):
|
||||
base = os.path.expanduser('~/.local/state')
|
||||
|
||||
return os.path.join(base, 'simple_backup', 'last_backup')
|
||||
|
||||
|
||||
def _write_state(user: str | None, destination: str) -> None:
|
||||
"""Record the time and directory of the last successful backup (two lines)"""
|
||||
|
||||
owner = _state_owner(user)
|
||||
path = _state_file(owner)
|
||||
tmp = ''
|
||||
euid, egid = os.geteuid(), os.getegid()
|
||||
|
||||
# Create the file (and any missing directories) as the user, not root
|
||||
if owner is not None:
|
||||
os.setegid(owner.pw_gid)
|
||||
os.seteuid(owner.pw_uid)
|
||||
|
||||
try:
|
||||
os.makedirs(os.path.dirname(path), exist_ok=True)
|
||||
|
||||
# Replace the file in one step, so readers (e.g. status bar scripts) never see it half written
|
||||
fd, tmp = mkstemp(prefix='.last_backup', dir=os.path.dirname(path), text=True)
|
||||
|
||||
with os.fdopen(fd, 'w', encoding='utf-8') as fp:
|
||||
fp.write(f'{datetime.now().astimezone().isoformat(timespec="seconds")}\n{destination}\n')
|
||||
|
||||
os.replace(tmp, path)
|
||||
except OSError as e:
|
||||
logger.warning('Cannot save the time of the last backup to %s: %s', path, e)
|
||||
|
||||
if tmp != '':
|
||||
try:
|
||||
os.remove(tmp)
|
||||
except OSError:
|
||||
pass
|
||||
finally:
|
||||
if owner is not None:
|
||||
os.seteuid(euid)
|
||||
os.setegid(egid)
|
||||
|
||||
|
||||
def _show_last_backup(user: str | None) -> int:
|
||||
path = _state_file(_state_owner(user))
|
||||
|
||||
try:
|
||||
with open(path, encoding='utf-8') as fp:
|
||||
time, destination = fp.read().splitlines()[:2]
|
||||
|
||||
completed = datetime.fromisoformat(time).astimezone()
|
||||
except FileNotFoundError:
|
||||
print('No backup recorded yet')
|
||||
|
||||
return 1
|
||||
except (OSError, ValueError) as e:
|
||||
print(f'Cannot read {path}: {e}', file=sys.stderr)
|
||||
|
||||
return 1
|
||||
|
||||
days = (datetime.now().astimezone().date() - completed.date()).days
|
||||
|
||||
if days == 0:
|
||||
age = 'today'
|
||||
elif days == 1:
|
||||
age = 'yesterday'
|
||||
else:
|
||||
age = f'{days} days ago'
|
||||
|
||||
print(f'Last backup: {completed:%Y-%m-%d %H:%M} ({age})')
|
||||
print(f'Directory: {destination}')
|
||||
|
||||
return 0
|
||||
|
||||
|
||||
def simple_backup() -> int:
|
||||
"""Main"""
|
||||
|
||||
@@ -1084,6 +1189,9 @@ def simple_backup() -> int:
|
||||
|
||||
user: str | None = args.user or _detect_user()
|
||||
|
||||
if args.last_backup:
|
||||
return _show_last_backup(user)
|
||||
|
||||
if not args.user and os.geteuid() != 0:
|
||||
homedir = os.getenv('HOME', '')
|
||||
elif user is not None:
|
||||
@@ -1167,8 +1275,13 @@ def simple_backup() -> int:
|
||||
try:
|
||||
return_code = backup.check_params(homedir)
|
||||
|
||||
if return_code != 0:
|
||||
return return_code
|
||||
|
||||
return_code = backup.run()
|
||||
|
||||
if return_code == 0:
|
||||
return backup.run()
|
||||
_write_state(user, backup.destination)
|
||||
|
||||
return return_code
|
||||
finally:
|
||||
|
||||
Reference in New Issue
Block a user