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') @admin_bp.route('/') def selection(): return render_template('admin/selection.html') @admin_bp.route('/login', methods=['GET', 'POST']) 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') @admin_bp.route('/dashboard') 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') @admin_bp.route('/logout') def logout(): session.clear() logger.info({"action": "logout", "user_type": "admin/assistant"}) return redirect(url_for('admin.login')) @admin_bp.route('/manage/doctors') 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) @admin_bp.route('/manage/patients') 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) @admin_bp.route('/manage/assistants') 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 @admin_bp.route('/add/doctor', methods=['POST']) 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')) @admin_bp.route('/add/patient', methods=['POST']) 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')) @admin_bp.route('/add/assistant', methods=['POST']) 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')) @admin_bp.route('/delete//') 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'))