Spaces:
Running
Running
| """ | |
| 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() | |