strat-gameplay-webapp/backend/clean_test_data.py

#!/usr/bin/env python3
"""
Database Cleanup Script

Clear out test data from database tables.
Use with caution - this will DELETE data!

Usage:
    python clean_test_data.py              # Interactive mode
    python clean_test_data.py --all        # Delete all data
    python clean_test_data.py --game <id>  # Delete specific game
"""

import asyncio
import logging
import sys
from uuid import UUID
from sqlalchemy import select, delete
from sqlalchemy.ext.asyncio import AsyncSession

from app.database.session import AsyncSessionLocal
from app.models.db_models import (
    Game,
    Play,
    Lineup,
    GameSession,
    RosterLink,
    GameCardsetLink
)

# Setup logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)


class DatabaseCleaner:
    """Clean test data from database."""

    async def count_records(self, session: AsyncSession) -> dict:
        """Count records in each table."""
        counts = {}

        # Count games
        result = await session.execute(select(Game))
        counts['games'] = len(list(result.scalars().all()))

        # Count plays
        result = await session.execute(select(Play))
        counts['plays'] = len(list(result.scalars().all()))

        # Count lineups
        result = await session.execute(select(Lineup))
        counts['lineups'] = len(list(result.scalars().all()))

        # Count roster links
        result = await session.execute(select(RosterLink))
        counts['roster_links'] = len(list(result.scalars().all()))

        # Count game sessions
        result = await session.execute(select(GameSession))
        counts['game_sessions'] = len(list(result.scalars().all()))

        # Count cardset links
        result = await session.execute(select(GameCardsetLink))
        counts['game_cardset_links'] = len(list(result.scalars().all()))

        return counts

    async def delete_game(self, game_id: UUID) -> None:
        """
        Delete a specific game and all related data.

        Cascade deletes will automatically remove:
        - Plays
        - Lineups
        - RosterLinks
        - GameSession
        - GameCardsetLinks
        """
        async with AsyncSessionLocal() as session:
            try:
                # Check if game exists
                result = await session.execute(
                    select(Game).where(Game.id == game_id)
                )
                game = result.scalar_one_or_none()

                if not game:
                    logger.warning(f"Game {game_id} not found")
                    return

                logger.info(f"Deleting game {game_id} ({game.league_id})...")

                # Delete game (cascade will handle related records)
                await session.delete(game)
                await session.commit()

                logger.info(f"✅ Successfully deleted game {game_id}")

            except Exception as e:
                await session.rollback()
                logger.error(f"❌ Error deleting game: {e}", exc_info=True)
                raise

    async def delete_all(self, confirm: bool = False) -> None:
        """
        Delete ALL data from all tables.

        WARNING: This is destructive and cannot be undone!
        """
        if not confirm:
            logger.warning("⚠️  delete_all() requires confirm=True parameter")
            return

        async with AsyncSessionLocal() as session:
            try:
                logger.info("🗑️  Deleting ALL data from database...")

                # Get counts before deletion
                before_counts = await self.count_records(session)
                logger.info(f"Current record counts: {before_counts}")

                # Delete in correct order (children first)
                logger.info("Deleting plays...")
                await session.execute(delete(Play))

                logger.info("Deleting lineups...")
                await session.execute(delete(Lineup))

                logger.info("Deleting roster links...")
                await session.execute(delete(RosterLink))

                logger.info("Deleting game sessions...")
                await session.execute(delete(GameSession))

                logger.info("Deleting game cardset links...")
                await session.execute(delete(GameCardsetLink))

                logger.info("Deleting games...")
                await session.execute(delete(Game))

                await session.commit()

                # Verify deletion
                after_counts = await self.count_records(session)
                logger.info(f"After deletion: {after_counts}")

                logger.info("✅ Successfully deleted all data")

            except Exception as e:
                await session.rollback()
                logger.error(f"❌ Error deleting data: {e}", exc_info=True)
                raise

    async def list_games(self) -> None:
        """List all games in the database."""
        async with AsyncSessionLocal() as session:
            result = await session.execute(
                select(Game).order_by(Game.created_at.desc())
            )
            games = result.scalars().all()

            if not games:
                logger.info("No games found in database")
                return

            logger.info(f"\n{'='*80}")
            logger.info(f"Found {len(games)} game(s) in database:")
            logger.info(f"{'='*80}")

            for game in games:
                logger.info(f"\nGame ID: {game.id}")
                logger.info(f"  League: {game.league_id}")
                logger.info(f"  Status: {game.status}")
                logger.info(f"  Mode: {game.game_mode}")
                logger.info(f"  Teams: {game.away_team_id} @ {game.home_team_id}")
                logger.info(f"  Score: {game.away_score}-{game.home_score}")
                if game.current_inning: # type: ignore
                    logger.info(f"  State: {game.current_half} {game.current_inning}")
                logger.info(f"  AI: Home={game.home_team_is_ai}, Away={game.away_team_is_ai}")
                logger.info(f"  Created: {game.created_at}")

    async def show_stats(self) -> None:
        """Show database statistics."""
        async with AsyncSessionLocal() as session:
            counts = await self.count_records(session)

            logger.info(f"\n{'='*80}")
            logger.info("DATABASE STATISTICS")
            logger.info(f"{'='*80}")
            logger.info(f"Games:            {counts['games']:>6}")
            logger.info(f"Plays:            {counts['plays']:>6}")
            logger.info(f"Lineups:          {counts['lineups']:>6}")
            logger.info(f"Roster Links:     {counts['roster_links']:>6}")
            logger.info(f"Game Sessions:    {counts['game_sessions']:>6}")
            logger.info(f"Cardset Links:    {counts['game_cardset_links']:>6}")
            logger.info(f"{'='*80}\n")


