86 lines
3.0 KiB
Python
86 lines
3.0 KiB
Python
import sqlite3
|
|
import logging
|
|
from contextlib import contextmanager
|
|
from datetime import datetime
|
|
|
|
class DBHandler:
|
|
def __init__(self, db_path='qr_codes.db'):
|
|
self.db_path = db_path
|
|
self.logger = logging.getLogger(__name__)
|
|
self.setup_database()
|
|
|
|
@contextmanager
|
|
def get_connection(self):
|
|
"""Administra la conexión a la base de datos"""
|
|
conn = None
|
|
try:
|
|
conn = sqlite3.connect(self.db_path)
|
|
yield conn
|
|
except sqlite3.Error as e:
|
|
self.logger.error(f"Error de conexión a la base de datos: {e}")
|
|
raise
|
|
finally:
|
|
if conn:
|
|
conn.close()
|
|
|
|
def setup_database(self):
|
|
"""Crea la tabla si no existe"""
|
|
create_table_sql = '''
|
|
CREATE TABLE IF NOT EXISTS qr_codes (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
qr_data TEXT NOT NULL UNIQUE,
|
|
camera_name TEXT NOT NULL,
|
|
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
|
|
);
|
|
'''
|
|
try:
|
|
with self.get_connection() as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute(create_table_sql)
|
|
conn.commit()
|
|
self.logger.info("Base de datos inicializada correctamente")
|
|
except sqlite3.Error as e:
|
|
self.logger.error(f"Error creando la tabla: {e}")
|
|
raise
|
|
|
|
def check_qr_exists(self, qr_data: str) -> bool:
|
|
"""Verifica si un código QR ya existe en la base de datos"""
|
|
query = 'SELECT COUNT(*) FROM qr_codes WHERE qr_data = ?'
|
|
try:
|
|
with self.get_connection() as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute(query, (qr_data,))
|
|
(count,) = cursor.fetchone()
|
|
return count > 0
|
|
except sqlite3.Error as e:
|
|
self.logger.error(f"Error verificando QR: {e}")
|
|
return False
|
|
|
|
def save_qr(self, qr_data: str,camera_name: str) -> bool:
|
|
"""Guarda un nuevo código QR en la base de datos"""
|
|
if self.check_qr_exists(qr_data):
|
|
return False
|
|
|
|
insert_sql = 'INSERT INTO qr_codes (qr_data,camera_name) VALUES (?, ?)'
|
|
try:
|
|
with self.get_connection() as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute(insert_sql, (qr_data,camera_name))
|
|
conn.commit()
|
|
self.logger.info(f"QR guardado en DB: {qr_data}")
|
|
return True
|
|
except sqlite3.Error as e:
|
|
self.logger.error(f"Error guardando QR: {e}")
|
|
return False
|
|
|
|
def get_all_qrs(self) -> list:
|
|
"""Obtiene todos los códigos QR almacenados"""
|
|
query = 'SELECT qr_data, created_at FROM qr_codes ORDER BY created_at DESC'
|
|
try:
|
|
with self.get_connection() as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute(query)
|
|
return cursor.fetchall()
|
|
except sqlite3.Error as e:
|
|
self.logger.error(f"Error obteniendo QRs: {e}")
|
|
return [] |