import sqlite3 import os DB_PATH = os.environ.get("LIC_DB_PATH", "recruits.db") def get_db_connection() -> sqlite3.Connection: """ Establish and return a connection to the SQLite database. Enforces row_factory to sqlite3.Row for dictionary-like row querying. """ conn = sqlite3.connect(DB_PATH) conn.row_factory = sqlite3.Row # Enable foreign keys conn.execute("PRAGMA foreign_keys = ON;") return conn def init_db() -> None: """ Initialize the database, creating the necessary tables if they do not exist. """ conn = get_db_connection() cursor = conn.cursor() # Create users table cursor.execute(""" CREATE TABLE IF NOT EXISTS users ( id INTEGER PRIMARY KEY AUTOINCREMENT, username TEXT UNIQUE NOT NULL, password_hash TEXT NOT NULL, role TEXT NOT NULL DEFAULT 'recruiter', rvp TEXT, email TEXT, phone TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); """) # Create sessions table cursor.execute(""" CREATE TABLE IF NOT EXISTS sessions ( session_id TEXT PRIMARY KEY, user_id INTEGER NOT NULL, csrf_token TEXT NOT NULL, expires_at TIMESTAMP NOT NULL, FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE ); """) # Create recruits table cursor.execute(""" CREATE TABLE IF NOT EXISTS recruits ( id INTEGER PRIMARY KEY AUTOINCREMENT, first_name TEXT NOT NULL, last_name TEXT NOT NULL, email TEXT, phone TEXT, iba_date TEXT, iba_status TEXT DEFAULT 'Pending', pol_account_logged_in INTEGER DEFAULT 0, whatsapp_invite_sent INTEGER DEFAULT 0, llqp_registered INTEGER DEFAULT 0, llqp_class_date TEXT, w1_sat_attended INTEGER DEFAULT 0, w1_sun_attended INTEGER DEFAULT 0, w2_sat_attended INTEGER DEFAULT 0, w2_sun_attended INTEGER DEFAULT 0, qual_exam_status TEXT DEFAULT 'Not Taken', qual_exam_date TEXT, prov_exam_booked INTEGER DEFAULT 0, prov_exam_booking_date TEXT, prov_exam_date TEXT, prov_exam_status TEXT DEFAULT 'Not Taken', license_app_submitted INTEGER DEFAULT 0, license_app_date TEXT, license_app_status TEXT DEFAULT 'Pending', reimbursement_submitted INTEGER DEFAULT 0, reimbursement_date TEXT, reimbursement_amount REAL DEFAULT 0.0, reimbursement_status TEXT DEFAULT 'Pending', reimbursement_notes TEXT, rvp_invoice_status TEXT DEFAULT 'Pending', rvp_invoice_date TEXT, rvp_invoice_amount REAL DEFAULT 50.0, mf_course_registered INTEGER DEFAULT 0, mf_registration_date TEXT, mf_study_completed INTEGER DEFAULT 0, mf_review_1_attended INTEGER DEFAULT 0, mf_review_2_attended INTEGER DEFAULT 0, mf_review_3_attended INTEGER DEFAULT 0, mf_review_4_attended INTEGER DEFAULT 0, mf_review_5_attended INTEGER DEFAULT 0, mf_review_6_attended INTEGER DEFAULT 0, mf_exam_booked INTEGER DEFAULT 0, mf_exam_date TEXT, mf_exam_status TEXT DEFAULT 'Not Taken', mf_nrd_application_completed INTEGER DEFAULT 0, mf_roadd_completed INTEGER DEFAULT 0, mf_nrd_access_date TEXT, mf_nrd_form4_completed INTEGER DEFAULT 0, mf_pfsl10_completed INTEGER DEFAULT 0, mf_approved_person_agreement_completed INTEGER DEFAULT 0, mf_license_received INTEGER DEFAULT 0, mf_license_received_date TEXT, mf_supervision_completed INTEGER DEFAULT 0, mf_district_leader_at_signup INTEGER DEFAULT 0, mf_district_leader_date TEXT, mf_reimbursement_status TEXT DEFAULT 'Pending', solution_number TEXT, rvp TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); """) # Add columns introduced after the first release to existing databases. migrations = { 'solution_number': 'TEXT', 'rvp': 'TEXT', 'pol_account_logged_in': 'INTEGER DEFAULT 0', 'whatsapp_invite_sent': 'INTEGER DEFAULT 0', "rvp_invoice_status": "TEXT DEFAULT 'Pending'", 'rvp_invoice_date': 'TEXT', 'rvp_invoice_amount': 'REAL DEFAULT 50.0', 'mf_course_registered': 'INTEGER DEFAULT 0', 'mf_registration_date': 'TEXT', 'mf_study_completed': 'INTEGER DEFAULT 0', **{f'mf_review_{number}_attended': 'INTEGER DEFAULT 0' for number in range(1, 7)}, 'mf_exam_booked': 'INTEGER DEFAULT 0', 'mf_exam_date': 'TEXT', "mf_exam_status": "TEXT DEFAULT 'Not Taken'", 'mf_nrd_application_completed': 'INTEGER DEFAULT 0', 'mf_roadd_completed': 'INTEGER DEFAULT 0', 'mf_nrd_access_date': 'TEXT', 'mf_nrd_form4_completed': 'INTEGER DEFAULT 0', 'mf_pfsl10_completed': 'INTEGER DEFAULT 0', 'mf_approved_person_agreement_completed': 'INTEGER DEFAULT 0', 'mf_license_received': 'INTEGER DEFAULT 0', 'mf_license_received_date': 'TEXT', 'mf_supervision_completed': 'INTEGER DEFAULT 0', 'mf_district_leader_at_signup': 'INTEGER DEFAULT 0', 'mf_district_leader_date': 'TEXT', "mf_reimbursement_status": "TEXT DEFAULT 'Pending'", } for column, definition in migrations.items(): try: cursor.execute(f'ALTER TABLE recruits ADD COLUMN {column} {definition};') except sqlite3.OperationalError: pass for column, definition in {'rvp': 'TEXT', 'email': 'TEXT', 'phone': 'TEXT'}.items(): try: cursor.execute(f'ALTER TABLE users ADD COLUMN {column} {definition};') except sqlite3.OperationalError: pass # Create exam_history table for tracking qualification and provincial rewrites cursor.execute(""" CREATE TABLE IF NOT EXISTS exam_history ( id INTEGER PRIMARY KEY AUTOINCREMENT, recruit_id INTEGER NOT NULL, exam_type TEXT NOT NULL, -- 'Qualification' or 'Provincial' exam_date TEXT NOT NULL, result TEXT NOT NULL, -- 'Pass' or 'Fail' notes TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (recruit_id) REFERENCES recruits(id) ON DELETE CASCADE ); """) # Bootstrap default user if users table is empty cursor.execute("SELECT COUNT(*) FROM users;") if cursor.fetchone()[0] == 0: from lic_webapp.auth import hash_password username = os.environ.get('LIC_ADMIN_USERNAME') password = os.environ.get('LIC_ADMIN_PASSWORD') if not username or not password or len(password) < 12 or len(password) > 128: conn.close() raise RuntimeError('Set LIC_ADMIN_USERNAME and LIC_ADMIN_PASSWORD (12–128 characters) for initial setup.') cursor.execute( "INSERT INTO users (username, password_hash, role) VALUES (?, ?, ?);", (username, hash_password(password), 'admin') ) cursor.execute('CREATE INDEX IF NOT EXISTS idx_exam_history_recruit_date ON exam_history(recruit_id, exam_date DESC, id DESC);') cursor.execute('CREATE INDEX IF NOT EXISTS idx_recruits_solution_number ON recruits(solution_number COLLATE NOCASE);') cursor.execute('CREATE INDEX IF NOT EXISTS idx_sessions_expires ON sessions(expires_at);') conn.commit() conn.close() if __name__ == "__main__": init_db() print("Database initialized successfully.")