2079 lines
80 KiB
Python
2079 lines
80 KiB
Python
from flask import Blueprint, jsonify, request, g
|
|
from datetime import datetime
|
|
import re
|
|
from app.utils.jwt_decorators import jwt_required
|
|
from app.models.user import User
|
|
from app.models.role import Role, Permission, SYSTEM_MODULES
|
|
from app.models.subject import Subject, Commission, CommissionTeacher
|
|
from app.models.career import Career
|
|
from app.models.classroom import Classroom
|
|
from app.models.reservation import Reservation, ReservationStatus
|
|
from app.models.academic_term import AcademicTerm
|
|
from app.models.milestone import MilestoneType, AcademicMilestone
|
|
from app.models.audit_log import AuditLog
|
|
from app.constants.document_types import DOCUMENT_TYPES, validate_document, format_document
|
|
from app.services.enrollment_service import EnrollmentService
|
|
from app.services.gradebook_service import GradebookService
|
|
from app import db
|
|
|
|
api_admin_bp = Blueprint('api_admin', __name__)
|
|
|
|
@api_admin_bp.route('/users', methods=['GET'])
|
|
@jwt_required
|
|
def get_users():
|
|
search = request.args.get('search', '').strip().lower()
|
|
role_id = request.args.get('role', type=int)
|
|
status = request.args.get('status', '').strip().lower()
|
|
|
|
query = User.query
|
|
if search:
|
|
query = query.filter(
|
|
(User.name.ilike(f'%{search}%')) |
|
|
(User.email.ilike(f'%{search}%')) |
|
|
(User.personal_email.ilike(f'%{search}%')) |
|
|
(User.document_number.ilike(f'%{search}%')) |
|
|
(User.first_name.ilike(f'%{search}%')) |
|
|
(User.last_name.ilike(f'%{search}%'))
|
|
)
|
|
if role_id:
|
|
query = query.filter(User.role_id == role_id)
|
|
if status == 'active':
|
|
query = query.filter(User.is_active == True)
|
|
elif status == 'inactive':
|
|
query = query.filter(User.is_active == False)
|
|
|
|
users = query.order_by(User.id.asc()).all()
|
|
roles = Role.query.order_by(Role.name.asc()).all()
|
|
|
|
users_data = []
|
|
for u in users:
|
|
d = u.to_dict()
|
|
d['role_obj'] = {
|
|
'id': u.role_obj.id,
|
|
'name': u.role_obj.name
|
|
} if u.role_obj else {'id': 1, 'name': u.role or 'Admin'}
|
|
users_data.append(d)
|
|
|
|
roles_data = [{'id': r.id, 'name': r.name} for r in roles]
|
|
|
|
return jsonify({
|
|
'status': 'success',
|
|
'total': len(users_data),
|
|
'users': users_data,
|
|
'roles': roles_data,
|
|
'document_types': DOCUMENT_TYPES
|
|
}), 200
|
|
|
|
@api_admin_bp.route('/roles', methods=['GET'])
|
|
@jwt_required
|
|
def get_roles():
|
|
roles = Role.query.order_by(Role.id.asc()).all()
|
|
modules = [
|
|
{'id': 'dashboard', 'name': 'Dashboard', 'icon': 'bi-speedometer2'},
|
|
{'id': 'classrooms', 'name': 'Aulas y Espacios', 'icon': 'bi-door-open'},
|
|
{'id': 'schedule', 'name': 'Cronograma', 'icon': 'bi-calendar3'},
|
|
{'id': 'subjects', 'name': 'Asignaturas', 'icon': 'bi-book'},
|
|
{'id': 'careers', 'name': 'Carreras', 'icon': 'bi-mortarboard'},
|
|
{'id': 'commissions', 'name': 'Comisiones', 'icon': 'bi-diagram-3'},
|
|
{'id': 'users', 'name': 'Usuarios', 'icon': 'bi-people'},
|
|
{'id': 'roles', 'name': 'Roles (RBAC)', 'icon': 'bi-shield-check'},
|
|
{'id': 'optimizer', 'name': 'Optimizador IA', 'icon': 'bi-cpu'},
|
|
{'id': 'audit', 'name': 'Auditoría', 'icon': 'bi-clipboard-data'}
|
|
]
|
|
|
|
roles_data = []
|
|
for r in roles:
|
|
perms_map = {}
|
|
for p in r.permissions:
|
|
perms_map[p.module] = p.access_level
|
|
|
|
roles_data.append({
|
|
'id': r.id,
|
|
'name': r.name,
|
|
'description': r.description or f'Rol institucional de {r.name}',
|
|
'is_system': getattr(r, 'is_system', True),
|
|
'users_count': len(r.users),
|
|
'users': [{'id': u.id, 'name': u.name} for u in r.users],
|
|
'permissions': perms_map
|
|
})
|
|
|
|
return jsonify({
|
|
'status': 'success',
|
|
'roles': roles_data,
|
|
'modules': modules
|
|
}), 200
|
|
|
|
@api_admin_bp.route('/subjects', methods=['GET'])
|
|
@jwt_required
|
|
def get_subjects():
|
|
search = request.args.get('search', '').strip().lower()
|
|
career_id = request.args.get('career_id', type=int)
|
|
active = request.args.get('active', '').strip().lower()
|
|
|
|
query = Subject.query
|
|
if search:
|
|
query = query.filter((Subject.name.ilike(f'%{search}%')) | (Subject.code.ilike(f'%{search}%')) | (Subject.department.ilike(f'%{search}%')))
|
|
if career_id:
|
|
query = query.filter(Subject.career_id == career_id)
|
|
if active == 'true':
|
|
query = query.filter(Subject.is_active == True)
|
|
elif active == 'false':
|
|
query = query.filter(Subject.is_active == False)
|
|
|
|
subjects = query.order_by(Subject.id.asc()).all()
|
|
careers = Career.query.order_by(Career.name.asc()).all()
|
|
|
|
subjects_data = []
|
|
for s in subjects:
|
|
subjects_data.append({
|
|
'id': s.id,
|
|
'code': s.code,
|
|
'name': s.name,
|
|
'career_id': s.career_id,
|
|
'career_name': s.career_obj.name if s.career_obj else (s.department or 'General'),
|
|
'department': s.department or 'General',
|
|
'credits': getattr(s, 'credits', 4),
|
|
'is_active': s.is_active,
|
|
'commissions_count': len(s.commissions),
|
|
'commissions': [{'id': c.id, 'code': c.code, 'schedule': c.schedule} for c in s.commissions]
|
|
})
|
|
|
|
careers_data = [{'id': c.id, 'name': c.name} for c in careers]
|
|
|
|
return jsonify({
|
|
'status': 'success',
|
|
'total': len(subjects_data),
|
|
'subjects': subjects_data,
|
|
'careers': careers_data
|
|
}), 200
|
|
|
|
@api_admin_bp.route('/careers', methods=['GET'])
|
|
@jwt_required
|
|
def get_careers():
|
|
search = request.args.get('search', '').strip().lower()
|
|
query = Career.query
|
|
if search:
|
|
query = query.filter(Career.name.ilike(f'%{search}%'))
|
|
|
|
careers = query.order_by(Career.name.asc()).all()
|
|
careers_data = []
|
|
for c in careers:
|
|
careers_data.append({
|
|
'id': c.id,
|
|
'name': c.name,
|
|
'code': getattr(c, 'code', f'CAR-{c.id}'),
|
|
'degree_level': getattr(c, 'degree_level', 'Grado / Tecnicatura'),
|
|
'is_active': getattr(c, 'is_active', True),
|
|
'subjects_count': len(c.subjects),
|
|
'subjects': [{'id': s.id, 'name': s.name, 'code': s.code} for s in c.subjects[:5]]
|
|
})
|
|
|
|
return jsonify({
|
|
'status': 'success',
|
|
'total': len(careers_data),
|
|
'careers': careers_data
|
|
}), 200
|
|
|
|
@api_admin_bp.route('/commissions', methods=['GET'])
|
|
@jwt_required
|
|
def get_commissions():
|
|
search = request.args.get('search', '').strip().lower()
|
|
subject_id = request.args.get('subject_id', type=int)
|
|
shift = request.args.get('shift', '').strip()
|
|
active = request.args.get('active', '').strip().lower()
|
|
|
|
query = Commission.query
|
|
if subject_id:
|
|
query = query.filter(Commission.subject_id == subject_id)
|
|
if shift:
|
|
query = query.filter(Commission.shift == shift)
|
|
if active == 'true':
|
|
query = query.filter(Commission.active == True)
|
|
elif active == 'false':
|
|
query = query.filter(Commission.active == False)
|
|
|
|
commissions = query.order_by(Commission.id.asc()).all()
|
|
subjects = Subject.query.order_by(Subject.name.asc()).all()
|
|
|
|
commissions_data = []
|
|
for c in commissions:
|
|
if search:
|
|
s_name = c.subject.name.lower() if c.subject else ''
|
|
c_code = c.code.lower()
|
|
if search not in s_name and search not in c_code:
|
|
continue
|
|
|
|
career_name = 'Carrera General'
|
|
if c.subject and hasattr(c.subject, 'career') and c.subject.career:
|
|
career_name = c.subject.career.name
|
|
|
|
commissions_data.append({
|
|
'id': c.id,
|
|
'code': c.code,
|
|
'full_code': c.get_full_code() if hasattr(c, 'get_full_code') else c.code,
|
|
'subject_id': c.subject_id,
|
|
'subject_name': c.subject.name if c.subject else 'Sin materia',
|
|
'subject_code': c.subject.code if c.subject else '-',
|
|
'career_name': career_name,
|
|
'semester': getattr(c, 'semester', '1C') or '1C',
|
|
'year': getattr(c, 'year', 2026) or 2026,
|
|
'teacher_id': c.teacher_id,
|
|
'teacher_name': c.teacher_name if hasattr(c, 'teacher_name') else (c.teacher.name if c.teacher else None),
|
|
'teachers': c.teachers if hasattr(c, 'teachers') else [],
|
|
'teacher_names': c.teacher_names if hasattr(c, 'teacher_names') else (c.teacher_name if hasattr(c, 'teacher_name') else 'Sin asignar'),
|
|
'schedule': c.schedule or 'A coordinar',
|
|
'shift': c.shift or 'Mañana',
|
|
'max_students': c.max_students or 35,
|
|
'current_students': getattr(c, 'current_students', 0) or 0,
|
|
'active': c.active,
|
|
'virtual_link': getattr(c, 'virtual_link', '') or ''
|
|
})
|
|
|
|
subjects_data = []
|
|
for s in subjects:
|
|
c_name = s.career.name if (hasattr(s, 'career') and s.career) else 'Carrera General'
|
|
subjects_data.append({
|
|
'id': s.id,
|
|
'name': s.name,
|
|
'code': s.code,
|
|
'career': c_name,
|
|
'career_name': c_name
|
|
})
|
|
|
|
return jsonify({
|
|
'status': 'success',
|
|
'total': len(commissions_data),
|
|
'commissions': commissions_data,
|
|
'subjects': subjects_data
|
|
}), 200
|
|
|
|
# ---------------------------------------------------------
|
|
# USERS CRUD
|
|
# ---------------------------------------------------------
|
|
|
|
@api_admin_bp.route('/users', methods=['POST'])
|
|
@jwt_required
|
|
def create_user():
|
|
data = request.get_json(silent=True) or request.form.to_dict() or {}
|
|
email = data.get('email', '').strip().lower()
|
|
first_name = data.get('first_name', '').strip()
|
|
last_name = data.get('last_name', '').strip()
|
|
name = data.get('name', '').strip()
|
|
if not name and (first_name or last_name):
|
|
name = f"{first_name} {last_name}".strip()
|
|
elif name and not first_name:
|
|
parts = name.split()
|
|
first_name = parts[0] if parts else ''
|
|
last_name = ' '.join(parts[1:]) if len(parts) > 1 else ''
|
|
|
|
password = data.get('password', '').strip()
|
|
role_id = data.get('role_id')
|
|
is_active = data.get('is_active', True)
|
|
if isinstance(is_active, str):
|
|
is_active = is_active.lower() in ['true', '1', 'on']
|
|
|
|
phone = data.get('phone', '').strip()
|
|
personal_email = data.get('personal_email', '').strip().lower()
|
|
address = data.get('address', '').strip()
|
|
document_type = data.get('document_type', 'DNI').strip().upper()
|
|
document_number = data.get('document_number', '').strip()
|
|
|
|
if not email or not name:
|
|
return jsonify({'error': 'ValidationError', 'message': 'El nombre y el email son obligatorios.'}), 400
|
|
|
|
if User.query.filter(User.email.ilike(email)).first():
|
|
return jsonify({'error': 'Conflict', 'message': f'Ya existe un usuario con el email {email}.'}), 409
|
|
|
|
if document_number:
|
|
is_valid, clean_doc, err_msg = validate_document(document_type, document_number)
|
|
if not is_valid:
|
|
return jsonify({'error': 'ValidationError', 'message': err_msg}), 400
|
|
document_number = clean_doc
|
|
|
|
existing_doc = User.query.filter(
|
|
User.document_type == document_type,
|
|
User.document_number == document_number
|
|
).first()
|
|
if existing_doc:
|
|
return jsonify({'error': 'Conflict', 'message': f'Ya existe un usuario con {document_type} {document_number}.'}), 409
|
|
|
|
role_obj = None
|
|
if role_id:
|
|
role_obj = Role.query.get(int(role_id))
|
|
role_name = role_obj.name if role_obj else data.get('role', 'Docente')
|
|
|
|
user = User(
|
|
email=email,
|
|
personal_email=personal_email,
|
|
name=name,
|
|
first_name=first_name,
|
|
last_name=last_name,
|
|
phone=phone,
|
|
address=address,
|
|
document_type=document_type,
|
|
document_number=document_number,
|
|
role=role_name,
|
|
role_id=int(role_id) if role_id else None,
|
|
is_active=bool(is_active)
|
|
)
|
|
user.set_password(password or 'edu-space2026')
|
|
db.session.add(user)
|
|
db.session.commit()
|
|
|
|
# Encolar sincronización con Moodle de forma asíncrona tolerante a fallos
|
|
try:
|
|
from app.services.moodle_queue_service import moodle_queue_service
|
|
username = user.email.split('@')[0].lower()
|
|
moodle_queue_service.enqueue_task(
|
|
action='CREATE_USER',
|
|
entity_type='user',
|
|
entity_id=user.id,
|
|
payload={
|
|
'username': username,
|
|
'email': user.email,
|
|
'firstname': user.first_name or (user.name.split()[0] if user.name else 'Docente'),
|
|
'lastname': user.last_name or (user.name.split()[1] if len(user.name.split()) > 1 else 'EduSpace'),
|
|
'password': password or 'EduSpace2026*'
|
|
}
|
|
)
|
|
except Exception:
|
|
pass
|
|
|
|
return jsonify({
|
|
'status': 'success',
|
|
'message': 'Usuario creado exitosamente.',
|
|
'user': user.to_dict()
|
|
}), 201
|
|
|
|
@api_admin_bp.route('/users/<int:id>', methods=['PUT'])
|
|
@jwt_required
|
|
def update_user(id):
|
|
user = User.query.get_or_404(id)
|
|
data = request.get_json(silent=True) or request.form.to_dict() or {}
|
|
|
|
first_name = data.get('first_name', user.first_name or '').strip()
|
|
last_name = data.get('last_name', user.last_name or '').strip()
|
|
name = data.get('name', '').strip()
|
|
|
|
if 'first_name' in data or 'last_name' in data:
|
|
user.first_name = first_name
|
|
user.last_name = last_name
|
|
user.name = f"{first_name} {last_name}".strip()
|
|
elif name:
|
|
user.name = name
|
|
parts = name.split()
|
|
user.first_name = parts[0] if parts else ''
|
|
user.last_name = ' '.join(parts[1:]) if len(parts) > 1 else ''
|
|
|
|
if 'phone' in data:
|
|
user.phone = (data['phone'] or '').strip()
|
|
if 'personal_email' in data:
|
|
user.personal_email = (data['personal_email'] or '').strip().lower()
|
|
if 'address' in data:
|
|
user.address = (data['address'] or '').strip()
|
|
|
|
if 'document_type' in data or 'document_number' in data:
|
|
doc_type = data.get('document_type', user.document_type or 'DNI').strip().upper()
|
|
doc_num = data.get('document_number', user.document_number or '').strip()
|
|
|
|
if doc_num:
|
|
is_valid, clean_doc, err_msg = validate_document(doc_type, doc_num)
|
|
if not is_valid:
|
|
return jsonify({'error': 'ValidationError', 'message': err_msg}), 400
|
|
doc_num = clean_doc
|
|
|
|
existing_doc = User.query.filter(
|
|
User.document_type == doc_type,
|
|
User.document_number == doc_num,
|
|
User.id != id
|
|
).first()
|
|
if existing_doc:
|
|
return jsonify({'error': 'Conflict', 'message': f'El documento {doc_type} {doc_num} ya está asignado a otro usuario.'}), 409
|
|
|
|
user.document_type = doc_type
|
|
user.document_number = doc_num
|
|
|
|
if 'email' in data and data['email']:
|
|
email = data['email'].strip().lower()
|
|
existing = User.query.filter(User.email.ilike(email), User.id != id).first()
|
|
if existing:
|
|
return jsonify({'error': 'Conflict', 'message': f'El email {email} ya está registrado por otro usuario.'}), 409
|
|
user.email = email
|
|
if 'password' in data and data['password']:
|
|
user.set_password(data['password'].strip())
|
|
if 'role_id' in data and data['role_id']:
|
|
role_obj = Role.query.get(int(data['role_id']))
|
|
if role_obj:
|
|
user.role_id = role_obj.id
|
|
user.role = role_obj.name
|
|
if 'is_active' in data:
|
|
val = data['is_active']
|
|
user.is_active = val in [True, 'true', '1', 'on']
|
|
|
|
db.session.commit()
|
|
|
|
# Encolar actualización con Moodle de forma asíncrona tolerante a fallos
|
|
try:
|
|
from app.services.moodle_queue_service import moodle_queue_service
|
|
username = user.email.split('@')[0].lower()
|
|
moodle_queue_service.enqueue_task(
|
|
action='UPDATE_USER',
|
|
entity_type='user',
|
|
entity_id=user.id,
|
|
payload={
|
|
'username': username,
|
|
'email': user.email,
|
|
'firstname': user.first_name or (user.name.split()[0] if user.name else 'Docente'),
|
|
'lastname': user.last_name or (user.name.split()[1] if len(user.name.split()) > 1 else 'EduSpace')
|
|
}
|
|
)
|
|
except Exception:
|
|
pass
|
|
|
|
return jsonify({
|
|
'status': 'success',
|
|
'message': 'Usuario actualizado correctamente.',
|
|
'user': user.to_dict()
|
|
}), 200
|
|
|
|
@api_admin_bp.route('/users/document-types', methods=['GET'])
|
|
@jwt_required
|
|
def get_document_types():
|
|
return jsonify({
|
|
'status': 'success',
|
|
'document_types': DOCUMENT_TYPES
|
|
}), 200
|
|
|
|
@api_admin_bp.route('/users/<int:id>/toggle', methods=['POST'])
|
|
@jwt_required
|
|
def toggle_user(id):
|
|
user = User.query.get_or_404(id)
|
|
user.is_active = not user.is_active
|
|
db.session.commit()
|
|
return jsonify({
|
|
'status': 'success',
|
|
'id': user.id,
|
|
'is_active': user.is_active,
|
|
'message': f'Usuario {"activado" if user.is_active else "desactivado"} correctamente.'
|
|
}), 200
|
|
|
|
@api_admin_bp.route('/users/<int:id>', methods=['DELETE'])
|
|
@jwt_required
|
|
def delete_user(id):
|
|
user = User.query.get_or_404(id)
|
|
has_reservations = len(user.reservations) if user.reservations else 0
|
|
if has_reservations > 0:
|
|
user.is_active = False
|
|
db.session.commit()
|
|
return jsonify({
|
|
'status': 'success',
|
|
'message': 'El usuario tiene reservas asociadas. Se ha desactivado en su lugar.'
|
|
}), 200
|
|
|
|
db.session.delete(user)
|
|
db.session.commit()
|
|
return jsonify({
|
|
'status': 'success',
|
|
'message': 'Usuario eliminado permanentemente.'
|
|
}), 200
|
|
|
|
# ---------------------------------------------------------
|
|
# ROLES CRUD
|
|
# ---------------------------------------------------------
|
|
|
|
@api_admin_bp.route('/roles', methods=['POST'])
|
|
@jwt_required
|
|
def create_role():
|
|
data = request.get_json(silent=True) or request.form.to_dict() or {}
|
|
name = data.get('name', '').strip()
|
|
description = data.get('description', '').strip()
|
|
permissions = data.get('permissions', {})
|
|
|
|
if not name:
|
|
return jsonify({'error': 'ValidationError', 'message': 'El nombre del rol es obligatorio.'}), 400
|
|
|
|
if Role.query.filter(Role.name.ilike(name)).first():
|
|
return jsonify({'error': 'Conflict', 'message': f'Ya existe un rol llamado {name}.'}), 409
|
|
|
|
role = Role(name=name, description=description, is_system=False)
|
|
db.session.add(role)
|
|
db.session.flush()
|
|
|
|
for mod, level in permissions.items():
|
|
role.set_permission(mod, level)
|
|
|
|
db.session.commit()
|
|
return jsonify({
|
|
'status': 'success',
|
|
'message': 'Rol creado exitosamente.',
|
|
'role': role.to_dict()
|
|
}), 201
|
|
|
|
@api_admin_bp.route('/roles/<int:id>', methods=['PUT'])
|
|
@jwt_required
|
|
def update_role(id):
|
|
role = Role.query.get_or_404(id)
|
|
data = request.get_json(silent=True) or request.form.to_dict() or {}
|
|
|
|
if 'name' in data and data['name']:
|
|
name = data['name'].strip()
|
|
existing = Role.query.filter(Role.name.ilike(name), Role.id != id).first()
|
|
if existing:
|
|
return jsonify({'error': 'Conflict', 'message': f'Ya existe otro rol llamado {name}.'}), 409
|
|
role.name = name
|
|
|
|
if 'description' in data:
|
|
role.description = data['description'].strip()
|
|
|
|
if 'permissions' in data and isinstance(data['permissions'], dict):
|
|
for mod, level in data['permissions'].items():
|
|
role.set_permission(mod, level)
|
|
|
|
db.session.commit()
|
|
return jsonify({
|
|
'status': 'success',
|
|
'message': 'Rol actualizado correctamente.',
|
|
'role': role.to_dict()
|
|
}), 200
|
|
|
|
@api_admin_bp.route('/roles/<int:id>', methods=['DELETE'])
|
|
@jwt_required
|
|
def delete_role(id):
|
|
role = Role.query.get_or_404(id)
|
|
if role.is_system:
|
|
return jsonify({'error': 'Forbidden', 'message': 'No se pueden eliminar roles de sistema.'}), 403
|
|
if role.users and len(role.users) > 0:
|
|
return jsonify({'error': 'Conflict', 'message': f'El rol tiene {len(role.users)} usuarios asignados. Reasígnelos primero.'}), 409
|
|
|
|
db.session.delete(role)
|
|
db.session.commit()
|
|
return jsonify({
|
|
'status': 'success',
|
|
'message': 'Rol eliminado exitosamente.'
|
|
}), 200
|
|
|
|
# ---------------------------------------------------------
|
|
# CAREERS CRUD
|
|
# ---------------------------------------------------------
|
|
|
|
@api_admin_bp.route('/careers', methods=['POST'])
|
|
@jwt_required
|
|
def create_career():
|
|
data = request.get_json(silent=True) or request.form.to_dict() or {}
|
|
name = data.get('name', '').strip()
|
|
code = data.get('code', '').strip()
|
|
description = data.get('description', '').strip()
|
|
|
|
if not name:
|
|
return jsonify({'error': 'ValidationError', 'message': 'El nombre de la carrera es obligatorio.'}), 400
|
|
|
|
if Career.query.filter(Career.name.ilike(name)).first():
|
|
return jsonify({'error': 'Conflict', 'message': f'Ya existe una carrera con el nombre {name}.'}), 409
|
|
|
|
career = Career(name=name, code=code or None, description=description or None)
|
|
db.session.add(career)
|
|
db.session.commit()
|
|
|
|
return jsonify({
|
|
'status': 'success',
|
|
'message': 'Carrera creada exitosamente.',
|
|
'career': career.to_dict()
|
|
}), 201
|
|
|
|
@api_admin_bp.route('/careers/<int:id>', methods=['PUT'])
|
|
@jwt_required
|
|
def update_career(id):
|
|
career = Career.query.get_or_404(id)
|
|
data = request.get_json(silent=True) or request.form.to_dict() or {}
|
|
|
|
if 'name' in data and data['name']:
|
|
name = data['name'].strip()
|
|
existing = Career.query.filter(Career.name.ilike(name), Career.id != id).first()
|
|
if existing:
|
|
return jsonify({'error': 'Conflict', 'message': f'Ya existe otra carrera con el nombre {name}.'}), 409
|
|
career.name = name
|
|
|
|
if 'code' in data:
|
|
career.code = data['code'].strip() or None
|
|
|
|
if 'description' in data:
|
|
career.description = data['description'].strip() or None
|
|
|
|
db.session.commit()
|
|
return jsonify({
|
|
'status': 'success',
|
|
'message': 'Carrera actualizada correctamente.',
|
|
'career': career.to_dict()
|
|
}), 200
|
|
|
|
@api_admin_bp.route('/careers/<int:id>', methods=['DELETE'])
|
|
@jwt_required
|
|
def delete_career(id):
|
|
career = Career.query.get_or_404(id)
|
|
if career.subjects and len(career.subjects) > 0:
|
|
return jsonify({
|
|
'error': 'Conflict',
|
|
'message': f'No se puede eliminar la carrera porque tiene {len(career.subjects)} asignaturas vinculadas.'
|
|
}), 409
|
|
|
|
db.session.delete(career)
|
|
db.session.commit()
|
|
return jsonify({
|
|
'status': 'success',
|
|
'message': 'Carrera eliminada permanentemente.'
|
|
}), 200
|
|
|
|
# ---------------------------------------------------------
|
|
# SUBJECTS CRUD
|
|
# ---------------------------------------------------------
|
|
|
|
@api_admin_bp.route('/subjects', methods=['POST'])
|
|
@jwt_required
|
|
def create_subject():
|
|
data = request.get_json(silent=True) or request.form.to_dict() or {}
|
|
code = data.get('code', '').strip().upper()
|
|
name = data.get('name', '').strip()
|
|
department = data.get('department', '').strip()
|
|
credits = int(data.get('credits', 4) or 4)
|
|
career_id = data.get('career_id')
|
|
career_id = int(career_id) if career_id else None
|
|
description = data.get('description', '').strip()
|
|
is_active = data.get('is_active', True)
|
|
if isinstance(is_active, str):
|
|
is_active = is_active.lower() in ['true', '1', 'on']
|
|
|
|
if not code or not name:
|
|
return jsonify({'error': 'ValidationError', 'message': 'El código y nombre de la asignatura son obligatorios.'}), 400
|
|
|
|
if Subject.query.filter_by(code=code).first():
|
|
return jsonify({'error': 'Conflict', 'message': f'Ya existe una asignatura con el código {code}.'}), 409
|
|
|
|
subject = Subject(
|
|
code=code,
|
|
name=name,
|
|
department=department,
|
|
credits=credits,
|
|
career_id=career_id,
|
|
description=description,
|
|
is_active=bool(is_active)
|
|
)
|
|
db.session.add(subject)
|
|
db.session.commit()
|
|
|
|
return jsonify({
|
|
'status': 'success',
|
|
'message': 'Asignatura creada exitosamente.',
|
|
'subject': subject.to_dict()
|
|
}), 201
|
|
|
|
@api_admin_bp.route('/subjects/<int:id>', methods=['PUT'])
|
|
@jwt_required
|
|
def update_subject(id):
|
|
subject = Subject.query.get_or_404(id)
|
|
data = request.get_json(silent=True) or request.form.to_dict() or {}
|
|
|
|
if 'code' in data and data['code']:
|
|
code = data['code'].strip().upper()
|
|
existing = Subject.query.filter(Subject.code == code, Subject.id != id).first()
|
|
if existing:
|
|
return jsonify({'error': 'Conflict', 'message': f'Ya existe otra asignatura con el código {code}.'}), 409
|
|
subject.code = code
|
|
|
|
if 'name' in data and data['name']:
|
|
subject.name = data['name'].strip()
|
|
|
|
if 'department' in data:
|
|
subject.department = data['department'].strip()
|
|
|
|
if 'credits' in data and data['credits'] is not None:
|
|
subject.credits = int(data['credits'])
|
|
|
|
if 'career_id' in data:
|
|
c_id = data['career_id']
|
|
subject.career_id = int(c_id) if c_id else None
|
|
|
|
if 'description' in data:
|
|
subject.description = data['description'].strip()
|
|
|
|
if 'is_active' in data:
|
|
val = data['is_active']
|
|
subject.is_active = val in [True, 'true', '1', 'on']
|
|
|
|
db.session.commit()
|
|
return jsonify({
|
|
'status': 'success',
|
|
'message': 'Asignatura actualizada correctamente.',
|
|
'subject': subject.to_dict()
|
|
}), 200
|
|
|
|
@api_admin_bp.route('/subjects/<int:id>/toggle', methods=['POST'])
|
|
@jwt_required
|
|
def toggle_subject(id):
|
|
subject = Subject.query.get_or_404(id)
|
|
subject.is_active = not subject.is_active
|
|
db.session.commit()
|
|
return jsonify({
|
|
'status': 'success',
|
|
'id': subject.id,
|
|
'is_active': subject.is_active,
|
|
'message': f'Asignatura {"activada" if subject.is_active else "desactivada"} correctamente.'
|
|
}), 200
|
|
|
|
@api_admin_bp.route('/subjects/<int:id>', methods=['DELETE'])
|
|
@jwt_required
|
|
def delete_subject(id):
|
|
subject = Subject.query.get_or_404(id)
|
|
if subject.commissions and len(subject.commissions) > 0:
|
|
subject.is_active = False
|
|
db.session.commit()
|
|
return jsonify({
|
|
'status': 'deactivated',
|
|
'action': 'deactivated',
|
|
'message': f'La asignatura "{subject.name}" tiene {len(subject.commissions)} comisión(es) vinculada(s). Se ha desactivado en su lugar para proteger el historial académico.'
|
|
}), 200
|
|
|
|
name = subject.name
|
|
db.session.delete(subject)
|
|
db.session.commit()
|
|
return jsonify({
|
|
'status': 'deleted',
|
|
'action': 'deleted',
|
|
'message': f'Asignatura "{name}" eliminada permanentemente del sistema.'
|
|
}), 200
|
|
|
|
# ---------------------------------------------------------
|
|
# COMMISSIONS CRUD
|
|
# ---------------------------------------------------------
|
|
|
|
@api_admin_bp.route('/commissions', methods=['POST'])
|
|
@jwt_required
|
|
def create_commission():
|
|
data = request.get_json(silent=True) or request.form.to_dict() or {}
|
|
subject_id = data.get('subject_id')
|
|
code = data.get('code', '').strip()
|
|
semester = (data.get('semester') or '1C').strip()
|
|
year = int(data.get('year', datetime.now().year) or datetime.now().year)
|
|
teacher_id = data.get('teacher_id')
|
|
teacher_id = int(teacher_id) if (teacher_id and str(teacher_id).strip()) else None
|
|
max_students = int(data.get('max_students', 35) or 35)
|
|
schedule = (data.get('schedule') or '').strip()
|
|
shift = (data.get('shift') or 'Mañana').strip()
|
|
virtual_link = (data.get('virtual_link') or '').strip()
|
|
active = data.get('active', True)
|
|
if isinstance(active, str):
|
|
active = active.lower() in ['true', '1', 'on']
|
|
|
|
if not subject_id or not code:
|
|
return jsonify({'error': 'ValidationError', 'message': 'La asignatura y código de comisión son obligatorios.'}), 400
|
|
|
|
subject = Subject.query.get(int(subject_id))
|
|
if not subject:
|
|
return jsonify({'error': 'NotFound', 'message': f'La asignatura con ID {subject_id} no existe.'}), 404
|
|
|
|
# Generar enlace automático si no fue provisto
|
|
if not virtual_link:
|
|
s_clean = re.sub(r'[^a-zA-Z0-9]', '', subject.code).lower()[:6] or 'subj'
|
|
c_clean = re.sub(r'[^a-zA-Z0-9]', '', code).lower()[:4] or 'c1'
|
|
virtual_link = f"https://meet.google.com/edu-{s_clean}-{c_clean}"
|
|
|
|
existing = Commission.query.filter_by(
|
|
subject_id=int(subject_id),
|
|
code=code,
|
|
semester=semester,
|
|
year=year
|
|
).first()
|
|
|
|
if existing:
|
|
existing.teacher_id = teacher_id
|
|
existing.max_students = max_students
|
|
existing.schedule = schedule
|
|
existing.shift = shift
|
|
if virtual_link:
|
|
existing.virtual_link = virtual_link
|
|
existing.active = bool(active)
|
|
if teacher_id:
|
|
ct = CommissionTeacher.query.filter_by(commission_id=existing.id, user_id=teacher_id).first()
|
|
if not ct:
|
|
db.session.add(CommissionTeacher(commission_id=existing.id, user_id=teacher_id, role='Titular', is_primary=True))
|
|
db.session.commit()
|
|
return jsonify({
|
|
'status': 'success',
|
|
'message': f'Comisión {code} asociada a {subject.name} actualizada exitosamente.',
|
|
'commission': existing.to_dict()
|
|
}), 200
|
|
|
|
commission = Commission(
|
|
subject_id=int(subject_id),
|
|
code=code,
|
|
semester=semester,
|
|
year=year,
|
|
teacher_id=teacher_id,
|
|
max_students=max_students,
|
|
schedule=schedule,
|
|
shift=shift,
|
|
virtual_link=virtual_link,
|
|
active=bool(active)
|
|
)
|
|
db.session.add(commission)
|
|
db.session.flush()
|
|
|
|
if teacher_id:
|
|
db.session.add(CommissionTeacher(commission_id=commission.id, user_id=teacher_id, role='Titular', is_primary=True))
|
|
|
|
# Soporte para docentes adicionales al crear comisión
|
|
extra_teachers = data.get('teacher_ids') or data.get('teachers') or []
|
|
for et in extra_teachers:
|
|
u_id = et.get('user_id') if isinstance(et, dict) else et
|
|
u_role = et.get('role', 'Adjunto') if isinstance(et, dict) else 'Adjunto'
|
|
if u_id and int(u_id) != teacher_id:
|
|
db.session.add(CommissionTeacher(commission_id=commission.id, user_id=int(u_id), role=u_role, is_primary=False))
|
|
|
|
db.session.commit()
|
|
|
|
return jsonify({
|
|
'status': 'success',
|
|
'message': f'Comisión {code} asociada exitosamente a {subject.name}.',
|
|
'commission': commission.to_dict()
|
|
}), 201
|
|
|
|
|
|
@api_admin_bp.route('/commissions/<int:id>', methods=['GET'])
|
|
@jwt_required
|
|
def get_commission_detail(id):
|
|
comm = Commission.query.get_or_404(id)
|
|
return jsonify({
|
|
'status': 'success',
|
|
'commission': comm.to_dict()
|
|
}), 200
|
|
|
|
|
|
@api_admin_bp.route('/commissions/<int:id>', methods=['PUT'])
|
|
@jwt_required
|
|
def update_commission(id):
|
|
comm = Commission.query.get_or_404(id)
|
|
data = request.get_json(silent=True) or request.form.to_dict() or {}
|
|
|
|
if 'code' in data and data['code']:
|
|
comm.code = data['code'].strip()
|
|
if 'semester' in data and data['semester']:
|
|
comm.semester = data['semester'].strip()
|
|
if 'year' in data and data['year']:
|
|
comm.year = int(data['year'])
|
|
if 'teacher_id' in data:
|
|
t_id = data['teacher_id']
|
|
comm.teacher_id = int(t_id) if t_id else None
|
|
if comm.teacher_id:
|
|
ct = CommissionTeacher.query.filter_by(commission_id=comm.id, user_id=comm.teacher_id).first()
|
|
if not ct:
|
|
db.session.add(CommissionTeacher(commission_id=comm.id, user_id=comm.teacher_id, role='Titular', is_primary=True))
|
|
if 'max_students' in data and data['max_students']:
|
|
comm.max_students = int(data['max_students'])
|
|
if 'schedule' in data:
|
|
comm.schedule = data['schedule'].strip()
|
|
if 'shift' in data:
|
|
comm.shift = data['shift'].strip()
|
|
if 'virtual_link' in data:
|
|
comm.virtual_link = data['virtual_link'].strip()
|
|
if 'active' in data:
|
|
val = data['active']
|
|
comm.active = val in [True, 'true', '1', 'on']
|
|
|
|
db.session.commit()
|
|
return jsonify({
|
|
'status': 'success',
|
|
'message': 'Comisión actualizada correctamente.',
|
|
'commission': comm.to_dict()
|
|
}), 200
|
|
|
|
@api_admin_bp.route('/commissions/<int:id>/toggle', methods=['POST'])
|
|
@jwt_required
|
|
def toggle_commission(id):
|
|
comm = Commission.query.get_or_404(id)
|
|
comm.active = not comm.active
|
|
db.session.commit()
|
|
return jsonify({
|
|
'status': 'success',
|
|
'id': comm.id,
|
|
'active': comm.active,
|
|
'message': f'Comisión {"activada" if comm.active else "desactivada"} correctamente.'
|
|
}), 200
|
|
|
|
@api_admin_bp.route('/commissions/<int:id>/assign-teacher', methods=['POST'])
|
|
@jwt_required
|
|
def assign_commission_teacher(id):
|
|
comm = Commission.query.get_or_404(id)
|
|
data = request.get_json(silent=True) or request.form.to_dict() or {}
|
|
t_id = data.get('teacher_id') or data.get('user_id')
|
|
role = (data.get('role') or 'Titular').strip()
|
|
|
|
if t_id:
|
|
t_id = int(t_id)
|
|
comm.teacher_id = t_id
|
|
ct = CommissionTeacher.query.filter_by(commission_id=comm.id, user_id=t_id).first()
|
|
if not ct:
|
|
ct = CommissionTeacher(commission_id=comm.id, user_id=t_id, role=role, is_primary=True)
|
|
db.session.add(ct)
|
|
else:
|
|
ct.role = role
|
|
ct.is_primary = True
|
|
else:
|
|
comm.teacher_id = None
|
|
|
|
db.session.commit()
|
|
return jsonify({
|
|
'status': 'success',
|
|
'id': comm.id,
|
|
'teacher_id': comm.teacher_id,
|
|
'teacher_name': comm.teacher_name,
|
|
'teachers': comm.teachers,
|
|
'message': 'Docente asignado correctamente.'
|
|
}), 200
|
|
|
|
@api_admin_bp.route('/commissions/<int:id>/teachers', methods=['POST'])
|
|
@jwt_required
|
|
def add_commission_teacher(id):
|
|
comm = Commission.query.get_or_404(id)
|
|
data = request.get_json(silent=True) or request.form.to_dict() or {}
|
|
user_id = data.get('user_id') or data.get('teacher_id')
|
|
if not user_id:
|
|
return jsonify({'error': 'ValidationError', 'message': 'El docente es requerido.'}), 400
|
|
|
|
user = User.query.get(int(user_id))
|
|
if not user:
|
|
return jsonify({'error': 'NotFound', 'message': f'El usuario docente con ID {user_id} no existe.'}), 404
|
|
|
|
role = (data.get('role') or 'Adjunto').strip()
|
|
is_primary = data.get('is_primary', False)
|
|
if isinstance(is_primary, str):
|
|
is_primary = is_primary.lower() in ['true', '1', 'on']
|
|
|
|
existing = CommissionTeacher.query.filter_by(commission_id=comm.id, user_id=user.id).first()
|
|
if existing:
|
|
existing.role = role
|
|
existing.is_primary = bool(is_primary)
|
|
else:
|
|
if is_primary or comm.teacher_id is None:
|
|
comm.teacher_id = user.id
|
|
is_primary = True
|
|
ct = CommissionTeacher(
|
|
commission_id=comm.id,
|
|
user_id=user.id,
|
|
role=role,
|
|
is_primary=bool(is_primary)
|
|
)
|
|
db.session.add(ct)
|
|
|
|
db.session.commit()
|
|
return jsonify({
|
|
'status': 'success',
|
|
'message': f'{user.name} asignado/a como {role} en la comisión.',
|
|
'teachers': comm.teachers,
|
|
'commission': comm.to_dict()
|
|
}), 200
|
|
|
|
@api_admin_bp.route('/commissions/<int:id>/teachers/<int:user_id>', methods=['DELETE'])
|
|
@jwt_required
|
|
def remove_commission_teacher(id, user_id):
|
|
comm = Commission.query.get_or_404(id)
|
|
ct = CommissionTeacher.query.filter_by(commission_id=comm.id, user_id=user_id).first()
|
|
if ct:
|
|
db.session.delete(ct)
|
|
|
|
# Si era el docente titular, reasignar a otro docente disponible o dejar en None
|
|
if comm.teacher_id == user_id:
|
|
other = CommissionTeacher.query.filter(CommissionTeacher.commission_id == comm.id, CommissionTeacher.user_id != user_id).first()
|
|
comm.teacher_id = other.user_id if other else None
|
|
if other:
|
|
other.is_primary = True
|
|
|
|
db.session.commit()
|
|
return jsonify({
|
|
'status': 'success',
|
|
'message': 'Docente desvinculado de la comisión correctamente.',
|
|
'teachers': comm.teachers,
|
|
'commission': comm.to_dict()
|
|
}), 200
|
|
|
|
# ---------------------------------------------------------
|
|
# COMMISSION STUDENT ENROLLMENTS (FASE 2 MVP)
|
|
# ---------------------------------------------------------
|
|
|
|
@api_admin_bp.route('/commissions/<int:id>/enrollments', methods=['GET'])
|
|
@jwt_required
|
|
def get_commission_enrollments(id):
|
|
comm = Commission.query.get_or_404(id)
|
|
enrollments = EnrollmentService.list_enrollments(comm.id)
|
|
return jsonify({
|
|
'commission_id': comm.id,
|
|
'commission_code': comm.code,
|
|
'count': len(enrollments),
|
|
'max_students': comm.max_students,
|
|
'current_students': comm.current_students,
|
|
'enrollments': enrollments
|
|
}), 200
|
|
|
|
@api_admin_bp.route('/commissions/<int:id>/enrollments', methods=['POST'])
|
|
@jwt_required
|
|
def enroll_commission_student(id):
|
|
data = request.get_json(silent=True) or {}
|
|
student_id = data.get('student_id')
|
|
notes = data.get('notes')
|
|
allow_same_day_exception = bool(data.get('allow_same_day_exception', False))
|
|
exception_reason = data.get('exception_reason')
|
|
|
|
if not student_id:
|
|
return jsonify({'error': 'BadRequest', 'message': 'student_id es obligatorio.'}), 400
|
|
|
|
try:
|
|
enrollment = EnrollmentService.enroll_student(
|
|
commission_id=id,
|
|
student_id=int(student_id),
|
|
notes=notes,
|
|
allow_same_day_exception=allow_same_day_exception,
|
|
exception_reason=exception_reason
|
|
)
|
|
return jsonify({
|
|
'status': 'success',
|
|
'message': 'Estudiante matriculado exitosamente en la comisión.',
|
|
'enrollment': enrollment.to_dict()
|
|
}), 201
|
|
except ValueError as e:
|
|
return jsonify({'error': 'ConflictOrValidation', 'message': str(e)}), 409
|
|
|
|
@api_admin_bp.route('/commissions/<int:id>/enrollments/<int:student_id>', methods=['DELETE'])
|
|
@jwt_required
|
|
def unenroll_commission_student(id, student_id):
|
|
success = EnrollmentService.unenroll_student(commission_id=id, student_id=student_id)
|
|
if not success:
|
|
return jsonify({'error': 'NotFound', 'message': 'Matrícula no encontrada para este estudiante.'}), 404
|
|
return jsonify({
|
|
'status': 'success',
|
|
'message': 'Estudiante desvinculado de la comisión exitosamente.'
|
|
}), 200
|
|
|
|
@api_admin_bp.route('/commissions/<int:id>/enrollments/<int:student_id>', methods=['PUT'])
|
|
@jwt_required
|
|
def update_commission_enrollment(id, student_id):
|
|
data = request.get_json(silent=True) or {}
|
|
status = data.get('status', 'activo')
|
|
notes = data.get('notes')
|
|
|
|
enrollment = EnrollmentService.update_enrollment_status(
|
|
commission_id=id,
|
|
student_id=student_id,
|
|
status=status,
|
|
notes=notes
|
|
)
|
|
if not enrollment:
|
|
return jsonify({'error': 'NotFound', 'message': 'Matrícula no encontrada.'}), 404
|
|
|
|
return jsonify({
|
|
'status': 'success',
|
|
'message': 'Estado de matrícula actualizado.',
|
|
'enrollment': enrollment.to_dict()
|
|
}), 200
|
|
|
|
# ---------------------------------------------------------
|
|
# COMMISSION GRADEBOOK & ACTAS (FASE 3 MVP)
|
|
# ---------------------------------------------------------
|
|
|
|
@api_admin_bp.route('/commissions/<int:id>/gradebook', methods=['GET'])
|
|
@jwt_required
|
|
def get_commission_gradebook(id):
|
|
"""Obtiene la matriz completa de calificaciones de la comisión."""
|
|
try:
|
|
data = GradebookService.get_commission_gradebook(id)
|
|
return jsonify({'status': 'success', 'data': data}), 200
|
|
except ValueError as e:
|
|
return jsonify({'error': 'NotFound', 'message': str(e)}), 404
|
|
except Exception as e:
|
|
return jsonify({'error': 'ServerError', 'message': str(e)}), 500
|
|
|
|
@api_admin_bp.route('/commissions/<int:id>/gradebook/grades', methods=['POST', 'PUT'])
|
|
@jwt_required
|
|
def save_commission_grades(id):
|
|
"""Carga masiva o individual de notas con autoguardado asíncrono."""
|
|
data = request.get_json(silent=True) or {}
|
|
grader_id = getattr(request, 'current_user_id', None)
|
|
|
|
try:
|
|
if 'grades' in data and isinstance(data['grades'], list):
|
|
res = GradebookService.bulk_save_grades(
|
|
commission_id=id,
|
|
grades_data=data['grades'],
|
|
grader_id=grader_id
|
|
)
|
|
return jsonify({'status': 'success', 'message': f"{res['saved_count']} calificaciones guardadas.", 'data': res}), 200
|
|
else:
|
|
milestone_id = data.get('milestone_id')
|
|
student_id = data.get('student_id')
|
|
score = data.get('score')
|
|
is_absent = bool(data.get('is_absent', False))
|
|
feedback = data.get('feedback', '')
|
|
|
|
if not milestone_id or not student_id:
|
|
return jsonify({'error': 'BadRequest', 'message': 'milestone_id y student_id son obligatorios.'}), 400
|
|
|
|
grade = GradebookService.save_single_grade(
|
|
commission_id=id,
|
|
milestone_id=int(milestone_id),
|
|
student_id=int(student_id),
|
|
score=float(score) if score is not None and str(score).strip() != '' else None,
|
|
is_absent=is_absent,
|
|
feedback=feedback,
|
|
grader_id=grader_id
|
|
)
|
|
return jsonify({'status': 'success', 'message': 'Calificación guardada.', 'grade': grade.to_dict()}), 200
|
|
|
|
except ValueError as e:
|
|
return jsonify({'error': 'ValidationError', 'message': str(e)}), 400
|
|
except Exception as e:
|
|
return jsonify({'error': 'ServerError', 'message': str(e)}), 500
|
|
|
|
@api_admin_bp.route('/commissions/<int:id>/gradebook/close', methods=['POST'])
|
|
@jwt_required
|
|
def close_commission_gradebook(id):
|
|
"""Cierre formal del acta de regularidad y promoción."""
|
|
data = request.get_json(silent=True) or {}
|
|
user_id = getattr(request, 'current_user_id', None) or 1
|
|
notes = data.get('notes', '')
|
|
|
|
try:
|
|
res = GradebookService.close_gradebook(commission_id=id, user_id=user_id, notes=notes)
|
|
return jsonify({'status': 'success', **res}), 200
|
|
except ValueError as e:
|
|
return jsonify({'error': 'BadRequest', 'message': str(e)}), 400
|
|
except Exception as e:
|
|
return jsonify({'error': 'ServerError', 'message': str(e)}), 500
|
|
|
|
@api_admin_bp.route('/commissions/<int:id>/gradebook/reopen', methods=['POST'])
|
|
@jwt_required
|
|
def reopen_commission_gradebook(id):
|
|
"""Reapertura administrativa del acta de calificaciones."""
|
|
data = request.get_json(silent=True) or {}
|
|
user_id = getattr(request, 'current_user_id', None) or 1
|
|
reason = data.get('reason', '')
|
|
|
|
try:
|
|
res = GradebookService.reopen_gradebook(commission_id=id, user_id=user_id, reason=reason)
|
|
return jsonify({'status': 'success', **res}), 200
|
|
except ValueError as e:
|
|
return jsonify({'error': 'BadRequest', 'message': str(e)}), 400
|
|
except Exception as e:
|
|
return jsonify({'error': 'ServerError', 'message': str(e)}), 500
|
|
|
|
# ---------------------------------------------------------
|
|
# ACADEMIC TERMS CRUD
|
|
# ---------------------------------------------------------
|
|
|
|
@api_admin_bp.route('/academic-terms', methods=['GET'])
|
|
@jwt_required
|
|
def get_academic_terms():
|
|
terms = AcademicTerm.query.order_by(AcademicTerm.year.desc(), AcademicTerm.code.desc()).all()
|
|
terms_data = []
|
|
for t in terms:
|
|
comm_count = Commission.query.filter_by(year=t.year).count()
|
|
terms_data.append({
|
|
'id': t.id,
|
|
'code': t.code,
|
|
'name': t.name,
|
|
'year': t.year,
|
|
'period_type': t.period_type,
|
|
'start_date': t.start_date.isoformat() if t.start_date else None,
|
|
'end_date': t.end_date.isoformat() if t.end_date else None,
|
|
'is_current': t.is_current,
|
|
'is_active': t.is_active,
|
|
'commissions_count': comm_count
|
|
})
|
|
return jsonify({
|
|
'status': 'success',
|
|
'total': len(terms_data),
|
|
'terms': terms_data
|
|
}), 200
|
|
|
|
@api_admin_bp.route('/academic-terms', methods=['POST'])
|
|
@jwt_required
|
|
def create_academic_term():
|
|
data = request.get_json(silent=True) or request.form.to_dict() or {}
|
|
code = data.get('code', '').strip()
|
|
name = data.get('name', '').strip()
|
|
year = int(data.get('year', datetime.now().year) or datetime.now().year)
|
|
period_type = data.get('period_type', 'Segundo Cuatrimestre').strip()
|
|
is_current = data.get('is_current', False)
|
|
if isinstance(is_current, str):
|
|
is_current = is_current.lower() in ['true', '1', 'on']
|
|
|
|
if not code or not name:
|
|
return jsonify({'error': 'ValidationError', 'message': 'Código y nombre son requeridos.'}), 400
|
|
|
|
existing = AcademicTerm.query.filter_by(code=code).first()
|
|
if existing:
|
|
return jsonify({'error': 'Conflict', 'message': f'Ya existe un ciclo con el código {code}.'}), 409
|
|
|
|
if is_current:
|
|
AcademicTerm.query.update({'is_current': False})
|
|
|
|
term = AcademicTerm(
|
|
code=code,
|
|
name=name,
|
|
year=year,
|
|
period_type=period_type,
|
|
is_current=bool(is_current),
|
|
is_active=True
|
|
)
|
|
db.session.add(term)
|
|
db.session.commit()
|
|
|
|
return jsonify({
|
|
'status': 'success',
|
|
'message': 'Ciclo lectivo creado exitosamente.',
|
|
'term': term.to_dict()
|
|
}), 201
|
|
|
|
@api_admin_bp.route('/academic-terms/<int:id>', methods=['PUT'])
|
|
@jwt_required
|
|
def update_academic_term(id):
|
|
term = AcademicTerm.query.get_or_404(id)
|
|
data = request.get_json(silent=True) or request.form.to_dict() or {}
|
|
|
|
if 'code' in data and data['code']:
|
|
code = data['code'].strip()
|
|
existing = AcademicTerm.query.filter(AcademicTerm.code == code, AcademicTerm.id != id).first()
|
|
if existing:
|
|
return jsonify({'error': 'Conflict', 'message': f'Ya existe otro ciclo con el código {code}.'}), 409
|
|
term.code = code
|
|
|
|
if 'name' in data and data['name']:
|
|
term.name = data['name'].strip()
|
|
if 'year' in data and data['year']:
|
|
term.year = int(data['year'])
|
|
if 'period_type' in data:
|
|
term.period_type = data['period_type'].strip()
|
|
if 'is_current' in data:
|
|
is_cur = data['is_current'] in [True, 'true', '1', 'on']
|
|
if is_cur:
|
|
AcademicTerm.query.update({'is_current': False})
|
|
term.is_current = is_cur
|
|
if 'is_active' in data:
|
|
term.is_active = data['is_active'] in [True, 'true', '1', 'on']
|
|
|
|
db.session.commit()
|
|
return jsonify({
|
|
'status': 'success',
|
|
'message': 'Ciclo lectivo actualizado correctamente.',
|
|
'term': term.to_dict()
|
|
}), 200
|
|
|
|
@api_admin_bp.route('/academic-terms/<int:id>/set-current', methods=['POST'])
|
|
@jwt_required
|
|
def set_current_academic_term(id):
|
|
term = AcademicTerm.query.get_or_404(id)
|
|
AcademicTerm.query.update({'is_current': False})
|
|
term.is_current = True
|
|
term.is_active = True
|
|
db.session.commit()
|
|
return jsonify({
|
|
'status': 'success',
|
|
'id': term.id,
|
|
'message': f'"{term.name}" establecido como ciclo activo.'
|
|
}), 200
|
|
|
|
# ---------------------------------------------------------
|
|
# MILESTONE TYPES CRUD
|
|
# ---------------------------------------------------------
|
|
|
|
@api_admin_bp.route('/milestone-types', methods=['GET'])
|
|
@jwt_required
|
|
def get_milestone_types():
|
|
types = MilestoneType.query.order_by(MilestoneType.sort_order.asc(), MilestoneType.id.asc()).all()
|
|
return jsonify({
|
|
'status': 'success',
|
|
'total': len(types),
|
|
'milestone_types': [t.to_dict() for t in types]
|
|
}), 200
|
|
|
|
@api_admin_bp.route('/milestone-types', methods=['POST'])
|
|
@jwt_required
|
|
def create_milestone_type():
|
|
data = request.get_json(silent=True) or request.form.to_dict() or {}
|
|
name = data.get('name', '').strip()
|
|
code = data.get('code', '').strip().lower().replace(' ', '_')
|
|
color = data.get('color', '#0d6efd').strip()
|
|
description = data.get('description', '').strip()
|
|
sort_order = int(data.get('sort_order', 0) or 0)
|
|
is_active = data.get('is_active', True)
|
|
if isinstance(is_active, str):
|
|
is_active = is_active.lower() in ['true', '1', 'on']
|
|
|
|
if not name or not code:
|
|
return jsonify({'error': 'ValidationError', 'message': 'Nombre y código son obligatorios.'}), 400
|
|
|
|
existing = MilestoneType.query.filter_by(code=code).first()
|
|
if existing:
|
|
return jsonify({'error': 'Conflict', 'message': f'Ya existe una tipificación con el código {code}.'}), 409
|
|
|
|
mt = MilestoneType(
|
|
name=name,
|
|
code=code,
|
|
color=color,
|
|
description=description,
|
|
sort_order=sort_order,
|
|
is_active=bool(is_active)
|
|
)
|
|
db.session.add(mt)
|
|
db.session.commit()
|
|
|
|
return jsonify({
|
|
'status': 'success',
|
|
'message': 'Tipificación de hito creada correctamente.',
|
|
'milestone_type': mt.to_dict()
|
|
}), 201
|
|
|
|
@api_admin_bp.route('/milestone-types/<int:id>', methods=['PUT'])
|
|
@jwt_required
|
|
def update_milestone_type(id):
|
|
mt = MilestoneType.query.get_or_404(id)
|
|
data = request.get_json(silent=True) or request.form.to_dict() or {}
|
|
|
|
if 'name' in data and data['name']:
|
|
mt.name = data['name'].strip()
|
|
if 'code' in data and data['code']:
|
|
code = data['code'].strip().lower().replace(' ', '_')
|
|
existing = MilestoneType.query.filter(MilestoneType.code == code, MilestoneType.id != id).first()
|
|
if existing:
|
|
return jsonify({'error': 'Conflict', 'message': f'Ya existe otra tipificación con el código {code}.'}), 409
|
|
mt.code = code
|
|
if 'color' in data:
|
|
mt.color = data['color'].strip()
|
|
if 'description' in data:
|
|
mt.description = data['description'].strip()
|
|
if 'sort_order' in data:
|
|
mt.sort_order = int(data['sort_order'])
|
|
if 'is_active' in data:
|
|
mt.is_active = data['is_active'] in [True, 'true', '1', 'on']
|
|
|
|
db.session.commit()
|
|
return jsonify({
|
|
'status': 'success',
|
|
'message': 'Tipificación actualizada correctamente.',
|
|
'milestone_type': mt.to_dict()
|
|
}), 200
|
|
|
|
@api_admin_bp.route('/milestone-types/<int:id>/toggle', methods=['POST'])
|
|
@jwt_required
|
|
def toggle_milestone_type(id):
|
|
mt = MilestoneType.query.get_or_404(id)
|
|
mt.is_active = not mt.is_active
|
|
db.session.commit()
|
|
return jsonify({
|
|
'status': 'success',
|
|
'id': mt.id,
|
|
'is_active': mt.is_active,
|
|
'message': f'Tipificación {"activada" if mt.is_active else "desactivada"} correctamente.'
|
|
}), 200
|
|
|
|
@api_admin_bp.route('/milestone-types/<int:id>', methods=['DELETE'])
|
|
@jwt_required
|
|
def delete_milestone_type(id):
|
|
mt = MilestoneType.query.get_or_404(id)
|
|
milestone_count = AcademicMilestone.query.filter_by(milestone_type_id=id).count()
|
|
if milestone_count > 0:
|
|
mt.is_active = False
|
|
db.session.commit()
|
|
return jsonify({
|
|
'status': 'success',
|
|
'message': f'La tipificación tiene {milestone_count} hitos registrados. Se ha desactivado en su lugar.'
|
|
}), 200
|
|
|
|
db.session.delete(mt)
|
|
db.session.commit()
|
|
return jsonify({
|
|
'status': 'success',
|
|
'message': 'Tipificación eliminada permanentemente.'
|
|
}), 200
|
|
|
|
# ---------------------------------------------------------
|
|
# GOOGLE SHEETS CONFIGURATION & ACADEMIC DATA RESET TOOL
|
|
# ---------------------------------------------------------
|
|
|
|
@api_admin_bp.route('/sheets-config', methods=['GET'])
|
|
@jwt_required
|
|
def get_sheets_config():
|
|
"""Obtiene la configuración actual del Google Sheet para sincronización."""
|
|
from app.models.setting import SystemSetting
|
|
from app.services.sheets_importer import BASE_CSV_URL, normalize_sheets_url
|
|
import os
|
|
|
|
db_url = SystemSetting.get_value('google_sheets_url')
|
|
env_url = os.getenv('GOOGLE_SHEETS_URL')
|
|
active_url = db_url or env_url or BASE_CSV_URL
|
|
source = 'database' if db_url else ('environment' if env_url else 'default')
|
|
|
|
return jsonify({
|
|
'status': 'success',
|
|
'config': {
|
|
'active_url': active_url,
|
|
'source': source,
|
|
'db_url': db_url,
|
|
'default_url': BASE_CSV_URL,
|
|
'normalized_url': normalize_sheets_url(active_url)
|
|
}
|
|
}), 200
|
|
|
|
@api_admin_bp.route('/sheets-config', methods=['POST'])
|
|
@jwt_required
|
|
def update_sheets_config():
|
|
"""Actualiza la URL del Google Sheet en SystemSetting."""
|
|
from app.models.setting import SystemSetting
|
|
from app.services.sheets_importer import normalize_sheets_url
|
|
|
|
data = request.get_json(silent=True) or request.form.to_dict() or {}
|
|
url = (data.get('url') or data.get('google_sheets_url') or '').strip()
|
|
|
|
if not url:
|
|
return jsonify({
|
|
'error': 'ValidationError',
|
|
'message': 'La URL de Google Sheets no puede estar vacía.'
|
|
}), 400
|
|
|
|
normalized = normalize_sheets_url(url)
|
|
SystemSetting.set_value('google_sheets_url', url, 'URL del Google Sheet académico configurada desde panel')
|
|
|
|
return jsonify({
|
|
'status': 'success',
|
|
'message': 'Configuración de Google Sheets guardada correctamente.',
|
|
'config': {
|
|
'active_url': url,
|
|
'normalized_url': normalized,
|
|
'source': 'database'
|
|
}
|
|
}), 200
|
|
|
|
@api_admin_bp.route('/reset-academic-data', methods=['POST'])
|
|
@jwt_required
|
|
def reset_academic_data_api():
|
|
"""
|
|
Herramienta de limpieza/purga de datos académicos para pruebas del importador.
|
|
Permite eliminar reservas, comisiones, asignaturas, carreras y aulas.
|
|
"""
|
|
from scripts.reset_academic_data import execute_academic_reset
|
|
data = request.get_json(silent=True) or request.form.to_dict() or {}
|
|
|
|
reset_all = data.get('all', False) in [True, 'true', '1', 'on']
|
|
reset_reservations = reset_all or data.get('reservations', True) in [True, 'true', '1', 'on']
|
|
reset_commissions = reset_all or data.get('commissions', False) in [True, 'true', '1', 'on']
|
|
reset_subjects = reset_all or data.get('subjects', False) in [True, 'true', '1', 'on']
|
|
reset_careers = reset_all or data.get('careers', False) in [True, 'true', '1', 'on']
|
|
reset_classrooms = reset_all or data.get('classrooms', False) in [True, 'true', '1', 'on']
|
|
|
|
if reset_all:
|
|
reset_reservations = True
|
|
reset_commissions = True
|
|
reset_subjects = True
|
|
reset_careers = True
|
|
reset_classrooms = True
|
|
|
|
result = execute_academic_reset(
|
|
reset_reservations=reset_reservations,
|
|
reset_commissions=reset_commissions,
|
|
reset_subjects=reset_subjects,
|
|
reset_careers=reset_careers,
|
|
reset_classrooms=reset_classrooms
|
|
)
|
|
|
|
if result.get('success'):
|
|
try:
|
|
from app.models.audit_log import AuditLog
|
|
audit = AuditLog(
|
|
user_id=getattr(request, 'current_user_id', None),
|
|
action_type='ACADEMIC_DATA_PURGE',
|
|
details=f"Purga académica ejecutada: {result.get('deleted')}",
|
|
ip_address=request.remote_addr
|
|
)
|
|
db.session.add(audit)
|
|
db.session.commit()
|
|
except Exception:
|
|
pass
|
|
|
|
return jsonify({
|
|
'status': 'success',
|
|
'message': result.get('message', 'Datos reiniciados con éxito.'),
|
|
'deleted': result.get('deleted', {})
|
|
}), 200
|
|
else:
|
|
return jsonify({
|
|
'status': 'error',
|
|
'message': result.get('error', 'Error durante la purga de datos.')
|
|
}), 500
|
|
|
|
# ---------------------------------------------------------
|
|
# FASE 4: MODO IMPERSONACIÓN & DRAG-AND-DROP (ESTADIO 4.1 & 4.2)
|
|
# ---------------------------------------------------------
|
|
|
|
@api_admin_bp.route('/admin/impersonate', methods=['POST'])
|
|
@jwt_required
|
|
def impersonate_user():
|
|
"""Iniciar sesión bajo la identidad de otro usuario (Superadmin / Bedelía)."""
|
|
admin_user = getattr(g, 'real_admin_user', g.jwt_user)
|
|
if not admin_user.is_admin():
|
|
return jsonify({'error': 'Forbidden', 'message': 'Solo administradores pueden impersonar usuarios.'}), 403
|
|
|
|
data = request.get_json(silent=True) or {}
|
|
target_id = data.get('target_user_id') or data.get('user_id')
|
|
target_email = data.get('target_email') or data.get('email')
|
|
|
|
target = None
|
|
if target_id:
|
|
target = db.session.get(User, int(target_id))
|
|
elif target_email:
|
|
target = User.query.filter_by(email=target_email.strip()).first()
|
|
|
|
if not target or not target.is_active:
|
|
return jsonify({'error': 'NotFound', 'message': 'Usuario objetivo no encontrado o inactivo.'}), 404
|
|
|
|
audit = AuditLog(
|
|
user_id=admin_user.id,
|
|
user_email=admin_user.email,
|
|
action='IMPERSONATE_START',
|
|
module='auth',
|
|
entity_id=target.id,
|
|
details=f"Administrador {admin_user.email} inició sesión temporal como {target.email} ({target.role})"
|
|
)
|
|
db.session.add(audit)
|
|
db.session.commit()
|
|
|
|
return jsonify({
|
|
'status': 'success',
|
|
'message': f"Impersonando exitosamente a {target.name or target.email}",
|
|
'target_user': target.to_dict(),
|
|
'real_admin_id': admin_user.id
|
|
}), 200
|
|
|
|
@api_admin_bp.route('/admin/stop-impersonating', methods=['POST'])
|
|
@jwt_required
|
|
def stop_impersonating():
|
|
"""Finalizar sesión de impersonación y restaurar cuenta de administrador."""
|
|
admin_user = getattr(g, 'real_admin_user', g.jwt_user)
|
|
current_target = g.jwt_user
|
|
|
|
audit = AuditLog(
|
|
user_id=admin_user.id,
|
|
user_email=admin_user.email,
|
|
action='IMPERSONATE_END',
|
|
module='auth',
|
|
entity_id=current_target.id if current_target else None,
|
|
details=f"Administrador {admin_user.email} finalizó la sesión de impersonación."
|
|
)
|
|
db.session.add(audit)
|
|
db.session.commit()
|
|
|
|
return jsonify({
|
|
'status': 'success',
|
|
'message': 'Sesión de impersonación finalizada.'
|
|
}), 200
|
|
|
|
@api_admin_bp.route('/reservations/drag-update', methods=['POST'])
|
|
@jwt_required
|
|
def drag_update_reservation():
|
|
"""Actualización o creación rápida de reserva mediante arrastre en la grilla semanal."""
|
|
data = request.get_json(silent=True) or {}
|
|
reservation_id = data.get('reservation_id')
|
|
commission_id = data.get('commission_id')
|
|
classroom_id = data.get('classroom_id')
|
|
start_time_str = data.get('start_time')
|
|
end_time_str = data.get('end_time')
|
|
|
|
if not classroom_id or not start_time_str or not end_time_str:
|
|
return jsonify({'error': 'ValidationError', 'message': 'Faltan parámetros obligatorios (aula, inicio, fin).'}), 400
|
|
|
|
try:
|
|
if isinstance(start_time_str, str):
|
|
start_dt = datetime.fromisoformat(start_time_str.replace('Z', '+00:00'))
|
|
else:
|
|
start_dt = start_time_str
|
|
|
|
if isinstance(end_time_str, str):
|
|
end_dt = datetime.fromisoformat(end_time_str.replace('Z', '+00:00'))
|
|
else:
|
|
end_dt = end_time_str
|
|
except Exception as e:
|
|
return jsonify({'error': 'ValidationError', 'message': f'Formato de fecha inválido: {str(e)}'}), 400
|
|
|
|
classroom = db.session.get(Classroom, classroom_id)
|
|
if not classroom:
|
|
return jsonify({'error': 'NotFound', 'message': 'Aula no encontrada.'}), 404
|
|
|
|
# Pre-validación de conflictos (físico, capacidad, docente)
|
|
from app.services.reservation_service import ReservationService
|
|
effective_comm_id = commission_id
|
|
if not effective_comm_id and reservation_id:
|
|
existing_res = db.session.get(Reservation, reservation_id)
|
|
if existing_res:
|
|
effective_comm_id = existing_res.commission_id
|
|
|
|
res_service = ReservationService()
|
|
matrix = res_service.check_full_conflicts_matrix(
|
|
classroom_id=classroom_id,
|
|
start_time=start_dt,
|
|
end_time=end_dt,
|
|
commission_id=effective_comm_id,
|
|
exclude_id=reservation_id
|
|
)
|
|
|
|
if matrix.get('is_blocked'):
|
|
return jsonify({
|
|
'status': 'conflict',
|
|
'error': 'ConflictDetected',
|
|
'message': ' '.join(matrix.get('block_reasons', ['Conflicto detectado en la asignación.'])),
|
|
'conflicts': matrix.get('block_reasons', [])
|
|
}), 409
|
|
|
|
admin_user = getattr(g, 'real_admin_user', g.jwt_user)
|
|
acting_user = g.jwt_user
|
|
|
|
if reservation_id:
|
|
res = db.session.get(Reservation, reservation_id)
|
|
if not res:
|
|
return jsonify({'error': 'NotFound', 'message': 'Reserva no encontrada.'}), 404
|
|
|
|
res.classroom_id = classroom_id
|
|
res.start_time = start_dt
|
|
res.end_time = end_dt
|
|
action = 'UPDATE_RESERVATION_DRAG'
|
|
msg = f"Reserva #{res.id} reprogramada al aula {classroom.code} ({start_dt.strftime('%H:%M')} a {end_dt.strftime('%H:%M')})"
|
|
else:
|
|
comm = db.session.get(Commission, commission_id) if commission_id else None
|
|
res = Reservation(
|
|
classroom_id=classroom_id,
|
|
commission_id=commission_id,
|
|
user_id=acting_user.id,
|
|
start_time=start_dt,
|
|
end_time=end_dt,
|
|
purpose=f"Cursada regular - {comm.subject.name if comm and comm.subject else 'Comisión'}",
|
|
expected_attendees=comm.max_students if comm else 30,
|
|
status=ReservationStatus.CONFIRMED.value
|
|
)
|
|
db.session.add(res)
|
|
action = 'CREATE_RESERVATION_DRAG'
|
|
msg = f"Comisión {comm.code if comm else ''} asignada al aula {classroom.code}"
|
|
|
|
details_str = msg
|
|
if getattr(g, 'is_impersonating', False):
|
|
details_str += f" [Ejecutado bajo impersonación por Admin #{admin_user.id}]"
|
|
|
|
audit = AuditLog(
|
|
user_id=admin_user.id,
|
|
user_email=admin_user.email,
|
|
action=action,
|
|
module='reservations',
|
|
entity_id=res.id,
|
|
details=details_str
|
|
)
|
|
db.session.add(audit)
|
|
db.session.commit()
|
|
return jsonify({
|
|
'status': 'success',
|
|
'message': msg,
|
|
'reservation': res.to_dict()
|
|
}), 200
|
|
|
|
|
|
# ==============================================================================
|
|
# CONFIGURACIÓN GLOBAL DEL SISTEMA (SMTP, AUTH PROVIDERS, GOOGLE OAUTH, MOODLE)
|
|
# ==============================================================================
|
|
|
|
@api_admin_bp.route('/settings/all', methods=['GET'])
|
|
@jwt_required
|
|
def get_all_settings():
|
|
"""Retorna todas las configuraciones agrupadas por módulo, enmascarando contraseñas."""
|
|
from app.models.setting import SystemSetting
|
|
|
|
smtp_config = {
|
|
'host': SystemSetting.get_value('smtp_host', 'smtp.gmail.com'),
|
|
'port': int(SystemSetting.get_value('smtp_port', 587)),
|
|
'user': SystemSetting.get_value('smtp_user', ''),
|
|
'password': SystemSetting.get_masked_value('smtp_password', ''),
|
|
'security': SystemSetting.get_value('smtp_security', 'STARTTLS'),
|
|
'sender_email': SystemSetting.get_value('smtp_sender_email', 'notificaciones@unicaba.edu.ar'),
|
|
'sender_name': SystemSetting.get_value('smtp_sender_name', 'Edu-Space UniCABA'),
|
|
'enabled': SystemSetting.get_value('smtp_enabled', 'true') in ['true', 'True', '1', True]
|
|
}
|
|
|
|
auth_providers = {
|
|
'local_enabled': SystemSetting.get_value('auth_local_enabled', 'true') in ['true', 'True', '1', True],
|
|
'google_enabled': SystemSetting.get_value('auth_google_enabled', 'false') in ['true', 'True', '1', True],
|
|
'moodle_enabled': SystemSetting.get_value('auth_moodle_enabled', 'false') in ['true', 'True', '1', True]
|
|
}
|
|
|
|
google_oauth = {
|
|
'client_id': SystemSetting.get_value('google_client_id', ''),
|
|
'client_secret': SystemSetting.get_masked_value('google_client_secret', ''),
|
|
'callback_url': SystemSetting.get_value('google_callback_url', '/auth/google/callback'),
|
|
'allowed_domains': SystemSetting.get_value('google_allowed_domains', 'unicaba.edu.ar,lasalle.edu.ar')
|
|
}
|
|
|
|
moodle_config = {
|
|
'server_url': SystemSetting.get_value('moodle_server_url', 'http://10.0.0.207/moodle'),
|
|
'ws_token': SystemSetting.get_masked_value('moodle_ws_token', ''),
|
|
'timeout': int(SystemSetting.get_value('moodle_timeout', 10)),
|
|
'auto_sync_enabled': SystemSetting.get_value('moodle_auto_sync_enabled', 'true') in ['true', 'True', '1', True],
|
|
'sync_interval_minutes': int(SystemSetting.get_value('moodle_sync_interval_minutes', 15))
|
|
}
|
|
|
|
return jsonify({
|
|
'status': 'success',
|
|
'settings': {
|
|
'smtp': smtp_config,
|
|
'auth_providers': auth_providers,
|
|
'google_oauth': google_oauth,
|
|
'moodle': moodle_config
|
|
}
|
|
}), 200
|
|
|
|
|
|
@api_admin_bp.route('/settings/smtp', methods=['POST'])
|
|
@jwt_required
|
|
def update_smtp_settings():
|
|
"""Actualiza los parámetros del servidor de correo SMTP en la base de datos."""
|
|
from app.models.setting import SystemSetting
|
|
from app.models.audit_log import AuditLog
|
|
|
|
acting_user = getattr(g, 'jwt_user', None)
|
|
if not (acting_user and (acting_user.is_admin() or getattr(acting_user, 'role', '').upper() == 'ADMIN')):
|
|
return jsonify({'error': 'Forbidden', 'message': 'Solo administradores pueden modificar la configuración SMTP.'}), 403
|
|
|
|
data = request.get_json(silent=True) or request.form.to_dict() or {}
|
|
|
|
host = str(data.get('host', '')).strip()
|
|
port = str(data.get('port', 587)).strip()
|
|
user = str(data.get('user', '')).strip()
|
|
password = str(data.get('password', '')).strip()
|
|
security = str(data.get('security', 'STARTTLS')).strip().upper()
|
|
sender_email = str(data.get('sender_email', '')).strip()
|
|
sender_name = str(data.get('sender_name', 'Edu-Space UniCABA')).strip()
|
|
enabled = 'true' if data.get('enabled') in [True, 'true', '1', 'on'] else 'false'
|
|
|
|
SystemSetting.set_value('smtp_host', host, 'Host del servidor SMTP', category='smtp')
|
|
SystemSetting.set_value('smtp_port', port, 'Puerto del servidor SMTP', category='smtp')
|
|
SystemSetting.set_value('smtp_user', user, 'Usuario o email de autenticación SMTP', category='smtp')
|
|
if password and password != '••••••••••••':
|
|
SystemSetting.set_encrypted_value('smtp_password', password, 'Contraseña de autenticación SMTP cifrada', category='smtp')
|
|
SystemSetting.set_value('smtp_security', security, 'Protocolo de seguridad SMTP (NONE, SSL, STARTTLS)', category='smtp')
|
|
SystemSetting.set_value('smtp_sender_email', sender_email, 'Email remitente oficial', category='smtp')
|
|
SystemSetting.set_value('smtp_sender_name', sender_name, 'Nombre remitente oficial', category='smtp')
|
|
SystemSetting.set_value('smtp_enabled', enabled, 'Habilitación del servicio SMTP', category='smtp')
|
|
|
|
try:
|
|
audit = AuditLog(
|
|
user_id=acting_user.id,
|
|
user_email=acting_user.email,
|
|
action='UPDATE_SMTP_SETTINGS',
|
|
module='settings',
|
|
details=f"Configuración SMTP actualizada (Host: {host}:{port}, Remitente: {sender_email})"
|
|
)
|
|
db.session.add(audit)
|
|
db.session.commit()
|
|
except Exception:
|
|
pass
|
|
|
|
return jsonify({
|
|
'status': 'success',
|
|
'message': 'Configuración de servidor SMTP guardada exitosamente.'
|
|
}), 200
|
|
|
|
|
|
@api_admin_bp.route('/settings/smtp/test', methods=['POST'])
|
|
@jwt_required
|
|
def test_smtp_connection():
|
|
"""Prueba en tiempo real la conexión al servidor SMTP y opcionalmente envía un email de prueba."""
|
|
import smtplib
|
|
from email.mime.text import MIMEText
|
|
from email.mime.multipart import MIMEMultipart
|
|
from app.models.setting import SystemSetting
|
|
|
|
data = request.get_json(silent=True) or request.form.to_dict() or {}
|
|
test_recipient = data.get('test_email') or g.jwt_user.email
|
|
|
|
host = data.get('host') or SystemSetting.get_value('smtp_host', 'smtp.gmail.com')
|
|
port = int(data.get('port') or SystemSetting.get_value('smtp_port', 587))
|
|
user = data.get('user') or SystemSetting.get_value('smtp_user', '')
|
|
password = data.get('password')
|
|
if not password or password == '••••••••••••':
|
|
password = SystemSetting.get_decrypted_value('smtp_password', '')
|
|
security = (data.get('security') or SystemSetting.get_value('smtp_security', 'STARTTLS')).upper()
|
|
sender_email = data.get('sender_email') or SystemSetting.get_value('smtp_sender_email', user)
|
|
sender_name = data.get('sender_name') or SystemSetting.get_value('smtp_sender_name', 'Edu-Space UniCABA')
|
|
|
|
if not host or not port:
|
|
return jsonify({'status': 'error', 'message': 'Host y puerto SMTP son requeridos.'}), 400
|
|
|
|
try:
|
|
if security == 'SSL':
|
|
server = smtplib.SMTP_SSL(host, port, timeout=10)
|
|
else:
|
|
server = smtplib.SMTP(host, port, timeout=10)
|
|
if security == 'STARTTLS':
|
|
server.ehlo()
|
|
server.starttls()
|
|
server.ehlo()
|
|
|
|
if user and password:
|
|
server.login(user, password)
|
|
|
|
if test_recipient:
|
|
msg = MIMEMultipart('alternative')
|
|
msg['Subject'] = '✔ Prueba de Conexión SMTP - Edu-Space UniCABA'
|
|
msg['From'] = f"{sender_name} <{sender_email}>"
|
|
msg['To'] = test_recipient
|
|
|
|
html_content = f"""
|
|
<div style="font-family: Arial, sans-serif; max-width: 550px; margin: auto; padding: 20px; border: 1px solid #e2e8f0; border-radius: 8px;">
|
|
<h2 style="color: #B43E8E; margin-bottom: 10px;">Edu-Space UniCABA</h2>
|
|
<h3 style="color: #1e293b; margin-top: 0;">Prueba de Conexión SMTP Exitosa</h3>
|
|
<p style="color: #475569;">Este es un mensaje de prueba para confirmar que los parámetros del servidor de correo han sido configurados correctamente.</p>
|
|
<div style="background-color: #f8fafc; padding: 12px; border-radius: 6px; font-size: 13px; color: #334155;">
|
|
<strong>Host:</strong> {host}:{port}<br>
|
|
<strong>Seguridad:</strong> {security}<br>
|
|
<strong>Usuario:</strong> {user or '(Sin autenticación)'}<br>
|
|
<strong>Remitente:</strong> {sender_email}
|
|
</div>
|
|
<p style="font-size: 11px; color: #94a3b8; margin-top: 20px;">Enviado desde el Panel de Administración de UniCABA.</p>
|
|
</div>
|
|
"""
|
|
msg.attach(MIMEText(html_content, 'html'))
|
|
server.sendmail(sender_email, [test_recipient], msg.as_string())
|
|
|
|
server.quit()
|
|
return jsonify({
|
|
'status': 'success',
|
|
'message': f'Conexión SMTP exitosa. Correo de prueba enviado a {test_recipient}.'
|
|
}), 200
|
|
|
|
except Exception as e:
|
|
return jsonify({
|
|
'status': 'error',
|
|
'message': f'Fallo en la prueba de conexión SMTP: {str(e)}'
|
|
}), 400
|
|
|
|
|
|
@api_admin_bp.route('/settings/auth-providers', methods=['POST'])
|
|
@jwt_required
|
|
def update_auth_providers():
|
|
"""Habilita o deshabilita los proveedores de autenticación del sistema."""
|
|
from app.models.setting import SystemSetting
|
|
from app.models.audit_log import AuditLog
|
|
|
|
acting_user = getattr(g, 'jwt_user', None)
|
|
if not (acting_user and (acting_user.is_admin() or getattr(acting_user, 'role', '').upper() == 'ADMIN')):
|
|
return jsonify({'error': 'Forbidden', 'message': 'Solo administradores pueden modificar los métodos de autenticación.'}), 403
|
|
|
|
data = request.get_json(silent=True) or request.form.to_dict() or {}
|
|
local_val = 'true' if data.get('local_enabled', True) in [True, 'true', '1', 'on'] else 'false'
|
|
google_val = 'true' if data.get('google_enabled', False) in [True, 'true', '1', 'on'] else 'false'
|
|
moodle_val = 'true' if data.get('moodle_enabled', False) in [True, 'true', '1', 'on'] else 'false'
|
|
|
|
SystemSetting.set_value('auth_local_enabled', local_val, 'Habilitar login nativo con usuario/contraseña', category='sso')
|
|
SystemSetting.set_value('auth_google_enabled', google_val, 'Habilitar login SSO con Google Workspace', category='sso')
|
|
SystemSetting.set_value('auth_moodle_enabled', moodle_val, 'Habilitar login delegado con Moodle', category='sso')
|
|
|
|
try:
|
|
audit = AuditLog(
|
|
user_id=acting_user.id,
|
|
user_email=acting_user.email,
|
|
action='UPDATE_AUTH_PROVIDERS',
|
|
module='settings',
|
|
details=f"Métodos de login actualizados: Local={local_val}, Google={google_val}, Moodle={moodle_val}"
|
|
)
|
|
db.session.add(audit)
|
|
db.session.commit()
|
|
except Exception:
|
|
pass
|
|
|
|
return jsonify({
|
|
'status': 'success',
|
|
'message': 'Métodos de autenticación actualizados correctamente.'
|
|
}), 200
|
|
|
|
|
|
@api_admin_bp.route('/settings/google-oauth', methods=['POST'])
|
|
@jwt_required
|
|
def update_google_oauth_settings():
|
|
"""Actualiza las credenciales de integración de Google OAuth2."""
|
|
from app.models.setting import SystemSetting
|
|
from app.models.audit_log import AuditLog
|
|
|
|
acting_user = getattr(g, 'jwt_user', None)
|
|
if not (acting_user and (acting_user.is_admin() or getattr(acting_user, 'role', '').upper() == 'ADMIN')):
|
|
return jsonify({'error': 'Forbidden', 'message': 'Acceso no autorizado.'}), 403
|
|
|
|
data = request.get_json(silent=True) or request.form.to_dict() or {}
|
|
client_id = str(data.get('client_id', '')).strip()
|
|
client_secret = str(data.get('client_secret', '')).strip()
|
|
allowed_domains = str(data.get('allowed_domains', 'unicaba.edu.ar')).strip()
|
|
callback_url = str(data.get('callback_url', '/auth/google/callback')).strip()
|
|
|
|
SystemSetting.set_value('google_client_id', client_id, 'Client ID de Google OAuth2', category='sso_google')
|
|
if client_secret and client_secret != '••••••••••••':
|
|
SystemSetting.set_encrypted_value('google_client_secret', client_secret, 'Client Secret de Google OAuth2 cifrado', category='sso_google')
|
|
SystemSetting.set_value('google_allowed_domains', allowed_domains, 'Dominios permitidos separados por coma', category='sso_google')
|
|
SystemSetting.set_value('google_callback_url', callback_url, 'Ruta de callback autorizada de Google OAuth2', category='sso_google')
|
|
|
|
try:
|
|
audit = AuditLog(
|
|
user_id=acting_user.id,
|
|
user_email=acting_user.email,
|
|
action='UPDATE_GOOGLE_OAUTH_SETTINGS',
|
|
module='settings',
|
|
details="Credenciales de Google OAuth2 actualizadas"
|
|
)
|
|
db.session.add(audit)
|
|
db.session.commit()
|
|
except Exception:
|
|
pass
|
|
|
|
return jsonify({
|
|
'status': 'success',
|
|
'message': 'Credenciales de Google OAuth2 guardadas correctamente.'
|
|
}), 200
|
|
|
|
|
|
@api_admin_bp.route('/settings/moodle', methods=['POST'])
|
|
@jwt_required
|
|
def update_moodle_settings():
|
|
"""Actualiza la configuración de integración y Web Services con Moodle 4.1."""
|
|
from app.models.setting import SystemSetting
|
|
from app.models.audit_log import AuditLog
|
|
|
|
acting_user = getattr(g, 'jwt_user', None)
|
|
if not (acting_user and (acting_user.is_admin() or getattr(acting_user, 'role', '').upper() == 'ADMIN')):
|
|
return jsonify({'error': 'Forbidden', 'message': 'Acceso no autorizado.'}), 403
|
|
|
|
data = request.get_json(silent=True) or request.form.to_dict() or {}
|
|
server_url = str(data.get('server_url', 'http://10.0.0.207/moodle')).strip().rstrip('/')
|
|
ws_token = str(data.get('ws_token', '')).strip()
|
|
timeout = str(data.get('timeout', 10)).strip()
|
|
auto_sync = 'true' if data.get('auto_sync_enabled', True) in [True, 'true', '1', 'on'] else 'false'
|
|
sync_interval = str(data.get('sync_interval_minutes', 15)).strip()
|
|
|
|
SystemSetting.set_value('moodle_server_url', server_url, 'URL base del servidor Moodle', category='sso_moodle')
|
|
if ws_token and ws_token != '••••••••••••':
|
|
SystemSetting.set_encrypted_value('moodle_ws_token', ws_token, 'Token de Web Services de Moodle cifrado', category='sso_moodle')
|
|
SystemSetting.set_value('moodle_timeout', timeout, 'Timeout en segundos para llamadas REST a Moodle', category='sso_moodle')
|
|
SystemSetting.set_value('moodle_auto_sync_enabled', auto_sync, 'Habilitación de sincronización periódica automática', category='sso_moodle')
|
|
SystemSetting.set_value('moodle_sync_interval_minutes', sync_interval, 'Intervalo en minutos para sincronización periódica', category='sso_moodle')
|
|
|
|
try:
|
|
audit = AuditLog(
|
|
user_id=acting_user.id,
|
|
user_email=acting_user.email,
|
|
action='UPDATE_MOODLE_SETTINGS',
|
|
module='settings',
|
|
details=f"Parámetros de Moodle actualizados (Servidor: {server_url})"
|
|
)
|
|
db.session.add(audit)
|
|
db.session.commit()
|
|
except Exception:
|
|
pass
|
|
|
|
return jsonify({
|
|
'status': 'success',
|
|
'message': 'Configuración de Moodle 4.1 guardada correctamente.'
|
|
}), 200
|
|
|
|
|
|
@api_admin_bp.route('/settings/moodle/test', methods=['POST'])
|
|
@jwt_required
|
|
def test_moodle_connection():
|
|
"""Prueba la conectividad y validez del token Web Services contra Moodle 4.1."""
|
|
from app.services.moodle_client import moodle_client
|
|
data = request.get_json(silent=True) or request.form.to_dict() or {}
|
|
server_url = data.get('server_url')
|
|
token = data.get('ws_token')
|
|
|
|
result = moodle_client.test_connection(server_url=server_url, token=token)
|
|
if result.get('success'):
|
|
return jsonify(result), 200
|
|
else:
|
|
return jsonify(result), 400
|
|
|
|
|
|
@api_admin_bp.route('/moodle/queue/stats', methods=['GET'])
|
|
@jwt_required
|
|
def get_moodle_queue_stats():
|
|
"""Retorna las estadísticas en tiempo real de la cola de sincronización con Moodle."""
|
|
from app.services.moodle_queue_service import moodle_queue_service
|
|
stats = moodle_queue_service.get_queue_summary()
|
|
return jsonify({
|
|
'status': 'success',
|
|
'data': stats
|
|
}), 200
|
|
|
|
|
|
@api_admin_bp.route('/moodle/queue/tasks', methods=['GET'])
|
|
@jwt_required
|
|
def get_moodle_queue_tasks():
|
|
"""Retorna la lista de tareas en cola con filtros por estado (ej: FAILED para DLQ)."""
|
|
from app.models.sync_task import MoodleSyncTask
|
|
status = request.args.get('status', '').strip().upper()
|
|
page = int(request.args.get('page', 1))
|
|
per_page = int(request.args.get('per_page', 20))
|
|
|
|
query = MoodleSyncTask.query
|
|
if status:
|
|
query = query.filter_by(status=status)
|
|
|
|
total = query.count()
|
|
tasks = query.order_by(MoodleSyncTask.created_at.desc()).offset((page - 1) * per_page).limit(per_page).all()
|
|
|
|
return jsonify({
|
|
'status': 'success',
|
|
'total': total,
|
|
'page': page,
|
|
'per_page': per_page,
|
|
'tasks': [t.to_dict() for t in tasks]
|
|
}), 200
|
|
|
|
|
|
@api_admin_bp.route('/moodle/queue/process-now', methods=['POST'])
|
|
@jwt_required
|
|
def process_moodle_queue_now():
|
|
"""Dispara de forma manual la ejecución inmediata de la cola de sincronización."""
|
|
from app.services.moodle_queue_service import moodle_queue_service
|
|
batch_size = int(request.json.get('batch_size', 50)) if request.is_json and request.json else 50
|
|
results = moodle_queue_service.process_pending_tasks(batch_size=batch_size)
|
|
return jsonify({
|
|
'status': 'success',
|
|
'message': f"Sincronización procesada: {results['succeeded']} exitosas, {results['failed']} a DLQ, {results['retrying']} reintentando.",
|
|
'results': results
|
|
}), 200
|
|
|
|
|
|
@api_admin_bp.route('/moodle/queue/tasks/<int:task_id>/retry', methods=['POST'])
|
|
@jwt_required
|
|
def retry_moodle_queue_task(task_id):
|
|
"""Reintenta manualmente una tarea específica desde la Dead Letter Queue."""
|
|
from app.services.moodle_queue_service import moodle_queue_service
|
|
success = moodle_queue_service.retry_task(task_id)
|
|
if success:
|
|
return jsonify({
|
|
'status': 'success',
|
|
'message': f'Tarea #{task_id} reiniciada a estado PENDIENTE para el próximo ciclo.'
|
|
}), 200
|
|
else:
|
|
return jsonify({
|
|
'status': 'error',
|
|
'message': f'No se encontró la tarea #{task_id}.'
|
|
}), 404
|
|
|
|
|
|
@api_admin_bp.route('/moodle/queue/retry-all', methods=['POST'])
|
|
@jwt_required
|
|
def retry_all_failed_moodle_tasks():
|
|
"""Reintenta todas las tareas en Dead Letter Queue (FAILED)."""
|
|
from app.services.moodle_queue_service import moodle_queue_service
|
|
count = moodle_queue_service.retry_all_failed()
|
|
return jsonify({
|
|
'status': 'success',
|
|
'message': f'Se reiniciaron {count} tareas fallidas a estado PENDIENTE.'
|
|
}), 200
|