""" config.py — Application configuration and DB connection manager. Update DB_CONFIG with your remote MySQL server credentials. """ import mysql.connector from mysql.connector import pooling import logging # ─── Logging Setup ──────────────────────────────────────────────────────────── import sys _file_handler = logging.FileHandler("app.log", encoding="utf-8") _stream_handler = logging.StreamHandler(stream=sys.stdout) # Force UTF-8 on the stream so Windows cp1252 consoles don't choke on # Unicode characters (arrows, em-dashes, etc.) in log messages. if hasattr(_stream_handler.stream, "reconfigure"): try: _stream_handler.stream.reconfigure(encoding="utf-8", errors="replace") except Exception: pass _formatter = logging.Formatter( "%(asctime)s [%(levelname)s] %(name)s - %(message)s" ) _file_handler.setFormatter(_formatter) _stream_handler.setFormatter(_formatter) logging.basicConfig( level=logging.INFO, handlers=[_file_handler, _stream_handler], ) logger = logging.getLogger("config") # ─── Config File Load / Save ────────────────────────────────────────────────── import configparser as _cp import os as _os CONFIG_FILE = "config.ini" APP_TITLE = "Website Checker" APP_VERSION = "1.0.0" def load_config() -> dict: """ Load DB settings from config.ini. Returns a dict with keys: host, port, database, user, password. Returns empty dict if the file does not exist or is incomplete. """ cfg = _cp.ConfigParser() if not _os.path.exists(CONFIG_FILE): return {} cfg.read(CONFIG_FILE, encoding="utf-8") if "database" not in cfg: return {} section = cfg["database"] return { "host": section.get("host", ""), "port": section.getint("port", 3306), "database": section.get("database", ""), "user": section.get("user", ""), "password": section.get("password", ""), } def save_config(host: str, port: int, database: str, user: str, password: str): """Persist DB connection settings to config.ini.""" cfg = _cp.ConfigParser() cfg["database"] = { "host": host, "port": str(port), "database": database, "user": user, "password": password, } with open(CONFIG_FILE, "w", encoding="utf-8") as fh: cfg.write(fh) logger.info(f"Configuration saved to {CONFIG_FILE}.") def config_exists() -> bool: """Return True if config.ini contains all required connection fields.""" ini = load_config() return bool(ini.get("host") and ini.get("database") and ini.get("user")) def reload_db_config(): """ Re-read config.ini and update DB_CONFIG in place. Also resets the connection pool so the next get_connection() uses new creds. """ global _pool, DB_CONFIG ini = load_config() DB_CONFIG.update({ "host": ini.get("host", DB_CONFIG["host"]), "port": ini.get("port", DB_CONFIG["port"]), "database": ini.get("database", DB_CONFIG["database"]), "user": ini.get("user", DB_CONFIG["user"]), "password": ini.get("password", DB_CONFIG["password"]), }) _pool = None # force pool recreation on next connection logger.info("DB_CONFIG reloaded from config.ini.") # ─── Database Configuration ─────────────────────────────────────────────────── # Populated from config.ini at runtime; falls back to placeholder strings so # the module is importable even before first-run setup has completed. _ini = load_config() DB_CONFIG = { "host": _ini.get("host", "your-mysql-host"), "port": _ini.get("port", 3306), "database": _ini.get("database", "website_checker"), "user": _ini.get("user", "your-db-user"), "password": _ini.get("password", "your-db-password"), "connection_timeout": 10, } # ─── Connection Pool ────────────────────────────────────────────────────────── _pool = None def get_connection_pool(): global _pool if _pool is None: try: _pool = pooling.MySQLConnectionPool( pool_name="app_pool", pool_size=5, **DB_CONFIG ) logger.info("Database connection pool initialised.") except mysql.connector.Error as e: logger.error(f"Failed to create connection pool: {e}") raise return _pool def get_connection(): """Return a connection from the pool.""" return get_connection_pool().get_connection() def initialize_database(): """Create all required tables if they do not exist.""" ddl_statements = [ """ CREATE TABLE IF NOT EXISTS users ( id INT AUTO_INCREMENT PRIMARY KEY, username VARCHAR(100) NOT NULL UNIQUE, password VARCHAR(255) NOT NULL, role ENUM('admin','user') NOT NULL DEFAULT 'user', full_name VARCHAR(200), is_active TINYINT(1) NOT NULL DEFAULT 1, created_at DATETIME DEFAULT CURRENT_TIMESTAMP, updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; """, """ CREATE TABLE IF NOT EXISTS websites ( id INT AUTO_INCREMENT PRIMARY KEY, name VARCHAR(200) NOT NULL, url TEXT NOT NULL, check_type ENUM('daily','weekly') NOT NULL DEFAULT 'daily', note TEXT, is_active TINYINT(1) NOT NULL DEFAULT 1, created_by INT, created_at DATETIME DEFAULT CURRENT_TIMESTAMP, updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, FOREIGN KEY (created_by) REFERENCES users(id) ON DELETE SET NULL ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; """, """ CREATE TABLE IF NOT EXISTS website_credentials ( id INT AUTO_INCREMENT PRIMARY KEY, website_id INT NOT NULL, username VARCHAR(200) NOT NULL, password VARCHAR(255) NOT NULL, label VARCHAR(100), FOREIGN KEY (website_id) REFERENCES websites(id) ON DELETE CASCADE ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; """, """ CREATE TABLE IF NOT EXISTS shift_checks ( id INT AUTO_INCREMENT PRIMARY KEY, website_id INT NOT NULL, user_id INT NOT NULL, checked_at DATETIME DEFAULT CURRENT_TIMESTAMP, user_note TEXT, FOREIGN KEY (website_id) REFERENCES websites(id) ON DELETE CASCADE, FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; """, """ CREATE TABLE IF NOT EXISTS activity_log ( id INT AUTO_INCREMENT PRIMARY KEY, user_id INT, action VARCHAR(100) NOT NULL, entity VARCHAR(100), entity_id INT, detail TEXT, logged_at DATETIME DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE SET NULL ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; """, """ CREATE TABLE IF NOT EXISTS shifts ( id INT AUTO_INCREMENT PRIMARY KEY, name VARCHAR(200) NOT NULL, days_of_week VARCHAR(20) NOT NULL DEFAULT '1234567', start_time TIME NOT NULL DEFAULT '00:00:00', end_time TIME NOT NULL DEFAULT '23:59:59', is_active TINYINT(1) NOT NULL DEFAULT 1, note TEXT, created_by INT, created_at DATETIME DEFAULT CURRENT_TIMESTAMP, updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, FOREIGN KEY (created_by) REFERENCES users(id) ON DELETE SET NULL ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; """, """ CREATE TABLE IF NOT EXISTS shift_users ( shift_id INT NOT NULL, user_id INT NOT NULL, PRIMARY KEY (shift_id, user_id), FOREIGN KEY (shift_id) REFERENCES shifts(id) ON DELETE CASCADE, FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; """, """ CREATE TABLE IF NOT EXISTS shift_websites ( shift_id INT NOT NULL, website_id INT NOT NULL, sort_order INT NOT NULL DEFAULT 0, PRIMARY KEY (shift_id, website_id), FOREIGN KEY (shift_id) REFERENCES shifts(id) ON DELETE CASCADE, FOREIGN KEY (website_id) REFERENCES websites(id) ON DELETE CASCADE ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; """, """ CREATE TABLE IF NOT EXISTS login_attempts ( id INT AUTO_INCREMENT PRIMARY KEY, username VARCHAR(100) NOT NULL, attempted_at DATETIME DEFAULT CURRENT_TIMESTAMP, ip_address VARCHAR(45), INDEX idx_username_time (username, attempted_at) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; """, """ CREATE TABLE IF NOT EXISTS website_users ( website_id INT NOT NULL, user_id INT NOT NULL, PRIMARY KEY (website_id, user_id), FOREIGN KEY (website_id) REFERENCES websites(id) ON DELETE CASCADE, FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; """, ] conn = None try: conn = get_connection() cursor = conn.cursor() for stmt in ddl_statements: cursor.execute(stmt) conn.commit() logger.info("Database schema initialised successfully.") # ── Safe migrations for existing deployments ─────────────────────── # Add check_type column to websites if it doesn't exist yet cursor.execute( """ SELECT COUNT(*) FROM information_schema.COLUMNS WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'websites' AND COLUMN_NAME = 'check_type' """ ) (has_col,) = cursor.fetchone() if not has_col: cursor.execute( "ALTER TABLE websites ADD COLUMN check_type ENUM('daily','weekly') " "NOT NULL DEFAULT 'daily' AFTER url" ) conn.commit() logger.info("Migration: added check_type column to websites table.") # Add failed_attempts column to users if it doesn't exist cursor.execute( """ SELECT COUNT(*) FROM information_schema.COLUMNS WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'users' AND COLUMN_NAME = 'failed_attempts' """ ) (has_fa,) = cursor.fetchone() if not has_fa: cursor.execute( "ALTER TABLE users " "ADD COLUMN failed_attempts TINYINT UNSIGNED NOT NULL DEFAULT 0 AFTER is_active, " "ADD COLUMN locked_until DATETIME NULL AFTER failed_attempts" ) conn.commit() logger.info("Migration: added failed_attempts and locked_until columns to users table.") # Add visibility column to websites if it doesn't exist cursor.execute( """ SELECT COUNT(*) FROM information_schema.COLUMNS WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'websites' AND COLUMN_NAME = 'visibility' """ ) (has_vis,) = cursor.fetchone() if not has_vis: cursor.execute( "ALTER TABLE websites ADD COLUMN visibility " "ENUM('all','assigned') NOT NULL DEFAULT 'all' AFTER check_type" ) conn.commit() logger.info("Migration: added visibility column to websites table.") # Seed default admin if users table is empty cursor.execute("SELECT COUNT(*) FROM users") (count,) = cursor.fetchone() if count == 0: import bcrypt as _bcrypt default_pw = _bcrypt.hashpw(b"admin123", _bcrypt.gensalt(rounds=12)).decode("utf-8") cursor.execute( "INSERT INTO users (username, password, role, full_name) VALUES (%s,%s,'admin','System Admin')", ("admin", default_pw) ) conn.commit() logger.info("Default admin account seeded (username: admin / password: admin123).") cursor.close() except mysql.connector.Error as e: logger.error(f"Database initialisation error: {e}") raise finally: if conn: conn.close()