#!/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() except Exception as e: print(f"Error insertando datos por defecto: {e}") # ------------------------------------------------------------------ # # 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 FROM Patients p JOIN Users u ON p.createdByUserID = u.userID ORDER BY p.creationDate DESC""" ).fetchall() else: rows = conn.execute( """SELECT p.*, u.username as doctorName FROM Patients p JOIN Users u ON p.createdByUserID = u.userID WHERE p.createdByUserID = ? 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 FROM Patients p JOIN Users u ON p.createdByUserID = u.userID WHERE p.name LIKE ? ORDER BY p.name""", (like,) ).fetchall() else: rows = conn.execute( """SELECT p.*, u.username as doctorName FROM Patients p JOIN Users u ON p.createdByUserID = u.userID WHERE p.name LIKE ? AND p.createdByUserID = ? 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_consultation_by_id(self, consultation_id: int, user_id: int, role: str) -> Optional[Dict]: """Devuelve una consulta completa con datos del paciente y factores de riesgo. Los doctores solo pueden ver sus propias consultas.""" conn = self.get_connection() try: query = """ SELECT c.consultationID, c.patientID, c.createdByUserID, c.diabeticRetinopathy, c.notes, c.consultationDate, c.imagePath, c.confidence, c.rawOutput, p.name AS patientName, p.birthDate AS birthDate, p.gender AS gender, p.diabetesType AS diabetesType, u.username AS doctorName FROM Consultations c JOIN Patients p ON c.patientID = p.patientID JOIN Users u ON c.createdByUserID = u.userID WHERE c.consultationID = ? """ params = [consultation_id] if role != "Admin": query += " AND c.createdByUserID = ?" params.append(user_id) row = conn.execute(query, params).fetchone() if not row: return None result = self._serialize(row) # Cargar factores de riesgo del paciente rf_rows = conn.execute( """SELECT rf.riskFactorID, rf.name, rf.description FROM RiskFactors rf JOIN PatientsRiskFactors prf ON rf.riskFactorID = prf.riskFactorID WHERE prf.patientID = ?""", (result["patientID"],) ).fetchall() result["riskFactors"] = [self._serialize(r) for r in rf_rows] return result 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, p.birthDate AS birthDate, p.gender AS gender, p.diabetesType AS diabetesType, 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()