59 lines
1.7 KiB
Python
59 lines
1.7 KiB
Python
import sqlite3
|
|
import shutil
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
import logging
|
|
from .. import get_db_path
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
def backup_database():
|
|
"""Create a backup of the SQLite database"""
|
|
try:
|
|
db_path = get_db_path()
|
|
if not db_path.exists():
|
|
logger.error("Database file not found")
|
|
return False
|
|
|
|
backup_dir = db_path.parent / 'backups'
|
|
backup_dir.mkdir(exist_ok=True)
|
|
|
|
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
|
|
backup_path = backup_dir / f'vpn_db_backup_{timestamp}.db'
|
|
|
|
shutil.copy2(db_path, backup_path)
|
|
logger.info(f"Database backed up successfully to {backup_path}")
|
|
|
|
# Keep only last 5 backups
|
|
backups = sorted(backup_dir.glob('vpn_db_backup_*.db'))
|
|
if len(backups) > 5:
|
|
for old_backup in backups[:-5]:
|
|
old_backup.unlink()
|
|
logger.info(f"Removed old backup: {old_backup}")
|
|
|
|
return True
|
|
|
|
except Exception as e:
|
|
logger.error(f"Backup failed: {str(e)}")
|
|
return False
|
|
|
|
def verify_database():
|
|
"""Check database integrity"""
|
|
try:
|
|
db_path = get_db_path()
|
|
conn = sqlite3.connect(db_path)
|
|
cursor = conn.cursor()
|
|
|
|
cursor.execute("PRAGMA integrity_check")
|
|
result = cursor.fetchone()[0]
|
|
|
|
if result != "ok":
|
|
logger.error(f"Database integrity check failed: {result}")
|
|
return False
|
|
|
|
logger.info("Database integrity verified")
|
|
return True
|
|
|
|
except Exception as e:
|
|
logger.error(f"Database verification failed: {str(e)}")
|
|
return False |