# 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'): 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': '' } 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: self.config['MYSQL']['host'] = form_data['mysql_host'] if 'mysql_port' in form_data: self.config['MYSQL']['port'] = str(form_data['mysql_port']) if 'mysql_database' in form_data: self.config['MYSQL']['database'] = form_data['mysql_database'] if 'mysql_username' in form_data: self.config['MYSQL']['username'] = form_data['mysql_username'] if 'mysql_password' in form_data: self.config['MYSQL']['password'] = form_data['mysql_password'] if 'mysql_charset' in form_data: self.config['MYSQL']['charset'] = form_data['mysql_charset'] # Update PostgreSQL settings if 'postgres_host' in form_data: self.config['POSTGRESQL']['host'] = form_data['postgres_host'] if 'postgres_port' in form_data: self.config['POSTGRESQL']['port'] = str(form_data['postgres_port']) if 'postgres_database' in form_data: self.config['POSTGRESQL']['database'] = form_data['postgres_database'] if 'postgres_username' in form_data: self.config['POSTGRESQL']['username'] = form_data['postgres_username'] if 'postgres_password' in form_data: self.config['POSTGRESQL']['password'] = form_data['postgres_password'] # Save to file with open(self.config_file, 'w') as f: self.config.write(f) # Update environment variables self._update_environment_variables() 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 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: club = Club(**club_data) session.add(club) # Create sample teams 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'}, {'club': 'HKFC', 'team': 'D', 'display_name': 'HKFC D', 'league': 'Division 3'}, ] for team_data in teams_data: team = Team(**team_data) session.add(team) # Create sample players players_data = [ {'player_number': 1, 'player_forenames': 'John', 'player_surname': 'Smith', 'player_nickname': 'Smithers', 'player_team': 'HKFC D'}, {'player_number': 2, 'player_forenames': 'Mike', 'player_surname': 'Jones', 'player_nickname': 'Jonesy', 'player_team': 'HKFC D'}, {'player_number': 3, 'player_forenames': 'David', 'player_surname': 'Brown', 'player_nickname': 'Brownie', 'player_team': 'HKFC D'}, {'player_number': 4, 'player_forenames': 'Chris', 'player_surname': 'Wilson', 'player_nickname': 'Willy', 'player_team': 'HKFC D'}, {'player_number': 5, 'player_forenames': 'Tom', 'player_surname': 'Taylor', 'player_nickname': 'Tayls', 'player_team': 'HKFC D'}, ] for player_data in players_data: player = Player(**player_data) session.add(player) # Create sample admin settings 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 fixtures_data = [ {'fixture_number': 1, 'date': datetime(2024, 1, 15), 'home_team': 'HKFC D', 'away_team': 'KCC A', 'venue': 'HKFC'}, {'fixture_number': 2, 'date': datetime(2024, 1, 22), 'home_team': 'USRC A', 'away_team': 'HKFC D', 'venue': 'USRC'}, {'fixture_number': 3, 'date': datetime(2024, 1, 29), 'home_team': 'HKFC D', 'away_team': 'Valley A', 'venue': 'HKFC'}, ] for fixture_data in fixtures_data: 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()