# encoding=utf-8 """ Database setup and configuration management module. Handles database initialization, configuration saving/loading, and sample data creation. """ import os import json import configparser from datetime import datetime from sqlalchemy import create_engine, text from sqlalchemy.exc import SQLAlchemyError from database import Base, db_config, init_database from database import ( Player, Club, Team, MatchSquad, HockeyFixture, AdminSettings, MotmVote, MatchComment, HockeyUser ) class DatabaseConfigManager: """Manages database configuration and setup.""" def __init__(self, config_file='database_config.ini'): # Use absolute path to ensure we save in the right location if not os.path.isabs(config_file): config_file = os.path.join(os.path.dirname(__file__), config_file) self.config_file = config_file self.config = configparser.ConfigParser() self.load_config() def load_config(self): """Load configuration from file.""" if os.path.exists(self.config_file): self.config.read(self.config_file) else: # Create default configuration self.config['DATABASE'] = { 'type': 'sqlite', 'sqlite_database_path': 'hockey_results.db' } self.config['MYSQL'] = { 'host': 'localhost', 'port': '3306', 'database': 'hockey_results', 'username': 'root', 'password': '', 'charset': 'utf8mb4' } self.config['POSTGRESQL'] = { 'host': 'localhost', 'port': '5432', 'database': 'hockey_results', 'username': 'postgres', 'password': '' } # Save the default configuration to file self._save_config_file() def save_config(self, form_data): """Save configuration from form data.""" # Update database type self.config['DATABASE']['type'] = form_data['database_type'] # Update SQLite settings if 'sqlite_database_path' in form_data: self.config['DATABASE']['sqlite_database_path'] = form_data['sqlite_database_path'] # Update MySQL settings if 'mysql_host' in form_data and form_data['mysql_host'] is not None: self.config['MYSQL']['host'] = str(form_data['mysql_host']) if 'mysql_port' in form_data and form_data['mysql_port'] is not None: self.config['MYSQL']['port'] = str(form_data['mysql_port']) if 'mysql_database' in form_data and form_data['mysql_database'] is not None: self.config['MYSQL']['database'] = str(form_data['mysql_database']) if 'mysql_username' in form_data and form_data['mysql_username'] is not None: self.config['MYSQL']['username'] = str(form_data['mysql_username']) if 'mysql_password' in form_data and form_data['mysql_password'] is not None: self.config['MYSQL']['password'] = str(form_data['mysql_password']) if 'mysql_charset' in form_data and form_data['mysql_charset'] is not None: self.config['MYSQL']['charset'] = str(form_data['mysql_charset']) # Update PostgreSQL settings if 'postgres_host' in form_data and form_data['postgres_host'] is not None: self.config['POSTGRESQL']['host'] = str(form_data['postgres_host']) if 'postgres_port' in form_data and form_data['postgres_port'] is not None: self.config['POSTGRESQL']['port'] = str(form_data['postgres_port']) if 'postgres_database' in form_data and form_data['postgres_database'] is not None: self.config['POSTGRESQL']['database'] = str(form_data['postgres_database']) if 'postgres_username' in form_data and form_data['postgres_username'] is not None: self.config['POSTGRESQL']['username'] = str(form_data['postgres_username']) if 'postgres_password' in form_data and form_data['postgres_password'] is not None: self.config['POSTGRESQL']['password'] = str(form_data['postgres_password']) # Save to file self._save_config_file() # Update environment variables self._update_environment_variables() def _save_config_file(self): """Save configuration to file.""" with open(self.config_file, 'w') as f: self.config.write(f) def _update_environment_variables(self): """Update environment variables based on configuration.""" db_type = self.config['DATABASE']['type'] if db_type == 'sqlite': os.environ['DATABASE_TYPE'] = 'sqlite' os.environ['SQLITE_DATABASE_PATH'] = self.config['DATABASE']['sqlite_database_path'] elif db_type == 'mysql': os.environ['DATABASE_TYPE'] = 'mysql' os.environ['MYSQL_HOST'] = self.config['MYSQL']['host'] os.environ['MYSQL_PORT'] = self.config['MYSQL']['port'] os.environ['MYSQL_DATABASE'] = self.config['MYSQL']['database'] os.environ['MYSQL_USER'] = self.config['MYSQL']['username'] os.environ['MYSQL_PASSWORD'] = self.config['MYSQL']['password'] os.environ['MYSQL_CHARSET'] = self.config['MYSQL']['charset'] elif db_type == 'postgresql': os.environ['DATABASE_TYPE'] = 'postgresql' os.environ['POSTGRES_HOST'] = self.config['POSTGRESQL']['host'] os.environ['POSTGRES_PORT'] = self.config['POSTGRESQL']['port'] os.environ['POSTGRES_DATABASE'] = self.config['POSTGRESQL']['database'] os.environ['POSTGRES_USER'] = self.config['POSTGRESQL']['username'] os.environ['POSTGRES_PASSWORD'] = self.config['POSTGRESQL']['password'] def test_connection(self, form_data): """Test database connection with provided settings.""" try: # Temporarily update environment variables old_env = {} for key in ['DATABASE_TYPE', 'SQLITE_DATABASE_PATH', 'MYSQL_HOST', 'MYSQL_PORT', 'MYSQL_DATABASE', 'MYSQL_USER', 'MYSQL_PASSWORD', 'MYSQL_CHARSET', 'POSTGRES_HOST', 'POSTGRES_PORT', 'POSTGRES_DATABASE', 'POSTGRES_USER', 'POSTGRES_PASSWORD']: old_env[key] = os.environ.get(key) # Set new environment variables os.environ['DATABASE_TYPE'] = form_data['database_type'] if form_data['database_type'] == 'sqlite': os.environ['SQLITE_DATABASE_PATH'] = form_data['sqlite_database_path'] elif form_data['database_type'] == 'mysql': os.environ['MYSQL_HOST'] = form_data['mysql_host'] os.environ['MYSQL_PORT'] = str(form_data['mysql_port']) os.environ['MYSQL_DATABASE'] = form_data['mysql_database'] os.environ['MYSQL_USER'] = form_data['mysql_username'] os.environ['MYSQL_PASSWORD'] = form_data['mysql_password'] os.environ['MYSQL_CHARSET'] = form_data['mysql_charset'] elif form_data['database_type'] == 'postgresql': os.environ['POSTGRES_HOST'] = form_data['postgres_host'] os.environ['POSTGRES_PORT'] = str(form_data['postgres_port']) os.environ['POSTGRES_DATABASE'] = form_data['postgres_database'] os.environ['POSTGRES_USER'] = form_data['postgres_username'] os.environ['POSTGRES_PASSWORD'] = form_data['postgres_password'] # Test connection from database import DatabaseConfig test_config = DatabaseConfig() engine = test_config.engine with engine.connect() as conn: result = conn.execute(text("SELECT 1")) result.fetchone() return True, "Connection successful!" except Exception as e: return False, f"Connection failed: {str(e)}" finally: # Restore original environment variables for key, value in old_env.items(): if value is None: os.environ.pop(key, None) else: os.environ[key] = value def initialize_database(self, create_sample_data=True): """Initialize database with tables and optionally sample data.""" try: # Create tables init_database() if create_sample_data: self._create_sample_data() return True, "Database initialized successfully!" except Exception as e: return False, f"Database initialization failed: {str(e)}" def _create_sample_data(self): """Create sample data for testing.""" from database import get_db_session session = get_db_session() try: # Create sample clubs (only if they don't exist) clubs_data = [ {'hockey_club': 'HKFC', 'logo_url': '/static/images/hkfc_logo.png'}, {'hockey_club': 'KCC', 'logo_url': '/static/images/kcc_logo.png'}, {'hockey_club': 'USRC', 'logo_url': '/static/images/usrc_logo.png'}, {'hockey_club': 'Valley', 'logo_url': '/static/images/valley_logo.png'}, ] for club_data in clubs_data: # Check if club already exists existing_club = session.query(Club).filter_by(hockey_club=club_data['hockey_club']).first() if not existing_club: club = Club(**club_data) session.add(club) # Create sample teams (only if they don't exist) teams_data = [ {'club': 'HKFC', 'team': 'A', 'display_name': 'HKFC A', 'league': 'Premier Division'}, {'club': 'HKFC', 'team': 'B', 'display_name': 'HKFC B', 'league': 'Division 1'}, {'club': 'HKFC', 'team': 'C', 'display_name': 'HKFC C', 'league': 'Division 2'}, ] for team_data in teams_data: # Check if team already exists existing_team = session.query(Team).filter_by(club=team_data['club'], team=team_data['team']).first() if not existing_team: team = Team(**team_data) session.add(team) # Create sample players (only if they don't exist) players_data = [ {'player_number': 1, 'player_forenames': 'John', 'player_surname': 'Smith', 'player_nickname': 'Smithers', 'player_team': 'HKFC C'}, {'player_number': 2, 'player_forenames': 'Mike', 'player_surname': 'Jones', 'player_nickname': 'Jonesy', 'player_team': 'HKFC C'}, {'player_number': 3, 'player_forenames': 'David', 'player_surname': 'Brown', 'player_nickname': 'Brownie', 'player_team': 'HKFC C'}, {'player_number': 4, 'player_forenames': 'Chris', 'player_surname': 'Wilson', 'player_nickname': 'Willy', 'player_team': 'HKFC C'}, {'player_number': 5, 'player_forenames': 'Tom', 'player_surname': 'Taylor', 'player_nickname': 'Tayls', 'player_team': 'HKFC C'}, ] for player_data in players_data: # Check if player already exists existing_player = session.query(Player).filter_by(player_number=player_data['player_number']).first() if not existing_player: player = Player(**player_data) session.add(player) # Create sample admin settings (only if they don't exist) existing_admin = session.query(AdminSettings).filter_by(userid='admin').first() if not existing_admin: admin_settings = AdminSettings( userid='admin', next_fixture=1, next_club='KCC', next_team='KCC A', curr_motm=1, curr_dotd=2, oppo_logo='/static/images/kcc_logo.png', hkfc_logo='/static/images/hkfc_logo.png', motm_url_suffix='abc123', prev_fixture=0 ) session.add(admin_settings) # Create sample fixtures (only if they don't exist) fixtures_data = [ {'fixture_number': 1, 'date': datetime(2025, 1, 15), 'home_team': 'HKFC C', 'away_team': 'KCC A', 'venue': 'HKFC'}, {'fixture_number': 2, 'date': datetime(2025, 1, 22), 'home_team': 'USRC A', 'away_team': 'HKFC C', 'venue': 'USRC'}, {'fixture_number': 3, 'date': datetime(2025, 1, 29), 'home_team': 'HKFC C', 'away_team': 'Valley A', 'venue': 'HKFC'}, ] for fixture_data in fixtures_data: # Check if fixture already exists existing_fixture = session.query(HockeyFixture).filter_by(fixture_number=fixture_data['fixture_number']).first() if not existing_fixture: fixture = HockeyFixture(**fixture_data) session.add(fixture) session.commit() except Exception as e: session.rollback() raise e finally: session.close() def get_config_dict(self): """Get configuration as dictionary for form population.""" config_dict = {} # Database type config_dict['database_type'] = self.config['DATABASE'].get('type', 'sqlite') config_dict['sqlite_database_path'] = self.config['DATABASE'].get('sqlite_database_path', 'hockey_results.db') # MySQL settings config_dict['mysql_host'] = self.config['MYSQL'].get('host', 'localhost') config_dict['mysql_port'] = int(self.config['MYSQL'].get('port', '3306')) config_dict['mysql_database'] = self.config['MYSQL'].get('database', 'hockey_results') config_dict['mysql_username'] = self.config['MYSQL'].get('username', 'root') config_dict['mysql_password'] = self.config['MYSQL'].get('password', '') config_dict['mysql_charset'] = self.config['MYSQL'].get('charset', 'utf8mb4') # PostgreSQL settings config_dict['postgres_host'] = self.config['POSTGRESQL'].get('host', 'localhost') config_dict['postgres_port'] = int(self.config['POSTGRESQL'].get('port', '5432')) config_dict['postgres_database'] = self.config['POSTGRESQL'].get('database', 'hockey_results') config_dict['postgres_username'] = self.config['POSTGRESQL'].get('username', 'postgres') config_dict['postgres_password'] = self.config['POSTGRESQL'].get('password', '') return config_dict # Global instance db_config_manager = DatabaseConfigManager()