from flask import Blueprint, request, jsonify from flask_jwt_extended import jwt_required, get_jwt_identity from db.db import get_db_connection from config.auth import role_required import mysql.connector from datetime import datetime from typing import Optional, List doctor_bp = Blueprint('doctor', __name__) VALID_DAYS = {'Monday', 'Tuesday', 'Wednesday', 'Thursday', 'Friday', 'Saturday', 'Sunday'} # ───────────────────────────────────────────── # 🛠️ HELPERS # ───────────────────────────────────────────── def _serialize_row(row: dict) -> dict: """Convert datetime / date / timedelta values to strings.""" for key, val in row.items(): if isinstance(val, datetime): row[key] = val.strftime('%Y-%m-%d %H:%M:%S') elif hasattr(val, 'isoformat'): # date row[key] = val.isoformat() elif hasattr(val, 'total_seconds'): # timedelta (TIME columns) total = int(val.total_seconds()) h, rem = divmod(total, 3600) m, s = divmod(rem, 60) row[key] = f"{h:02d}:{m:02d}:{s:02d}" return row def _validate_schedule(schedule: list) -> Optional[str]: """ Validate the schedule array sent from the frontend. Returns an error string or None if valid. Expected format: [ { "day": "Monday", "sessions": [ { "start": "09:00", "end": "12:00", "max_patients": 20 }, ... ] }, ... ] """ if not isinstance(schedule, list) or len(schedule) == 0: return "Schedule must be a non-empty list." seen_days = set() for entry in schedule: day = entry.get('day', '').strip() if day not in VALID_DAYS: return f"Invalid day: '{day}'." if day in seen_days: return f"Duplicate day in schedule: '{day}'." seen_days.add(day) sessions = entry.get('sessions', []) if not isinstance(sessions, list) or len(sessions) == 0: return f"At least one session is required for '{day}'." for i, s in enumerate(sessions, 1): start = s.get('start') or '' end = s.get('end') or '' if not start or not end: return f"Session {i} on {day}: start and end times are required." if start >= end: return f"Session {i} on {day}: end time must be after start time." return None def _insert_schedules(cursor, doctor_id: int, schedule: list): """Insert rows into doctor_schedules for the given doctor.""" for entry in schedule: day = entry['day'] sessions = entry['sessions'] for idx, s in enumerate(sessions, 1): cursor.execute( """ INSERT INTO doctor_schedules (doctor_id, day, session_no, start_time, end_time, max_patients) VALUES (%s, %s, %s, %s, %s, %s) """, ( doctor_id, day, idx, s['start'], s['end'], int(s['max_patients']) if s.get('max_patients') else None, ) ) def _fetch_schedules(cursor, doctor_id: int) -> list: """Return schedule rows for a doctor, grouped by day.""" cursor.execute( """ SELECT day, session_no, start_time, end_time, max_patients FROM doctor_schedules WHERE doctor_id = %s ORDER BY FIELD(day,'Monday','Tuesday','Wednesday','Thursday','Friday','Saturday','Sunday'), session_no """, (doctor_id,) ) rows = cursor.fetchall() # Serialise timedelta -> "HH:MM:SS" for r in rows: for col in ('start_time', 'end_time'): val = r[col] if hasattr(val, 'total_seconds'): total = int(val.total_seconds()) h, rem = divmod(total, 3600) m, s = divmod(rem, 60) r[col] = f"{h:02d}:{m:02d}:{s:02d}" # Group by day grouped = {} for r in rows: d = r['day'] if d not in grouped: grouped[d] = [] grouped[d].append({ 'session_no': r['session_no'], 'start_time': r['start_time'], 'end_time': r['end_time'], 'max_patients': r['max_patients'], }) return [{'day': day, 'sessions': sessions} for day, sessions in grouped.items()] # ───────────────────────────────────────────── # ➕ ADD DOCTOR # ───────────────────────────────────────────── @doctor_bp.route('/add_doctor', methods=['POST']) @jwt_required() @role_required('admin', 'cashier', 'receptionist') def add_doctor(): try: data = request.get_json() if not data: return jsonify({"success": False, "error": "No data provided."}), 400 # ── Required fields ────────────────────────────────── name = data.get('name', '').strip() reg_no = data.get('reg_no', '').strip() specialization = data.get('specialization', '').strip() qualification = data.get('qualification', '').strip() contact = data.get('contact', '').strip() channeling_fee = data.get('channeling_fee') schedule = data.get('schedule') # ── Optional fields ────────────────────────────────── email = data.get('email') or None address = data.get('address') or None status = data.get('status', 'Active') dob = data.get('dob') or None gender = data.get('gender') or None notes = data.get('notes') or None store_id = data.get('store_id') or None photo_url = data.get('photo_url') or None # ── Basic validation ────────────────────────────────── if not name: return jsonify({"success": False, "error": "Doctor name is required."}), 400 if not reg_no: return jsonify({"success": False, "error": "Registration number is required."}), 400 if not specialization: return jsonify({"success": False, "error": "Specialization is required."}), 400 if not qualification: return jsonify({"success": False, "error": "Qualification is required."}), 400 if not contact: return jsonify({"success": False, "error": "Contact number is required."}), 400 if channeling_fee is None: return jsonify({"success": False, "error": "Channeling fee is required."}), 400 try: channeling_fee = float(channeling_fee) if channeling_fee < 0: raise ValueError except (ValueError, TypeError): return jsonify({"success": False, "error": "Invalid channeling fee."}), 400 if status not in ('Active', 'Inactive'): status = 'Active' if store_id is not None: try: store_id = int(store_id) except (ValueError, TypeError): return jsonify({"success": False, "error": "Invalid store_id."}), 400 # ── Schedule validation ─────────────────────────────── if not schedule: return jsonify({"success": False, "error": "Please select at least one available day and session."}), 400 err = _validate_schedule(schedule) if err: return jsonify({"success": False, "error": err}), 400 # ── DB operations ───────────────────────────────────── conn = get_db_connection() cursor = conn.cursor(dictionary=True) # Duplicate reg_no check cursor.execute("SELECT id FROM doctors WHERE reg_no = %s", (reg_no,)) if cursor.fetchone(): return jsonify({ "success": False, "error": f"A doctor with registration number '{reg_no}' already exists." }), 409 # Store existence check if store_id is not None: cursor.execute( "SELECT id FROM stores WHERE id = %s AND is_active = 1", (store_id,) ) if not cursor.fetchone(): return jsonify({ "success": False, "error": "Selected store does not exist or is not active." }), 400 created_by = get_jwt_identity() now = datetime.now() cursor.execute( """ INSERT INTO doctors ( name, reg_no, specialization, qualification, email, contact, address, channeling_fee, status, dob, gender, notes, store_id, photo_url, created_by, created_at, updated_at ) VALUES ( %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s ) """, ( name, reg_no, specialization, qualification, email, contact, address, channeling_fee, status, dob, gender, notes, store_id, photo_url, created_by, now, now, ) ) conn.commit() new_doctor_id = cursor.lastrowid # Insert full schedule into doctor_schedules _insert_schedules(cursor, new_doctor_id, schedule) conn.commit() return jsonify({ "success": True, "message": f"Doctor '{name}' registered successfully.", "doctor_id": new_doctor_id }), 201 except mysql.connector.Error as db_err: return jsonify({"success": False, "error": f"Database error: {str(db_err)}"}), 500 except Exception as e: return jsonify({"success": False, "error": f"Server error: {str(e)}"}), 500 finally: if 'cursor' in locals(): cursor.close() if 'conn' in locals(): conn.close() # ───────────────────────────────────────────── # 📋 GET ALL DOCTORS # ───────────────────────────────────────────── @doctor_bp.route('/all_doctors', methods=['GET']) @jwt_required() @role_required('admin', 'cashier', 'receptionist') def get_all_doctors(): try: store_id = request.args.get('store_id', type=int) status = request.args.get('status') search = request.args.get('search', '').strip() conn = get_db_connection() cursor = conn.cursor(dictionary=True) query = """ SELECT d.*, s.store_name FROM doctors d LEFT JOIN stores s ON s.id = d.store_id WHERE 1=1 """ params = [] if store_id: query += " AND d.store_id = %s" params.append(store_id) if status in ('Active', 'Inactive'): query += " AND d.status = %s" params.append(status) if search: query += """ AND ( d.name LIKE %s OR d.reg_no LIKE %s OR d.specialization LIKE %s OR d.contact LIKE %s )""" like = f"%{search}%" params.extend([like, like, like, like]) query += " ORDER BY d.created_at DESC" cursor.execute(query, params) doctors = cursor.fetchall() for doc in doctors: _serialize_row(doc) doc['schedule'] = _fetch_schedules(cursor, doc['id']) return jsonify({"success": True, "doctors": doctors, "count": len(doctors)}), 200 except mysql.connector.Error as db_err: return jsonify({"success": False, "error": f"Database error: {str(db_err)}"}), 500 except Exception as e: return jsonify({"success": False, "error": f"Server error: {str(e)}"}), 500 finally: if 'cursor' in locals(): cursor.close() if 'conn' in locals(): conn.close() # ───────────────────────────────────────────── # 🔍 GET SINGLE DOCTOR # ───────────────────────────────────────────── @doctor_bp.route('/get_doctor/', methods=['GET']) @jwt_required() @role_required('admin', 'cashier', 'receptionist') def get_doctor(doctor_id): try: conn = get_db_connection() cursor = conn.cursor(dictionary=True) cursor.execute( """ SELECT d.*, s.store_name FROM doctors d LEFT JOIN stores s ON s.id = d.store_id WHERE d.id = %s """, (doctor_id,) ) doctor = cursor.fetchone() if not doctor: return jsonify({"success": False, "error": "Doctor not found."}), 404 _serialize_row(doctor) doctor['schedule'] = _fetch_schedules(cursor, doctor_id) return jsonify({"success": True, "doctor": doctor}), 200 except mysql.connector.Error as db_err: return jsonify({"success": False, "error": f"Database error: {str(db_err)}"}), 500 except Exception as e: return jsonify({"success": False, "error": f"Server error: {str(e)}"}), 500 finally: if 'cursor' in locals(): cursor.close() if 'conn' in locals(): conn.close() # ───────────────────────────────────────────── # ✏️ UPDATE DOCTOR # ───────────────────────────────────────────── @doctor_bp.route('/update_doctor/', methods=['PUT']) @jwt_required() @role_required('admin', 'cashier', 'receptionist') def update_doctor(doctor_id): try: data = request.get_json() if not data: return jsonify({"success": False, "error": "No data provided."}), 400 conn = get_db_connection() cursor = conn.cursor(dictionary=True) # Doctor must exist cursor.execute("SELECT id FROM doctors WHERE id = %s", (doctor_id,)) if not cursor.fetchone(): return jsonify({"success": False, "error": "Doctor not found."}), 404 # Duplicate reg_no check (exclude self) new_reg_no = data.get('reg_no', '').strip() if new_reg_no: cursor.execute( "SELECT id FROM doctors WHERE reg_no = %s AND id != %s", (new_reg_no, doctor_id) ) if cursor.fetchone(): return jsonify({ "success": False, "error": f"Registration number '{new_reg_no}' is already used by another doctor." }), 409 # Store existence check if 'store_id' in data and data['store_id'] is not None: try: new_store_id = int(data['store_id']) except (ValueError, TypeError): return jsonify({"success": False, "error": "Invalid store_id."}), 400 cursor.execute( "SELECT id FROM stores WHERE id = %s AND is_active = 1", (new_store_id,) ) if not cursor.fetchone(): return jsonify({ "success": False, "error": "Selected store does not exist or is not active." }), 400 # ── Validate and replace schedule if provided ───────── schedule = data.pop('schedule', None) if schedule is not None: err = _validate_schedule(schedule) if err: return jsonify({"success": False, "error": err}), 400 # ── Build dynamic SET clause ────────────────────────── allowed_fields = [ 'name', 'reg_no', 'specialization', 'qualification', 'email', 'contact', 'address', 'channeling_fee', 'status', 'dob', 'gender', 'notes', 'store_id', 'photo_url', ] set_clauses, values = [], [] for field in allowed_fields: if field in data: set_clauses.append(f"`{field}` = %s") values.append(data[field] if data[field] != '' else None) if not set_clauses and schedule is None: return jsonify({"success": False, "error": "No valid fields to update."}), 400 if set_clauses: set_clauses.append("`updated_at` = %s") values.extend([datetime.now(), doctor_id]) cursor.execute( f"UPDATE doctors SET {', '.join(set_clauses)} WHERE id = %s", values ) conn.commit() # ── Replace schedule ────────────────────────────────── if schedule is not None: cursor.execute( "DELETE FROM doctor_schedules WHERE doctor_id = %s", (doctor_id,) ) _insert_schedules(cursor, doctor_id, schedule) conn.commit() return jsonify({ "success": True, "message": "Doctor updated successfully.", "doctor_id": doctor_id }), 200 except mysql.connector.Error as db_err: return jsonify({"success": False, "error": f"Database error: {str(db_err)}"}), 500 except Exception as e: return jsonify({"success": False, "error": f"Server error: {str(e)}"}), 500 finally: if 'cursor' in locals(): cursor.close() if 'conn' in locals(): conn.close() # ───────────────────────────────────────────── # 🔄 TOGGLE DOCTOR STATUS # ───────────────────────────────────────────── @doctor_bp.route('/toggle_doctor_status/', methods=['PATCH']) @jwt_required() @role_required('admin', 'cashier', 'receptionist') def toggle_doctor_status(doctor_id): try: conn = get_db_connection() cursor = conn.cursor(dictionary=True) cursor.execute( "SELECT id, name, status FROM doctors WHERE id = %s", (doctor_id,) ) doctor = cursor.fetchone() if not doctor: return jsonify({"success": False, "error": "Doctor not found."}), 404 new_status = 'Inactive' if doctor['status'] == 'Active' else 'Active' cursor.execute( "UPDATE doctors SET status = %s, updated_at = %s WHERE id = %s", (new_status, datetime.now(), doctor_id) ) conn.commit() return jsonify({ "success": True, "message": f"Doctor '{doctor['name']}' status changed to '{new_status}'.", "new_status": new_status, "doctor_id": doctor_id }), 200 except mysql.connector.Error as db_err: return jsonify({"success": False, "error": f"Database error: {str(db_err)}"}), 500 except Exception as e: return jsonify({"success": False, "error": f"Server error: {str(e)}"}), 500 finally: if 'cursor' in locals(): cursor.close() if 'conn' in locals(): conn.close() # ───────────────────────────────────────────── # 🗑️ DELETE DOCTOR # ───────────────────────────────────────────── @doctor_bp.route('/delete_doctor/', methods=['DELETE']) @jwt_required() @role_required('admin') def delete_doctor(doctor_id): try: conn = get_db_connection() cursor = conn.cursor(dictionary=True) cursor.execute( "SELECT id, name FROM doctors WHERE id = %s", (doctor_id,) ) doctor = cursor.fetchone() if not doctor: return jsonify({"success": False, "error": "Doctor not found."}), 404 # Block delete if appointments exist try: cursor.execute( "SELECT COUNT(*) AS cnt FROM appointments WHERE doctor_id = %s", (doctor_id,) ) row = cursor.fetchone() if row and row['cnt'] > 0: return jsonify({ "success": False, "error": ( f"Cannot delete doctor '{doctor['name']}'. " f"{row['cnt']} appointment(s) are linked to this doctor. " "Please deactivate instead." ) }), 409 except mysql.connector.Error: pass # appointments table may not exist yet # doctor_schedules rows cascade-delete via FK cursor.execute("DELETE FROM doctors WHERE id = %s", (doctor_id,)) conn.commit() return jsonify({ "success": True, "message": f"Doctor '{doctor['name']}' deleted successfully.", "doctor_id": doctor_id }), 200 except mysql.connector.Error as db_err: return jsonify({"success": False, "error": f"Database error: {str(db_err)}"}), 500 except Exception as e: return jsonify({"success": False, "error": f"Server error: {str(e)}"}), 500 finally: if 'cursor' in locals(): cursor.close() if 'conn' in locals(): conn.close() # ───────────────────────────────────────────── # 📅 GET DOCTORS BY SPECIALIZATION # ───────────────────────────────────────────── @doctor_bp.route('/doctors_by_specialization', methods=['GET']) @jwt_required() @role_required('admin', 'cashier', 'receptionist') def get_doctors_by_specialization(): try: specialization = request.args.get('specialization', '').strip() store_id = request.args.get('store_id', type=int) if not specialization: return jsonify({"success": False, "error": "Specialization is required."}), 400 conn = get_db_connection() cursor = conn.cursor(dictionary=True) query = """ SELECT d.id, d.name, d.reg_no, d.specialization, d.qualification, d.channeling_fee, d.status, d.photo_url, s.store_name FROM doctors d LEFT JOIN stores s ON s.id = d.store_id WHERE d.specialization = %s AND d.status = 'Active' """ params = [specialization] if store_id: query += " AND d.store_id = %s" params.append(store_id) query += " ORDER BY d.name ASC" cursor.execute(query, params) doctors = cursor.fetchall() for doc in doctors: _serialize_row(doc) doc['schedule'] = _fetch_schedules(cursor, doc['id']) return jsonify({ "success": True, "specialization": specialization, "doctors": doctors, "count": len(doctors) }), 200 except mysql.connector.Error as db_err: return jsonify({"success": False, "error": f"Database error: {str(db_err)}"}), 500 except Exception as e: return jsonify({"success": False, "error": f"Server error: {str(e)}"}), 500 finally: if 'cursor' in locals(): cursor.close() if 'conn' in locals(): conn.close() # ───────────────────────────────────────────── # 🏪 GET DOCTORS BY STORE # ───────────────────────────────────────────── @doctor_bp.route('/doctors_by_store/', methods=['GET']) @jwt_required() @role_required('admin', 'cashier', 'receptionist') def get_doctors_by_store(store_id): try: status = request.args.get('status', 'Active') conn = get_db_connection() cursor = conn.cursor(dictionary=True) cursor.execute( "SELECT id, store_name FROM stores WHERE id = %s", (store_id,) ) store = cursor.fetchone() if not store: return jsonify({"success": False, "error": "Store not found."}), 404 query = """ SELECT d.id, d.name, d.reg_no, d.specialization, d.qualification, d.email, d.contact, d.address, d.channeling_fee, d.status, d.dob, d.gender, d.photo_url, d.created_at FROM doctors d WHERE d.store_id = %s """ params = [store_id] if status in ('Active', 'Inactive'): query += " AND d.status = %s" params.append(status) query += " ORDER BY d.name ASC" cursor.execute(query, params) doctors = cursor.fetchall() for doc in doctors: _serialize_row(doc) doc['schedule'] = _fetch_schedules(cursor, doc['id']) return jsonify({ "success": True, "store_id": store_id, "store_name": store['store_name'], "doctors": doctors, "count": len(doctors) }), 200 except mysql.connector.Error as db_err: return jsonify({"success": False, "error": f"Database error: {str(db_err)}"}), 500 except Exception as e: return jsonify({"success": False, "error": f"Server error: {str(e)}"}), 500 finally: if 'cursor' in locals(): cursor.close() if 'conn' in locals(): conn.close()