Add final runtime and disk usage summary to mail reports

lib/usage.py

@@ -0,0 +1,72 @@
+"""Best-effort, bounded disk usage summary for backup and restore mail reports."""
+
+from pathlib import Path
+import os
+import subprocess
+
+
+def human(value):
+    for unit in ('B', 'KiB', 'MiB', 'GiB', 'TiB'):
+        if value < 1024 or unit == 'TiB':
+            return f'{value:.1f} {unit}'
+        value /= 1024
+
+
+def duration(seconds):
+    hours, remainder = divmod(max(0, int(seconds)), 3600)
+    minutes, seconds = divmod(remainder, 60)
+    return f'{hours}h {minutes:02d}m {seconds:02d}s'
+
+
+def allocated(path):
+    """Count allocated bytes without following symlinks or crossing filesystems."""
+    try:
+        result = subprocess.run(['du', '-sx', '-B1', '--', str(path)],
+                                capture_output=True, text=True, timeout=30, check=True)
+        return int(result.stdout.split()[0])
+    except (OSError, ValueError, IndexError, subprocess.SubprocessError):
+        return None
+
+
+def free_space(path):
+    try:
+        stat = os.statvfs(path)
+        return stat.f_bavail * stat.f_frsize
+    except OSError:
+        return None
+
+
+def summary(seconds, workdir, archives, staging, restore, logs):
+    """Measure the final local state, after temporary-file cleanup."""
+    lines = ['', 'Laufzeit und Speicher am Ende', '----------------------------',
+             f'Verarbeitung: {duration(seconds)} ({max(0, int(seconds))}s)']
+    paths = [('Arbeitsbereich gesamt', workdir), ('Backups (lokale Archive)', archives),
+             ('Backup-Staging', staging), ('Restore-Arbeitsbereich', restore),
+             ('Restore-Test', str(Path(workdir) / 'restore-test')), ('Protokolle und Berichte', logs)]
+    for label, name in paths:
+        if not name:
+            continue
+        path = Path(name)
+        amount = allocated(path) if path.exists() and not path.is_symlink() else None
+        lines.append(f'{label}: {human(amount) if amount is not None else "nicht verfügbar"} ({path})')
+    devices = set()
+    for label, name in [('Arbeitsbereich', workdir), ('Archive', archives),
+                        ('Staging', staging), ('Restore', restore), ('Restore-Test', str(Path(workdir) / 'restore-test')),
+                        ('Protokolle', logs)]:
+        if not name:
+            continue
+        try:
+            path = Path(name)
+            if path.is_symlink():
+                continue
+            device = path.stat().st_dev
+        except OSError:
+            continue
+        if device in devices:
+            continue
+        devices.add(device)
+        amount = free_space(path)
+        lines.append(f'Frei auf Datenträger {label}: '
+                     f'{human(amount) if amount is not None else "nicht verfügbar"} ({path})')
+    lines.append('Teilbereiche sind im Arbeitsbereich enthalten; getrennte Datenträger werden separat gemessen.')
+    return lines