""" EduClone AI — Authentication & User Management SQLite WAL mode, bcrypt passwords, roles: admin / user """ import sqlite3, bcrypt, datetime, os, re, threading DB_PATH = os.environ.get("EDUCLONE_DB", "educlone_users.db") ADMIN_USERNAME = "admin" ADMIN_EMAIL = "admin@educlone.ai" ADMIN_PASSWORD = "admin123" _lock = threading.Lock() def _conn(): """Return a fresh connection each call — WAL so reads never block.""" c = sqlite3.connect(DB_PATH, timeout=30) c.row_factory = sqlite3.Row c.execute("PRAGMA journal_mode=WAL") c.execute("PRAGMA busy_timeout=10000") return c def init_db(): with _lock: c = _conn() c.execute("""CREATE TABLE IF NOT EXISTS users ( id INTEGER PRIMARY KEY AUTOINCREMENT, username TEXT UNIQUE NOT NULL, email TEXT UNIQUE NOT NULL, password_hash TEXT NOT NULL, role TEXT NOT NULL DEFAULT 'user', full_name TEXT DEFAULT '', institution TEXT DEFAULT '', is_approved INTEGER DEFAULT 1, is_active INTEGER DEFAULT 1, created_at TEXT NOT NULL, last_login TEXT)""") c.execute("""CREATE TABLE IF NOT EXISTS activity_log ( id INTEGER PRIMARY KEY AUTOINCREMENT, user_id INTEGER, username TEXT, action TEXT, detail TEXT, timestamp TEXT)""") # seed admin if not c.execute("SELECT id FROM users WHERE username=?", (ADMIN_USERNAME,)).fetchone(): ph = bcrypt.hashpw(ADMIN_PASSWORD.encode(), bcrypt.gensalt()).decode() c.execute("""INSERT INTO users (username,email,password_hash,role,full_name,institution,is_approved,created_at) VALUES(?,?,?,?,?,?,?,?)""", (ADMIN_USERNAME, ADMIN_EMAIL, ph, "admin", "Abdulqayyum MBA", "EduClone AI", 1, datetime.datetime.now().isoformat())) c.commit(); c.close() def _valid_email(e): return bool(re.match(r'^[^@\s]+@[^@\s]+\.[^@\s]+$', e)) def _valid_user(u): return bool(re.match(r'^[a-zA-Z0-9_]{3,30}$', u)) def _valid_pass(p): return len(p)>=8 and re.search(r'[A-Z]',p) and re.search(r'[0-9]',p) def register_user(username, email, password, full_name="", institution=""): username = username.strip().lower(); email = email.strip().lower() if not _valid_user(username): return False,"Username: 3–30 chars, letters/numbers/underscore only." if not _valid_email(email): return False,"Please enter a valid email address." if not _valid_pass(password): return False,"Password: 8+ chars, 1 uppercase, 1 number." with _lock: c = _conn() try: if c.execute("SELECT id FROM users WHERE username=?",(username,)).fetchone(): return False,"Username already taken." if c.execute("SELECT id FROM users WHERE email=?",(email,)).fetchone(): return False,"Email already registered." ph = bcrypt.hashpw(password.encode(), bcrypt.gensalt()).decode() c.execute("""INSERT INTO users (username,email,password_hash,role,full_name,institution,is_approved,created_at) VALUES(?,?,?,?,?,?,?,?)""", (username,email,ph,"user",full_name.strip(),institution.strip(),1, datetime.datetime.now().isoformat())) c.commit() return True, f"✅ Account created! Welcome, {username}. You can now log in." except Exception as ex: return False, f"Registration error: {ex}" finally: c.close() def login_user(username, password): username = username.strip().lower() with _lock: c = _conn() try: row = c.execute( "SELECT * FROM users WHERE username=? OR email=?", (username, username)).fetchone() if not row: return False, "❌ Username or password incorrect.", None if not row["is_active"]: return False, "❌ Account deactivated. Contact admin.", None if not bcrypt.checkpw(password.encode(), row["password_hash"].encode()): return False, "❌ Username or password incorrect.", None c.execute("UPDATE users SET last_login=? WHERE id=?", (datetime.datetime.now().isoformat(), row["id"])) c.commit() user = {k: row[k] for k in row.keys() if k != "password_hash"} return True, f"✅ Welcome back, {row['full_name'] or row['username']}!", user except Exception as ex: return False, f"Login error: {ex}", None finally: c.close() def log_activity(user_id, username, action, detail=""): with _lock: c = _conn() try: c.execute("""INSERT INTO activity_log(user_id,username,action,detail,timestamp) VALUES(?,?,?,?,?)""", (user_id, username, action, detail, datetime.datetime.now().isoformat())) c.commit() finally: c.close() def get_all_users(): with _lock: c = _conn() try: rows = c.execute("""SELECT id,username,email,role,full_name,institution, is_approved,is_active,created_at,last_login FROM users ORDER BY created_at DESC""").fetchall() return [dict(r) for r in rows] finally: c.close() def toggle_user_active(user_id, active): with _lock: c = _conn() try: c.execute("UPDATE users SET is_active=? WHERE id=?", (1 if active else 0, user_id)) c.commit() return "✅ User status updated." finally: c.close() def change_user_role(user_id, role): if role not in ("admin","user"): return "❌ Invalid role." with _lock: c = _conn() try: c.execute("UPDATE users SET role=? WHERE id=?", (role, user_id)) c.commit() return f"✅ Role updated to {role}." finally: c.close() def delete_user(user_id): with _lock: c = _conn() try: c.execute("DELETE FROM users WHERE id=? AND role != 'admin'", (user_id,)) c.commit() return "✅ User deleted." finally: c.close() def get_activity_log(limit=200): with _lock: c = _conn() try: rows = c.execute( "SELECT * FROM activity_log ORDER BY timestamp DESC LIMIT ?", (limit,)).fetchall() return [dict(r) for r in rows] finally: c.close() def get_stats(): with _lock: c = _conn() try: today = datetime.date.today().isoformat() return { "total": c.execute("SELECT COUNT(*) FROM users").fetchone()[0], "admins": c.execute("SELECT COUNT(*) FROM users WHERE role='admin'").fetchone()[0], "active": c.execute("SELECT COUNT(*) FROM users WHERE is_active=1").fetchone()[0], "new_today": c.execute("SELECT COUNT(*) FROM users WHERE created_at LIKE ?", (today+"%",)).fetchone()[0], "logins_today": c.execute("SELECT COUNT(*) FROM activity_log WHERE action='login' AND timestamp LIKE ?", (today+"%",)).fetchone()[0], } finally: c.close() init_db()