309 lines
10 KiB
Python
309 lines
10 KiB
Python
#!/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:
|
|
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)
|