36 lines
1.3 KiB
Python
36 lines
1.3 KiB
Python
#!/usr/bin/env python3
|
|
"""Create and verify a consistent SQLite backup for ProspectOS.
|
|
|
|
Usage inside the app container:
|
|
python scripts/backup_sqlite.py
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import sqlite3
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
db_path = os.environ.get("PROSPECT_DB", "/data/prospects.db")
|
|
backup_dir = Path(os.environ.get("BACKUP_DIR", "") or (Path(db_path).parent / "backups")).resolve()
|
|
retention = max(1, min(100, int(os.environ.get("BACKUP_RETENTION", "14"))))
|
|
backup_dir.mkdir(parents=True, exist_ok=True)
|
|
name = "prospects-" + datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") + ".db"
|
|
target = backup_dir / name
|
|
source = sqlite3.connect(db_path, timeout=30)
|
|
destination = sqlite3.connect(str(target), timeout=30)
|
|
try:
|
|
source.backup(destination)
|
|
check = destination.execute("PRAGMA integrity_check").fetchone()[0]
|
|
if str(check).lower() != "ok":
|
|
raise SystemExit(f"integrity_check_failed:{check}")
|
|
destination.commit()
|
|
finally:
|
|
destination.close()
|
|
source.close()
|
|
items = sorted(backup_dir.glob("prospects-*.db"), key=lambda path: path.stat().st_mtime, reverse=True)
|
|
for old in items[retention:]:
|
|
old.unlink(missing_ok=True)
|
|
print(json.dumps({"name": name, "path": str(target), "size_bytes": target.stat().st_size, "integrity": "ok"}))
|