833 lines
36 KiB
Python
833 lines
36 KiB
Python
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
|
|
|