import sqlite3
DB_PATH = "[Link]"
class UsuariosDB:
def __init__(self):
self._init_db()
self._add_columns_if_missing()
def _get_conn(self):
conn = [Link](DB_PATH)
[Link]("PRAGMA foreign_keys = ON")
return conn
def _init_db(self):
conn = self._get_conn()
cur = [Link]()
[Link]("""
CREATE TABLE IF NOT EXISTS usuarios (
id INTEGER PRIMARY KEY AUTOINCREMENT,
dni INTEGER UNIQUE NOT NULL,
nombre TEXT NOT NULL,
contraseña TEXT NOT NULL,
tipo TEXT CHECK(tipo IN ('Administrador','Usuario')) NOT NULL
);
""")
[Link]()
[Link]()
# ================= AÑADIR COLUMNAS SI FALTAN =================
def _add_columns_if_missing(self):
conn = self._get_conn()
cur = [Link]()
# Pregunta
try:
[Link]("ALTER TABLE usuarios ADD COLUMN pregunta TEXT;")
except [Link]:
pass # ya existe
# Respuesta
try:
[Link]("ALTER TABLE usuarios ADD COLUMN respuesta TEXT;")
except [Link]:
pass # ya existe
[Link]()
[Link]()
# ================= ADMINISTRADORES =================
def contar_admins(self):
conn = self._get_conn()
cur = [Link]()
[Link]("SELECT COUNT(*) FROM usuarios WHERE tipo='Administrador';")
count = [Link]()[0]
[Link]()
return count
# ================= REGISTRO =================
def insertar_usuario(self, dni, nombre, contraseña, tipo,
pregunta=None, respuesta=None):
if tipo == "Administrador" and self.contar_admins() >= 2:
return False, "Ya se ha alcanzado el número máximo de administradores"
conn = self._get_conn()
cur = [Link]()
try:
[Link]("""
INSERT INTO usuarios
(dni, nombre, contraseña, tipo, pregunta, respuesta)
VALUES (?, ?, ?, ?, ?, ?);
""", (dni, nombre, contraseña, tipo, pregunta, respuesta))
[Link]()
return True, "Usuario registrado correctamente"
except [Link]:
return False, "DNI ya existe"
finally:
[Link]()
# ================= CONSULTAS =================
def buscar_por_dni(self, dni):
conn = self._get_conn()
cur = [Link]()
[Link]("SELECT * FROM usuarios WHERE dni=?;", (dni,))
row = [Link]()
[Link]()
return row
def listar_usuarios(self):
conn = self._get_conn()
cur = [Link]()
[Link]("SELECT id, dni, nombre, tipo FROM usuarios;")
rows = [Link]()
[Link]()
return rows
def eliminar_usuario(self, usuario_id):
conn = self._get_conn()
cur = [Link]()
[Link]("DELETE FROM usuarios WHERE id=?;", (usuario_id,))
[Link]()
[Link]()
# ================= LOGIN =================
def validar_login(self, dni, contraseña):
conn = self._get_conn()
cur = [Link]()
[Link]("""
SELECT id, nombre, tipo
FROM usuarios
WHERE dni=? AND contraseña=?;
""", (dni, contraseña))
user = [Link]()
[Link]()
return user
# ================= RECUPERAR CONTRASEÑA =================
def obtener_pregunta(self, dni):
conn = self._get_conn()
cur = [Link]()
[Link]("""
SELECT pregunta, respuesta
FROM usuarios
WHERE dni=?;
""", (dni,))
data = [Link]()
[Link]()
return data
def actualizar_contraseña(self, dni, nueva_contraseña):
conn = self._get_conn()
cur = [Link]()
[Link]("""
UPDATE usuarios
SET contraseña=?
WHERE dni=?;
""", (nueva_contraseña, dni))
[Link]()
[Link]()