async def interactive_menu():
    """Interactive menu for cleanup operations."""
    cleaner = DatabaseCleaner()

    while True:
        print("\n" + "="*60)
        print("DATABASE CLEANUP MENU")
        print("="*60)
        print("1. Show database statistics")
        print("2. List all games")
        print("3. Delete specific game")
        print("4. Delete ALL data (⚠️  DANGER)")
        print("5. Exit")
        print("="*60)

        choice = input("\nEnter your choice (1-5): ").strip()

        if choice == "1":
            await cleaner.show_stats()

        elif choice == "2":
            await cleaner.list_games()

        elif choice == "3":
            game_id_str = input("Enter game UUID: ").strip()
            try:
                game_id = UUID(game_id_str)
                confirm = input(f"Delete game {game_id}? (yes/no): ").strip().lower()
                if confirm == "yes":
                    await cleaner.delete_game(game_id)
                else:
                    logger.info("Cancelled")
            except ValueError:
                logger.error("Invalid UUID format")

        elif choice == "4":
            await cleaner.show_stats()
            print("\n⚠️  WARNING: This will DELETE ALL DATA from the database!")
            confirm1 = input("Are you absolutely sure? (type 'DELETE ALL'): ").strip()
            if confirm1 == "DELETE ALL":
                confirm2 = input("Type 'yes' to confirm: ").strip().lower()
                if confirm2 == "yes":
                    await cleaner.delete_all(confirm=True)
                else:
                    logger.info("Cancelled")
            else:
                logger.info("Cancelled")

        elif choice == "5":
            logger.info("Exiting...")
            break

        else:
            logger.warning("Invalid choice")


async def main():
    """Main entry point."""
    cleaner = DatabaseCleaner()

    # Parse command line arguments
    if len(sys.argv) == 1:
        # No arguments - run interactive mode
        await interactive_menu()

    elif "--help" in sys.argv or "-h" in sys.argv:
        print(__doc__)

    elif "--stats" in sys.argv:
        await cleaner.show_stats()

    elif "--list" in sys.argv:
        await cleaner.list_games()

    elif "--all" in sys.argv:
        await cleaner.show_stats()
        print("\n⚠️  WARNING: This will DELETE ALL DATA!")
        confirm = input("Type 'yes' to confirm: ").strip().lower()
        if confirm == "yes":
            await cleaner.delete_all(confirm=True)
        else:
            logger.info("Cancelled")

    elif "--game" in sys.argv:
        idx = sys.argv.index("--game")
        if idx + 1 < len(sys.argv):
            try:
                game_id = UUID(sys.argv[idx + 1])
                await cleaner.delete_game(game_id)
            except ValueError:
                logger.error("Invalid UUID format")
        else:
            logger.error("--game requires a UUID argument")

    else:
        print("Unknown arguments. Use --help for usage information.")


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        logger.info("\nInterrupted by user")
    except Exception as e:
        logger.error(f"Error: {e}", exc_info=True)
        sys.exit(1)