amo-server/scripts/recover_database.py

#!/usr/bin/env python3
"""
Database Recovery Script for Corrupted SQLite Database

This script attempts to recover data from a corrupted SQLite database
by dumping recoverable data to SQL format and creating a new database.
"""
import sqlite3
import shutil
import sys
from pathlib import Path
from datetime import datetime


def backup_corrupted_db(db_path: str) -> str:
    """Create a backup of the corrupted database."""
    backup_path = f"{db_path}.corrupted.{datetime.now().strftime('%Y%m%d_%H%M%S')}"
    print(f"Creating backup of corrupted database: {backup_path}")
    shutil.copy2(db_path, backup_path)
    return backup_path


def dump_database(db_path: str, output_path: str) -> bool:
    """
    Attempt to dump the database using SQLite's .dump command.
    This will skip corrupted pages and recover what it can.
    """
    print(f"Attempting to dump database from {db_path}...")
    try:
        # Use sqlite3 command line to dump with recovery mode
        import subprocess

        dump_cmd = [
            "sqlite3",
            db_path,
            ".recover"  # Use .recover instead of .dump for corrupted databases
        ]

        print("Running: sqlite3 with .recover mode")
        result = subprocess.run(
            dump_cmd,
            capture_output=True,
            text=True,
            check=False
        )

        if result.returncode == 0 or result.stdout:
            # Write recovered SQL to file
            with open(output_path, 'w', encoding='utf-8') as f:
                f.write(result.stdout)
            print(f"✓ Recovery dump saved to: {output_path}")

            if result.stderr:
                print(f"⚠ Warnings during recovery:\n{result.stderr}")
            return True
        else:
            print(f"✗ Recovery failed: {result.stderr}")
            return False

    except Exception as e:
        print(f"✗ Error during dump: {e}")
        return False


def recreate_database(db_path: str, sql_dump_path: str) -> bool:
    """Recreate the database from the SQL dump."""
    print(f"Recreating database at {db_path}...")

    try:
        # Remove the corrupted database
        if Path(db_path).exists():
            Path(db_path).unlink()
            print("✓ Removed corrupted database")

        # Create new database and import SQL dump
        conn = sqlite3.connect(db_path)
        cursor = conn.cursor()

        print("Importing SQL dump into new database...")
        with open(sql_dump_path, 'r', encoding='utf-8') as f:
            sql_script = f.read()

        # Execute the SQL script
        cursor.executescript(sql_script)
        conn.commit()
        conn.close()

        print("✓ Database recreated successfully")
        return True

    except Exception as e:
        print(f"✗ Error recreating database: {e}")
        return False


def verify_database(db_path: str) -> bool:
    """Run integrity check on the new database."""
    print("Verifying new database integrity...")

    try:
        conn = sqlite3.connect(db_path)
        cursor = conn.cursor()

        cursor.execute("PRAGMA integrity_check;")
        result = cursor.fetchall()
        conn.close()

        if result[0][0] == "ok":
            print("✓ Database integrity check: PASSED")
            return True
        else:
            print("⚠ Database integrity check: ISSUES FOUND")
            for row in result:
                print(f"  - {row[0]}")
            return False

    except Exception as e:
        print(f"✗ Error verifying database: {e}")
        return False


def get_table_stats(db_path: str) -> None:
    """Get statistics about recovered tables."""
    print("\nDatabase Statistics:")
    print("-" * 60)

    try:
        conn = sqlite3.connect(db_path)
        cursor = conn.cursor()

        # Get all tables
        cursor.execute("""
            SELECT name FROM sqlite_master
            WHERE type='table' AND name NOT LIKE 'sqlite_%'
            ORDER BY name;
        """)
        tables = cursor.fetchall()

        for (table_name,) in tables:
            cursor.execute(f"SELECT COUNT(*) FROM {table_name}")
            count = cursor.fetchone()[0]
            print(f"  {table_name:30} {count:>10} rows")

        conn.close()

    except Exception as e:
        print(f"Error getting statistics: {e}")


def main() -> None:
    """Main recovery process."""
    db_path = "amo_data.db"

    print("=" * 60)
    print("SQLite Database Recovery Tool")
    print("=" * 60)
    print()

    if not Path(db_path).exists():
        print(f"✗ Database not found: {db_path}")
        sys.exit(1)

    # Step 1: Backup corrupted database
    backup_path = backup_corrupted_db(db_path)

    # Step 2: Attempt to dump database
    dump_path = f"{db_path}.recovered.sql"
    if not dump_database(db_path, dump_path):
        print("\n⚠ Could not recover database using .recover mode")
        print("Manual recovery may be needed.")
        sys.exit(1)

    # Step 3: Recreate database from dump
    if not recreate_database(db_path, dump_path):
        print("\n✗ Failed to recreate database")
        print(f"Restoring backup from: {backup_path}")
        shutil.copy2(backup_path, db_path)
        sys.exit(1)

    # Step 4: Verify new database
    verify_database(db_path)

    # Step 5: Show statistics
    get_table_stats(db_path)

    print("\n" + "=" * 60)
    print("Recovery Complete!")
    print("=" * 60)
    print(f"✓ Corrupted database backed up to: {backup_path}")
    print(f"✓ Recovery SQL saved to: {dump_path}")
    print(f"✓ New database created at: {db_path}")
    print("\nNext steps:")
    print("1. Test the application with the recovered database")
    print("2. If data is missing, check if you can re-import from AMO CRM")
    print("3. Keep the backup and SQL dump for reference")
    print()


if __name__ == "__main__":
    main()