8a1d15911f
Version control Claude Code configuration including: - Global instructions (CLAUDE.md) - User settings (settings.json) - Custom agents (architect, designer, engineer, etc.) - Custom skills (create-skill templates and workflows) Excludes session data, secrets, cache, and temporary files per .gitignore. Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
265 lines
9.3 KiB
Python
Executable File
265 lines
9.3 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
"""
|
||
Database Validation Script
|
||
|
||
Validates Major Domo database integrity and checks for common issues.
|
||
|
||
Usage:
|
||
python validate_database.py --env prod
|
||
python validate_database.py --env dev --verbose
|
||
"""
|
||
|
||
import sys
|
||
import os
|
||
import argparse
|
||
from typing import List, Dict, Tuple
|
||
|
||
# Add parent directory to path for imports
|
||
sys.path.insert(0, os.path.dirname(os.path.dirname(__file__)))
|
||
from api_client import MajorDomoAPI
|
||
|
||
|
||
class DatabaseValidator:
|
||
"""Database validation checks"""
|
||
|
||
def __init__(self, api: MajorDomoAPI):
|
||
self.api = api
|
||
self.issues: List[str] = []
|
||
self.warnings: List[str] = []
|
||
|
||
def add_issue(self, message: str):
|
||
"""Add a critical issue"""
|
||
self.issues.append(f"❌ {message}")
|
||
|
||
def add_warning(self, message: str):
|
||
"""Add a warning"""
|
||
self.warnings.append(f"⚠️ {message}")
|
||
|
||
def check_current_status(self):
|
||
"""Validate current season/week status"""
|
||
print("\n📋 Checking current season/week status...")
|
||
|
||
try:
|
||
current = self.api.get_current()
|
||
|
||
# Basic validation
|
||
if current['season'] < 1:
|
||
self.add_issue("Invalid season number")
|
||
if current['week'] < 0 or current['week'] > 18:
|
||
self.add_warning(f"Week {current['week']} outside normal range (0-18)")
|
||
|
||
# Trade deadline validation
|
||
if current['trade_deadline'] > 18:
|
||
self.add_warning(f"Trade deadline week {current['trade_deadline']} beyond season end")
|
||
|
||
# Playoffs validation
|
||
if current['playoffs_begin'] < current['trade_deadline']:
|
||
self.add_warning("Playoffs begin before trade deadline")
|
||
|
||
print(f" ✓ Season {current['season']}, Week {current['week']}")
|
||
print(f" ✓ Trade Deadline: Week {current['trade_deadline']}")
|
||
print(f" ✓ Playoffs Begin: Week {current['playoffs_begin']}")
|
||
|
||
except Exception as e:
|
||
self.add_issue(f"Failed to retrieve current status: {e}")
|
||
|
||
def check_teams(self, season: int):
|
||
"""Validate team data"""
|
||
print(f"\n👥 Checking teams for season {season}...")
|
||
|
||
try:
|
||
teams = self.api.list_teams(season=season)
|
||
|
||
if not teams:
|
||
self.add_issue(f"No teams found for season {season}")
|
||
return
|
||
|
||
# Check for duplicate abbreviations
|
||
abbrevs = [t['abbrev'] for t in teams]
|
||
duplicates = set([x for x in abbrevs if abbrevs.count(x) > 1])
|
||
if duplicates:
|
||
self.add_issue(f"Duplicate team abbreviations: {duplicates}")
|
||
|
||
# Check for teams without managers
|
||
no_manager = [t for t in teams if not t.get('manager1') and not t.get('manager2')]
|
||
if no_manager:
|
||
self.add_warning(f"{len(no_manager)} teams without managers")
|
||
|
||
# Check active teams
|
||
active_teams = [t for t in teams if not t['abbrev'].endswith('IL') and not t['abbrev'].endswith('MiL')]
|
||
|
||
print(f" ✓ Total teams: {len(teams)}")
|
||
print(f" ✓ Active teams: {len(active_teams)}")
|
||
print(f" ✓ IL/MiL teams: {len(teams) - len(active_teams)}")
|
||
|
||
if no_manager:
|
||
print(f" ⚠️ Teams without managers: {len(no_manager)}")
|
||
|
||
except Exception as e:
|
||
self.add_issue(f"Failed to retrieve teams: {e}")
|
||
|
||
def check_players(self, season: int):
|
||
"""Validate player data"""
|
||
print(f"\n⚾ Checking players for season {season}...")
|
||
|
||
try:
|
||
# Get sample of players
|
||
players = self.api.list_players(season=season, short_output=True)
|
||
|
||
if not players:
|
||
self.add_issue(f"No players found for season {season}")
|
||
return
|
||
|
||
# Check for players without positions
|
||
no_position = [p for p in players if not p.get('pos_1')]
|
||
if no_position:
|
||
self.add_warning(f"{len(no_position)} players without positions")
|
||
|
||
# Check for players with invalid WARA
|
||
invalid_wara = [p for p in players if p.get('wara', 0) < 0]
|
||
if invalid_wara:
|
||
self.add_warning(f"{len(invalid_wara)} players with negative WARA")
|
||
|
||
# Check injured players
|
||
injured = self.api.list_players(season=season, is_injured=True, short_output=True)
|
||
|
||
print(f" ✓ Total players: {len(players)}")
|
||
print(f" ✓ Injured players: {len(injured)}")
|
||
|
||
if no_position:
|
||
print(f" ⚠️ Players without positions: {len(no_position)}")
|
||
if invalid_wara:
|
||
print(f" ⚠️ Players with negative WARA: {len(invalid_wara)}")
|
||
|
||
except Exception as e:
|
||
self.add_issue(f"Failed to retrieve players: {e}")
|
||
|
||
def check_standings(self, season: int):
|
||
"""Validate standings data"""
|
||
print(f"\n🏆 Checking standings for season {season}...")
|
||
|
||
try:
|
||
standings = self.api.get_standings(season=season)
|
||
|
||
if not standings:
|
||
self.add_warning(f"No standings found for season {season}")
|
||
return
|
||
|
||
# Check for negative records
|
||
negative_records = [s for s in standings if s.get('wins', 0) < 0 or s.get('losses', 0) < 0]
|
||
if negative_records:
|
||
self.add_issue(f"{len(negative_records)} teams with negative records")
|
||
|
||
# Calculate total games
|
||
total_games = sum(s.get('wins', 0) + s.get('losses', 0) for s in standings)
|
||
avg_games = total_games / len(standings) if standings else 0
|
||
|
||
print(f" ✓ Teams in standings: {len(standings)}")
|
||
print(f" ✓ Average games played: {avg_games:.1f}")
|
||
|
||
if negative_records:
|
||
print(f" ❌ Teams with negative records: {len(negative_records)}")
|
||
|
||
except Exception as e:
|
||
self.add_issue(f"Failed to retrieve standings: {e}")
|
||
|
||
def check_transactions(self, season: int):
|
||
"""Validate transaction data"""
|
||
print(f"\n💼 Checking transactions for season {season}...")
|
||
|
||
try:
|
||
transactions = self.api.get_transactions(season=season, short_output=True)
|
||
|
||
if not transactions:
|
||
print(f" ℹ️ No transactions found for season {season}")
|
||
return
|
||
|
||
# Check for cancelled transactions
|
||
cancelled = [t for t in transactions if t.get('cancelled')]
|
||
frozen = [t for t in transactions if t.get('frozen')]
|
||
|
||
print(f" ✓ Total transactions: {len(transactions)}")
|
||
print(f" ℹ️ Cancelled: {len(cancelled)}")
|
||
print(f" ℹ️ Frozen: {len(frozen)}")
|
||
|
||
except Exception as e:
|
||
self.add_issue(f"Failed to retrieve transactions: {e}")
|
||
|
||
def run_all_checks(self, season: Optional[int] = None):
|
||
"""Run all validation checks"""
|
||
print(f"\n{'='*60}")
|
||
print(f"Major Domo Database Validation")
|
||
print(f"Environment: {self.api.env.upper()}")
|
||
print(f"{'='*60}")
|
||
|
||
# Get current season if not provided
|
||
if season is None:
|
||
try:
|
||
current = self.api.get_current()
|
||
season = current['season']
|
||
except Exception as e:
|
||
print(f"❌ Failed to get current season: {e}")
|
||
return False
|
||
|
||
# Run checks
|
||
self.check_current_status()
|
||
self.check_teams(season)
|
||
self.check_players(season)
|
||
self.check_standings(season)
|
||
self.check_transactions(season)
|
||
|
||
# Print summary
|
||
print(f"\n{'='*60}")
|
||
print("Validation Summary")
|
||
print(f"{'='*60}")
|
||
|
||
if self.issues:
|
||
print(f"\n❌ Critical Issues ({len(self.issues)}):")
|
||
for issue in self.issues:
|
||
print(f" {issue}")
|
||
|
||
if self.warnings:
|
||
print(f"\n⚠️ Warnings ({len(self.warnings)}):")
|
||
for warning in self.warnings:
|
||
print(f" {warning}")
|
||
|
||
if not self.issues and not self.warnings:
|
||
print("\n✅ All checks passed! Database is healthy.")
|
||
return True
|
||
elif not self.issues:
|
||
print(f"\n✅ No critical issues found. {len(self.warnings)} warnings to review.")
|
||
return True
|
||
else:
|
||
print(f"\n❌ Validation failed with {len(self.issues)} critical issues.")
|
||
return False
|
||
|
||
|
||
def main():
|
||
parser = argparse.ArgumentParser(description='Validate Major Domo database integrity')
|
||
parser.add_argument('--env', choices=['prod', 'dev'], default='prod', help='Environment')
|
||
parser.add_argument('--season', type=int, help='Season to validate (defaults to current)')
|
||
parser.add_argument('--verbose', action='store_true', help='Verbose output')
|
||
args = parser.parse_args()
|
||
|
||
try:
|
||
# Initialize API client
|
||
api = MajorDomoAPI(environment=args.env, verbose=args.verbose)
|
||
|
||
# Run validation
|
||
validator = DatabaseValidator(api)
|
||
success = validator.run_all_checks(season=args.season)
|
||
|
||
# Exit with appropriate code
|
||
sys.exit(0 if success else 1)
|
||
|
||
except Exception as e:
|
||
print(f"\n❌ Validation failed: {e}")
|
||
if args.verbose:
|
||
import traceback
|
||
traceback.print_exc()
|
||
sys.exit(1)
|
||
|
||
|
||
if __name__ == '__main__':
|
||
main()
|