slaguru666/AIPass
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(backup): restore full 9-stage rich CLI output (FPLAN-0263) — new backup_timestamps state handler + display.py 5-stage pipeline (last_backups panel -> boxed header -> live Rich progress bar -> result summary -> backups_now panel), extend BackupResult with files_checked/files_skipped/backup_path, emit on_progress callbacks from copy/snapshot+versioned, rewire snapshot/versioned/all to the rich pipeline. Faithful port from gold archive source. Verified live under pty: full color + animated transient progress bar. seedgo 95%, ruff clean.

Browse Source
This commit is contained in:
AIOSAI
2026-06-12 11:35:18 -07:00
parent 1f3727d4fc
commit 1747a23e5c
8 changed files with 315 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/aipass/backup/apps/handlers/copy/snapshot.py
+4
View File
@@ -24,6 +24,7 @@ def copy_snapshot(
files: list[tuple[str, str]],
dest: str,
project_root: str,
on_progress=None,
) -> dict:
"""Copy files into a snapshot destination.
@@ -31,6 +32,7 @@ def copy_snapshot(
files: List of (absolute_path, relative_path) tuples.
dest: Absolute destination directory path (.backup_system/snapshots/).
project_root: Project root for logging context.
on_progress: Optional callback called after each file is processed.
Returns:
Dict with files_copied, bytes_copied, errors.
@@ -54,6 +56,8 @@ def copy_snapshot(
except OSError as e:
logger.warning(f"Failed to copy {rel_path}: {e}")
errors.append(f"{rel_path}: {e}")
if on_progress:
on_progress()
result = {
"files_copied": files_copied,
src/aipass/backup/apps/handlers/copy/versioned.py
+8
View File
@@ -25,6 +25,7 @@ def copy_versioned(
dest: str,
previous_timestamps: dict,
project_root: str,
on_progress=None,
) -> dict:
"""Copy changed files into a versioned destination.
@@ -33,6 +34,7 @@ def copy_versioned(
dest: Absolute destination directory for this version.
previous_timestamps: Mapping of relative_path to last-known mtime.
project_root: Project root for logging context.
on_progress: Optional callback called after each file is processed.
Returns:
Dict with files_copied, bytes_copied, files_unchanged, errors, new_timestamps.
@@ -51,6 +53,8 @@ def copy_versioned(
except OSError as e:
logger.warning(f"Cannot stat {rel_path}: {e}")
errors.append(f"{rel_path}: {e}")
if on_progress:
on_progress()
continue
new_timestamps[rel_path] = current_mtime
@@ -58,6 +62,8 @@ def copy_versioned(
if current_mtime <= prev_mtime:
files_unchanged += 1
if on_progress:
on_progress()
continue
target = os.path.join(dest, rel_path)
@@ -69,6 +75,8 @@ def copy_versioned(
except OSError as e:
logger.warning(f"Failed to copy {rel_path}: {e}")
errors.append(f"{rel_path}: {e}")
if on_progress:
on_progress()
result = {
"files_copied": files_copied,
src/aipass/backup/apps/handlers/report/result.py
+3
View File
@@ -24,8 +24,11 @@ class BackupResult:
mode: str
project_root: str = ""
files_copied: int = 0
files_checked: int = 0
files_skipped: int = 0
bytes_copied: int = 0
duration_seconds: float = 0.0
backup_path: str = ""
errors: list[str] = field(default_factory=list)
src/aipass/backup/apps/handlers/state/backup_timestamps.py
+93
View File
@@ -0,0 +1,93 @@
# =================== AIPass ====================
# Name: backup_timestamps.py
# Description: Tracks last-run timestamps for all backup modes
# Version: 1.0.0
# Created: 2026-06-12
# Modified: 2026-06-12
# =============================================
"""Backup timestamps — tracks when each backup mode was last run."""
import json
import os
import tempfile
from datetime import datetime
from pathlib import Path
from aipass.prax import logger
from ..json import json_handler
_BACKUP_ROOT = Path(__file__).resolve().parents[3]
TIMESTAMPS_FILE = _BACKUP_ROOT / "backup_json" / "backup_timestamps.json"
MODES = ["snapshot", "versioned", "drive_sync"]
def get_timestamps() -> dict:
"""Read all backup timestamps from disk."""
data = {}
if TIMESTAMPS_FILE.exists():
try:
data = json.loads(TIMESTAMPS_FILE.read_text(encoding="utf-8"))
except (json.JSONDecodeError, OSError) as e:
logger.warning(f"[backup_timestamps] Failed to read timestamps file: {e}")
data = {}
return {mode: data.get(mode) for mode in MODES}
def update_timestamp(mode: str) -> None:
"""Update the timestamp for a backup mode to now."""
json_handler.log_operation("timestamp_updated", {"mode": mode})
data = {}
if TIMESTAMPS_FILE.exists():
try:
data = json.loads(TIMESTAMPS_FILE.read_text(encoding="utf-8"))
except (json.JSONDecodeError, OSError) as e:
logger.warning(f"[backup_timestamps] Failed to read timestamps for update: {e}")
data = {}
data[mode] = datetime.now().isoformat()
TIMESTAMPS_FILE.parent.mkdir(parents=True, exist_ok=True)
fd, tmp_path = tempfile.mkstemp(suffix=".tmp", dir=str(TIMESTAMPS_FILE.parent))
try:
with os.fdopen(fd, "w", encoding="utf-8") as f:
json.dump(data, f, indent=2)
os.replace(tmp_path, str(TIMESTAMPS_FILE))
except Exception:
try:
os.unlink(tmp_path)
except OSError as cleanup_err:
logger.warning(f"[backup_timestamps] Failed to clean temp file: {cleanup_err}")
raise
def format_age(iso_str: str | None) -> str:
"""Format an ISO timestamp as a human-readable age string."""
if not iso_str:
return "never"
try:
then = datetime.fromisoformat(iso_str)
except (ValueError, TypeError) as e:
logger.info(f"[backup_timestamps] Could not parse timestamp '{iso_str}': {e}")
return "unknown"
delta = datetime.now() - then
seconds = int(delta.total_seconds())
if seconds < 60:
return "just now"
if seconds < 3600:
mins = seconds // 60
return f"{mins} min{'s' if mins != 1 else ''} ago"
if seconds < 86400:
hours = seconds // 3600
return f"{hours} hour{'s' if hours != 1 else ''} ago"
days = seconds // 86400
return f"{days} day{'s' if days != 1 else ''} ago"
# =============================================
src/aipass/backup/apps/modules/all.py
+3 -6
View File
@@ -1,9 +1,9 @@
# =================== AIPass ====================
# Name: all.py
# Description: All module — full cycle: snapshot + versioned
# Version: 1.0.0
# Version: 2.0.0
# Created: 2026-04-17
# Modified: 2026-04-23
# Modified: 2026-06-12
# =============================================
"""All Module — runs snapshot then versioned backup in sequence."""
@@ -14,7 +14,6 @@ from aipass.prax import logger
from aipass.cli.apps.modules import console
from aipass.backup.apps.handlers.json import json_handler
from aipass.backup.apps.handlers.report.formatter import format_result
from aipass.backup.apps.modules.snapshot import run_snapshot
from aipass.backup.apps.modules.versioned import run_versioned
@@ -43,11 +42,9 @@ def handle_command(command: str, args: list) -> bool:
logger.info(f"[backup] Running full backup cycle for {project_root}")
snap_result = run_snapshot(project_root)
console.print(format_result(snap_result))
console.print("")
console.print()
ver_result = run_versioned(project_root)
console.print(format_result(ver_result))
json_handler.log_operation(
"all_complete",
src/aipass/backup/apps/modules/display.py
+114
View File
@@ -0,0 +1,114 @@
# =================== AIPass ====================
# Name: display.py
# Description: Rich CLI rendering for backup results (full 9-stage output)
# Version: 2.0.0
# Created: 2026-06-12
# Modified: 2026-06-12
# =============================================
"""Rich CLI rendering for backup — full output pipeline faithfully ported from gold source."""
from rich.progress import BarColumn, Progress, TextColumn, TimeRemainingColumn
from aipass.prax import logger
from aipass.cli.apps.modules import console, error, header, success, warning
from aipass.backup.apps.handlers.json import json_handler
from aipass.backup.apps.handlers.report.formatter import _human_bytes
from aipass.backup.apps.handlers.report.result import BackupResult
from aipass.backup.apps.handlers.state.backup_timestamps import (
format_age,
get_timestamps,
update_timestamp,
)
MODULE_NAME = "display"
def print_introspection():
"""Display module info."""
console.print(f"[bold cyan]{MODULE_NAME} Module[/bold cyan]")
console.print(" Rich CLI rendering for backup results (full 9-stage output)")
console.print(" Not a command module — used by snapshot/versioned/all")
def show_last_backups() -> None:
"""Stage 1: Show 'Last backups:' panel with dim ages."""
ts = get_timestamps()
console.print()
console.print("[dim]Last backups:[/dim]")
console.print(f" [dim]Snapshot: {format_age(ts.get('snapshot'))}[/dim]")
console.print(f" [dim]Versioned: {format_age(ts.get('versioned'))}[/dim]")
console.print(f" [dim]Drive sync: {format_age(ts.get('drive_sync'))}[/dim]")
def show_run_header(result: BackupResult) -> None:
"""Stage 3: Show run header with boxed panel."""
header(
f"Backup — {result.mode.title()}",
{
"Project": result.project_root,
"Mode": result.mode,
},
)
def create_progress_bar():
"""Stage 5: Create and return a Rich Progress context for the copy loop."""
return Progress(
TextColumn("[progress.description]{task.description}"),
BarColumn(),
TextColumn("[progress.percentage]{task.percentage:>3.0f}%"),
TimeRemainingColumn(),
console=console,
transient=True,
)
def show_result_summary(result: BackupResult) -> None:
"""Stage 6+7: Show rich result summary (stats + completion status)."""
console.print()
if result.errors:
if len(result.errors) > 5:
error(
f"{result.mode.title()} backup FAILED",
suggestion="Check file permissions and disk space",
)
else:
warning(
f"{result.mode.title()} completed with {len(result.errors)} errors",
details="; ".join(result.errors[:3]),
)
for err in result.errors[:5]:
console.print(f" [dim]- {err}[/dim]")
if len(result.errors) > 5:
console.print(f" [dim]... and {len(result.errors) - 5} more[/dim]")
else:
success(
f"{result.mode.title()} backup complete",
files_copied=result.files_copied,
files_checked=result.files_checked,
files_skipped=result.files_skipped,
size=_human_bytes(result.bytes_copied),
)
location = result.backup_path if result.backup_path else result.project_root
console.print(f" [dim]Duration: {result.duration_seconds:.1f}s | Location: {location}[/dim]")
json_handler.log_operation("render_result", {"mode": result.mode})
logger.info(f"[backup] Rendered {result.mode} result: {result.files_copied} files")
def show_backups_now(mode: str) -> None:
"""Stage 8: Update timestamp and show 'Backups now:' panel with updated dim ages."""
update_timestamp(mode)
ts = get_timestamps()
console.print()
console.print("[dim]Backups now:[/dim]")
console.print(f" [dim]Snapshot: {format_age(ts.get('snapshot'))}[/dim]")
console.print(f" [dim]Versioned: {format_age(ts.get('versioned'))}[/dim]")
console.print(f" [dim]Drive sync: {format_age(ts.get('drive_sync'))}[/dim]")
# =============================================
src/aipass/backup/apps/modules/snapshot.py
+42 -15
View File
@@ -1,9 +1,9 @@
# =================== AIPass ====================
# Name: snapshot.py
# Description: Snapshot module — full-copy backup of a project
# Version: 1.0.0
# Version: 2.0.0
# Created: 2026-04-17
# Modified: 2026-04-23
# Modified: 2026-06-12
# =============================================
"""Snapshot Module — full mirror backup of a project directory."""
@@ -22,13 +22,19 @@ from aipass.backup.apps.handlers.json import json_handler
from aipass.backup.apps.handlers.path.builder import build_snapshot_path
from aipass.backup.apps.handlers.project.config import load_project_config
from aipass.backup.apps.handlers.project.setup import create_backup_dir
from aipass.backup.apps.handlers.report.formatter import format_result
from aipass.backup.apps.handlers.report.result import BackupResult
from aipass.backup.apps.handlers.scan.filter import filter_paths
from aipass.backup.apps.handlers.scan.walk import walk_project
from aipass.backup.apps.handlers.state.changelog import append_changelog
from aipass.backup.apps.handlers.state.metadata import build_metadata
from aipass.backup.apps.handlers.state.timestamps import save_timestamps
from aipass.backup.apps.modules.display import (
create_progress_bar,
show_backups_now,
show_last_backups,
show_result_summary,
show_run_header,
)
MODULE_NAME = "snapshot"
PRIMARY_COMMAND = "snapshot"
@@ -42,10 +48,13 @@ def print_introspection():
console.print(" Handlers: scan, copy/snapshot, state, ignore, path, report")
def run_snapshot(project_root: str) -> BackupResult:
def run_snapshot(project_root: str, show_panels: bool = True) -> BackupResult:
"""Run a full snapshot backup for a project."""
start = time.time()
if show_panels:
show_last_backups()
create_backup_dir(project_root)
config = load_project_config(project_root)
@@ -57,21 +66,34 @@ def run_snapshot(project_root: str) -> BackupResult:
filtered = filter_paths(all_files, patterns, whitelist_entries, max_size)
dest = str(build_snapshot_path(project_root))
copy_result = copy_snapshot(filtered, dest, project_root)
result = BackupResult(
mode="snapshot",
project_root=project_root,
files_checked=len(filtered),
backup_path=dest,
)
if show_panels:
show_run_header(result)
progress = create_progress_bar()
with progress:
task = progress.add_task("Processing files...", total=len(filtered))
copy_result = copy_snapshot(filtered, dest, project_root, on_progress=lambda: progress.advance(task))
console.print(f"Processing completed: {len(filtered)}/{len(filtered)} files checked")
duration = time.time() - start
timestamps = {rel: os.path.getmtime(abs_p) for abs_p, rel in filtered}
save_timestamps(project_root, timestamps)
result = BackupResult(
mode="snapshot",
project_root=project_root,
files_copied=copy_result.get("files_copied", 0),
bytes_copied=copy_result.get("bytes_copied", 0),
duration_seconds=duration,
errors=copy_result.get("errors", []),
)
result.files_copied = copy_result.get("files_copied", 0)
result.files_skipped = result.files_checked - result.files_copied
result.bytes_copied = copy_result.get("bytes_copied", 0)
result.duration_seconds = duration
result.errors = copy_result.get("errors", [])
metadata = build_metadata(result)
append_changelog(project_root, metadata)
@@ -81,6 +103,12 @@ def run_snapshot(project_root: str) -> BackupResult:
{"project_root": project_root, "files": result.files_copied},
)
logger.info(f"[backup] Snapshot complete: {result.files_copied} files")
if show_panels:
show_result_summary(result)
if not result.errors:
show_backups_now("snapshot")
return result
@@ -94,8 +122,7 @@ def handle_command(command: str, args: list) -> bool:
return True
project_root = args[0]
result = run_snapshot(project_root)
console.print(format_result(result))
run_snapshot(project_root)
return True
src/aipass/backup/apps/modules/versioned.py
+48 -15
View File
@@ -1,9 +1,9 @@
# =================== AIPass ====================
# Name: versioned.py
# Description: Versioned module — timestamped incremental backup with diffs
# Version: 1.0.0
# Version: 2.0.0
# Created: 2026-04-17
# Modified: 2026-04-23
# Modified: 2026-06-12
# =============================================
"""Versioned Module — incremental timestamped backup of a project directory."""
@@ -24,13 +24,19 @@ from aipass.backup.apps.handlers.json import json_handler
from aipass.backup.apps.handlers.path.builder import build_versioned_path
from aipass.backup.apps.handlers.project.config import load_project_config
from aipass.backup.apps.handlers.project.setup import create_backup_dir
from aipass.backup.apps.handlers.report.formatter import format_result
from aipass.backup.apps.handlers.report.result import BackupResult
from aipass.backup.apps.handlers.scan.filter import filter_paths
from aipass.backup.apps.handlers.scan.walk import walk_project
from aipass.backup.apps.handlers.state.changelog import append_changelog
from aipass.backup.apps.handlers.state.metadata import build_metadata
from aipass.backup.apps.handlers.state.timestamps import load_timestamps, save_timestamps
from aipass.backup.apps.modules.display import (
create_progress_bar,
show_backups_now,
show_last_backups,
show_result_summary,
show_run_header,
)
MODULE_NAME = "versioned"
PRIMARY_COMMAND = "versioned"
@@ -59,10 +65,13 @@ def _prune_old_versions(project_root: str, max_versions: int) -> None:
logger.info(f"[backup] Pruned old version: {oldest.name}")
def run_versioned(project_root: str) -> BackupResult:
def run_versioned(project_root: str, show_panels: bool = True) -> BackupResult:
"""Run an incremental versioned backup for a project."""
start = time.time()
if show_panels:
show_last_backups()
create_backup_dir(project_root)
config = load_project_config(project_root)
@@ -77,7 +86,29 @@ def run_versioned(project_root: str) -> BackupResult:
ts = datetime.now(timezone.utc).strftime("%Y%m%d-%H%M%S")
dest = str(build_versioned_path(project_root, ts))
copy_result = copy_versioned(filtered, dest, prev_timestamps, project_root)
result = BackupResult(
mode="versioned",
project_root=project_root,
files_checked=len(filtered),
backup_path=dest,
)
if show_panels:
show_run_header(result)
progress = create_progress_bar()
with progress:
task = progress.add_task("Processing files...", total=len(filtered))
copy_result = copy_versioned(
filtered,
dest,
prev_timestamps,
project_root,
on_progress=lambda: progress.advance(task),
)
console.print(f"Processing completed: {len(filtered)}/{len(filtered)} files checked")
new_timestamps = copy_result.get("new_timestamps", {})
save_timestamps(project_root, new_timestamps)
@@ -91,14 +122,11 @@ def run_versioned(project_root: str) -> BackupResult:
duration = time.time() - start
result = BackupResult(
mode="versioned",
project_root=project_root,
files_copied=copy_result.get("files_copied", 0),
bytes_copied=copy_result.get("bytes_copied", 0),
duration_seconds=duration,
errors=copy_result.get("errors", []),
)
result.files_copied = copy_result.get("files_copied", 0)
result.files_skipped = result.files_checked - result.files_copied
result.bytes_copied = copy_result.get("bytes_copied", 0)
result.duration_seconds = duration
result.errors = copy_result.get("errors", [])
metadata = build_metadata(result)
append_changelog(project_root, metadata)
@@ -114,6 +142,12 @@ def run_versioned(project_root: str) -> BackupResult:
logger.info(
f"[backup] Versioned complete: {result.files_copied} changed, {copy_result.get('files_unchanged', 0)} unchanged"
)
if show_panels:
show_result_summary(result)
if not result.errors:
show_backups_now("versioned")
return result
@@ -127,8 +161,7 @@ def handle_command(command: str, args: list) -> bool:
return True
project_root = args[0]
result = run_versioned(project_root)
console.print(format_result(result))
run_versioned(project_root)
return True