#!/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()