Spaces:

CYV28
/

RetinAI

Sleeping

RetinAI / database.py
Update database.py
409c0a6 verified 8 months ago
34.4 kB
#!/usr/bin/env python3
"""
BASE DE DATOS SQLITE - APLICACIÓN MÉDICA
Retinopatía Diabética - Sistema de Diagnóstico
Versión Web (Flask) - Con aislamiento de datos por usuario
"""
import sqlite3
import os
import hashlib
from datetime import datetime
from typing import Optional, List, Dict


# Usar /data si existe y tiene permisos, sino usar directorio local
def _get_db_path():
    data_dir = os.environ.get("DB_PATH", "/data/medical_app.db")
    parent = os.path.dirname(data_dir)
    try:
        os.makedirs(parent, exist_ok=True)
        # Test de escritura
        test = os.path.join(parent, ".write_test")
        with open(test, "w") as f:
            f.write("ok")
        os.remove(test)
        return data_dir
    except Exception:
        # Fallback: guardar junto al script
        return os.path.join(os.path.dirname(os.path.abspath(__file__)), "medical_app.db")

DB_PATH = _get_db_path()


class DatabaseManager:
    def __init__(self, db_path: str = DB_PATH):
        self.db_path = db_path
        os.makedirs(os.path.dirname(db_path), exist_ok=True)
        self.init_database()

    def get_connection(self) -> sqlite3.Connection:
        conn = sqlite3.connect(self.db_path)
        conn.row_factory = sqlite3.Row
        conn.execute("PRAGMA foreign_keys = ON")
        return conn

    def init_database(self):
        conn = self.get_connection()
        try:
            conn.execute('''
                CREATE TABLE IF NOT EXISTS Users (
                    userID    INTEGER PRIMARY KEY AUTOINCREMENT,
                    username  VARCHAR(30) NOT NULL UNIQUE,
                    password  VARCHAR(255) NOT NULL,
                    role      VARCHAR(20) DEFAULT 'Doctor',
                    creationDate DATETIME DEFAULT CURRENT_TIMESTAMP
                )
            ''')

            conn.execute('''
                CREATE TABLE IF NOT EXISTS Patients (
                    patientID    INTEGER PRIMARY KEY AUTOINCREMENT,
                    createdByUserID INTEGER NOT NULL,
                    name         VARCHAR(50) NOT NULL,
                    birthDate    DATE,
                    gender       VARCHAR(1),
                    diabetesType VARCHAR(20),
                    creationDate DATETIME DEFAULT CURRENT_TIMESTAMP,
                    FOREIGN KEY (createdByUserID) REFERENCES Users(userID)
                )
            ''')

            conn.execute('''
                CREATE TABLE IF NOT EXISTS RiskFactors (
                    riskFactorID INTEGER PRIMARY KEY AUTOINCREMENT,
                    name         VARCHAR(30) NOT NULL,
                    description  TEXT,
                    creationDate DATETIME DEFAULT CURRENT_TIMESTAMP
                )
            ''')

            conn.execute('''
                CREATE TABLE IF NOT EXISTS PatientsRiskFactors (
                    patientID    INTEGER NOT NULL,
                    riskFactorID INTEGER NOT NULL,
                    creationDate DATETIME DEFAULT CURRENT_TIMESTAMP,
                    PRIMARY KEY (patientID, riskFactorID),
                    FOREIGN KEY (patientID)    REFERENCES Patients(patientID)    ON DELETE CASCADE,
                    FOREIGN KEY (riskFactorID) REFERENCES RiskFactors(riskFactorID) ON DELETE CASCADE
                )
            ''')

            conn.execute('''
                CREATE TABLE IF NOT EXISTS Consultations (
                    consultationID      INTEGER PRIMARY KEY AUTOINCREMENT,
                    patientID           INTEGER NOT NULL,
                    createdByUserID     INTEGER NOT NULL,
                    diabeticRetinopathy BOOLEAN DEFAULT FALSE,
                    notes               TEXT,
                    consultationDate    DATETIME DEFAULT CURRENT_TIMESTAMP,
                    imagePath           TEXT,
                    confidence          REAL,
                    rawOutput           REAL,
                    FOREIGN KEY (patientID)       REFERENCES Patients(patientID) ON DELETE CASCADE,
                    FOREIGN KEY (createdByUserID) REFERENCES Users(userID)
                )
            ''')

            conn.execute('''
                CREATE TABLE IF NOT EXISTS Tasks (
                    taskID       INTEGER PRIMARY KEY AUTOINCREMENT,
                    userID       INTEGER NOT NULL,
                    text         TEXT NOT NULL,
                    completed    BOOLEAN DEFAULT FALSE,
                    creationDate DATETIME DEFAULT CURRENT_TIMESTAMP,
                    FOREIGN KEY (userID) REFERENCES Users(userID) ON DELETE CASCADE
                )
            ''')

            conn.commit()
            self._insert_default_data(conn)
            print("Base de datos inicializada correctamente")
        except Exception as e:
            print(f"Error inicializando base de datos: {e}")
            conn.rollback()
        finally:
            conn.close()

    def _insert_default_data(self, conn):
        try:
            cursor = conn.execute("SELECT COUNT(*) FROM Users WHERE username = 'admin'")
            if cursor.fetchone()[0] == 0:
                admin_password = self.hash_password("admin123")
                conn.execute(
                    "INSERT INTO Users (username, password, role) VALUES (?, ?, ?)",
                    ("admin", admin_password, "Admin")
                )
                print("Usuario administrador creado: admin / admin123")

            cursor = conn.execute("SELECT COUNT(*) FROM RiskFactors")
            if cursor.fetchone()[0] == 0:
                risk_factors = [
                    ("Hipertensión",            "Presión arterial alta"),
                    ("Diabetes Tipo 1",          "Diabetes mellitus dependiente de insulina"),
                    ("Diabetes Tipo 2",          "Diabetes mellitus no dependiente de insulina"),
                    ("Obesidad",                 "Índice de masa corporal elevado"),
                    ("Tabaquismo",               "Consumo de cigarrillos o tabaco"),
                    ("Sedentarismo",             "Falta de actividad física regular"),
                    ("Antecedentes Familiares",  "Historia familiar de diabetes o cardiovascular"),
                    ("Edad Avanzada",            "Mayor de 65 años"),
                    ("Colesterol Alto",          "Niveles elevados de colesterol"),
                    ("Nefropatía",               "Enfermedad renal relacionada con diabetes"),
                ]
                conn.executemany(
                    "INSERT INTO RiskFactors (name, description) VALUES (?, ?)",
                    risk_factors
                )
                print("Factores de riesgo insertados")

            conn.commit()

            # Cargar datos de demo si la BD está recién creada (sin pacientes)
            n_patients = conn.execute("SELECT COUNT(*) FROM Patients").fetchone()[0]
            if n_patients == 0:
                self._seed_demo_data(conn)

        except Exception as e:
            print(f"Error insertando datos por defecto: {e}")

    def _seed_demo_data(self, conn):
        """Inserta datos de demostración al arrancar con una BD vacía."""
        import random
        from datetime import datetime, timedelta

        print("=== Insertando datos de demostración ===")

        def rand_consult_date(days_max=180):
            days  = random.randint(0, days_max)
            hours = random.randint(7, 18)
            mins  = random.randint(0, 59)
            return (datetime.now() - timedelta(days=days, hours=hours, minutes=mins)
                    ).strftime("%Y-%m-%d %H:%M:%S")

        # ── Usuarios de prueba ────────────────────────────────────────────
        demo_users = [
            ("dr_ramirez",  "doctor123", "Doctor"),
            ("dr_perez",    "doctor123", "Doctor"),
            ("dr_martinez", "doctor123", "Doctor"),
            ("supervisor",  "admin456",  "Admin"),
        ]
        user_ids = {}
        for username, password, role in demo_users:
            cur = conn.execute(
                "INSERT OR IGNORE INTO Users (username, password, role) VALUES (?,?,?)",
                (username, self.hash_password(password), role)
            )
            row = conn.execute("SELECT userID FROM Users WHERE username=?", (username,)).fetchone()
            user_ids[username] = row["userID"]
        conn.commit()
        print(f"  + {len(demo_users)} usuarios de prueba creados")

        doctor_names = ["dr_ramirez", "dr_perez", "dr_martinez"]
        doctor_ids   = [user_ids[n] for n in doctor_names]

        # ── Pacientes ─────────────────────────────────────────────────────
        patients_data = [
            ("Juan Carlos Medina Castillo",    "1958-03-14", "M", "Tipo 2"),
            ("María Elena Reyes Sánchez",      "1965-07-22", "F", "Tipo 2"),
            ("Roberto Antonio Pérez Núñez",    "1972-11-05", "M", "Tipo 1"),
            ("Carmen Altagracia Díaz López",   "1949-01-30", "F", "Tipo 2"),
            ("Luis Fernando Vargas Torres",    "1980-09-18", "M", "Tipo 2"),
            ("Ana Milagros Santos Herrera",    "1955-04-12", "F", "Tipo 2"),
            ("Pedro Emilio Guzmán Rosario",    "1963-12-28", "M", "Tipo 1"),
            ("Josefa Alicia Mora Taveras",     "1970-06-03", "F", "Gestacional"),
            ("Francisco Javier Marte Féliz",   "1945-08-17", "M", "Tipo 2"),
            ("Rosa María Almonte Encarnación", "1968-02-09", "F", "Tipo 2"),
            ("Miguel Ángel Cabrera Paulino",   "1976-10-25", "M", "MODY"),
            ("Esperanza del Carmen Pichardo",  "1952-05-31", "F", "Tipo 2"),
            ("Carlos Daniel Tejada Ureña",     "1988-03-07", "M", "Tipo 1"),
            ("Yolanda Beatriz Novas Acosta",   "1961-09-14", "F", "Tipo 2"),
            ("Ramón Ernesto Jiménez Clase",    "1957-07-19", "M", "Tipo 2"),
            ("Gloria Inés Santana Bido",       "1973-11-22", "F", "Tipo 2"),
            ("Héctor Manuel Ogando Matos",     "1966-04-08", "M", "Tipo 2"),
            ("Milagros Concepción Báez Rijo",  "1982-01-16", "F", "Tipo 1"),
            ("Víctor Hugo Severino Cepeda",    "1948-08-04", "M", "Tipo 2"),
            ("Luz Marina Polanco Espinal",     "1959-06-27", "F", "Tipo 2"),
            ("Andrés Felipe Castillo Grullón", "1990-12-11", "M", "Tipo 1"),
            ("Natividad Eugenia Pena Arias",   "1944-03-23", "F", "Tipo 2"),
            ("Domingo Rafael Flores Medina",   "1967-10-01", "M", "Otro"),
            ("Sandra Margarita Cruz Lizardo",  "1978-05-18", "F", "Tipo 2"),
            ("Eugenio Antonio Lora Ferreira",  "1953-09-30", "M", "Tipo 2"),
        ]
        patient_ids = []
        for i, (name, birth, gender, diab) in enumerate(patients_data):
            cur = conn.execute(
                "INSERT INTO Patients (createdByUserID, name, birthDate, gender, diabetesType) VALUES (?,?,?,?,?)",
                (doctor_ids[i % len(doctor_ids)], name, birth, gender, diab)
            )
            patient_ids.append(cur.lastrowid)
        conn.commit()
        print(f"  + {len(patient_ids)} pacientes creados")

        # ── Factores de riesgo ────────────────────────────────────────────
        rf_map = {r["name"]: r["riskFactorID"]
                  for r in conn.execute("SELECT riskFactorID, name FROM RiskFactors").fetchall()}

        patient_risk_factors = {
            "Hipertensión":            [0, 3, 4, 5, 8, 9, 13, 14, 16, 18, 19, 21, 24],
            "Diabetes Tipo 1":         [2, 6, 12, 17, 20],
            "Diabetes Tipo 2":         [0, 1, 3, 4, 5, 7, 8, 9, 11, 13, 14, 15, 16, 18, 19, 21, 22, 23, 24],
            "Obesidad":                [0, 3, 8, 9, 14, 18, 21, 24],
            "Tabaquismo":              [6, 8, 14, 18, 22],
            "Sedentarismo":            [1, 3, 5, 9, 13, 19, 21],
            "Antecedentes Familiares": [0, 2, 4, 6, 10, 12, 15, 17, 20],
            "Edad Avanzada":           [3, 8, 18, 21],
            "Colesterol Alto":         [0, 4, 8, 13, 16, 18, 24],
            "Nefropatía":              [3, 8, 18],
        }
        rf_count = 0
        for rf_name, indices in patient_risk_factors.items():
            rf_id = rf_map.get(rf_name)
            if rf_id is None:
                continue
            for idx in indices:
                if idx < len(patient_ids):
                    conn.execute(
                        "INSERT OR IGNORE INTO PatientsRiskFactors (patientID, riskFactorID) VALUES (?,?)",
                        (patient_ids[idx], rf_id)
                    )
                    rf_count += 1
        conn.commit()
        print(f"  + {rf_count} factores de riesgo asignados")

        # ── Consultas ─────────────────────────────────────────────────────
        consultations = [
            (0,  92.3, True,  "Lesiones hemorrágicas visibles en polo posterior. Evaluación urgente con retinólogo."),
            (0,  87.1, True,  "Seguimiento: persisten microaneurismas. Control en 3 meses."),
            (3,  95.8, True,  "Retinopatía diabética no proliferativa severa. Derivación inmediata."),
            (3,  88.4, True,  "Exudados duros periféricos. Edema macular presente."),
            (5,  79.2, True,  "Microaneurismas múltiples en arcada temporal superior."),
            (8,  96.1, True,  "Retinopatía proliferativa. Neovascularización de disco. Requiere fotocoagulación."),
            (8,  91.7, True,  "Post-fotocoagulación. Reducción de neovascularización. Seguimiento mensual."),
            (13, 84.5, True,  "Hemorragias en llama en cuadrante nasal inferior."),
            (14, 88.9, True,  "Exudados algodonosos y microaneurismas. RDNP moderada."),
            (18, 93.2, True,  "Retinopatía diabética avanzada. Tracción vitreorretiniana incipiente."),
            (21, 90.6, True,  "Microaneurismas y hemorragias puntiformes bilaterales."),
            (24, 82.3, True,  "RDNP leve. Primera detección. Inicio de seguimiento cada 6 meses."),
            (1,  91.5, False, "Retina sin signos de retinopatía. Control anual programado."),
            (1,  94.2, False, "Sin cambios respecto a control anterior. Fondo de ojo normal."),
            (2,  88.7, False, "Retina en buen estado. Se refuerza control glucémico."),
            (4,  96.3, False, "Sin evidencia de retinopatía. Paciente con buen control metabólico."),
            (6,  85.1, False, "Fondo de ojo sin alteraciones. Seguimiento semestral indicado."),
            (7,  92.8, False, "Retina normal. Paciente en primer trimestre, control mensual."),
            (9,  89.4, False, "Sin signos de RD. Se mejora el control de HbA1c."),
            (10, 93.7, False, "Sin alteraciones retinianas. Paciente joven con buen pronóstico."),
            (11, 87.6, False, "Retina sin lesiones. Control semestral."),
            (12, 91.2, False, "Sin retinopatía. Visión borrosa ocasional descartada como causa retiniana."),
            (15, 94.8, False, "Fondo de ojo normal. Mantener HbA1c < 7%."),
            (16, 86.3, False, "Sin signos de RD. Control en 6 meses."),
            (17, 93.1, False, "Retina sin alteraciones. Control glucémico óptimo."),
            (19, 88.9, False, "Sin retinopatía. Se orienta sobre importancia del ejercicio físico."),
            (20, 95.4, False, "Fondo de ojo normal. Primer control, paciente de reciente diagnóstico."),
            (23, 90.1, False, "Retina sin lesiones. Excelente adherencia al tratamiento."),
        ]
        for pat_idx, confidence, has_dr, notes in consultations:
            if pat_idx >= len(patient_ids):
                continue
            pid = patient_ids[pat_idx]
            doc_id = conn.execute(
                "SELECT createdByUserID FROM Patients WHERE patientID=?", (pid,)
            ).fetchone()["createdByUserID"]
            conn.execute(
                """INSERT INTO Consultations
                   (patientID, createdByUserID, diabeticRetinopathy, notes,
                    confidence, rawOutput, consultationDate)
                   VALUES (?,?,?,?,?,?,?)""",
                (pid, doc_id, 1 if has_dr else 0, notes,
                 confidence, confidence / 100.0, rand_consult_date())
            )
        conn.commit()
        print(f"  + {len(consultations)} consultas creadas")
        print("=== Datos de demostración listos ===")

    # ------------------------------------------------------------------ #
    #  UTILIDADES
    # ------------------------------------------------------------------ #
    @staticmethod
    def hash_password(password: str) -> str:
        return hashlib.sha256(password.encode()).hexdigest()

    @staticmethod
    def _serialize(row) -> Dict:
        """Convierte sqlite3.Row a dict y serializa fechas."""
        d = dict(row)
        for key, val in d.items():
            if isinstance(val, (datetime,)):
                d[key] = str(val)
        return d

    # ------------------------------------------------------------------ #
    #  USUARIOS
    # ------------------------------------------------------------------ #
    def authenticate_user(self, username: str, password: str) -> Optional[Dict]:
        conn = self.get_connection()
        try:
            hashed = self.hash_password(password)
            cursor = conn.execute(
                "SELECT userID, username, role, creationDate FROM Users WHERE username=? AND password=?",
                (username, hashed)
            )
            row = cursor.fetchone()
            return self._serialize(row) if row else None
        finally:
            conn.close()

    def create_user(self, username: str, password: str, role: str = "Doctor") -> bool:
        conn = self.get_connection()
        try:
            conn.execute(
                "INSERT INTO Users (username, password, role) VALUES (?, ?, ?)",
                (username, self.hash_password(password), role)
            )
            conn.commit()
            return True
        except sqlite3.IntegrityError:
            return False
        finally:
            conn.close()

    def get_all_users(self) -> List[Dict]:
        conn = self.get_connection()
        try:
            rows = conn.execute(
                "SELECT userID, username, role, creationDate FROM Users ORDER BY creationDate DESC"
            ).fetchall()
            return [self._serialize(r) for r in rows]
        finally:
            conn.close()

    def get_user(self, user_id: int) -> Optional[Dict]:
        conn = self.get_connection()
        try:
            row = conn.execute(
                "SELECT userID, username, role, creationDate FROM Users WHERE userID=?",
                (user_id,)
            ).fetchone()
            return self._serialize(row) if row else None
        finally:
            conn.close()

    def update_user(self, user_id: int, username: str = None, role: str = None, password: str = None) -> bool:
        conn = self.get_connection()
        try:
            if password:
                conn.execute(
                    "UPDATE Users SET username=?, role=?, password=? WHERE userID=?",
                    (username, role, self.hash_password(password), user_id)
                )
            else:
                conn.execute(
                    "UPDATE Users SET username=?, role=? WHERE userID=?",
                    (username, role, user_id)
                )
            conn.commit()
            return True
        except Exception:
            return False
        finally:
            conn.close()

    def delete_user(self, user_id: int) -> bool:
        conn = self.get_connection()
        try:
            conn.execute("DELETE FROM Users WHERE userID=?", (user_id,))
            conn.commit()
            return True
        except Exception:
            return False
        finally:
            conn.close()

    # ------------------------------------------------------------------ #
    #  PACIENTES  (con filtrado por usuario según rol)
    # ------------------------------------------------------------------ #
    def create_patient(self, created_by_user_id: int, name: str,
                       birth_date: str = None, gender: str = None,
                       diabetes_type: str = None) -> Optional[int]:
        conn = self.get_connection()
        try:
            cursor = conn.execute(
                "INSERT INTO Patients (createdByUserID, name, birthDate, gender, diabetesType) VALUES (?,?,?,?,?)",
                (created_by_user_id, name, birth_date, gender, diabetes_type)
            )
            conn.commit()
            return cursor.lastrowid
        except Exception as e:
            print(f"Error creando paciente: {e}")
            return None
        finally:
            conn.close()

    def get_patients(self, user_id: int, role: str) -> List[Dict]:
        """Admin ve todos; Doctor ve solo los suyos."""
        conn = self.get_connection()
        try:
            if role == "Admin":
                rows = conn.execute(
                    """SELECT p.*, u.username as doctorName,
                           COUNT(c.consultationID) as consultationCount,
                           MAX(c.consultationDate) as lastConsultationDate
                       FROM Patients p
                       JOIN Users u ON p.createdByUserID = u.userID
                       LEFT JOIN Consultations c ON c.patientID = p.patientID
                       GROUP BY p.patientID
                       ORDER BY p.creationDate DESC"""
                ).fetchall()
            else:
                rows = conn.execute(
                    """SELECT p.*, u.username as doctorName,
                           COUNT(c.consultationID) as consultationCount,
                           MAX(c.consultationDate) as lastConsultationDate
                       FROM Patients p
                       JOIN Users u ON p.createdByUserID = u.userID
                       LEFT JOIN Consultations c ON c.patientID = p.patientID
                       WHERE p.createdByUserID = ?
                       GROUP BY p.patientID
                       ORDER BY p.creationDate DESC""",
                    (user_id,)
                ).fetchall()
            return [self._serialize(r) for r in rows]
        finally:
            conn.close()

    def get_patient(self, patient_id: int) -> Optional[Dict]:
        conn = self.get_connection()
        try:
            row = conn.execute(
                """SELECT p.*, u.username as doctorName
                   FROM Patients p JOIN Users u ON p.createdByUserID = u.userID
                   WHERE p.patientID = ?""",
                (patient_id,)
            ).fetchone()
            return self._serialize(row) if row else None
        finally:
            conn.close()

    def search_patients(self, search_term: str, user_id: int, role: str) -> List[Dict]:
        conn = self.get_connection()
        try:
            like = f"%{search_term}%"
            if role == "Admin":
                rows = conn.execute(
                    """SELECT p.*, u.username as doctorName,
                           COUNT(c.consultationID) as consultationCount,
                           MAX(c.consultationDate) as lastConsultationDate
                       FROM Patients p
                       JOIN Users u ON p.createdByUserID = u.userID
                       LEFT JOIN Consultations c ON c.patientID = p.patientID
                       WHERE p.name LIKE ?
                       GROUP BY p.patientID
                       ORDER BY p.name""",
                    (like,)
                ).fetchall()
            else:
                rows = conn.execute(
                    """SELECT p.*, u.username as doctorName,
                           COUNT(c.consultationID) as consultationCount,
                           MAX(c.consultationDate) as lastConsultationDate
                       FROM Patients p
                       JOIN Users u ON p.createdByUserID = u.userID
                       LEFT JOIN Consultations c ON c.patientID = p.patientID
                       WHERE p.name LIKE ? AND p.createdByUserID = ?
                       GROUP BY p.patientID
                       ORDER BY p.name""",
                    (like, user_id)
                ).fetchall()
            return [self._serialize(r) for r in rows]
        finally:
            conn.close()

    def update_patient(self, patient_id: int, **kwargs) -> bool:
        conn = self.get_connection()
        try:
            fields = {k: v for k, v in kwargs.items() if v is not None}
            if not fields:
                return False
            set_clause = ", ".join(f"{k}=?" for k in fields)
            conn.execute(
                f"UPDATE Patients SET {set_clause} WHERE patientID=?",
                list(fields.values()) + [patient_id]
            )
            conn.commit()
            return True
        except Exception:
            return False
        finally:
            conn.close()

    def delete_patient(self, patient_id: int) -> bool:
        conn = self.get_connection()
        try:
            conn.execute("DELETE FROM Patients WHERE patientID=?", (patient_id,))
            conn.commit()
            return True
        except Exception:
            return False
        finally:
            conn.close()

    # ------------------------------------------------------------------ #
    #  FACTORES DE RIESGO
    # ------------------------------------------------------------------ #
    def get_all_risk_factors(self) -> List[Dict]:
        conn = self.get_connection()
        try:
            rows = conn.execute("SELECT * FROM RiskFactors ORDER BY name").fetchall()
            return [self._serialize(r) for r in rows]
        finally:
            conn.close()

    def get_patient_risk_factors(self, patient_id: int) -> List[Dict]:
        conn = self.get_connection()
        try:
            rows = conn.execute(
                """SELECT rf.* FROM RiskFactors rf
                   JOIN PatientsRiskFactors prf ON rf.riskFactorID = prf.riskFactorID
                   WHERE prf.patientID = ?""",
                (patient_id,)
            ).fetchall()
            return [self._serialize(r) for r in rows]
        finally:
            conn.close()

    def add_patient_risk_factor(self, patient_id: int, risk_factor_id: int) -> bool:
        conn = self.get_connection()
        try:
            conn.execute(
                "INSERT OR IGNORE INTO PatientsRiskFactors (patientID, riskFactorID) VALUES (?,?)",
                (patient_id, risk_factor_id)
            )
            conn.commit()
            return True
        except Exception:
            return False
        finally:
            conn.close()

    def remove_patient_risk_factor(self, patient_id: int, risk_factor_id: int) -> bool:
        conn = self.get_connection()
        try:
            conn.execute(
                "DELETE FROM PatientsRiskFactors WHERE patientID=? AND riskFactorID=?",
                (patient_id, risk_factor_id)
            )
            conn.commit()
            return True
        except Exception:
            return False
        finally:
            conn.close()

    # ------------------------------------------------------------------ #
    #  CONSULTAS
    # ------------------------------------------------------------------ #
    def create_consultation(self, patient_id: int, created_by_user_id: int,
                            has_dr: bool, confidence: float, raw_output: float,
                            notes: str = "") -> Optional[int]:
        conn = self.get_connection()
        try:
            cursor = conn.execute(
                """INSERT INTO Consultations
                   (patientID, createdByUserID, diabeticRetinopathy, notes, confidence, rawOutput)
                   VALUES (?,?,?,?,?,?)""",
                (patient_id, created_by_user_id, has_dr, notes, confidence, raw_output)
            )
            conn.commit()
            return cursor.lastrowid
        except Exception as e:
            print(f"Error creando consulta: {e}")
            return None
        finally:
            conn.close()

    def get_patient_consultations(self, patient_id: int) -> List[Dict]:
        conn = self.get_connection()
        try:
            rows = conn.execute(
                """SELECT c.*, u.username as doctorName
                   FROM Consultations c JOIN Users u ON c.createdByUserID = u.userID
                   WHERE c.patientID = ?
                   ORDER BY c.consultationDate DESC""",
                (patient_id,)
            ).fetchall()
            return [self._serialize(r) for r in rows]
        finally:
            conn.close()

    def get_consultations(self, user_id: int, role: str,
                          page: int = 1, per_page: int = 10,
                          search: str = "", filter_type: str = "all") -> Dict:
        conn = self.get_connection()
        try:
            base = """FROM Consultations c
                      JOIN Patients p ON c.patientID = p.patientID
                      JOIN Users u ON c.createdByUserID = u.userID"""
            conditions = []
            params = []

            if role != "Admin":
                conditions.append("c.createdByUserID = ?")
                params.append(user_id)

            if search.strip():
                conditions.append("(p.name LIKE ? OR c.notes LIKE ?)")
                params.extend([f"%{search}%", f"%{search}%"])

            if filter_type == "positive":
                conditions.append("c.diabeticRetinopathy = 1")
            elif filter_type == "negative":
                conditions.append("c.diabeticRetinopathy = 0")

            where = ("WHERE " + " AND ".join(conditions)) if conditions else ""

            total = conn.execute(f"SELECT COUNT(*) {base} {where}", params).fetchone()[0]
            total_pages = max(1, (total + per_page - 1) // per_page)
            offset = (page - 1) * per_page

            rows = conn.execute(
                f"""SELECT c.*, p.name as patientName, u.username as doctorName
                    {base} {where}
                    ORDER BY c.consultationDate DESC
                    LIMIT ? OFFSET ?""",
                params + [per_page, offset]
            ).fetchall()

            consultations = [self._serialize(r) for r in rows]
            return {
                "success": True,
                "consultations": consultations,
                "pagination": {
                    "current_page": page,
                    "per_page": per_page,
                    "total_pages": total_pages,
                    "total_records": total,
                    "has_previous": page > 1,
                    "has_next": page < total_pages
                }
            }
        except Exception as e:
            return {"success": False, "error": str(e), "consultations": [], "pagination": {}}
        finally:
            conn.close()

    def get_dashboard_stats(self, user_id: int, role: str) -> Dict:
        conn = self.get_connection()
        try:
            filter_clause = "" if role == "Admin" else "AND c.createdByUserID = ?"
            params = [] if role == "Admin" else [user_id]

            today = datetime.now().strftime('%Y-%m-%d')

            stats = conn.execute(f"""
                SELECT
                    COUNT(DISTINCT c.patientID) as total_patients,
                    COUNT(*) as total_consultations,
                    SUM(CASE WHEN c.diabeticRetinopathy=1 THEN 1 ELSE 0 END) as positive_cases,
                    SUM(CASE WHEN c.diabeticRetinopathy=0 THEN 1 ELSE 0 END) as negative_cases,
                    SUM(CASE WHEN date(c.consultationDate)=? THEN 1 ELSE 0 END) as today_consultations
                FROM Consultations c
                WHERE 1=1 {filter_clause}
            """, [today] + params).fetchone()

            row = self._serialize(stats)
            total = row.get("total_consultations") or 0
            pos = row.get("positive_cases") or 0
            row["positivity_rate"] = round((pos / total * 100), 1) if total > 0 else 0
            return {"success": True, "stats": row}
        except Exception as e:
            return {"success": False, "error": str(e)}
        finally:
            conn.close()

    # ------------------------------------------------------------------ #
    #  TAREAS (por usuario)
    # ------------------------------------------------------------------ #
    def get_tasks(self, user_id: int) -> List[Dict]:
        conn = self.get_connection()
        try:
            rows = conn.execute(
                "SELECT * FROM Tasks WHERE userID=? ORDER BY completed, creationDate DESC",
                (user_id,)
            ).fetchall()
            return [self._serialize(r) for r in rows]
        finally:
            conn.close()

    def add_task(self, user_id: int, text: str) -> Optional[Dict]:
        conn = self.get_connection()
        try:
            cursor = conn.execute(
                "INSERT INTO Tasks (userID, text) VALUES (?,?)",
                (user_id, text)
            )
            conn.commit()
            row = conn.execute("SELECT * FROM Tasks WHERE taskID=?", (cursor.lastrowid,)).fetchone()
            return self._serialize(row)
        finally:
            conn.close()

    def toggle_task(self, task_id: int, user_id: int) -> bool:
        conn = self.get_connection()
        try:
            conn.execute(
                "UPDATE Tasks SET completed = NOT completed WHERE taskID=? AND userID=?",
                (task_id, user_id)
            )
            conn.commit()
            return True
        except Exception:
            return False
        finally:
            conn.close()

    def delete_task(self, task_id: int, user_id: int) -> bool:
        conn = self.get_connection()
        try:
            conn.execute("DELETE FROM Tasks WHERE taskID=? AND userID=?", (task_id, user_id))
            conn.commit()
            return True
        except Exception:
            return False
        finally:
            conn.close()