""" app/routes/cambios.py – Las 3 transacciones del control de cambios. Incluye endpoint AJAX para cargar usuarios por sistema. """ import logging from datetime import date from io import BytesIO from flask import (Blueprint, render_template, redirect, url_for, flash, request, jsonify, send_file) from flask_login import login_required, current_user from app import db from app.models import Cambio, Rol, Sistema, Usuario, UsuarioSistema from app.forms import AltaCambioForm, AplicacionCambioForm, LiberacionCambioForm from app.utils.decorators import role_required from app.utils.mailer import enviar_correo_liberacion from app.utils.pdf_generator import generar_pdf_cambio_individual cambios_bp = Blueprint('cambios', __name__, url_prefix='/cambios') logger = logging.getLogger(__name__) # --------------------------------------------------------------- # API: usuarios por sistema (AJAX para el combo dependiente) # --------------------------------------------------------------- @cambios_bp.route('/api/usuarios-por-sistema/') @login_required def usuarios_por_sistema(sistema_id): """Devuelve JSON con los Promotores asignados a un sistema.""" relaciones = (UsuarioSistema.query .filter_by(sistema_id=sistema_id) .join(Usuario, UsuarioSistema.usuario_id == Usuario.id) .join(Rol, Usuario.rol_id == Rol.id) .filter(Usuario.activo == True, Rol.nombre == 'Promotor') .all()) usuarios = [ {'id': r.usuario_id, 'nombre': r.usuario.nombre_completo} for r in relaciones ] return jsonify(usuarios) # --------------------------------------------------------------- # Listado general de cambios # --------------------------------------------------------------- @cambios_bp.route('/') @login_required def lista(): page = request.args.get('page', 1, type=int) q = Cambio.query.order_by(Cambio.creado_en.desc()) # Promotores solo ven sus propios cambios if current_user.es_promotor: q = q.filter_by(usuario_solicitante_id=current_user.id) cambios = q.paginate(page=page, per_page=15, error_out=False) return render_template('cambios/list.html', cambios=cambios) # --------------------------------------------------------------- # Transacción 1: Alta de Cambio # --------------------------------------------------------------- @cambios_bp.route('/alta', methods=['GET', 'POST']) @login_required @role_required('Promotor', 'Administrador') def alta(): form = AltaCambioForm() sistemas = Sistema.query.filter_by(activo=True).order_by(Sistema.nombre).all() form.sistema_id.choices = [(s.id, s.nombre) for s in sistemas] # Poblar choices del solicitante ANTES de validate_on_submit. # WTForms rechaza valores que no estén en choices, así que leemos # el sistema_id del POST para cargar los usuarios correctos. sistema_id_post = request.form.get('sistema_id', type=int) if sistema_id_post: relaciones = (UsuarioSistema.query .filter_by(sistema_id=sistema_id_post) .join(Usuario, UsuarioSistema.usuario_id == Usuario.id) .join(Rol, Usuario.rol_id == Rol.id) .filter(Usuario.activo == True, Rol.nombre == 'Promotor') .all()) form.usuario_solicitante_id.choices = [ (r.usuario_id, r.usuario.nombre_completo) for r in relaciones ] else: form.usuario_solicitante_id.choices = [(0, '-- Seleccionar sistema primero --')] if form.validate_on_submit(): # Verificar que el usuario seleccionado es Promotor y pertenece al sistema sistema_id = form.sistema_id.data usr_id = form.usuario_solicitante_id.data rel = (UsuarioSistema.query .filter_by(sistema_id=sistema_id, usuario_id=usr_id) .join(Usuario, UsuarioSistema.usuario_id == Usuario.id) .join(Rol, Usuario.rol_id == Rol.id) .filter(Usuario.activo == True, Rol.nombre == 'Promotor') .first()) if not rel: flash('El usuario seleccionado no está asignado a ese sistema.', 'danger') else: cambio = Cambio( sistema_id=sistema_id, usuario_solicitante_id=usr_id, numero_incidencia=form.numero_incidencia.data, descripcion_breve=form.descripcion_breve.data.strip(), urgente=form.urgente.data == '1', fecha_pruebas=form.fecha_pruebas.data, estatus='Solicitado', ) db.session.add(cambio) db.session.commit() logger.info('Cambio #%s creado por %s', cambio.id, current_user.correo) flash(f'Solicitud de cambio #{cambio.id} registrada correctamente.', 'success') return redirect(url_for('cambios.lista')) return render_template('cambios/alta.html', form=form) # --------------------------------------------------------------- # Transacción 2: Aplicación de Cambio # --------------------------------------------------------------- @cambios_bp.route('/aplicacion') @login_required @role_required('Desarrollador', 'Administrador') def aplicacion_lista(): """Lista de cambios en estatus Solicitado para aplicar.""" cambios = (Cambio.query .filter_by(estatus='Solicitado') .order_by(Cambio.creado_en) .all()) return render_template('cambios/aplicacion_lista.html', cambios=cambios) @cambios_bp.route('/aplicacion/', methods=['GET', 'POST']) @login_required @role_required('Desarrollador', 'Administrador') def aplicacion_detalle(id): cambio = Cambio.query.get_or_404(id) if cambio.estatus != 'Solicitado': flash('Este cambio ya no está en estado Solicitado.', 'warning') return redirect(url_for('cambios.aplicacion_lista')) form = AplicacionCambioForm() if form.validate_on_submit(): # La fecha propuesta no puede ser anterior a fecha de pruebas if form.fecha_propuesta_produccion.data < cambio.fecha_pruebas: flash('La fecha propuesta no puede ser anterior a la fecha de pruebas.', 'danger') return render_template('cambios/aplicacion_detalle.html', form=form, cambio=cambio) cambio.descripcion_tecnica = form.descripcion_tecnica.data.strip() cambio.fecha_propuesta_produccion = form.fecha_propuesta_produccion.data cambio.desarrollador_id = current_user.id archivo = form.archivo_cambio.data if archivo and archivo.filename: cambio.archivo_nombre = archivo.filename cambio.archivo_cambio = archivo.read() cambio.estatus = 'Programado' db.session.commit() logger.info('Cambio #%s programado por %s', id, current_user.correo) flash(f'Cambio #{id} marcado como Programado.', 'success') return redirect(url_for('cambios.aplicacion_lista')) return render_template('cambios/aplicacion_detalle.html', form=form, cambio=cambio) # --------------------------------------------------------------- # Transacción 3: Liberación a Producción # --------------------------------------------------------------- @cambios_bp.route('/liberacion') @login_required @role_required('Administrador') def liberacion_lista(): """Lista de cambios en estatus Programado para liberar.""" cambios = (Cambio.query .filter_by(estatus='Programado') .order_by(Cambio.creado_en) .all()) return render_template('cambios/liberacion_lista.html', cambios=cambios) @cambios_bp.route('/liberacion/', methods=['GET', 'POST']) @login_required @role_required('Administrador') def liberacion_detalle(id): cambio = Cambio.query.get_or_404(id) if cambio.estatus != 'Programado': flash('Este cambio ya no está en estado Programado.', 'warning') return redirect(url_for('cambios.liberacion_lista')) form = LiberacionCambioForm() usuarios_activos = Usuario.query.filter_by(activo=True).order_by(Usuario.nombre_completo).all() form.usuario_valido_id.choices = [(u.id, u.nombre_completo) for u in usuarios_activos] if form.validate_on_submit(): cambio.fecha_real_aplicacion = form.fecha_real_aplicacion.data cambio.usuario_valido_id = form.usuario_valido_id.data cambio.estado_cambio = form.estado_cambio.data cambio.observaciones = form.observaciones.data cambio.administrador_libero_id = current_user.id cambio.estatus = 'Liberado' db.session.commit() logger.info('Cambio #%s liberado por %s', id, current_user.correo) # Generar PDF y enviar correo try: pdf_buffer = generar_pdf_cambio_individual(cambio, current_user) enviado, detalle = enviar_correo_liberacion(cambio, pdf_buffer) if enviado: flash(f'Cambio liberado y notificación enviada a: {detalle}', 'success') else: flash(f'Cambio liberado. Correo no enviado: {detalle}', 'warning') except Exception as exc: logger.error('Error en liberación de cambio #%s: %s', id, exc) flash(f'Cambio liberado, pero hubo un error al generar el PDF/correo: {exc}', 'warning') return redirect(url_for('cambios.liberacion_lista')) return render_template('cambios/liberacion_detalle.html', form=form, cambio=cambio) # --------------------------------------------------------------- # Descarga de PDF de un cambio individual # --------------------------------------------------------------- @cambios_bp.route('//pdf') @login_required def descargar_pdf(id): cambio = Cambio.query.get_or_404(id) # Promotores solo pueden descargar PDF de sus propias solicitudes if current_user.es_promotor and cambio.usuario_solicitante_id != current_user.id: flash('No tienes permiso para descargar este PDF.', 'danger') return redirect(url_for('cambios.lista')) from datetime import datetime buffer = generar_pdf_cambio_individual(cambio, current_user) nombre_arch = f'Cambio_{id}_{datetime.now().strftime("%Y%m%d")}.pdf' logger.info('PDF generado para cambio #%s por %s', id, current_user.correo) return send_file( buffer, mimetype='application/pdf', as_attachment=True, download_name=nombre_arch, ) # --------------------------------------------------------------- # Descarga del archivo de objetos/scripts (BLOB) de un cambio # --------------------------------------------------------------- @cambios_bp.route('//archivo') @login_required @role_required('Administrador') def descargar_archivo(id): cambio = Cambio.query.get_or_404(id) if not cambio.archivo_cambio: flash('Este cambio no tiene un archivo adjunto.', 'warning') return redirect(url_for('cambios.liberacion_lista')) nombre = cambio.archivo_nombre or f'cambio_{id}.zip' if nombre.lower().endswith('.sql'): mimetype = 'application/sql' else: mimetype = 'application/zip' logger.info('Archivo descargado para cambio #%s por %s', id, current_user.correo) return send_file( BytesIO(cambio.archivo_cambio), mimetype=mimetype, as_attachment=True, download_name=nombre, )