my-project-public

repository

loading code, commits, and activity

repositories

loading repo index

my-project-public — gitlawb

#1	"""
#2	Mnemosyne Disaster Recovery System
#3
#4	Comprehensive backup, restore, and integrity verification for Mnemosyne.
#5	"""
#6
#7	import gzip
#8	import json
#9	import hashlib
#10	import shutil
#11	from datetime import datetime
#12	from pathlib import Path
#13	from typing import Dict, List, Optional
#14
#15
#16	def get_default_paths():
#17	"""Get default Mnemosyne paths"""
#18	data_dir = Path.home() / ".mnemosyne" / "data"
#19	backup_dir = Path.home() / ".mnemosyne" / "backups"
#20	db_path = data_dir / "mnemosyne.db"
#21	return data_dir, backup_dir, db_path
#22
#23
#24	def create_backup(db_path: Path = None, backup_dir: Path = None) -> Dict:
#25	"""
#26	Create a compressed backup of the database.
#27
#28	Returns:
#29	Dict with backup_path, size, checksum, and timestamp
#30	"""
#31	_, default_backup_dir, default_db = get_default_paths()
#32	db_path = db_path or default_db
#33	backup_dir = backup_dir or default_backup_dir
#34
#35	if not db_path.exists():
#36	raise FileNotFoundError(f"Database not found: {db_path}")
#37
#38	backup_dir.mkdir(parents=True, exist_ok=True)
#39
#40	timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
#41	backup_name = f"mnemosyne_backup_{timestamp}.db.gz"
#42	backup_path = backup_dir / backup_name
#43
#44	# Compress database
#45	with open(db_path, 'rb') as f_in:
#46	with gzip.open(backup_path, 'wb') as f_out:
#47	shutil.copyfileobj(f_in, f_out)
#48
#49	# Calculate checksums
#50	db_checksum = hashlib.sha256(db_path.read_bytes()).hexdigest()[:16]
#51	backup_checksum = hashlib.sha256(backup_path.read_bytes()).hexdigest()[:16]
#52
#53	# Create metadata
#54	metadata = {
#55	"timestamp": timestamp,
#56	"original_size": db_path.stat().st_size,
#57	"backup_size": backup_path.stat().st_size,
#58	"db_checksum": db_checksum,
#59	"backup_checksum": backup_checksum,
#60	"compressed": True
#61	}
#62
#63	# Save metadata
#64	meta_path = backup_path.with_suffix('.gz.json')
#65	with open(meta_path, 'w') as f:
#66	json.dump(metadata, f, indent=2)
#67
#68	return {
#69	"backup_path": str(backup_path),
#70	"metadata_path": str(meta_path),
#71	**metadata
#72	}
#73
#74
#75	def restore_backup(backup_path: Path, db_path: Path = None) -> Dict:
#76	"""
#77	Restore database from a compressed backup.
#78
#79	Args:
#80	backup_path: Path to the .gz backup file
#81	db_path: Destination database path (default: ~/.mnemosyne/data/mnemosyne.db)
#82
#83	Returns:
#84	Dict with restore status and details
#85	"""
#86	_, _, default_db = get_default_paths()
#87	db_path = db_path or default_db
#88
#89	if not backup_path.exists():
#90	raise FileNotFoundError(f"Backup not found: {backup_path}")
#91
#92	# Create emergency backup of current DB
#93	if db_path.exists():
#94	emergency_path = db_path.with_suffix('.emergency_backup.db')
#95	shutil.copy2(db_path, emergency_path)
#96
#97	# Decompress and restore
#98	db_path.parent.mkdir(parents=True, exist_ok=True)
#99
#100	with gzip.open(backup_path, 'rb') as f_in:
#101	with open(db_path, 'wb') as f_out:
#102	shutil.copyfileobj(f_in, f_out)
#103
#104	# Verify restored database
#105	is_valid = verify_integrity(db_path)
#106
#107	return {
#108	"restored": True,
#109	"backup_used": str(backup_path),
#110	"database_path": str(db_path),
#111	"integrity_check": is_valid
#112	}
#113
#114
#115	def emergency_restore(backup_dir: Path = None, db_path: Path = None) -> Dict:
#116	"""
#117	Automatically restore from the most recent valid backup.
#118
#119	Returns:
#120	Dict with restore status
#121	"""
#122	_, default_backup_dir, default_db = get_default_paths()
#123	backup_dir = backup_dir or default_backup_dir
#124	db_path = db_path or default_db
#125
#126	# Find all backups
#127	backups = sorted(backup_dir.glob("mnemosyne_backup_*.db.gz"), reverse=True)
#128
#129	if not backups:
#130	raise FileNotFoundError("No backups found in " + str(backup_dir))
#131
#132	# Try each backup until one works
#133	for backup in backups:
#134	try:
#135	result = restore_backup(backup, db_path)
#136	if result["integrity_check"]:
#137	return {
#138	"restored": True,
#139	"backup_used": str(backup),
#140	"attempts": 1
#141	}
#142	except Exception as e:
#143	continue
#144
#145	raise RuntimeError("All backups failed integrity check")
#146
#147
#148	def verify_integrity(db_path: Path = None) -> bool:
#149	"""
#150	Verify SQLite database integrity.
#151
#152	Returns:
#153	True if database is valid, False otherwise
#154	"""
#155	import sqlite3
#156
#157	_, _, default_db = get_default_paths()
#158	db_path = db_path or default_db
#159
#160	if not db_path.exists():
#161	return False
#162
#163	try:
#164	conn = sqlite3.connect(str(db_path))
#165	cursor = conn.cursor()
#166
#167	# Run PRAGMA integrity_check
#168	cursor.execute("PRAGMA integrity_check")
#169	result = cursor.fetchone()
#170
#171	conn.close()
#172
#173	return result[0] == "ok"
#174	except Exception:
#175	return False
#176
#177
#178	def list_backups(backup_dir: Path = None) -> List[Dict]:
#179	"""
#180	List all available backups with metadata.
#181
#182	Returns:
#183	List of backup information dictionaries
#184	"""
#185	_, default_backup_dir, _ = get_default_paths()
#186	backup_dir = backup_dir or default_backup_dir
#187
#188	backups = []
#189	for backup_file in sorted(backup_dir.glob("mnemosyne_backup_*.db.gz"), reverse=True):
#190	meta_file = backup_file.with_suffix('.gz.json')
#191
#192	info = {
#193	"file": str(backup_file),
#194	"name": backup_file.name,
#195	"size": backup_file.stat().st_size,
#196	"modified": datetime.fromtimestamp(backup_file.stat().st_mtime).isoformat()
#197	}
#198
#199	if meta_file.exists():
#200	with open(meta_file) as f:
#201	info["metadata"] = json.load(f)
#202
#203	backups.append(info)
#204
#205	return backups
#206
#207
#208	def rotate_backups(backup_dir: Path = None, keep: int = 10) -> Dict:
#209	"""
#210	Rotate backups, keeping only the most recent N.
#211
#212	Args:
#213	keep: Number of backups to retain
#214
#215	Returns:
#216	Dict with rotation results
#217	"""
#218	_, default_backup_dir, _ = get_default_paths()
#219	backup_dir = backup_dir or default_backup_dir
#220
#221	backups = sorted(backup_dir.glob("mnemosyne_backup_*.db.gz"))
#222
#223	to_delete = backups[:-keep] if len(backups) > keep else []
#224	deleted = []
#225
#226	for backup in to_delete:
#227	# Delete backup and metadata
#228	backup.unlink()
#229	meta = backup.with_suffix('.gz.json')
#230	if meta.exists():
#231	meta.unlink()
#232	deleted.append(backup.name)
#233
#234	return {
#235	"total_backups": len(backups),
#236	"kept": keep,
#237	"deleted": len(deleted),
#238	"deleted_files": deleted
#239	}
#240
#241
#242	def health_check() -> Dict:
#243	"""
#244	Comprehensive health check of Mnemosyne system.
#245
#246	Returns:
#247	Dict with health status of all components
#248	"""
#249	data_dir, backup_dir, db_path = get_default_paths()
#250
#251	# Check database
#252	db_exists = db_path.exists()
#253	db_valid = verify_integrity(db_path) if db_exists else False
#254
#255	# Check backups
#256	backups = list(backup_dir.glob("mnemosyne_backup_*.db.gz")) if backup_dir.exists() else []
#257
#258	return {
#259	"database": {
#260	"exists": db_exists,
#261	"valid": db_valid,
#262	"path": str(db_path),
#263	"message": "Database integrity verified" if db_valid else "Database missing or corrupt"
#264	},
#265	"backups": {
#266	"total": len(backups),
#267	"latest": str(backups[-1]) if backups else None,
#268	"directory": str(backup_dir)
#269	},
#270	"status": "healthy" if db_valid else "unhealthy"
#271	}
#272
#273
#274	# CLI interface
#275	if __name__ == "__main__":
#276	import sys
#277
#278	if len(sys.argv) < 2:
#279	print("Usage: python -m mnemosyne.dr [backup\|restore\|emergency\|verify\|list\|health\|rotate]")
#280	sys.exit(1)
#281
#282	cmd = sys.argv[1]
#283
#284	if cmd == "backup":
#285	result = create_backup()
#286	print(json.dumps(result, indent=2))
#287
#288	elif cmd == "restore" and len(sys.argv) > 2:
#289	result = restore_backup(Path(sys.argv[2]))
#290	print(json.dumps(result, indent=2))
#291
#292	elif cmd == "emergency":
#293	result = emergency_restore()
#294	print(json.dumps(result, indent=2))
#295
#296	elif cmd == "verify":
#297	valid = verify_integrity()
#298	print(json.dumps({"valid": valid}))
#299
#300	elif cmd == "list":
#301	backups = list_backups()
#302	print(json.dumps(backups, indent=2))
#303
#304	elif cmd == "health":
#305	status = health_check()
#306	print(json.dumps(status, indent=2))
#307
#308	elif cmd == "rotate":
#309	result = rotate_backups()
#310	print(json.dumps(result, indent=2))
#311
#312	else:
#313	print(f"Unknown command: {cmd}")
#314	sys.exit(1)
#315

z6Mkq5mY3JWtxoxUobWcfNHm7AkRubgSWEZTkBVqZXJviFZ5/my-project-public