62 lines
2.1 KiB
Python
62 lines
2.1 KiB
Python
#!/usr/bin/env python3
|
|
"""Backup-Task fuer Second Brain - isoliert, persistent.
|
|
|
|
Local retention is intentionally small on the OpenClaw host: keep only the
|
|
newest two JSONL exports. Longer-term/off-host retention should be handled by
|
|
an encrypted artifact/package workflow, not by committing raw brain exports to
|
|
Git.
|
|
"""
|
|
import json, os, sys
|
|
from pathlib import Path
|
|
from datetime import datetime, timezone
|
|
|
|
BRAIN_DIR = Path("/root/.openclaw/workspace/second-brain")
|
|
sys.path.insert(0, str(BRAIN_DIR))
|
|
from src.store import EngramStore
|
|
|
|
DEFAULT_KEEP_LOCAL_BACKUPS = 2
|
|
|
|
def backup_sort_key(path: Path) -> tuple[float, str]:
|
|
return (path.stat().st_mtime, path.name)
|
|
|
|
def prune_local_backups(data_dir: Path, keep: int = DEFAULT_KEEP_LOCAL_BACKUPS) -> list[str]:
|
|
"""Keep only the newest local backup files.
|
|
|
|
Both uncompressed `.jsonl` and compressed `.jsonl.gz` exports are counted,
|
|
so historical compressed backups do not silently accumulate again.
|
|
"""
|
|
keep = max(1, keep)
|
|
backups = sorted(
|
|
list(data_dir.glob("backup_*.jsonl")) + list(data_dir.glob("backup_*.jsonl.gz")),
|
|
key=backup_sort_key,
|
|
reverse=True,
|
|
)
|
|
removed: list[str] = []
|
|
for backup in backups[keep:]:
|
|
backup.unlink()
|
|
removed.append(str(backup))
|
|
return removed
|
|
|
|
def main():
|
|
brain_db = os.environ.get("BRAIN_DB", str(BRAIN_DIR / "data" / "brain.sqlite"))
|
|
keep_local = int(os.environ.get("SECOND_BRAIN_BACKUP_KEEP_LOCAL", str(DEFAULT_KEEP_LOCAL_BACKUPS)))
|
|
store = EngramStore(brain_db)
|
|
ts = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S")
|
|
backup_path = Path(brain_db).parent / f"backup_{ts}.jsonl"
|
|
count = store.export_jsonl(str(backup_path))
|
|
removed = prune_local_backups(Path(brain_db).parent, keep_local)
|
|
result = {
|
|
"timestamp": datetime.now(timezone.utc).isoformat(),
|
|
"backup_path": str(backup_path),
|
|
"count": count,
|
|
"removed_old_backups": removed,
|
|
"keep_local": keep_local,
|
|
"success": True,
|
|
}
|
|
print(f"BACKUP: {count} Engramme -> {backup_path}")
|
|
print(json.dumps(result, ensure_ascii=True))
|
|
return 0
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|