Spaces:
Runtime error
Runtime error
| from flask import Blueprint, render_template, request, redirect, url_for, flash, session, jsonify | |
| from utils.logging.logger_setup import logger | |
| from utils.db.db import mongo | |
| import bcrypt | |
| from datetime import datetime | |
| admin_bp = Blueprint('admin', __name__, url_prefix='/admin') | |
| def selection(): | |
| return render_template('admin/selection.html') | |
| def login(): | |
| # Helper to redirect logged-in users | |
| if session.get('user_type'): | |
| user_type = session.get('user_type') | |
| if user_type == 'admin': | |
| return redirect(url_for('admin.dashboard')) | |
| elif user_type == 'assistant': | |
| # Redirect to generic assistant dashboard which handles role splitting | |
| return redirect(url_for('assistant.dashboard')) | |
| if request.method == 'POST': | |
| username = request.form['username'] | |
| password = request.form['password'] | |
| # Check Admin | |
| admin = mongo.db.admin.find_one({"username": username}) | |
| if admin: | |
| # Check password | |
| stored_password = admin.get('password') | |
| is_valid = False | |
| try: | |
| if stored_password: | |
| is_valid = bcrypt.checkpw(password.encode('utf-8'), stored_password.encode('utf-8')) | |
| except ValueError: | |
| is_valid = (stored_password == password) | |
| if is_valid: | |
| session['admin_id'] = admin.get('sql_id', admin.get('id')) | |
| session['user_type'] = 'admin' | |
| logger.info({"action": "admin_login", "status": "success", "username": username}) | |
| return redirect(url_for('admin.dashboard')) | |
| # Check Assistant | |
| assistant = mongo.db.administrativeassistant.find_one({"username": username}) | |
| if assistant: | |
| stored_password = assistant.get('password') | |
| is_valid = False | |
| try: | |
| if stored_password: | |
| is_valid = bcrypt.checkpw(password.encode('utf-8'), stored_password.encode('utf-8')) | |
| except ValueError: | |
| is_valid = (stored_password == password) | |
| if is_valid: | |
| session['assistant_id'] = assistant.get('sql_id', assistant.get('id')) | |
| session['user_type'] = 'assistant' | |
| session['role'] = assistant.get('role') | |
| session['user_name'] = assistant.get('username') | |
| if assistant.get('role') == 'account_creation': | |
| return redirect(url_for('assistant.account_creation_dashboard')) | |
| elif assistant.get('role') == 'data_entry': | |
| return redirect(url_for('assistant.data_entry_dashboard')) | |
| else: | |
| return redirect(url_for('admin.dashboard')) | |
| logger.warning({"action": "admin_login", "status": "failure", "username": username}) | |
| flash('Invalid credentials') | |
| return redirect(url_for('admin.login')) | |
| return render_template('admin/admin_login.html') | |
| def dashboard(): | |
| if 'user_type' not in session: | |
| return redirect(url_for('admin.login')) | |
| # If an assistant tries to access admin dashboard, send them to their place | |
| if session.get('user_type') == 'assistant': | |
| return redirect(url_for('assistant.dashboard')) | |
| return render_template('admin/admin_dashboard.html') | |
| def logout(): | |
| session.clear() | |
| logger.info({"action": "logout", "user_type": "admin/assistant"}) | |
| return redirect(url_for('admin.login')) | |
| def manage_doctors(): | |
| if session.get('user_type') != 'admin': | |
| return redirect(url_for('admin.login')) | |
| doctors = list(mongo.db.doctor.find()) | |
| return render_template('admin/manage_doctors.html', doctors=doctors) | |
| def manage_patients(): | |
| if session.get('user_type') != 'admin': | |
| return redirect(url_for('admin.login')) | |
| patients = list(mongo.db.patient.find()) | |
| return render_template('admin/manage_patients.html', patients=patients) | |
| def manage_assistants(): | |
| if session.get('user_type') != 'admin': | |
| return redirect(url_for('admin.login')) | |
| assistants = list(mongo.db.administrativeassistant.find()) | |
| return render_template('admin/manage_assistants.html', assistants=assistants) | |
| def get_next_id(collection_name): | |
| """Helper to simulate auto-increment ID for compatibility""" | |
| last_item = mongo.db[collection_name].find_one(sort=[("sql_id", -1)]) | |
| if last_item and 'sql_id' in last_item: | |
| return last_item['sql_id'] + 1 | |
| return 1 | |
| def add_doctor(): | |
| """ | |
| Create a new doctor account. | |
| """ | |
| if session.get('user_type') != 'admin': | |
| return redirect(url_for('admin.login')) | |
| try: | |
| name = request.form['name'] | |
| specialization = request.form['specialization'] | |
| email = request.form['email'] | |
| phone = request.form['phone'] | |
| password = request.form['password'] | |
| hashed_password = bcrypt.hashpw(password.encode('utf-8'), bcrypt.gensalt()).decode('utf-8') | |
| new_id = get_next_id('doctor') | |
| new_doctor = { | |
| "sql_id": new_id, | |
| "id": new_id, | |
| "name": name, | |
| "specialization": specialization, | |
| "email": email, | |
| "phone": phone, | |
| "password": hashed_password, | |
| "bio": None | |
| } | |
| mongo.db.doctor.insert_one(new_doctor) | |
| logger.info({"action": "add_doctor", "performed_by": "admin", "doctor_email": email}) | |
| flash("Doctor added successfully.") | |
| except Exception as e: | |
| logger.error(f"Error adding doctor: {e}") | |
| flash("Error adding doctor. Please check logs.") | |
| return redirect(url_for('admin.manage_doctors')) | |
| def add_patient(): | |
| """ | |
| Create a new patient account with a structured ID. | |
| """ | |
| if session.get('user_type') != 'admin': | |
| return redirect(url_for('admin.login')) | |
| try: | |
| name = request.form['name'] | |
| email = request.form['email'] | |
| password = request.form['password'] | |
| hashed_password = bcrypt.hashpw(password.encode('utf-8'), bcrypt.gensalt()).decode('utf-8') | |
| new_id = get_next_id('patient') | |
| # Generate Custom Patient ID: SQ-YYYY-COUNT | |
| current_year = datetime.now().year | |
| # Count existing patients with IDs starting with SQ-YYYY | |
| count = mongo.db.patient.count_documents({"patient_id": {"$regex": f"^SQ-{current_year}-"}}) | |
| new_patient_id = f"SQ-{current_year}-{count + 1:03d}" | |
| new_patient = { | |
| "sql_id": new_id, | |
| "id": new_id, | |
| "patient_id": new_patient_id, # New Structured ID | |
| "name": name, | |
| "email": email, | |
| "password": hashed_password, | |
| "created_at": datetime.now() | |
| } | |
| mongo.db.patient.insert_one(new_patient) | |
| logger.info({"action": "add_patient", "performed_by": "admin", "patient_email": email, "new_patient_id": new_patient_id}) | |
| flash(f"Patient added successfully! ID: {new_patient_id}") | |
| except Exception as e: | |
| logger.error(f"Error adding patient: {e}") | |
| flash("Error adding patient.") | |
| return redirect(url_for('admin.manage_patients')) | |
| def add_assistant(): | |
| """ | |
| Create a new administrative assistant account. | |
| """ | |
| if session.get('user_type') != 'admin': | |
| return redirect(url_for('admin.login')) | |
| try: | |
| username = request.form['username'] | |
| password = request.form['password'] | |
| role = request.form['role'] | |
| hashed_password = bcrypt.hashpw(password.encode('utf-8'), bcrypt.gensalt()).decode('utf-8') | |
| new_id = get_next_id('administrativeassistant') | |
| new_assistant = { | |
| "sql_id": new_id, | |
| "id": new_id, | |
| "username": username, | |
| "password": hashed_password, | |
| "role": role | |
| } | |
| mongo.db.administrativeassistant.insert_one(new_assistant) | |
| logger.info({"action": "add_assistant", "performed_by": "admin", "username": username, "role": role}) | |
| flash("Assistant added successfully.") | |
| except Exception as e: | |
| logger.error(f"Error adding assistant: {e}") | |
| flash("Error adding assistant.") | |
| return redirect(url_for('admin.manage_assistants')) | |
| def delete_user(user_type, user_id): | |
| """ | |
| Delete a user (doctor, patient, or assistant) by ID. | |
| """ | |
| if session.get('user_type') != 'admin': | |
| return redirect(url_for('admin.login')) | |
| collection_name = None | |
| redirect_url = 'admin.dashboard' | |
| if user_type == 'doctor': | |
| collection_name = 'doctor' | |
| redirect_url = 'admin.manage_doctors' | |
| elif user_type == 'patient': | |
| collection_name = 'patient' | |
| redirect_url = 'admin.manage_patients' | |
| elif user_type == 'assistant': | |
| collection_name = 'administrativeassistant' | |
| redirect_url = 'admin.manage_assistants' | |
| if collection_name: | |
| try: | |
| # Delete by sql_id or id | |
| result = mongo.db[collection_name].delete_one({"$or": [{"sql_id": user_id}, {"id": user_id}]}) | |
| if result.deleted_count > 0: | |
| logger.info({"action": "delete_user", "target_type": user_type, "target_id": user_id, "status": "success"}) | |
| flash(f"{user_type.capitalize()} deleted successfully.") | |
| else: | |
| logger.warning({"action": "delete_user", "target_type": user_type, "target_id": user_id, "status": "not_found"}) | |
| flash("User not found.") | |
| except Exception as e: | |
| logger.error(f"Error deleting user: {e}") | |
| flash("Error deleting user.") | |
| return redirect(url_for(redirect_url)) | |
| return redirect(url_for('admin.dashboard')) | |