team-tryouts/app/supporting_scripts/backup.py

"""Database backup script for the Team Tryouts application.

This module provides a simple backup mechanism for the SQLite database
and uploaded contract documents. Designed to be run as a scheduled task
(Windows Task Scheduler) or cron job.

Usage:
    python backup.py
    
Configuration via environment variables:
    BACKUP_DIR: Directory to store backups (default: ./backups)
    BACKUP_RETENTION_DAYS: Number of days to keep backups (default: 30)
"""

import os
import shutil
import sqlite3
from datetime import datetime, timedelta

# Configuration
BACKUP_DIR = os.getenv('BACKUP_DIR', os.path.join(os.getcwd(), 'backups'))
BACKUP_RETENTION_DAYS = int(os.getenv('BACKUP_RETENTION_DAYS', 30))
DATABASE_PATH = os.getenv('DATABASE_PATH', os.path.join(os.getcwd(), 'instance', 'team_tryouts.db'))
DOCUMENTS_DIR = os.path.join(os.getcwd(), 'documents')


def create_backup_dir():
    """Create the backup directory if it doesn't exist."""
    os.makedirs(BACKUP_DIR, exist_ok=True)


def backup_database():
    """Backup the SQLite database using sqlite3's built-in backup API.
    
    Returns:
        str: Path to the created backup file, or None if failed.
    """
    if not os.path.exists(DATABASE_PATH):
        print(f'[WARNING] Database not found at {DATABASE_PATH}. Skipping database backup.')
        return None

    timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
    backup_filename = f'db_backup_{timestamp}.db'
    backup_path = os.path.join(BACKUP_DIR, backup_filename)

    try:
        source = sqlite3.connect(DATABASE_PATH)
        destination = sqlite3.connect(backup_path)
        source.backup(destination)
        source.close()
        destination.close()
        print(f'[OK] Database backed up to: {backup_path}')
        return backup_path
    except Exception as e:
        print(f'[ERROR] Database backup failed: {e}')
        return None


def backup_documents():
    """Backup the uploaded contract documents directory.
    
    Returns:
        str: Path to the created archive, or None if no documents exist.
    """
    if not os.path.exists(DOCUMENTS_DIR):
        print('[INFO] No documents directory found. Skipping document backup.')
        return None

    timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
    archive_basename = f'documents_backup_{timestamp}'
    archive_path = os.path.join(BACKUP_DIR, archive_basename)

    try:
        shutil.make_archive(archive_path, 'zip', DOCUMENTS_DIR)
        zip_path = f'{archive_path}.zip'
        print(f'[OK] Documents backed up to: {zip_path}')
        return zip_path
    except Exception as e:
        print(f'[ERROR] Document backup failed: {e}')
        return None


def cleanup_old_backups():
    """Remove backup files older than BACKUP_RETENTION_DAYS."""
    if not os.path.exists(BACKUP_DIR):
        return

    cutoff = datetime.now() - timedelta(days=BACKUP_RETENTION_DAYS)
    removed_count = 0

    for filename in os.listdir(BACKUP_DIR):
        file_path = os.path.join(BACKUP_DIR, filename)
        if os.path.isfile(file_path):
            file_time = datetime.fromtimestamp(os.path.getmtime(file_path))
            if file_time < cutoff:
                try:
                    os.remove(file_path)
                    removed_count += 1
                    print(f'[CLEANUP] Removed old backup: {filename}')
                except OSError as e:
                    print(f'[WARNING] Could not remove {filename}: {e}')

    if removed_count > 0:
        print(f'[CLEANUP] Removed {removed_count} old backup(s).')
    else:
        print('[CLEANUP] No old backups to remove.')


def verify_backup(backup_path):
    """Verify a database backup by running a quick integrity check.
    
    Args:
        backup_path: Path to the backup file to verify.
        
    Returns:
        bool: True if backup is valid, False otherwise.
    """
    if not backup_path or not os.path.exists(backup_path):
        return False

    try:
        conn = sqlite3.connect(backup_path)
        cursor = conn.cursor()
        cursor.execute('PRAGMA integrity_check')
        result = cursor.fetchone()
        conn.close()
        is_valid = result[0] == 'ok'
        if is_valid:
            print(f'[OK] Backup integrity verified: {backup_path}')
        else:
            print(f'[ERROR] Backup integrity check failed: {backup_path} - {result[0]}')
        return is_valid
    except Exception as e:
        print(f'[ERROR] Backup verification failed: {e}')
        return False


def main():
    """Run the full backup process.
    
    Steps:
        1. Create backup directory
        2. Backup database
        3. Backup documents (if any)
        4. Verify database backup
        5. Clean up old backups
        
    Returns:
        int: 0 on success, 1 on failure.
    """
    print('=== Team Tryouts Backup ===')
    print(f'Started at: {datetime.now().strftime("%Y-%m-%d %H:%M:%S")}')
    print(f'Backup directory: {BACKUP_DIR}')
    print(f'Retention period: {BACKUP_RETENTION_DAYS} days')
    print()

    create_backup_dir()

    # 1. Backup database
    db_backup_path = backup_database()
    success = True

    # 2. Verify database backup
    if db_backup_path:
        if not verify_backup(db_backup_path):
            success = False

    # 3. Backup documents
    backup_documents()

    # 4. Cleanup old backups
    cleanup_old_backups()

    print()
    if success:
        print('=== Backup completed successfully ===')
    else:
        print('=== Backup completed with warnings ===')

    return 0 if success else 1


if __name__ == '__main__':
    exit(main())