# Para subir archivo tipo foto al servidor from werkzeug.utils import secure_filename import uuid # Modulo de python para crear un string from conexion.conexionBD import connectionBD # Conexión a BD import datetime import re import os from os import remove # Modulo para remover archivo from os import path # Modulo para obtener la ruta o directorio import openpyxl # Para generar el excel # biblioteca o modulo send_file para forzar la descarga from flask import send_file def procesar_form_empleado(dataForm, foto_perfil): # Formateando Salario salario_sin_puntos = re.sub('[^0-9]+', '', dataForm['salario_empleado']) # convertir salario a float for DECIMAL salario_decimal = float(salario_sin_puntos) # Map sexo: 1 -> 'Masculino', 2 -> 'Femenino' sexo_map = {'1': 'Masculino', '2': 'Femenino'} sexo_empleado = sexo_map.get(dataForm['sexo_empleado'], 'Masculino') result_foto_perfil = procesar_imagen_perfil(foto_perfil) try: with connectionBD() as conexion_MySQLdb: with conexion_MySQLdb.cursor(dictionary=True) as cursor: sql = "INSERT INTO tbl_empleados (nombre_empleado, apellido_empleado, sexo_empleado, telefono_empleado, email_empleado, profesion_empleado, foto_empleado, salario_empleado) VALUES (%s, %s, %s, %s, %s, %s, %s, %s)" # Creando una tupla con los valores del INSERT valores = (dataForm['nombre_empleado'], dataForm['apellido_empleado'], sexo_empleado, dataForm['telefono_empleado'], dataForm['email_empleado'], dataForm['profesion_empleado'], result_foto_perfil, salario_decimal) cursor.execute(sql, valores) conexion_MySQLdb.commit() resultado_insert = cursor.rowcount return resultado_insert except Exception as e: return f'Se produjo un error en procesar_form_empleado: {str(e)}' def procesar_imagen_perfil(foto): try: # Validar que el archivo no esté vacío if not foto or not foto.filename: raise ValueError("No se seleccionó ningún archivo") # Validar tamaño del archivo (200MB = 200 * 1024 * 1024 bytes) foto.seek(0, 2) # Ir al final del archivo file_size = foto.tell() # Obtener el tamaño en bytes foto.seek(0) # Volver al inicio del archivo if file_size > 200 * 1024 * 1024: # 200MB raise ValueError(f"El archivo es demasiado grande. Tamaño actual: {file_size / (1024*1024):.2f}MB. Tamaño máximo permitido: 200MB") # Validar extensión del archivo allowed_extensions = {'.jpg', '.jpeg', '.png', '.gif', '.bmp', '.webp'} filename = secure_filename(foto.filename) extension = os.path.splitext(filename)[1].lower() if extension not in allowed_extensions: raise ValueError(f"Extensión de archivo no permitida. Tipos permitidos: {', '.join(allowed_extensions)}") # Creando un string único para el nombre del archivo nuevoNameFile = (uuid.uuid4().hex + uuid.uuid4().hex)[:100] nombreFile = nuevoNameFile + extension # Construir la ruta completa de subida del archivo basepath = os.path.abspath(os.path.dirname(__file__)) upload_dir = os.path.join(basepath, f'../static/fotos_empleados/') # Validar si existe la ruta y crearla si no existe if not os.path.exists(upload_dir): os.makedirs(upload_dir) # Dando permiso a la carpeta os.chmod(upload_dir, 0o755) # Construir la ruta completa de subida del archivo upload_path = os.path.join(upload_dir, nombreFile) foto.save(upload_path) return nombreFile except ValueError as ve: print(f"Error de validación en archivo: {ve}") raise ve except Exception as e: print("Error al procesar archivo:", e) raise Exception(f"Error al procesar el archivo: {str(e)}") # Lista de Empleados def sql_lista_empleadosBD(): try: with connectionBD() as conexion_MySQLdb: with conexion_MySQLdb.cursor(dictionary=True) as cursor: querySQL = (f""" SELECT e.id_empleado, e.nombre_empleado, e.apellido_empleado, e.salario_empleado, e.foto_empleado, e.sexo_empleado FROM tbl_empleados AS e ORDER BY e.id_empleado DESC """) cursor.execute(querySQL,) empleadosBD = cursor.fetchall() return empleadosBD except Exception as e: print( f"Errro en la función sql_lista_empleadosBD: {e}") return None # Detalles del Empleado def sql_detalles_empleadosBD(idEmpleado): try: with connectionBD() as conexion_MySQLdb: with conexion_MySQLdb.cursor(dictionary=True) as cursor: querySQL = (""" SELECT e.id_empleado, e.nombre_empleado, e.apellido_empleado, e.salario_empleado, e.sexo_empleado, e.telefono_empleado, e.email_empleado, e.profesion_empleado, e.foto_empleado, DATE_FORMAT(e.fecha_registro, '%Y-%m-%d %h:%i %p') AS fecha_registro FROM tbl_empleados AS e WHERE id_empleado =%s ORDER BY e.id_empleado DESC """) cursor.execute(querySQL, (idEmpleado,)) empleadosBD = cursor.fetchone() return empleadosBD except Exception as e: print( f"Errro en la función sql_detalles_empleadosBD: {e}") return None # Funcion Profesores Documentos Informe (Reporte) def documentosProfesoresReporte(codmodular): try: with connectionBD() as conexion_MySQLdb: with conexion_MySQLdb.cursor(dictionary=True) as cursor: querySQL = (""" SELECT p.apaterno, p.amaterno, p.nombres, d.curso, d.sesion, d.mes, d.trimestre, d.archivo, d.fechasubida FROM tbl_docprof AS d INNER JOIN profesor AS p ON d.dniprof = p.dniprof WHERE p.codmodular = %s ORDER BY d.fechasubida DESC """) cursor.execute(querySQL, (codmodular,)) documentosBD = cursor.fetchall() return documentosBD except Exception as e: print( f"Error en la función documentosProfesoresReporte: {e}") return None def generarReporteExcel(codmodular): dataDocumentos = documentosProfesoresReporte(codmodular) wb = openpyxl.Workbook() # Hoja principal con lista de documentos hoja = wb.active hoja.title = "Documentos" # Agregar la fila de encabezado con los títulos cabeceraExcel = ("Apellido Paterno", "Apellido Materno", "Nombres", "Curso", "Sesión", "Mes", "Trimestre", "Archivo", "Fecha Subida") hoja.append(cabeceraExcel) # Agregar los registros a la hoja for registro in dataDocumentos: apaterno = registro['apaterno'] amaterno = registro['amaterno'] nombres = registro['nombres'] curso = registro['curso'] sesion = registro['sesion'] mes = registro['mes'] trimestre = registro['trimestre'] archivo = registro['archivo'] fechasubida = registro['fechasubida'] # Agregar los valores a la hoja hoja.append((apaterno, amaterno, nombres, curso, sesion, mes, trimestre, archivo, fechasubida)) # Crear hoja de estadísticas hoja_stats = wb.create_sheet(title="Estadísticas") # Estadísticas por profesor from collections import defaultdict prof_count = defaultdict(int) month_count = defaultdict(int) for reg in dataDocumentos: prof_name = f"{reg['apaterno']} {reg['amaterno']} {reg['nombres']}" prof_count[prof_name] += 1 if reg['fechasubida']: try: if isinstance(reg['fechasubida'], str): fecha = datetime.datetime.strptime(reg['fechasubida'], '%Y-%m-%d %H:%M:%S') else: fecha = reg['fechasubida'] month = fecha.strftime('%Y-%m') month_count[month] += 1 except: pass # Encabezados para estadísticas de profesores hoja_stats.append(("Profesor", "Número de Documentos")) for prof, count in prof_count.items(): hoja_stats.append((prof, count)) # Espacio hoja_stats.append(("", "")) # Encabezados para estadísticas por mes hoja_stats.append(("Mes", "Número de Documentos")) for month, count in sorted(month_count.items()): hoja_stats.append((month, count)) fecha_actual = datetime.datetime.now() archivoExcel = f"Reporte_documentos_profesores_{fecha_actual.strftime('%Y_%m_%d')}.xlsx" carpeta_descarga = "../static/downloads-excel" ruta_descarga = os.path.join(os.path.dirname( os.path.abspath(__file__)), carpeta_descarga) if not os.path.exists(ruta_descarga): os.makedirs(ruta_descarga) # Dando permisos a la carpeta os.chmod(ruta_descarga, 0o755) ruta_archivo = os.path.join(ruta_descarga, archivoExcel) wb.save(ruta_archivo) # Enviar el archivo como respuesta HTTP return send_file(ruta_archivo, as_attachment=True) def buscarEmpleadoBD(search): try: with connectionBD() as conexion_MySQLdb: with conexion_MySQLdb.cursor(dictionary=True) as mycursor: querySQL = (""" SELECT e.id_empleado, e.nombre_empleado, e.apellido_empleado, e.salario_empleado, e.sexo_empleado AS sexo_empleado FROM tbl_empleados AS e WHERE e.nombre_empleado LIKE %s ORDER BY e.id_empleado DESC """) search_pattern = f"%{search}%" # Agregar "%" alrededor del término de búsqueda mycursor.execute(querySQL, (search_pattern,)) resultado_busqueda = mycursor.fetchall() return resultado_busqueda except Exception as e: print(f"Ocurrió un error en def buscarEmpleadoBD: {e}") return [] def buscarEmpleadoUnico(id): try: with connectionBD() as conexion_MySQLdb: with conexion_MySQLdb.cursor(dictionary=True) as mycursor: querySQL = (""" SELECT e.id_empleado, e.nombre_empleado, e.apellido_empleado, e.sexo_empleado, e.telefono_empleado, e.email_empleado, e.profesion_empleado, e.salario_empleado, e.foto_empleado FROM tbl_empleados AS e WHERE e.id_empleado =%s LIMIT 1 """) mycursor.execute(querySQL, (id,)) empleado = mycursor.fetchone() return empleado except Exception as e: print(f"Ocurrió un error en def buscarEmpleadoUnico: {e}") return [] def procesar_actualizacion_form(data): try: with connectionBD() as conexion_MySQLdb: with conexion_MySQLdb.cursor(dictionary=True) as cursor: nombre_empleado = data.form['nombre_empleado'] apellido_empleado = data.form['apellido_empleado'] sexo_empleado = data.form['sexo_empleado'] telefono_empleado = data.form['telefono_empleado'] email_empleado = data.form['email_empleado'] profesion_empleado = data.form['profesion_empleado'] salario_sin_puntos = re.sub( '[^0-9]+', '', data.form['salario_empleado']) salario_empleado = float(salario_sin_puntos) id_empleado = data.form['id_empleado'] if data.files['foto_empleado']: file = data.files['foto_empleado'] fotoForm = procesar_imagen_perfil(file) querySQL = """ UPDATE tbl_empleados SET nombre_empleado = %s, apellido_empleado = %s, sexo_empleado = %s, telefono_empleado = %s, email_empleado = %s, profesion_empleado = %s, salario_empleado = %s, foto_empleado = %s WHERE id_empleado = %s """ values = (nombre_empleado, apellido_empleado, sexo_empleado, telefono_empleado, email_empleado, profesion_empleado, salario_empleado, fotoForm, id_empleado) else: querySQL = """ UPDATE tbl_empleados SET nombre_empleado = %s, apellido_empleado = %s, sexo_empleado = %s, telefono_empleado = %s, email_empleado = %s, profesion_empleado = %s, salario_empleado = %s WHERE id_empleado = %s """ values = (nombre_empleado, apellido_empleado, sexo_empleado, telefono_empleado, email_empleado, profesion_empleado, salario_empleado, id_empleado) cursor.execute(querySQL, values) conexion_MySQLdb.commit() return cursor.rowcount or [] except Exception as e: print(f"Ocurrió un error en procesar_actualizacion_form: {e}") return None # Lista de Usuarios creados def lista_usuariosBD(): try: with connectionBD() as conexion_MySQLdb: with conexion_MySQLdb.cursor(dictionary=True) as cursor: querySQL = "SELECT id, name_surname, email_user, created_user FROM users" cursor.execute(querySQL,) usuariosBD = cursor.fetchall() return usuariosBD except Exception as e: print(f"Error en lista_usuariosBD : {e}") return [] # Eliminar uEmpleado def eliminarEmpleado(id_empleado, foto_empleado): try: with connectionBD() as conexion_MySQLdb: with conexion_MySQLdb.cursor(dictionary=True) as cursor: querySQL = "DELETE FROM tbl_empleados WHERE id_empleado=%s" cursor.execute(querySQL, (id_empleado,)) conexion_MySQLdb.commit() resultado_eliminar = cursor.rowcount if resultado_eliminar: # Eliminadon foto_empleado desde el directorio basepath = path.dirname(__file__) url_File = path.join( basepath, '../static/fotos_empleados', foto_empleado) if path.exists(url_File): remove(url_File) # Borrar foto desde la carpeta return resultado_eliminar except Exception as e: print(f"Error en eliminarEmpleado : {e}") return [] # Eliminar usuario def eliminarUsuario(id): try: with connectionBD() as conexion_MySQLdb: with conexion_MySQLdb.cursor(dictionary=True) as cursor: querySQL = "DELETE FROM users WHERE id=%s" cursor.execute(querySQL, (id,)) conexion_MySQLdb.commit() resultado_eliminar = cursor.rowcount return resultado_eliminar except Exception as e: print(f"Error en eliminarUsuario : {e}") return []