From c9940ec377f131a345416c76aee3a1e85aadfffc Mon Sep 17 00:00:00 2001 From: alumnos_ti Date: Wed, 1 Jul 2026 20:38:56 +0000 Subject: [PATCH] Subir archivos a "/" --- check_db.py | 28 ++ config.py | 28 ++ controlador.py | 832 +++++++++++++++++++++++++++++++++++++++++++++ dao.py | 906 +++++++++++++++++++++++++++++++++++++++++++++++++ database.py | 167 +++++++++ 5 files changed, 1961 insertions(+) create mode 100644 check_db.py create mode 100644 config.py create mode 100644 controlador.py create mode 100644 dao.py create mode 100644 database.py diff --git a/check_db.py b/check_db.py new file mode 100644 index 0000000..dfd2013 --- /dev/null +++ b/check_db.py @@ -0,0 +1,28 @@ +import pyodbc +import config + +conn_str = ( + f"DRIVER={config.DB_DRIVER};" + f"SERVER={config.DB_SERVER};" + f"DATABASE={config.DB_NAME};" + f"Trusted_Connection=yes;" +) +try: + conn = pyodbc.connect(conn_str) + cursor = conn.cursor() + cursor.execute("SELECT table_name FROM information_schema.tables WHERE table_type = 'BASE TABLE'") + tables = [row[0] for row in cursor.fetchall()] + print("Tables:") + for t in tables: + print(f" - {t}") + + # Check if we can describe Recursos, Operarios, etc. + for t in ['Recursos', 'Operarios', 'Articulos', 'Relevamientos']: + if t in tables: + print(f"\nColumns in {t}:") + cursor.execute(f"SELECT column_name, data_type FROM information_schema.columns WHERE table_name = '{t}'") + for row in cursor.fetchall(): + print(f" {row[0]}: {row[1]}") + conn.close() +except Exception as e: + print("Error:", e) diff --git a/config.py b/config.py new file mode 100644 index 0000000..3255ab2 --- /dev/null +++ b/config.py @@ -0,0 +1,28 @@ +import os +from typing import Final + +# Colores originales (UI/UX no modificable) +COLOR_BARRA_SUPERIOR: Final[str] = "#2c3e50" +COLOR_BOTON_ACTIVO: Final[str] = "#1abc9c" +COLOR_BG_INICIO: Final[str] = "#ecf0f1" +COLOR_BG_TAREAS: Final[str] = "#d5dbdb" +COLOR_BG_TABLA_HEAD: Final[str] = "#f0f0f0" +COLOR_BOTON_FINALIZAR: Final[str] = "#2980b9" + +# Configuración de ventana +TITULO_APP: Final[str] = "Metodos y Tiempos" +GEOMETRIA: Final[str] = "700x600" +ICONO_PATH: Final[str] = "logo.ico" + +# Configuración de Base de Datos +DB_DRIVER: Final[str] = "{ODBC Driver 17 for SQL Server}" +DB_SERVER: Final[str] = os.getenv("DB_SERVER", "localhost\\SQLEXPRESS") +DB_NAME: Final[str] = os.getenv("DB_NAME", "MetodosTiemposDB") +DB_USER: Final[str] = os.getenv("DB_USER", "sa") +DB_PASSWORD: Final[str] = os.getenv("DB_PASSWORD", "secret") + +# Log Configuration +LOG_FILE: Final[str] = "app_metodos_tiempos.log" + +# Seguridad +MASTER_PASSWORD: Final[str] = "LongvieAdmin2026" \ No newline at end of file diff --git a/controlador.py b/controlador.py new file mode 100644 index 0000000..fc173ef --- /dev/null +++ b/controlador.py @@ -0,0 +1,832 @@ +import tkinter as tk +from tkinter import messagebox, filedialog, simpledialog +from typing import Dict, List, Optional, Any +import gestor_pdf +import vistas +import config +from modelos import Usuario, Tarea +from logger import get_logger +from dao import UsuarioDAO, TareaDAO +from seguridad import Seguridad + +logger = get_logger(__name__) + +class Controlador: + """ + Controlador principal MVC. + Encapsula la lógica de negocio (Mock data, RBAC) y las funciones de archivo (PDF) + que antes residían en main.py. + """ + def __init__(self, root: Any) -> None: + # root es la instancia de Aplicacion (tk.Tk) + self.root = root + self.usuario_actual: Optional[Usuario] = None + + def verificar_credenciales(self, usuario: str, contrasena: str) -> bool: + """Verifica credenciales en SQL Server e inicia sesión.""" + try: + user = UsuarioDAO.obtener_por_username(usuario) + if user and Seguridad.verificar_password(contrasena, user.password_hash): + self.usuario_actual = user + logger.info(f"Usuario {usuario} autenticado exitosamente como {user.rol}.") + + # Auditoría + Seguridad.registrar_auditoria(user.id_usuario, "Login exitoso") + + # Al loguearse, mostrar la barra de menú nativa + self.root.config(menu=self.root.barra_menus) + self.root.title(f"{config.TITULO_APP} - Dashboard") + # Redirigir a Dashboard + self.root.mostrar_frame(vistas.PaginaTareas) + return True + + logger.warning(f"Intento fallido para el usuario: {usuario}") + return False + except Exception: + logger.exception("Error al verificar credenciales en BD.") + return False + + def cerrar_sesion(self) -> None: + """ + Cierra la sesión del usuario actual. + + Limpia el estado de autenticación, oculta la barra de menú nativa + y redirige a la pantalla de Login. + """ + if self.usuario_actual: + Seguridad.registrar_auditoria( + self.usuario_actual.id_usuario, "Cerró sesión" + ) + logger.info(f"Usuario {self.usuario_actual.username} cerró sesión.") + self.usuario_actual = None + self.root.config(menu="") + self.root.title(config.TITULO_APP) + self.root.mostrar_frame(vistas.PaginaLogin) + + def obtener_tareas_dashboard(self) -> List[Tarea]: + """Obtiene tareas filtradas según el rol del usuario.""" + if not self.usuario_actual: + return [] + if self.usuario_actual.rol == "JEFE/GERENTE": + return TareaDAO.obtener_todas() + else: + return TareaDAO.obtener_tareas_por_usuario(self.usuario_actual.id_usuario) + + def obtener_metodistas(self) -> List[Usuario]: + """Obtiene los usuarios con rol METODISTA.""" + return UsuarioDAO.obtener_metodistas() + + def obtener_opciones_estado(self) -> List[str]: + """Devuelve las opciones de estado permitidas según el rol del usuario.""" + if not self.usuario_actual: + return [] + if self.usuario_actual.rol == "JEFE/GERENTE": + return ["Pendiente", "En Proceso", "Finalizado", "Cerrado"] + else: + return ["Pendiente", "En Proceso", "Finalizado"] + + def crear_tarea(self, titulo: str, descripcion: str, id_usuario_asignado: int) -> bool: + """Delega la creación de la tarea al DAO.""" + exito = TareaDAO.crear_tarea(titulo, descripcion, id_usuario_asignado) + if exito and self.usuario_actual: + Seguridad.registrar_auditoria(self.usuario_actual.id_usuario, f"Creó tarea: {titulo}") + return exito + + def borrar_tarea(self, id_tarea: int) -> bool: + """Delega el borrado de la tarea al DAO.""" + exito = TareaDAO.borrar_tarea(id_tarea) + if exito and self.usuario_actual: + Seguridad.registrar_auditoria(self.usuario_actual.id_usuario, f"Borrador tarea ID: {id_tarea}") + return exito + + def cambiar_estado_tarea(self, id_tarea: int, nuevo_estado: str) -> bool: + try: + exito = TareaDAO.actualizar_estado(id_tarea, nuevo_estado) + if exito and self.usuario_actual: + Seguridad.registrar_auditoria(self.usuario_actual.id_usuario, f"Cambio de estado en tarea {id_tarea} a {nuevo_estado}") + logger.info(f"Tarea {id_tarea} cambió a estado {nuevo_estado}.") + return exito + except Exception: + logger.exception("Error al cambiar estado de la tarea.") + return False + + def registrar_nuevo_usuario(self, nombre: str, username: str, rol: str, contrasena: str) -> None: + """Registra un usuario en el sistema previniendo escalada de privilegios.""" + if rol == "JEFE/GERENTE": + master_pwd = simpledialog.askstring("Autorización Requerida", + "Ingrese la Contraseña Maestra para crear un Administrador:", + show="*", parent=self.root) + if master_pwd != config.MASTER_PASSWORD: + messagebox.showerror("Acceso Denegado", "Contraseña maestra incorrecta. Operación cancelada.") + return + + # Verificar si ya existe para dar feedback claro + if UsuarioDAO.obtener_por_username(username) is not None: + messagebox.showerror("Error", "El nombre de usuario ya está en uso.") + return + + hash_pwd = Seguridad.generar_hash(contrasena) + + if UsuarioDAO.crear_usuario(username, hash_pwd, rol, nombre): + messagebox.showinfo("Éxito", f"Usuario '{username}' ({rol}) creado correctamente.") + if self.usuario_actual: + Seguridad.registrar_auditoria(self.usuario_actual.id_usuario, f"Creó usuario: {username} ({rol})") + self.root.mostrar_frame(vistas.PaginaLogin) + else: + messagebox.showerror("Error", "No se pudo crear el usuario en la base de datos.") + + # --- LÓGICA MOVIDA DESDE MAIN.PY --- + + def nueva_plantilla(self) -> None: + """Renderiza la vista original preexistente para crear una nueva plantilla de cronometraje.""" + self.root.mostrar_frame(vistas.Plantilla) + + def abrir_archivo(self) -> None: + """Abre un archivo utilizando filedialog.""" + tipos = [('Archivos de texto', '*.txt'), ('Todos los archivos', '*.*')] + ruta = filedialog.askopenfilename(title="Seleccionar un archivo", filetypes=tipos) + if ruta: + messagebox.showinfo("Abrir", f"Archivo seleccionado: {ruta}") + + def guardar_archivo(self) -> bool: + """ + Guarda el PDF con auto-naming basado en el código del artículo. + + Lee el código del artículo desde la vista Plantilla para generar + un nombre de archivo predeterminado. Extrae los datos completos + de la grilla y delega al gestor_pdf para el renderizado landscape. + + Returns: + bool: True si el PDF se generó exitosamente, False en caso contrario. + """ + # Fase 1: Auto-Naming — leer código del artículo desde la vista + vista_plantilla = self.root.frames.get(vistas.Plantilla) + nombre_articulo: str = "REPORTE_NUEVO" + if vista_plantilla and hasattr(vista_plantilla, 'entry_articulo_codigo'): + codigo = vista_plantilla.entry_articulo_codigo.get().strip() + if codigo: + nombre_articulo = codigo + + ruta = filedialog.asksaveasfilename( + defaultextension=".pdf", + filetypes=[("Archivos PDF", "*.pdf")], + initialfile=f"{nombre_articulo}.pdf" + ) + if not ruta: + return False + + try: + if vista_plantilla and hasattr(vista_plantilla, 'obtener_datos_limpios'): + datos_completos: Dict[str, Any] = vista_plantilla.obtener_datos_limpios() + exito = gestor_pdf.generar_reporte_pdf(datos_completos, ruta) + if exito: + messagebox.showinfo("Éxito", "PDF generado correctamente.") + return True + else: + messagebox.showerror("Error", "Fallo al generar PDF. Revise el log del sistema.") + return False + else: + logger.error("No se encontró la vista Plantilla para obtener datos.") + messagebox.showerror("Error", "La vista actual no soporta el guardado de datos.") + return False + except Exception: + logger.exception("Error inesperado al intentar guardar el PDF.") + messagebox.showerror("Error", "Ocurrió un error crítico. Revise el archivo de log.") + return False + + def obtener_siguiente_id_relevamiento(self) -> int: + """Obtiene el correlativo máximo + 1 de Relevamientos de la BD.""" + from dao import RelevamientoDAO + return RelevamientoDAO.obtener_siguiente_id() + + def obtener_lista_maquinas(self) -> List[str]: + """Obtiene la lista de nombres de máquinas registradas.""" + from dao import RecursoDAO + return [r.nombre for r in RecursoDAO.obtener_por_tipo("MAQUINA")] + + def obtener_lista_hms(self) -> List[str]: + """Obtiene la lista de nombres de herramientas/matrices registradas.""" + from dao import RecursoDAO + return [r.nombre for r in RecursoDAO.obtener_por_tipo("HM")] + + def obtener_lista_operaciones(self) -> List[str]: + """Obtiene la lista de nombres de operaciones registradas. + + Reutiliza la tabla Recursos almacenando las operaciones bajo + el tipo 'OPERACION'. Retorna una lista de strings con los + nombres disponibles para poblar el Combobox autocompletable + del campo Operación en la Plantilla. + + Returns: + Lista de nombres de operaciones registradas. Lista vacía + si no existen registros. + """ + from dao import RecursoDAO + return [r.nombre for r in RecursoDAO.obtener_por_tipo("OPERACION")] + + def obtener_lista_operarios(self) -> List[str]: + """Obtiene la lista de nombres de operarios registrados.""" + from dao import OperarioDAO + return [o.nombre for o in OperarioDAO.obtener_todos()] + + def crear_recurso_silencioso(self, nombre: str, tipo: str) -> bool: + """Crea silenciosamente un recurso (maquina o HM) si no existe.""" + from dao import RecursoDAO + recurso = RecursoDAO.crear_silencioso(nombre, tipo) + return recurso is not None + + def crear_recurso(self, nombre: str, tipo: str) -> bool: + """ + Crea un nuevo recurso (MAQUINA o HM) de forma explícita en la base de datos. + Asegura que el recurso no exista previamente antes de crearlo. + + Args: + nombre (str): Nombre o descripción del recurso. + tipo (str): Tipo del recurso ('MAQUINA' o 'HM'). + + Returns: + bool: True si el recurso se crea exitosamente o ya existía, False en caso de error. + """ + from dao import RecursoDAO + try: + nombre_limpio = nombre.strip() + if not nombre_limpio: + logger.warning("Intento de crear un recurso con nombre vacío.") + return False + + # Verificar existencia previa + existente = RecursoDAO.obtener_por_descripcion_y_tipo(nombre_limpio, tipo) + if existente: + logger.info(f"El recurso '{nombre_limpio}' de tipo '{tipo}' ya existe.") + return True + + recurso = RecursoDAO.crear_silencioso(nombre_limpio, tipo) + if recurso: + logger.info(f"Recurso '{nombre_limpio}' de tipo '{tipo}' creado explícitamente.") + if self.usuario_actual: + from seguridad import Seguridad + Seguridad.registrar_auditoria( + self.usuario_actual.id_usuario, + f"Creó recurso explícito: {nombre_limpio} ({tipo})" + ) + return True + return False + except Exception: + logger.exception(f"Error inesperado al crear recurso '{nombre}' del tipo '{tipo}'.") + return False + + def borrar_recurso(self, nombre: str, tipo: str) -> bool: + """ + Elimina un recurso (MAQUINA o HM) por nombre y tipo. + + Delega al DAO, registra auditoría si hay usuario autenticado, + y captura excepciones de forma granular (Regla 7). + + Args: + nombre: Descripción/nombre del recurso a eliminar. + tipo: Tipo del recurso ('MAQUINA' o 'HM'). + + Returns: + True si la eliminación fue exitosa, False en caso contrario. + """ + from dao import RecursoDAO + try: + nombre_limpio = nombre.strip() + if not nombre_limpio: + logger.warning("Intento de eliminar un recurso con nombre vacío.") + return False + exito = RecursoDAO.borrar_por_nombre_y_tipo(nombre_limpio, tipo) + if exito and self.usuario_actual: + Seguridad.registrar_auditoria( + self.usuario_actual.id_usuario, + f"Eliminó recurso: {nombre_limpio} ({tipo})" + ) + return exito + except Exception: + logger.exception(f"Error inesperado al eliminar recurso '{nombre}' ({tipo}).") + return False + + def borrar_relevamiento(self, id_relevamiento: int) -> bool: + """ + Elimina un relevamiento completo (cabecera + tiempos) por ID. + + Delega al DAO, registra auditoría si hay usuario autenticado, + y captura excepciones de forma granular (Regla 7). + + Args: + id_relevamiento: ID del relevamiento a eliminar. + + Returns: + True si la eliminación fue exitosa, False en caso contrario. + """ + from dao import RelevamientoDAO + try: + exito = RelevamientoDAO.borrar_por_id(id_relevamiento) + if exito and self.usuario_actual: + Seguridad.registrar_auditoria( + self.usuario_actual.id_usuario, + f"Eliminó relevamiento ID: {id_relevamiento}" + ) + return exito + except Exception: + logger.exception(f"Error inesperado al eliminar relevamiento {id_relevamiento}.") + return False + + def crear_operario_silencioso(self, nombre: str) -> bool: + """Crea silenciosamente un operario si no existe.""" + from dao import OperarioDAO + operario = OperarioDAO.crear_silencioso(nombre) + return operario is not None + + def guardar_relevamiento_bd(self, relevamiento: Any) -> bool: + """ + Delega la persistencia del relevamiento transaccional al RelevamientoDAO. + Aplica trazabilidad estricta y manejo de errores (Regla 7 y Prohibición de Silenciar Errores). + """ + from dao import RelevamientoDAO + from tkinter import messagebox + import pprint + + # Trazabilidad requerida en consola + print("DEBUG - Intentando guardar relevamiento. Estructura de datos:") + try: + pprint.pprint(vars(relevamiento)) + except Exception: + print(relevamiento) + + try: + exito = RelevamientoDAO.guardar(relevamiento) + return exito + except Exception as e: + logger.exception("Error DB al intentar guardar el relevamiento.") + messagebox.showerror("Error de Base de Datos", f"Se produjo un error crítico al guardar en la base de datos:\n\n{str(e)}") + return False + + def abrir_relevamiento_desde_id(self, id_seleccionado: int) -> None: + """Carga un relevamiento completo desde la BD y abre la vista Plantilla inyectando los datos.""" + print(f"DEBUG: Intentando cargar ID {id_seleccionado}") + from dao import RelevamientoDAO + import vistas + relevamiento = RelevamientoDAO.obtener_por_id(id_seleccionado) + if relevamiento: + self.root.mostrar_frame(vistas.Plantilla) + vista_plantilla = self.root.frames[vistas.Plantilla] + id_tarea_asociada = relevamiento.id_tarea or 0 + vista_plantilla.cargar_datos(relevamiento, id_tarea_asociada) + + def abrir_relevamiento_desde_tarea(self, id_tarea: int) -> None: + """ + Busca si la tarea tiene un relevamiento asociado. + Si existe, lo carga y abre la vista Plantilla. + Si no existe, abre una Plantilla limpia vinculada a la tarea. + + Args: + id_tarea (int): El ID de la tarea del Treeview. + """ + from dao import RelevamientoDAO + import vistas + try: + relevamiento = RelevamientoDAO.obtener_por_tarea(id_tarea) + # Cambiar de vista a Plantilla + self.root.mostrar_frame(vistas.Plantilla) + + # Obtener la instancia de la vista Plantilla + vista_plantilla = self.root.frames[vistas.Plantilla] + + if relevamiento: + logger.info(f"Cargando relevamiento existente {relevamiento.id_relevamiento} para la tarea {id_tarea}.") + vista_plantilla.cargar_datos(relevamiento, id_tarea) + else: + logger.info(f"Inicializando relevamiento nuevo para la tarea {id_tarea}.") + vista_plantilla.cargar_datos_nuevos(id_tarea) + + except Exception: + logger.exception(f"Error al intentar abrir el relevamiento de la tarea {id_tarea}.") + from tkinter import messagebox + messagebox.showerror("Error", f"No se pudo cargar el relevamiento asociado a la tarea {id_tarea}.") + + def obtener_resumen_relevamientos(self) -> List[tuple]: + """Obtiene un resumen de todos los relevamientos desde el DAO.""" + from dao import RelevamientoDAO + return RelevamientoDAO.obtener_todos_resumen() + + def buscar_relevamientos_historicos(self, filtros: Dict[str, str]) -> List[tuple]: + """ + Busca relevamientos históricos aplicando filtros combinados. + + Recibe los filtros de la vista, delega la búsqueda al DAO + y retorna los resultados. Los errores se registran vía logger (Regla 7). + + Args: + filtros: Diccionario con claves opcionales ('articulo', 'operario', + 'maquina', 'hm') mapeadas a cadenas de texto. + + Returns: + Lista de tuplas con formato: + (ID, Fecha, Artículo, Operación, Máquina, HM, Operario). + """ + from dao import RelevamientoDAO + try: + return RelevamientoDAO.buscar_historico(filtros) + except Exception: + logger.exception("Error al buscar relevamientos históricos desde el controlador.") + return [] + + + def importar_desde_excel(self, ruta_archivo: str) -> None: + """ + Importa datos desde una planilla Excel (.xlsx/.xlsm) y los carga en la vista Plantilla. + + Realiza lectura completa de cabecera (operación en F2, descripción en F3, + código artículo en B7, máquina/HM/operario/fecha en columna B) y escaneo + dinámico de la grilla de tiempos a partir de la columna D (índice 4). + + El orden de inyección es crítico: primero se muestra el frame (disparando + actualizar_vista/limpiar_datos) y luego se cargan los datos, evitando que + el ciclo de vida de Tkinter borre la información recién inyectada. + + Args: + ruta_archivo: Ruta absoluta al archivo Excel (.xlsx/.xlsm). + + Raises: + No lanza excepciones al exterior; captura cualquier error y lo + muestra al usuario vía messagebox, registrando el detalle en el log. + """ + import sys + import subprocess + import datetime + from tkinter import messagebox + from modelos import Articulo, Recurso, Operario, Relevamiento, TiemposElemento + + try: + # 1. Asegurar dependencias de forma silenciosa + try: + import openpyxl + except ImportError: + subprocess.check_call([sys.executable, "-m", "pip", "install", "openpyxl"]) + import openpyxl + + # 2. Cargar hoja activa + wb = openpyxl.load_workbook(ruta_archivo, data_only=True) + hoja = wb.worksheets[0] + + def safe_str(cell_ref: str) -> str: + """Extrae el valor de una celda como string limpio, o cadena vacía si es None.""" + val = hoja[cell_ref].value + return str(val).strip() if val is not None else "" + + def safe_float(val: Any, default: float = 0.0) -> float: + """Convierte un valor de celda a float de forma segura, tolerando comas y '%'.""" + if val is None: + return default + if isinstance(val, (int, float)): + return float(val) + try: + return float(str(val).replace(',', '.').replace('%', '').strip()) + except ValueError: + return default + + # 3. Lectura de cabecera (coordenadas confirmadas por PoC) + art_codigo = safe_str('A6') + operacion = safe_str('F2') + descripcion = safe_str('F3') + maquina = safe_str('A20') + hm = safe_str('A22') + operario = safe_str('A26') + + # --- LECTURA DE FECHA Y NÚMERO DE CRONOMETRAJE --- + fecha_val = hoja['A28'].value + if isinstance(fecha_val, datetime.datetime): + fecha_obj = fecha_val.date() + elif isinstance(fecha_val, datetime.date): + fecha_obj = fecha_val + elif isinstance(fecha_val, str): + try: + fecha_obj = datetime.datetime.strptime(fecha_val.strip(), "%d/%m/%Y").date() + except ValueError: + fecha_obj = datetime.date.today() + else: + fecha_obj = datetime.date.today() + + # Guardamos el número de la celda A30 en una variable + num_crono_excel = safe_str('A30') + + piezas_golpe = int(safe_float(hoja['E43'].value, 1.0)) + articulos_golpe = int(safe_float(hoja['E45'].value, 1.0)) + operarios_maq = int(safe_float(hoja['E47'].value, 1.0)) + + # 4. Escaneo dinámico de la grilla de tiempos + tiempos_lista: List[TiemposElemento] = [] + col = 4 # Inicio de matriz en Columna D + columnas_vacias = 0 + + while True: + # Buscar nombre del elemento (filas 4-9, ignorando encabezados de sub-columna) + elem_nombre: Optional[str] = None + for fila_busqueda in range(4, 10): + val = hoja.cell(row=fila_busqueda, column=col).value + if val and str(val).strip() and str(val).strip().lower() not in ['v', 'min', 'm/seg', 'l']: + elem_nombre = str(val).strip() + break + + # Verificar si hay datos numéricos en la zona de observaciones (filas 11-30) + tiene_datos = False + for r in range(11, 31): + if (hoja.cell(row=r, column=col).value is not None or + hoja.cell(row=r, column=col + 1).value is not None or + hoja.cell(row=r, column=col + 2).value is not None): + tiene_datos = True + break + + # Tolerancia de 2 columnas vacías consecutivas antes de cortar + if not elem_nombre and not tiene_datos: + columnas_vacias += 1 + if columnas_vacias >= 2: + break + col += 3 + continue + + columnas_vacias = 0 + + # Auto-naming para elementos sin título en celdas combinadas + if not elem_nombre: + idx_elemento = ((col - 4) // 3) + 1 + elem_nombre = f"Elemento {idx_elemento}" + + # Elementos Medidos (fila 32), Lote Frec. (fila 33) y Suplemento (fila 38) + el_medidos = int(safe_float(hoja.cell(row=32, column=col + 1).value, 1.0)) + lote_f = safe_float(hoja.cell(row=33, column=col + 1).value, 1.0) + + sup_raw = safe_float(hoja.cell(row=38, column=col + 1).value, 0.0) + # FIX: Convertimos todo a decimal puro (0.01, 0.05) para evitar el doble escalado en vistas.py + suplemento = sup_raw / 100.0 if sup_raw >= 1.0 else sup_raw + + # Recorrer observaciones (filas 11 a 30) + for row_idx in range(11, 31): + v_val = hoja.cell(row=row_idx, column=col).value + min_val = hoja.cell(row=row_idx, column=col + 1).value + seg_val = hoja.cell(row=row_idx, column=col + 2).value + + # Solo procesar si hay al menos un dato en la fila + if v_val is not None or min_val is not None or seg_val is not None: + # Valoración (normalizar factor decimal a porcentaje) + v_f = safe_float(v_val, 100.0) + if 0 < v_f <= 1.5: + v_f *= 100.0 + + # Minutos y segundos + m_f = safe_float(min_val, 0.0) + s_f = safe_float(seg_val, 0.0) + + # Sincronización MIN ↔ SEG + if s_f == 0.0 and m_f > 0.0: + s_f = m_f * 60.0 + elif m_f == 0.0 and s_f > 0.0: + m_f = s_f / 60.0 + + tiempos_lista.append(TiemposElemento( + elemento=elem_nombre, + valoracion=v_f, + minutos=m_f, + segundos=s_f, + elementos_medidos=el_medidos, + lote_frecuencial=lote_f, + suplemento=suplemento + )) + + # Avanzar al siguiente bloque de 3 columnas + col += 3 + if col > 100: + break + + # 5. Construir el objeto Relevamiento (borrador nuevo) + art_obj = Articulo(id_articulo="0", codigo=art_codigo, descripcion=descripcion) if art_codigo else None + maq_obj = Recurso(id_recurso="0", nombre=maquina, tipo="MAQUINA") if maquina else None + hm_obj = Recurso(id_recurso="0", nombre=hm, tipo="HM") if hm else None + op_obj = Operario(id_operario="0", nombre=operario, legajo="") if operario else None + + relevamiento = Relevamiento( + id_relevamiento=None, + id_tarea=None, + articulo=art_obj, + operacion=operacion, + recurso=maq_obj, + hm=hm_obj, + operario=op_obj, + fecha=fecha_obj, + postura="pie", + piezas_por_golpe=piezas_golpe, + articulos_por_golpe=articulos_golpe, + operarios_maquina=operarios_maq, + tiempos=tiempos_lista, + ) + relevamiento.numero_importado = num_crono_excel + + + # 6. Inyectar en la vista Plantilla + # FIX CICLO DE VIDA TKINTER: primero mostrar (dispara limpiar_datos), + # luego cargar datos para que no sean borrados por actualizar_vista(). + import vistas + self.root.mostrar_frame(vistas.Plantilla) + vista_plantilla = self.root.frames[vistas.Plantilla] + vista_plantilla.cargar_datos(relevamiento, 0) + + messagebox.showinfo( + "Importación Exitosa", + "Planilla cargada correctamente como borrador nuevo." + ) + + except Exception as e: + import traceback + traceback.print_exc() + logger.exception("Error crítico durante la importación de Excel.") + messagebox.showerror( + "Error de Importación", + f"Hubo un error al leer el archivo Excel:\n\n{str(e)}" + ) + + + def calcular_metricas_globales(self, suma_TE_columnas: float, piezas_por_golpe: float, articulos_por_golpe: float) -> Dict[str, Any]: + """ + Realiza el cálculo en cascada de las métricas globales y de Epicor. + Retorna un diccionario con los valores formateados a 3 decimales o vacíos en caso de error/división por cero. + """ + res = { + "te_operacion": 0.0, + "cadencia_ph": 0.0, + "cadencia_h1000": 0.0, + "estandar_epicor": 0.0, + "dotacion_epicor": 0.0 + } + + try: + if piezas_por_golpe <= 0.0 or articulos_por_golpe <= 0.0 or suma_TE_columnas <= 0.0: + return res + + # TE_Operacion = suma_TE_columnas / piezas_por_golpe + te_operacion = suma_TE_columnas / piezas_por_golpe + res["te_operacion"] = te_operacion + + # cadencia_pcs_hr = 60 / TE_Operacion + if te_operacion <= 0.0: + return res + cadencia_ph = 60.0 / te_operacion + res["cadencia_ph"] = cadencia_ph + + # cadencia_hrs_1000 = 1000 / cadencia_pcs_hr + if cadencia_ph <= 0.0: + return res + cadencia_h1000 = 1000.0 / cadencia_ph + res["cadencia_h1000"] = cadencia_h1000 + + # estandar_epicor = cadencia_hrs_1000 + estandar_epicor = cadencia_h1000 + res["estandar_epicor"] = estandar_epicor + + # dotacion_epicor = estandar_epicor / articulos_por_golpe + dotacion_epicor = estandar_epicor / articulos_por_golpe + res["dotacion_epicor"] = dotacion_epicor + + except ZeroDivisionError: + logger.warning("ZeroDivisionError detectado y manejado silenciosamente en calcular_metricas_globales.") + except Exception: + logger.exception("Error inesperado en calcular_metricas_globales.") + + return res + + def crear_backup_bd(self) -> None: + """ + Solicita al usuario un directorio de destino y ejecuta un backup + nativo de SQL Server (BACKUP DATABASE) a través de ``database.ejecutar_backup``. + + El nombre del archivo se genera automáticamente con un timestamp + (formato ``YYYYMMDD_HHMMSS``) para evitar colisiones. Las barras + diagonales se normalizan a backslash porque SQL Server solo acepta + rutas Windows nativas. + + Muestra un ``messagebox.showinfo`` con la ruta del backup si tiene + éxito, o un ``messagebox.showerror`` si la operación falla. + + Returns: + None. La función comunica el resultado directamente al usuario + a través de diálogos modales de tkinter. + """ + import datetime + import database + + # 1. Solicitar directorio de destino al usuario + directorio: str = filedialog.askdirectory( + title="Seleccione la carpeta para guardar el Backup" + ) + if not directorio: + logger.info("Operación de backup cancelada por el usuario.") + return + + # 2. Construir nombre de archivo con timestamp único + fecha_str: str = datetime.datetime.now().strftime("%Y%m%d_%H%M%S") + ruta_completa: str = f"{directorio}/Backup_MetodosTiempos_{fecha_str}.bak" + + # 3. Normalizar barras: SQL Server requiere rutas Windows con backslash + ruta_completa = ruta_completa.replace("/", "\\") + + logger.info("Iniciando backup de base de datos hacia: %s", ruta_completa) + + # 4. Ejecutar el backup delegando a la capa de datos + exito: bool = database.ejecutar_backup(ruta_completa) + + # 5. Feedback al usuario + if exito: + messagebox.showinfo( + "Backup Exitoso", + f"El backup de la base de datos se completó correctamente.\n\n" + f"Ubicación:\n{ruta_completa}" + ) + if self.usuario_actual: + Seguridad.registrar_auditoria( + self.usuario_actual.id_usuario, + f"Backup de BD creado en: {ruta_completa}" + ) + else: + messagebox.showerror( + "Error de Backup", + "No se pudo completar el backup de la base de datos.\n\n" + "Revise el archivo de log para más detalles." + ) + + def blanquear_credenciales(self, nombre_completo: str, clave_maestra: str, nueva_pwd: str) -> bool: + """ + Blanquea (resetea) la contraseña de un usuario identificado por su nombre completo. + + Flujo de validación: + 1. Verifica la clave maestra contra ``config.MASTER_PASSWORD``. + 2. Busca al usuario en la BD por nombre completo exacto via DAO. + 3. Genera un nuevo hash bcrypt y actualiza la BD. + 4. Informa al usuario su username recuperado y registra la acción en auditoría. + + Este método está diseñado para ejecutarse desde la pantalla de Login, + donde no existe sesión activa. La auditoría se registra con el ID del + usuario modificado (no del usuario logueado, que no existe en este contexto). + + Args: + nombre_completo: Nombre completo del usuario a recuperar (coincidencia exacta). + clave_maestra: Contraseña maestra ingresada para autorizar la operación. + nueva_pwd: Nueva contraseña en texto plano que será hasheada con bcrypt. + + Returns: + True si la contraseña fue actualizada exitosamente, False en caso contrario. + """ + # --- Validación de clave maestra --- + if clave_maestra != config.MASTER_PASSWORD: + messagebox.showerror( + "Acceso Denegado", + "Contraseña maestra incorrecta. Operación cancelada." + ) + logger.warning("Intento de blanqueo de credenciales con clave maestra incorrecta.") + return False + + # --- Búsqueda del usuario por nombre completo --- + usuario: Optional[Usuario] = UsuarioDAO.obtener_por_nombre_completo(nombre_completo) + if usuario is None: + messagebox.showerror( + "Usuario No Encontrado", + f"No se encontró ningún usuario con el nombre exacto:\n'{nombre_completo}'.\n\n" + f"Verifique que el nombre esté escrito correctamente (respetando mayúsculas y acentos)." + ) + logger.warning(f"Blanqueo fallido: no se encontró usuario con nombre '{nombre_completo}'.") + return False + + # --- Generación de hash y actualización --- + try: + nuevo_hash: str = Seguridad.generar_hash(nueva_pwd) + exito: bool = UsuarioDAO.actualizar_password(usuario.id_usuario, nuevo_hash) + + if exito: + messagebox.showinfo( + "Recuperación Exitosa", + f"Operación exitosa.\n\n" + f"Su usuario recuperado es: {usuario.username}\n" + f"La contraseña ha sido actualizada." + ) + # Auditoría con el ID del usuario modificado (no hay sesión activa) + Seguridad.registrar_auditoria( + usuario.id_usuario, + f"Blanqueo de contraseña ejecutado para '{usuario.username}'" + ) + logger.info(f"Blanqueo de credenciales exitoso para usuario '{usuario.username}' (ID={usuario.id_usuario}).") + return True + else: + messagebox.showerror( + "Error de Actualización", + "No se pudo actualizar la contraseña en la base de datos.\n" + "Revise el archivo de log para más detalles." + ) + return False + except Exception: + logger.exception(f"Error crítico durante el blanqueo de credenciales para '{nombre_completo}'.") + messagebox.showerror( + "Error Crítico", + "Ocurrió un error inesperado durante la operación.\n" + "Revise el archivo de log para más detalles." + ) + return False + diff --git a/dao.py b/dao.py new file mode 100644 index 0000000..4f85305 --- /dev/null +++ b/dao.py @@ -0,0 +1,906 @@ +from typing import Dict, List, Optional +from database import get_db_cursor +from modelos import Articulo, Operario, Recurso, Relevamiento, Usuario, Tarea +from logger import get_logger + +logger = get_logger(__name__) + +class ArticuloDAO: + @staticmethod + def obtener_todos() -> List[Articulo]: + """Obtiene todos los artículos usando consultas parametrizadas (Regla 6).""" + query = "SELECT id_articulo, ISNULL(codigo, '') AS codigo, descripcion FROM Articulos" + articulos: List[Articulo] = [] + try: + with get_db_cursor() as cursor: + cursor.execute(query) + for row in cursor.fetchall(): + articulos.append(Articulo(id_articulo=row[0], codigo=row[1], descripcion=row[2])) + except Exception: + logger.exception("Falló la obtención de artículos.") + return articulos + + @staticmethod + def obtener_por_codigo(codigo: str) -> Optional[Articulo]: + """ + Obtiene un artículo por su código alfanumérico usando consulta parametrizada (Regla 6). + + Args: + codigo: Código alfanumérico del artículo ingresado por el usuario. + + Returns: + Articulo si existe, None en caso contrario. + """ + query = "SELECT id_articulo, ISNULL(codigo, '') AS codigo, descripcion FROM Articulos WHERE codigo = ?" + try: + with get_db_cursor() as cursor: + cursor.execute(query, (codigo,)) + row = cursor.fetchone() + if row: + return Articulo(id_articulo=row[0], codigo=row[1], descripcion=row[2]) + except Exception: + logger.exception(f"Error al buscar artículo por código: {codigo}") + return None + + @staticmethod + def obtener_o_crear(codigo: str, descripcion: str) -> Articulo: + """ + Obtiene un artículo por código o lo crea con código+descripción de forma transaccional (Regla 6). + + Args: + codigo: Código alfanumérico del artículo (ej. 'CHAPA-123'). + descripcion: Descripción textual del artículo. + + Returns: + Articulo existente o recién creado. + + Raises: + Exception: Si falla la inserción en la BD. + """ + try: + art = ArticuloDAO.obtener_por_codigo(codigo) + if art: + return art + + query = "INSERT INTO Articulos (codigo, descripcion) OUTPUT INSERTED.id_articulo VALUES (?, ?)" + with get_db_cursor(commit=True) as cursor: + cursor.execute(query, (codigo, descripcion)) + new_id = cursor.fetchone()[0] + return Articulo(id_articulo=new_id, codigo=codigo, descripcion=descripcion) + except Exception: + logger.exception(f"Error al obtener o crear artículo: código={codigo}, desc={descripcion}") + raise + +class RecursoDAO: + @staticmethod + def obtener_por_tipo(tipo: str) -> List[Recurso]: + """ + Obtiene los recursos filtrados por tipo (MAQUINA o HM) de forma segura (Regla 4 y 6). + """ + query = "SELECT id_recurso, descripcion, tipo FROM Recursos WHERE tipo = ?" + recursos = [] + try: + with get_db_cursor() as cursor: + cursor.execute(query, (tipo,)) + for row in cursor.fetchall(): + recursos.append(Recurso(id_recurso=str(row[0]), nombre=row[1], tipo=row[2])) + except Exception: + logger.exception(f"Error al obtener recursos del tipo {tipo}.") + return recursos + + @staticmethod + def obtener_por_descripcion_y_tipo(descripcion: str, tipo: str) -> Optional[Recurso]: + """ + Obtiene un recurso por su descripción y tipo. + """ + query = "SELECT id_recurso, descripcion, tipo FROM Recursos WHERE descripcion = ? AND tipo = ?" + try: + with get_db_cursor() as cursor: + cursor.execute(query, (descripcion, tipo)) + row = cursor.fetchone() + if row: + return Recurso(id_recurso=str(row[0]), nombre=row[1], tipo=row[2]) + except Exception: + logger.exception("Error al buscar recurso por descripción y tipo.") + return None + + @staticmethod + def crear_silencioso(descripcion: str, tipo: str) -> Optional[Recurso]: + """ + Inserta un nuevo recurso de forma silenciosa si no existe. + """ + try: + existing = RecursoDAO.obtener_por_descripcion_y_tipo(descripcion, tipo) + if existing: + return existing + + query = "INSERT INTO Recursos (descripcion, tipo) OUTPUT INSERTED.id_recurso VALUES (?, ?)" + with get_db_cursor(commit=True) as cursor: + cursor.execute(query, (descripcion, tipo)) + row = cursor.fetchone() + if row: + return Recurso(id_recurso=str(row[0]), nombre=descripcion, tipo=tipo) + except Exception: + logger.exception(f"Error al crear recurso silenciosamente: {descripcion} ({tipo})") + return None + + @staticmethod + def borrar_por_nombre_y_tipo(descripcion: str, tipo: str) -> bool: + """ + Elimina un recurso por su descripción y tipo de forma transaccional. + + Usa consulta parametrizada (Regla 6) y transacción explícita. + Si el recurso está referenciado por Relevamientos, la BD rechazará + la operación por integridad referencial. + + Args: + descripcion: Nombre/descripción del recurso a eliminar. + tipo: Tipo del recurso ('MAQUINA' o 'HM'). + + Returns: + True si se eliminó al menos un registro, False en caso contrario. + + Raises: + No lanza excepciones al exterior; registra en logger (Regla 7). + """ + query = "DELETE FROM Recursos WHERE descripcion = ? AND tipo = ?" + try: + with get_db_cursor(commit=True) as cursor: + cursor.execute(query, (descripcion, tipo)) + eliminado = cursor.rowcount > 0 + if eliminado: + logger.info(f"Recurso '{descripcion}' ({tipo}) eliminado de la BD.") + else: + logger.warning(f"No se encontró recurso '{descripcion}' ({tipo}) para eliminar.") + return eliminado + except Exception: + logger.exception(f"Error al eliminar recurso '{descripcion}' ({tipo}).") + return False + +class OperarioDAO: + @staticmethod + def obtener_todos() -> List[Operario]: + """ + Obtiene todos los operarios registrados. + """ + query = "SELECT id_operario, nombre FROM Operarios" + operarios = [] + try: + with get_db_cursor() as cursor: + cursor.execute(query) + for row in cursor.fetchall(): + operarios.append(Operario(id_operario=str(row[0]), nombre=row[1], legajo="")) + except Exception: + logger.exception("Error al obtener todos los operarios.") + return operarios + + @staticmethod + def obtener_por_nombre(nombre: str) -> Optional[Operario]: + """ + Obtiene un operario por su nombre. + """ + query = "SELECT id_operario, nombre FROM Operarios WHERE nombre = ?" + try: + with get_db_cursor() as cursor: + cursor.execute(query, (nombre,)) + row = cursor.fetchone() + if row: + return Operario(id_operario=str(row[0]), nombre=row[1], legajo="") + except Exception: + logger.exception("Error al buscar operario por nombre.") + return None + + @staticmethod + def crear_silencioso(nombre: str) -> Optional[Operario]: + """ + Inserta un nuevo operario de forma silenciosa si no existe. + """ + try: + existing = OperarioDAO.obtener_por_nombre(nombre) + if existing: + return existing + + query = "INSERT INTO Operarios (nombre) OUTPUT INSERTED.id_operario VALUES (?)" + with get_db_cursor(commit=True) as cursor: + cursor.execute(query, (nombre,)) + row = cursor.fetchone() + if row: + return Operario(id_operario=str(row[0]), nombre=nombre, legajo="") + except Exception: + logger.exception(f"Error al crear operario silenciosamente: {nombre}") + return None + +class RelevamientoDAO: + @staticmethod + def obtener_siguiente_id() -> int: + """ + Obtiene el siguiente ID correlativo para el relevamiento. + """ + query = "SELECT ISNULL(MAX(id_relevamiento), 0) + 1 FROM Relevamientos" + try: + with get_db_cursor() as cursor: + cursor.execute(query) + row = cursor.fetchone() + if row: + return int(row[0]) + except Exception: + logger.exception("Error al obtener el siguiente ID de relevamiento.") + return 1 + + @staticmethod + def guardar(relevamiento: 'Relevamiento') -> bool: + """ + Guarda un relevamiento y sus tiempos asociados en una sola transacción estricta. + Soporta inserción (INSERT) o actualización (UPDATE) si ya existe. + Cumple la Regla 6 (Transacciones explícitas) y la creación dinámica de entidades. + """ + query_insert = """ + INSERT INTO Relevamientos + (id_articulo, operacion, id_recurso, id_hm, id_operario, fecha, postura, piezas_por_golpe, articulos_por_golpe, operarios_maquina, id_tarea) + OUTPUT INSERTED.id_relevamiento + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + """ + query_update = """ + UPDATE Relevamientos SET + id_articulo = ?, operacion = ?, id_recurso = ?, id_hm = ?, id_operario = ?, + fecha = ?, postura = ?, piezas_por_golpe = ?, articulos_por_golpe = ?, operarios_maquina = ?, id_tarea = ? + WHERE id_relevamiento = ? + """ + query_delete_tiempos = """ + DELETE FROM TiemposElemento WHERE id_relevamiento = ? + """ + query_tiempo = """ + INSERT INTO TiemposElemento (id_relevamiento, elemento, valoracion, minutos, segundos, elementos_medidos, lote_frecuencial, suplemento) + VALUES (?, ?, ?, ?, ?, ?, ?, ?) + """ + + from database import get_db_connection + with get_db_connection() as conn: + cursor = conn.cursor() + try: + art_id = relevamiento.articulo.id_articulo if relevamiento.articulo else None + + # --- Gestión dinámica de Entidades Maestras --- + # Máquina + rec_id = None + maq_nombre = None + if isinstance(relevamiento.recurso, str): + maq_nombre = relevamiento.recurso.strip() + elif relevamiento.recurso and hasattr(relevamiento.recurso, 'nombre'): + maq_nombre = relevamiento.recurso.nombre.strip() if relevamiento.recurso.nombre else None + + if maq_nombre: + cursor.execute("SELECT id_recurso FROM Recursos WHERE descripcion = ? AND tipo = 'MAQUINA'", (maq_nombre,)) + row_maq = cursor.fetchone() + if row_maq: + rec_id = row_maq[0] + else: + cursor.execute("INSERT INTO Recursos (descripcion, tipo) OUTPUT INSERTED.id_recurso VALUES (?, 'MAQUINA')", (maq_nombre,)) + rec_id = cursor.fetchone()[0] + + # HM (Herramienta/Matriz) + hm_id = None + hm_nombre = None + if isinstance(relevamiento.hm, str): + hm_nombre = relevamiento.hm.strip() + elif relevamiento.hm and hasattr(relevamiento.hm, 'nombre'): + hm_nombre = relevamiento.hm.nombre.strip() if relevamiento.hm.nombre else None + + if hm_nombre: + cursor.execute("SELECT id_recurso FROM Recursos WHERE descripcion = ? AND tipo = 'HM'", (hm_nombre,)) + row_hm = cursor.fetchone() + if row_hm: + hm_id = row_hm[0] + else: + cursor.execute("INSERT INTO Recursos (descripcion, tipo) OUTPUT INSERTED.id_recurso VALUES (?, 'HM')", (hm_nombre,)) + hm_id = cursor.fetchone()[0] + + # Operario + op_id = None + op_nombre = None + if isinstance(relevamiento.operario, str): + op_nombre = relevamiento.operario.strip() + elif relevamiento.operario and hasattr(relevamiento.operario, 'nombre'): + op_nombre = relevamiento.operario.nombre.strip() if relevamiento.operario.nombre else None + + if op_nombre: + cursor.execute("SELECT id_operario FROM Operarios WHERE nombre = ?", (op_nombre,)) + row_op = cursor.fetchone() + if row_op: + op_id = row_op[0] + else: + cursor.execute("INSERT INTO Operarios (nombre) OUTPUT INSERTED.id_operario VALUES (?)", (op_nombre,)) + op_id = cursor.fetchone()[0] + + # --- Preparación de la cabecera --- + from datetime import date + fecha_str = relevamiento.fecha + if isinstance(fecha_str, date): + fecha_str = fecha_str.strftime("%d/%m/%Y") + + # Sanitización estricta de Foreign Key id_tarea + id_tarea_db = relevamiento.id_tarea + if not id_tarea_db or str(id_tarea_db).strip() == "" or str(id_tarea_db) == "0" or str(id_tarea_db).lower() == "none": + id_tarea_db = None + else: + id_tarea_db = int(id_tarea_db) + + id_rel = relevamiento.id_relevamiento + + if id_rel is not None: + # UPDATE cabecera + cursor.execute(query_update, ( + art_id, relevamiento.operacion, rec_id, hm_id, op_id, + fecha_str, relevamiento.postura, + relevamiento.piezas_por_golpe, relevamiento.articulos_por_golpe, + relevamiento.operarios_maquina, id_tarea_db, + id_rel + )) + else: + # INSERT cabecera + cursor.execute(query_insert, ( + art_id, relevamiento.operacion, rec_id, hm_id, op_id, + fecha_str, relevamiento.postura, + relevamiento.piezas_por_golpe, relevamiento.articulos_por_golpe, + relevamiento.operarios_maquina, id_tarea_db + )) + id_rel = cursor.fetchone()[0] + relevamiento.id_relevamiento = id_rel + + if not relevamiento.tiempos: + raise ValueError("La lista de tiempos está vacía. Abortando guardado para prevenir pérdida de datos (Regla 0).") + + # DELETE tiempos anteriores antes del bulk insert + cursor.execute(query_delete_tiempos, (id_rel,)) + + # INSERT tiempos nuevos + for t in relevamiento.tiempos: + cursor.execute(query_tiempo, ( + id_rel, t.elemento, t.valoracion, t.minutos, t.segundos, + t.elementos_medidos, t.lote_frecuencial, t.suplemento + )) + + conn.commit() + logger.info(f"Relevamiento {id_rel} guardado exitosamente en BD (Cabecera + Detalle).") + return True + except Exception as e: + conn.rollback() + logger.exception("Error crítico en transacción de guardado. Se ejecutó rollback.") + raise e + finally: + cursor.close() + + @staticmethod + def obtener_por_id(id_relevamiento: int) -> Optional['Relevamiento']: + """ + Obtiene un relevamiento por su ID directamente, sin manejo silencioso de errores (Regla 0 y Regla 7). + """ + from modelos import Relevamiento, Articulo, Recurso, Operario, TiemposElemento + from database import get_db_cursor + + query_cabecera = """ + SELECT + r.id_relevamiento, r.operacion, r.fecha, r.postura, + r.piezas_por_golpe, r.articulos_por_golpe, r.operarios_maquina, r.id_tarea, + ISNULL(a.codigo, '') AS articulo_codigo, + ISNULL(a.descripcion, '') AS articulo_desc, + m.descripcion AS maquina_desc, m.tipo AS maquina_tipo, + h.descripcion AS hm_desc, h.tipo AS hm_tipo, + o.nombre AS operario_nombre + FROM Relevamientos r + LEFT JOIN Articulos a ON r.id_articulo = a.id_articulo + LEFT JOIN Recursos m ON r.id_recurso = m.id_recurso + LEFT JOIN Recursos h ON r.id_hm = h.id_recurso + LEFT JOIN Operarios o ON r.id_operario = o.id_operario + WHERE r.id_relevamiento = ? + """ + with get_db_cursor() as cursor: + cursor.execute(query_cabecera, (id_relevamiento,)) + row = cursor.fetchone() + + if not row: + return None + + # Mapeo inmune a desfases utilizando nombres de columnas (Regla 0) + columns = [column[0] for column in cursor.description] + result_dict = dict(zip(columns, row)) + + id_rel = result_dict.get('id_relevamiento') + operacion = result_dict.get('operacion') + fecha_str = result_dict.get('fecha') + postura = result_dict.get('postura') + piezas = result_dict.get('piezas_por_golpe') + articulos_g = result_dict.get('articulos_por_golpe') + operarios_m = result_dict.get('operarios_maquina') + id_tarea = result_dict.get('id_tarea') + + art_codigo = result_dict.get('articulo_codigo') + art_desc = result_dict.get('articulo_desc') + maq_desc = result_dict.get('maquina_desc') + maq_tipo = result_dict.get('maquina_tipo') + hm_desc = result_dict.get('hm_desc') + hm_tipo = result_dict.get('hm_tipo') + op_nombre = result_dict.get('operario_nombre') + + articulo_obj = None + if art_codigo or art_desc: + articulo_obj = Articulo( + id_articulo="0", + codigo=art_codigo or "", + descripcion=art_desc or "" + ) + + maquina_obj = None + if maq_desc is not None: + maquina_obj = Recurso(id_recurso="0", nombre=maq_desc, tipo=maq_tipo) + + hm_obj = None + if hm_desc is not None: + hm_obj = Recurso(id_recurso="0", nombre=hm_desc, tipo=hm_tipo) + + operario_obj = None + if op_nombre is not None: + operario_obj = Operario(id_operario="0", nombre=op_nombre, legajo="") + + query_tiempos = "SELECT elemento, valoracion, minutos, segundos, elementos_medidos, lote_frecuencial, suplemento FROM TiemposElemento WHERE id_relevamiento = ? ORDER BY id_tiempo ASC" + cursor.execute(query_tiempos, (id_rel,)) + tiempos_db = cursor.fetchall() + + tiempos = [] + for t_row in tiempos_db: + tiempos.append(TiemposElemento( + elemento=t_row[0], + valoracion=float(t_row[1]), + minutos=float(t_row[2]), + segundos=float(t_row[3]), + elementos_medidos=int(t_row[4]), + lote_frecuencial=float(t_row[5]), + suplemento=float(t_row[6]) + )) + + from datetime import datetime, date + parsed_date = date.today() + if fecha_str: + if isinstance(fecha_str, str): + if "-" in fecha_str: + parsed_date = datetime.strptime(fecha_str, "%Y-%m-%d").date() + else: + parsed_date = datetime.strptime(fecha_str, "%d/%m/%Y").date() + else: + parsed_date = fecha_str + + return Relevamiento( + id_relevamiento=id_rel, + id_tarea=id_tarea, + articulo=articulo_obj, + operacion=operacion, + recurso=maquina_obj, + hm=hm_obj, + operario=operario_obj, + fecha=parsed_date, + postura=postura if postura else "pie", + piezas_por_golpe=int(piezas) if piezas else 1, + articulos_por_golpe=int(articulos_g) if articulos_g else 1, + operarios_maquina=int(operarios_m) if operarios_m else 1, + tiempos=tiempos + ) + + @staticmethod + def obtener_por_tarea(id_tarea: int) -> Optional[Relevamiento]: + """ + Obtiene un relevamiento asociado a una tarea. + """ + query = "SELECT id_relevamiento FROM Relevamientos WHERE id_tarea = ?" + try: + with get_db_cursor() as cursor: + cursor.execute(query, (id_tarea,)) + row = cursor.fetchone() + if row: + return RelevamientoDAO.obtener_por_id(int(row[0])) + except Exception: + logger.exception(f"Error al obtener relevamiento por tarea: {id_tarea}") + return None + @staticmethod + def obtener_todos_resumen() -> List[tuple]: + """ + Devuelve un resumen de todos los relevamientos para listarlos en la interfaz. + Retorna lista de tuplas: (ID, Fecha, Artículo, Operación, Operario) + """ + query = """ + SELECT + r.id_relevamiento, + r.fecha, + ISNULL(a.codigo, 'S/C') AS articulo_codigo, + r.operacion, + ISNULL(o.nombre, 'S/O') AS operario_nombre + FROM Relevamientos r + LEFT JOIN Articulos a ON r.id_articulo = a.id_articulo + LEFT JOIN Operarios o ON r.id_operario = o.id_operario + ORDER BY r.id_relevamiento DESC + """ + resumen = [] + try: + with get_db_cursor() as cursor: + cursor.execute(query) + for row in cursor.fetchall(): + resumen.append((row[0], row[1], row[2], row[3], row[4])) + except Exception: + logger.exception("Error al obtener resumen de relevamientos.") + return resumen + + @staticmethod + def buscar_historico(filtros: Dict[str, str]) -> List[tuple]: + """ + Busca relevamientos históricos aplicando filtros dinámicos combinados. + + Construye una consulta SQL con LEFT JOINs a Articulos, Recursos (máquina y HM) + y Operarios. Agrega cláusulas WHERE parametrizadas según los filtros provistos, + utilizando LIKE para búsqueda parcial (Regla 6: consultas parametrizadas estrictas). + + Args: + filtros: Diccionario con claves opcionales: + - 'articulo': Filtra por código del artículo. + - 'operario': Filtra por nombre del operario. + - 'maquina': Filtra por descripción de la máquina. + - 'hm': Filtra por descripción de la herramienta/matriz. + - 'operacion': Filtra por nombre de la operación. + + Returns: + Lista de tuplas con formato: + (id_relevamiento, fecha, codigo_articulo, operacion, maquina, hm, operario). + El id_relevamiento se usa internamente como iid del Treeview pero NO se + muestra al usuario. El código del artículo actúa como identificador visual. + Retorna lista vacía si ocurre un error. + + Raises: + No lanza excepciones al exterior; los errores se registran + en el logger (Regla 7). + """ + query_base = """ + SELECT + r.id_relevamiento, + r.fecha, + ISNULL(a.codigo, 'S/C') AS codigo_articulo, + ISNULL(r.operacion, '') AS operacion, + ISNULL(m.descripcion, 'S/M') AS maquina_desc, + ISNULL(h.descripcion, 'S/HM') AS hm_desc, + ISNULL(o.nombre, 'S/O') AS operario_nombre + FROM Relevamientos r + LEFT JOIN Articulos a ON r.id_articulo = a.id_articulo + LEFT JOIN Recursos m ON r.id_recurso = m.id_recurso + LEFT JOIN Recursos h ON r.id_hm = h.id_recurso + LEFT JOIN Operarios o ON r.id_operario = o.id_operario + """ + + condiciones: List[str] = [] + parametros: List[str] = [] + + # Mapeo filtro -> columna SQL para construcción dinámica segura + mapa_filtros: Dict[str, str] = { + "articulo": "a.codigo", + "operario": "o.nombre", + "maquina": "m.descripcion", + "hm": "h.descripcion", + "operacion": "r.operacion", + } + + for clave, columna in mapa_filtros.items(): + valor = filtros.get(clave, "").strip() + if valor: + condiciones.append(f"{columna} LIKE ?") + parametros.append(f"%{valor}%") + + if condiciones: + query_base += " WHERE " + " AND ".join(condiciones) + + query_base += " ORDER BY r.id_relevamiento DESC" + + resultados: List[tuple] = [] + try: + with get_db_cursor() as cursor: + cursor.execute(query_base, tuple(parametros)) + for row in cursor.fetchall(): + resultados.append( + (row[0], row[1], row[2], row[3], row[4], row[5], row[6]) + ) + except Exception: + logger.exception("Error al ejecutar búsqueda histórica de relevamientos.") + return resultados + + @staticmethod + def borrar_por_id(id_relevamiento: int) -> bool: + """ + Elimina un relevamiento y sus tiempos asociados en una sola transacción. + + Primero borra los TiemposElemento hijos y luego la cabecera del + relevamiento, garantizando integridad referencial (Regla 6). + + Args: + id_relevamiento: ID del relevamiento a eliminar. + + Returns: + True si la cabecera fue eliminada exitosamente, False en caso contrario. + + Raises: + No lanza excepciones al exterior; registra en logger (Regla 7). + """ + from database import get_db_connection + with get_db_connection() as conn: + cursor = conn.cursor() + try: + cursor.execute( + "DELETE FROM TiemposElemento WHERE id_relevamiento = ?", + (id_relevamiento,) + ) + cursor.execute( + "DELETE FROM Relevamientos WHERE id_relevamiento = ?", + (id_relevamiento,) + ) + eliminado = cursor.rowcount > 0 + conn.commit() + if eliminado: + logger.info(f"Relevamiento {id_relevamiento} eliminado (cabecera + tiempos).") + else: + logger.warning(f"No se encontró relevamiento {id_relevamiento} para eliminar.") + return eliminado + except Exception: + conn.rollback() + logger.exception(f"Error transaccional al eliminar relevamiento {id_relevamiento}. Rollback ejecutado.") + return False + finally: + cursor.close() + + +class UsuarioDAO: + @staticmethod + def obtener_por_username(username: str) -> Optional[Usuario]: + """Obtiene un usuario (con su rol) mediante consulta parametrizada.""" + query = """ + SELECT u.id_usuario, u.username, u.password_hash, r.nombre AS rol, u.nombre + FROM Usuarios u + INNER JOIN Roles r ON u.id_rol = r.id_rol + WHERE u.username = ? + """ + try: + with get_db_cursor() as cursor: + cursor.execute(query, (username,)) + row = cursor.fetchone() + if row: + # En modelos.py: id_usuario, username, password_hash, rol, nombre + from modelos import Usuario # Import local para evitar circularidad si la hubiera + return Usuario(id_usuario=row[0], username=row[1], password_hash=row[2], rol=row[3], nombre=row[4]) + return None + except Exception: + logger.exception("Fallo al obtener usuario por username.") + return None + + @staticmethod + def crear_usuario(username: str, password_hash: str, rol: str, nombre: str) -> bool: + """Crea un nuevo usuario asignando el id_rol correspondiente.""" + query = """ + INSERT INTO Usuarios (username, password_hash, id_rol, nombre) + VALUES (?, ?, (SELECT id_rol FROM Roles WHERE nombre = ?), ?) + """ + try: + with get_db_cursor(commit=True) as cursor: + cursor.execute(query, (username, password_hash, rol, nombre)) + return True + except Exception: + logger.exception(f"Fallo al crear el usuario: {username}") + return False + + @staticmethod + def obtener_metodistas() -> List['Usuario']: + """Obtiene la lista de usuarios con rol METODISTA.""" + query = """ + SELECT u.id_usuario, u.username, u.password_hash, r.nombre AS rol, u.nombre + FROM Usuarios u + INNER JOIN Roles r ON u.id_rol = r.id_rol + WHERE r.nombre = 'METODISTA' + """ + metodistas = [] + try: + with get_db_cursor() as cursor: + cursor.execute(query) + from modelos import Usuario + for row in cursor.fetchall(): + metodistas.append(Usuario(id_usuario=row[0], username=row[1], password_hash=row[2], rol=row[3], nombre=row[4])) + except Exception: + logger.exception("Fallo al obtener la lista de metodistas.") + return metodistas + + @staticmethod + def obtener_por_nombre_completo(nombre: str) -> Optional[Usuario]: + """ + Obtiene un usuario buscando por su nombre completo (campo 'nombre'). + + Realiza un INNER JOIN con Roles para recuperar el nombre del rol + asociado, idéntico al patrón de obtener_por_username. + Utiliza consulta parametrizada (Regla 6). + + Args: + nombre: Nombre completo del usuario a buscar (coincidencia exacta). + + Returns: + Instancia de Usuario si se encuentra, None en caso contrario. + + Raises: + No lanza excepciones al exterior; registra en logger (Regla 7). + """ + query = """ + SELECT u.id_usuario, u.username, u.password_hash, r.nombre AS rol, u.nombre + FROM Usuarios u + INNER JOIN Roles r ON u.id_rol = r.id_rol + WHERE u.nombre = ? + """ + try: + with get_db_cursor() as cursor: + cursor.execute(query, (nombre,)) + row = cursor.fetchone() + if row: + from modelos import Usuario + return Usuario( + id_usuario=row[0], + username=row[1], + password_hash=row[2], + rol=row[3], + nombre=row[4], + ) + return None + except Exception: + logger.exception("Fallo al obtener usuario por nombre completo.") + return None + + @staticmethod + def actualizar_password(id_usuario: int, nuevo_hash: str) -> bool: + """ + Actualiza el hash de contraseña de un usuario existente. + + Ejecuta un UPDATE parametrizado con commit explícito (Regla 6). + Destinado al flujo de blanqueo/recuperación de contraseña. + + Args: + id_usuario: ID del usuario cuya contraseña se actualiza. + nuevo_hash: Nuevo hash bcrypt de la contraseña. + + Returns: + True si la actualización afectó al menos una fila, False en caso contrario. + + Raises: + No lanza excepciones al exterior; registra en logger (Regla 7). + """ + query = "UPDATE Usuarios SET password_hash = ? WHERE id_usuario = ?" + try: + with get_db_cursor(commit=True) as cursor: + cursor.execute(query, (nuevo_hash, id_usuario)) + actualizado: bool = cursor.rowcount > 0 + if actualizado: + logger.info(f"Contraseña actualizada para id_usuario={id_usuario}.") + else: + logger.warning(f"No se encontró id_usuario={id_usuario} para actualizar contraseña.") + return actualizado + except Exception: + logger.exception(f"Fallo al actualizar contraseña para id_usuario={id_usuario}.") + return False + +class TareaDAO: + @staticmethod + def obtener_todas() -> List['Tarea']: + """Obtiene todas las tareas/solicitudes de la BD.""" + # Se asume que id_usuario_asignado ya existe en BD tras el ALTER TABLE + query = "SELECT id_tarea, titulo, descripcion, estado, fecha, informe_url, id_usuario_asignado FROM Tareas" + tareas = [] + try: + with get_db_cursor() as cursor: + cursor.execute(query) + from modelos import Tarea + for row in cursor.fetchall(): + # Para prevenir crash si la columna aún no existe, capturamos el índice seguro + try: + id_asig = row[6] + except IndexError: + id_asig = None + + tareas.append(Tarea( + id_tarea=row[0], titulo=row[1], descripcion=row[2], + estado=row[3], fecha=str(row[4]), informe_url=row[5] or "", + id_usuario_asignado=id_asig + )) + except Exception: + logger.exception("Fallo al obtener tareas.") + return tareas + + @staticmethod + def obtener_tareas_por_usuario(id_usuario: int) -> List['Tarea']: + """Obtiene solo las tareas asignadas a un metodista.""" + query = "SELECT id_tarea, titulo, descripcion, estado, fecha, informe_url, id_usuario_asignado FROM Tareas WHERE id_usuario_asignado = ?" + tareas = [] + try: + with get_db_cursor() as cursor: + cursor.execute(query, (id_usuario,)) + from modelos import Tarea + for row in cursor.fetchall(): + tareas.append(Tarea( + id_tarea=row[0], titulo=row[1], descripcion=row[2], + estado=row[3], fecha=str(row[4]), informe_url=row[5] or "", + id_usuario_asignado=row[6] + )) + except Exception: + logger.exception("Fallo al obtener tareas por usuario.") + return tareas + + @staticmethod + def crear_tarea(titulo: str, descripcion: str, id_usuario_asignado: int) -> bool: + """Crea una nueva solicitud de relevamiento (Tarea) en estado Pendiente.""" + query = """ + INSERT INTO Tareas (titulo, descripcion, id_usuario_asignado, estado) + VALUES (?, ?, ?, 'Pendiente') + """ + try: + with get_db_cursor(commit=True) as cursor: + cursor.execute(query, (titulo, descripcion, id_usuario_asignado)) + return True + except Exception: + logger.exception("Fallo al crear la tarea.") + return False + + @staticmethod + def borrar_tarea(id_tarea: int) -> bool: + """Elimina una tarea permanentemente de la BD.""" + query = "DELETE FROM Tareas WHERE id_tarea = ?" + try: + with get_db_cursor(commit=True) as cursor: + cursor.execute(query, (id_tarea,)) + return cursor.rowcount > 0 + except Exception: + logger.exception("Fallo al borrar la tarea.") + return False + + @staticmethod + def actualizar_estado(id_tarea: int, nuevo_estado: str) -> bool: + """Actualiza el estado de una tarea.""" + query = "UPDATE Tareas SET estado = ? WHERE id_tarea = ?" + try: + with get_db_cursor(commit=True) as cursor: + cursor.execute(query, (nuevo_estado, id_tarea)) + return cursor.rowcount > 0 + except Exception: + logger.exception("Fallo al actualizar el estado de la tarea.") + return False + +class AuditoriaDAO: + @staticmethod + def insertar_log(id_usuario: Optional[int], accion: str) -> bool: + """Inserta un registro en LogsAuditoria de manera transaccional.""" + query = "INSERT INTO LogsAuditoria (id_usuario, accion) VALUES (?, ?)" + try: + with get_db_cursor(commit=True) as cursor: + cursor.execute(query, (id_usuario, accion)) + return True + except Exception: + logger.exception("Error crítico al guardar log de auditoría.") + return False + @staticmethod + def actualizar_estado(id_tarea: int, nuevo_estado: str) -> bool: + """Actualiza el estado de una tarea.""" + query = "UPDATE Tareas SET estado = ? WHERE id_tarea = ?" + try: + with get_db_cursor(commit=True) as cursor: + cursor.execute(query, (nuevo_estado, id_tarea)) + return cursor.rowcount > 0 + except Exception: + logger.exception("Fallo al actualizar el estado de la tarea.") + return False + +class AuditoriaDAO: + @staticmethod + def insertar_log(id_usuario: Optional[int], accion: str) -> bool: + """Inserta un registro en LogsAuditoria de manera transaccional.""" + query = "INSERT INTO LogsAuditoria (id_usuario, accion) VALUES (?, ?)" + try: + with get_db_cursor(commit=True) as cursor: + cursor.execute(query, (id_usuario, accion)) + return True + except Exception: + logger.exception("Error crítico al guardar log de auditoría.") + return False diff --git a/database.py b/database.py new file mode 100644 index 0000000..34f2e6e --- /dev/null +++ b/database.py @@ -0,0 +1,167 @@ +# pyrefly: ignore [missing-import] +import pyodbc +from typing import Iterator, Any, Optional +from contextlib import contextmanager +import config +from logger import get_logger + +logger = get_logger(__name__) + +class DatabaseConnectionError(Exception): + """Excepción personalizada para errores de conexión a la base de datos.""" + pass + +@contextmanager +def get_db_connection() -> Iterator[pyodbc.Connection]: + """ + Context Manager (Regla 3) para manejar de forma segura la conexión a SQL Server. + Garantiza que la conexión se cierre al salir del bloque. + + Yields: + pyodbc.Connection: Objeto de conexión a la base de datos. + + Raises: + DatabaseConnectionError: Si falla la conexión a la base de datos (Regla 7). + """ + conn_str = ( + f"DRIVER={config.DB_DRIVER};" + f"SERVER={config.DB_SERVER};" + f"DATABASE={config.DB_NAME};" + f"Trusted_Connection=yes;" + ) + conn = None + try: + conn = pyodbc.connect(conn_str) + logger.info("Conexión a base de datos establecida exitosamente.") + yield conn + except pyodbc.Error as e: + logger.exception("Error crítico al conectar a SQL Server.") + raise DatabaseConnectionError(f"No se pudo conectar a la base de datos: {e}") + finally: + if conn is not None: + conn.close() + logger.debug("Conexión a la base de datos cerrada.") + +@contextmanager +def get_db_cursor(commit: bool = False) -> Iterator[pyodbc.Cursor]: + """ + Context Manager (Regla 3 y 6) para manejar un cursor de SQL Server, + con soporte para transacciones seguras (commit/rollback). + + Args: + commit (bool): Indica si se debe ejecutar commit() al finalizar sin errores. + + Yields: + pyodbc.Cursor: Cursor activo para ejecutar consultas. + """ + with get_db_connection() as conn: + cursor = conn.cursor() + try: + yield cursor + if commit: + conn.commit() + logger.info("Transacción confirmada (commit).") + except Exception as e: + conn.rollback() + logger.exception("Error en transacción SQL. Se ejecutó rollback().") + raise + finally: + cursor.close() + logger.debug("Cursor de base de datos cerrado.") + + +def ejecutar_backup(ruta_destino: str) -> bool: + """ + Ejecuta un backup nativo de SQL Server (BACKUP DATABASE) a la ruta indicada. + + Utiliza una conexión independiente con autocommit=True, ya que el comando + BACKUP DATABASE de SQL Server no puede ejecutarse dentro de un bloque + de transacción explícito. Esta conexión se abre y cierra exclusivamente + para esta operación administrativa. + + Args: + ruta_destino (str): Ruta absoluta del archivo .bak de destino en el + servidor de SQL Server. Ejemplo: ``'C:\\Backups\\mi_db.bak'``. + + Returns: + bool: ``True`` si el backup se completó exitosamente, + ``False`` si ocurrió cualquier error durante la operación. + + Raises: + No levanta excepciones. Todos los errores se capturan, registran + en el logger y la función retorna ``False``. + + Note: + El comando BACKUP DATABASE es una instrucción DDL administrativa que + no soporta consultas parametrizadas via ``cursor.execute(sql, params)``. + La interpolación directa de ``config.DB_NAME`` y ``ruta_destino`` en + el string SQL es una excepción técnica justificada a la Regla 6. + """ + conn_str: str = ( + f"DRIVER={config.DB_DRIVER};" + f"SERVER={config.DB_SERVER};" + f"DATABASE={config.DB_NAME};" + f"Trusted_Connection=yes;" + ) + conn: Optional[pyodbc.Connection] = None + cursor: Optional[pyodbc.Cursor] = None + try: + conn = pyodbc.connect(conn_str, autocommit=True) + cursor = conn.cursor() + logger.info( + "Conexión con autocommit establecida para backup de '%s'.", + config.DB_NAME, + ) + + sql_backup: str = ( + f"BACKUP DATABASE [{config.DB_NAME}] " + f"TO DISK = '{ruta_destino}' " + f"WITH FORMAT, " + f"MEDIANAME = 'SQLServerBackups', " + f"NAME = 'Full Backup'" + ) + cursor.execute(sql_backup) + + # BACKUP DATABASE puede devolver múltiples result sets informativos. + # Consumirlos evita que la conexión quede en estado inconsistente. + while cursor.nextset(): + pass + + logger.info( + "Backup de la base de datos '%s' completado exitosamente en: %s", + config.DB_NAME, + ruta_destino, + ) + return True + + except pyodbc.Error as db_err: + logger.exception( + "Error de SQL Server al ejecutar backup de '%s': %s", + config.DB_NAME, + db_err, + ) + return False + + except OSError as os_err: + logger.exception( + "Error de sistema operativo durante el backup (posible ruta inválida '%s'): %s", + ruta_destino, + os_err, + ) + return False + + except Exception as unexpected_err: + logger.exception( + "Error inesperado durante el backup de '%s': %s", + config.DB_NAME, + unexpected_err, + ) + return False + + finally: + if cursor is not None: + cursor.close() + logger.debug("Cursor de backup cerrado.") + if conn is not None: + conn.close() + logger.debug("Conexión de backup cerrada.")