alumnos_ti/MT
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Subir archivos a "/"

Browse Source
This commit is contained in:
alumnos_ti
2026-07-01 20:38:56 +00:00
commit c9940ec377
5 changed files with 1961 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
check_db.py
+28
View File
@@ -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)
config.py
+28
View File
@@ -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"
controlador.py
+832
View File
@@ -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
dao.py
+906
View File
@@ -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
database.py
+167
View File
@@ -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.")