Files
admin-edu-space/backend/init_db.py
T

515 lines
26 KiB
Python

import sys
# Asegurar codificación UTF-8 en consola de Windows
if hasattr(sys.stdout, 'reconfigure'):
sys.stdout.reconfigure(encoding='utf-8')
if hasattr(sys.stderr, 'reconfigure'):
sys.stderr.reconfigure(encoding='utf-8')
from app import create_app, db
from app.models.user import User
from app.models.role import Role, Permission, SYSTEM_MODULES
from sqlalchemy import text
def init_database():
app = create_app()
with app.app_context():
try:
print("[*] Conectando a la base de datos y sincronizando esquema...")
# 1. Crear tablas que no existan
db.create_all()
# 2. Migraciones incrementales seguras para bases de datos existentes
if db.engine.url.drivername != 'sqlite':
# Columnas extendidas de users
db.session.execute(text("""
ALTER TABLE users ADD COLUMN IF NOT EXISTS preferred_language VARCHAR(10) DEFAULT NULL;
ALTER TABLE users ADD COLUMN IF NOT EXISTS theme_preference VARCHAR(10) DEFAULT 'auto' NOT NULL;
ALTER TABLE users ADD COLUMN IF NOT EXISTS role_id INTEGER REFERENCES roles(id) ON DELETE SET NULL;
ALTER TABLE users ADD COLUMN IF NOT EXISTS first_name VARCHAR(100);
ALTER TABLE users ADD COLUMN IF NOT EXISTS last_name VARCHAR(100);
ALTER TABLE users ADD COLUMN IF NOT EXISTS phone VARCHAR(50);
ALTER TABLE users ADD COLUMN IF NOT EXISTS personal_email VARCHAR(255);
ALTER TABLE users ADD COLUMN IF NOT EXISTS address VARCHAR(255);
ALTER TABLE users ADD COLUMN IF NOT EXISTS document_type VARCHAR(20) DEFAULT 'DNI';
ALTER TABLE users ADD COLUMN IF NOT EXISTS document_number VARCHAR(30);
CREATE INDEX IF NOT EXISTS ix_users_document_number ON users (document_number);
"""))
# Columnas extendidas de subjects
db.session.execute(text("""
ALTER TABLE subjects ADD COLUMN IF NOT EXISTS career_id INTEGER REFERENCES careers(id) ON DELETE SET NULL;
ALTER TABLE subjects ALTER COLUMN code TYPE VARCHAR(50);
ALTER TABLE subjects ADD COLUMN IF NOT EXISTS is_short_course BOOLEAN DEFAULT FALSE;
"""))
# Infraestructura: buildings y classrooms
db.session.execute(text("""
CREATE TABLE IF NOT EXISTS buildings (
id SERIAL PRIMARY KEY,
name VARCHAR(100) UNIQUE NOT NULL,
code VARCHAR(50) UNIQUE,
address VARCHAR(255),
floors INTEGER DEFAULT 1 NOT NULL,
description TEXT,
is_active BOOLEAN DEFAULT TRUE NOT NULL,
created_at TIMESTAMP WITHOUT TIME ZONE DEFAULT NOW(),
updated_at TIMESTAMP WITHOUT TIME ZONE DEFAULT NOW()
);
ALTER TABLE classrooms ADD COLUMN IF NOT EXISTS building_id INTEGER REFERENCES buildings(id) ON DELETE SET NULL;
"""))
# Columnas extendidas de commissions
db.session.execute(text("""
ALTER TABLE commissions ADD COLUMN IF NOT EXISTS shift VARCHAR(30);
ALTER TABLE commissions ADD COLUMN IF NOT EXISTS virtual_link VARCHAR(500);
ALTER TABLE commissions ADD COLUMN IF NOT EXISTS grades_closed BOOLEAN DEFAULT FALSE;
ALTER TABLE commissions ADD COLUMN IF NOT EXISTS grades_closed_at TIMESTAMP WITHOUT TIME ZONE;
ALTER TABLE commissions ADD COLUMN IF NOT EXISTS grades_closed_by INTEGER REFERENCES users(id) ON DELETE SET NULL;
ALTER TABLE commissions ADD COLUMN IF NOT EXISTS acta_number VARCHAR(50);
"""))
# Columnas extendidas de reservations
db.session.execute(text("""
ALTER TABLE reservations ADD COLUMN IF NOT EXISTS shift VARCHAR(30);
ALTER TABLE reservations ADD COLUMN IF NOT EXISTS virtual_link VARCHAR(500);
"""))
# Tablas de soporte: audit_logs
db.session.execute(text("""
CREATE TABLE IF NOT EXISTS audit_logs (
id SERIAL PRIMARY KEY,
user_id INTEGER REFERENCES users(id) ON DELETE SET NULL,
user_email VARCHAR(255),
action VARCHAR(50) NOT NULL,
module VARCHAR(50) NOT NULL,
entity_id INTEGER,
details TEXT,
ip_address VARCHAR(45),
created_at TIMESTAMP WITHOUT TIME ZONE DEFAULT CURRENT_TIMESTAMP NOT NULL
);
CREATE INDEX IF NOT EXISTS ix_audit_logs_action ON audit_logs (action);
CREATE INDEX IF NOT EXISTS ix_audit_logs_module ON audit_logs (module);
CREATE INDEX IF NOT EXISTS ix_audit_logs_created_at ON audit_logs (created_at);
"""))
# Tablas de soporte: student_enrollments
db.session.execute(text("""
CREATE TABLE IF NOT EXISTS student_enrollments (
id SERIAL PRIMARY KEY,
student_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
commission_id INTEGER NOT NULL REFERENCES commissions(id) ON DELETE CASCADE,
status VARCHAR(20) NOT NULL DEFAULT 'activo',
enrolled_at TIMESTAMP WITHOUT TIME ZONE DEFAULT NOW() NOT NULL,
updated_at TIMESTAMP WITHOUT TIME ZONE DEFAULT NOW(),
notes TEXT,
CONSTRAINT uq_student_commission UNIQUE (student_id, commission_id)
);
CREATE INDEX IF NOT EXISTS ix_student_enrollments_student_id ON student_enrollments (student_id);
CREATE INDEX IF NOT EXISTS ix_student_enrollments_commission_id ON student_enrollments (commission_id);
"""))
# Tablas de soporte: commission_teachers
db.session.execute(text("""
CREATE TABLE IF NOT EXISTS commission_teachers (
id SERIAL PRIMARY KEY,
commission_id INTEGER NOT NULL REFERENCES commissions(id) ON DELETE CASCADE,
user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
role VARCHAR(50) DEFAULT 'Titular' NOT NULL,
is_primary BOOLEAN DEFAULT FALSE NOT NULL,
created_at TIMESTAMP WITHOUT TIME ZONE DEFAULT NOW(),
CONSTRAINT uq_commission_teacher UNIQUE (commission_id, user_id)
);
CREATE INDEX IF NOT EXISTS ix_commission_teachers_commission_id ON commission_teachers (commission_id);
CREATE INDEX IF NOT EXISTS ix_commission_teachers_user_id ON commission_teachers (user_id);
"""))
# Tablas de soporte: system_settings
db.session.execute(text("""
CREATE TABLE IF NOT EXISTS system_settings (
id SERIAL PRIMARY KEY,
key VARCHAR(100) UNIQUE NOT NULL,
value TEXT,
description VARCHAR(255),
category VARCHAR(50) DEFAULT 'system',
is_encrypted BOOLEAN DEFAULT FALSE,
updated_at TIMESTAMP WITHOUT TIME ZONE DEFAULT NOW()
);
CREATE INDEX IF NOT EXISTS ix_system_settings_key ON system_settings (key);
"""))
# Tablas de soporte: moodle_sync_tasks
db.session.execute(text("""
CREATE TABLE IF NOT EXISTS moodle_sync_tasks (
id SERIAL PRIMARY KEY,
action VARCHAR(50) NOT NULL,
entity_type VARCHAR(50) NOT NULL,
entity_id VARCHAR(100),
payload TEXT NOT NULL,
status VARCHAR(20) DEFAULT 'PENDING',
attempts INTEGER DEFAULT 0,
max_attempts INTEGER DEFAULT 5,
error_message TEXT,
next_retry_at TIMESTAMP WITHOUT TIME ZONE DEFAULT NOW(),
created_at TIMESTAMP WITHOUT TIME ZONE DEFAULT NOW(),
updated_at TIMESTAMP WITHOUT TIME ZONE DEFAULT NOW()
);
CREATE INDEX IF NOT EXISTS ix_moodle_sync_tasks_action ON moodle_sync_tasks (action);
CREATE INDEX IF NOT EXISTS ix_moodle_sync_tasks_status ON moodle_sync_tasks (status);
CREATE INDEX IF NOT EXISTS ix_moodle_sync_tasks_next_retry_at ON moodle_sync_tasks (next_retry_at);
"""))
db.session.commit()
else:
try:
cursor = db.session.connection().connection.cursor()
cursor.execute("PRAGMA table_info(users)")
existing_cols = [row[1] for row in cursor.fetchall()]
for col_name, col_def in [
('preferred_language', 'VARCHAR(10)'),
('theme_preference', "VARCHAR(10) DEFAULT 'auto'"),
('role_id', 'INTEGER'),
('first_name', 'VARCHAR(100)'),
('last_name', 'VARCHAR(100)'),
('phone', 'VARCHAR(50)'),
('personal_email', 'VARCHAR(255)'),
('address', 'VARCHAR(255)'),
('document_type', "VARCHAR(20) DEFAULT 'DNI'"),
('document_number', 'VARCHAR(30)')
]:
if col_name not in existing_cols:
cursor.execute(f"ALTER TABLE users ADD COLUMN {col_name} {col_def}")
db.session.commit()
except Exception:
pass
print("[OK] Esquema y tablas actualizadas con éxito.")
# 3. Crear o sincronizar roles predeterminados y matriz de permisos
print("[*] Configurando roles institucionales y permisos RBAC...")
default_roles = [
{
'name': 'Admin',
'description': 'Control total institucional, infraestructura, gestion academica y accesos RBAC',
'is_system': True,
'permissions': {
'classrooms': 'read_write',
'buildings': 'read_write',
'reservations': 'read_write',
'schedule': 'read_write',
'academic': 'read_write',
'import': 'read_write',
'optimizer': 'read_write',
'users': 'read_write',
'milestone_types': 'read_write',
'metrics': 'read_write',
'enrollment': 'read_write'
}
},
{
'name': 'Docente',
'description': 'Gestion de reservas de cursada, consulta de cronograma y materias',
'is_system': False,
'permissions': {
'classrooms': 'read',
'buildings': 'read',
'reservations': 'read_write',
'schedule': 'read_write',
'academic': 'read',
'import': 'none',
'optimizer': 'read',
'users': 'none',
'milestone_types': 'read',
'metrics': 'none',
'enrollment': 'read'
}
},
{
'name': 'Bedelia',
'description': 'Gestion operativa de aulas, sedes, reservas, comisiones, importacion y cartelera',
'is_system': False,
'permissions': {
'classrooms': 'read_write',
'buildings': 'read_write',
'reservations': 'read_write',
'schedule': 'read_write',
'academic': 'read_write',
'import': 'read_write',
'optimizer': 'none',
'users': 'none',
'milestone_types': 'read_write',
'metrics': 'read_write',
'enrollment': 'read_write'
}
},
{
'name': 'Alumno',
'description': 'Acceso de solo lectura al cronograma general, asignaturas y aulas',
'is_system': False,
'permissions': {
'classrooms': 'read',
'buildings': 'read',
'reservations': 'read',
'schedule': 'read',
'academic': 'read',
'import': 'none',
'optimizer': 'none',
'users': 'none',
'milestone_types': 'read',
'metrics': 'none',
'enrollment': 'read'
}
}
]
module_ids = [m['id'] for m in SYSTEM_MODULES]
admin_role = None
for r_data in default_roles:
role = Role.query.filter_by(name=r_data['name']).first()
if not role:
role = Role(
name=r_data['name'],
description=r_data['description'],
is_system=r_data['is_system']
)
db.session.add(role)
db.session.flush()
print(f" [+] Rol '{role.name}' creado.")
else:
role.description = r_data['description']
role.is_system = r_data['is_system']
if role.name == 'Admin':
admin_role = role
# Sincronizar permisos del rol
for mod in module_ids:
level = r_data['permissions'].get(mod, 'none')
role.set_permission(mod, level)
db.session.commit()
print("[OK] Roles y matriz de permisos configurados correctamente.")
# 3b. Deduplicar rol Bedelia con acento (Bedelía) si existe
print("[*] Verificando duplicados del rol Bedelia...")
bedelia_accented = Role.query.filter(
Role.name.in_(['Bedelía', 'Bedel\u00eda', 'BEDELÍA'])
).first()
bedelia_canonical = Role.query.filter_by(name='Bedelia').first()
if bedelia_accented and bedelia_canonical and bedelia_accented.id != bedelia_canonical.id:
affected = User.query.filter_by(role_id=bedelia_accented.id).all()
for u in affected:
u.role_id = bedelia_canonical.id
u.role = 'BEDELIA'
print(f" [~] Usuario {u.email} migrado de '{bedelia_accented.name}' a 'Bedelia'")
db.session.flush()
db.session.delete(bedelia_accented)
db.session.commit()
print(f" [OK] Rol duplicado '{bedelia_accented.name}' eliminado. {len(affected)} usuario(s) migrado(s).")
else:
print(" [OK] Sin duplicados de Bedelia.")
# 4. Verificar o crear usuarios institucionales de cada rol
print("[*] Verificando usuarios institucionales y credenciales...")
roles_by_name = {r.name: r for r in Role.query.all()}
institutional_users = [
{
'email': 'admin@edu-space.com',
'name': 'Administrador del Sistema',
'role_name': 'Admin',
'role_code': 'ADMIN'
},
{
'email': 'bedelia@edu-space.com',
'name': 'Operador de Bedelía',
'role_name': 'Bedelia',
'role_code': 'BEDELIA'
},
{
'email': 'docente@edu-space.com',
'name': 'Profesor Titular',
'role_name': 'Docente',
'role_code': 'DOCENTE'
},
{
'email': 'alumno@edu-space.com',
'name': 'Estudiante Regular',
'role_name': 'Alumno',
'role_code': 'ALUMNO'
}
]
for u_info in institutional_users:
u_obj = User.query.filter_by(email=u_info['email']).first()
r_match = roles_by_name.get(u_info['role_name'])
if not u_obj:
u_obj = User(
email=u_info['email'],
name=u_info['name'],
role=u_info['role_code'],
role_id=r_match.id if r_match else None,
is_active=True,
preferred_language="es",
theme_preference="auto"
)
u_obj.set_password("admin123")
db.session.add(u_obj)
db.session.commit()
print(f" [+] Usuario creado: {u_info['email']} | Clave: admin123 | Rol: {u_info['role_name']}")
else:
# Garantizar sincronización de rol, estado activo y clave
if r_match and u_obj.role_id != r_match.id:
u_obj.role_id = r_match.id
u_obj.role = u_info['role_code']
u_obj.is_active = True
u_obj.set_password("admin123")
if not u_obj.preferred_language:
u_obj.preferred_language = "es"
if not u_obj.theme_preference:
u_obj.theme_preference = "auto"
db.session.commit()
print(f" [OK] Usuario {u_info['email']} validado (Rol: {u_info['role_name']}, Clave: admin123).")
# 5. Inicializar edificios institucionales oficiales
print("[*] Verificando edificios institucionales...")
from app.models.building import Building
b_central = Building.query.filter_by(name='Edificio Central').first()
if not b_central:
b_central = Building(
name='Edificio Central',
code='CENTRAL',
address='Tte. Gral. Juan Domingo Peron 802, CABA',
floors=4,
description='Sede central institucional de UniCABA. Aulas de grado, auditorio y laboratorios.'
)
db.session.add(b_central)
else:
b_central.address = 'Tte. Gral. Juan Domingo Peron 802, CABA'
b_virtual = Building.query.filter_by(name='Campus Virtual').first()
if not b_virtual:
b_virtual = Building(
name='Campus Virtual',
code='VIRTUAL',
address='Entorno Digital / Microsoft Teams - Moodle',
floors=0,
description='Entorno de aulas virtuales y clases a distancia'
)
db.session.add(b_virtual)
db.session.commit()
print("[OK] Edificios institucionales configurados con dirección oficial.")
# 6. Inicializar catalogo dinamico de tipificaciones de hitos academicos
print("[*] Verificando catalogo de tipificaciones de hitos academicos...")
from app.models.milestone import MilestoneType, BASE_MILESTONE_TYPES
for item in BASE_MILESTONE_TYPES:
mt = MilestoneType.query.filter_by(code=item['code']).first()
if not mt:
mt = MilestoneType(
name=item['name'],
code=item['code'],
color=item['color'],
description=item['description'],
sort_order=item['sort_order'],
is_active=True
)
db.session.add(mt)
else:
if not mt.color:
mt.color = item['color']
if not mt.sort_order:
mt.sort_order = item['sort_order']
db.session.commit()
print(f"[OK] Catalogo de tipificaciones inicializado ({MilestoneType.query.count()} tipos disponibles).")
# 7. Inicializar aulas, materias, comisiones y reservas de muestra
from app.models.classroom import Classroom
from app.models.subject import Subject, Commission
from app.models.reservation import Reservation, ReservationStatus
from datetime import datetime, timedelta, date
if Classroom.query.count() == 0:
print("[*] Generando catalogo de aulas...")
for floor in ['PB', '1', '2', '3']:
for num in range(1, 8):
code = f'{floor}0{num}' if floor != 'PB' else f'PB-0{num}'
c = Classroom(
code=code,
building=b_central.name,
building_id=b_central.id,
floor=floor,
capacity=30 + (num * 5),
description=f'Aula física estándar {code} en {b_central.name}',
is_active=True
)
db.session.add(c)
for v in range(1, 4):
vc = Classroom(
code=f'VIRTUAL-0{v}',
building=b_virtual.name,
building_id=b_virtual.id,
floor='0',
capacity=100,
description=f'Sala virtual Microsoft Teams {v}',
is_active=True
)
db.session.add(vc)
db.session.commit()
print(f"[OK] {Classroom.query.count()} aulas creadas.")
# Materia y Comisión
s = Subject.query.filter_by(code='ARQ-101').first()
if not s:
s = Subject(name='Arquitectura de Software', code='ARQ-101', is_active=True)
db.session.add(s)
db.session.commit()
comm = Commission.query.filter_by(code='COM-A').first()
if not comm:
comm = Commission(code='COM-A', subject_id=s.id, shift='Noche', semester=2, year=2026, max_students=40, is_active=True)
db.session.add(comm)
db.session.commit()
# Reservas de Hoy
if Reservation.query.count() == 0:
print("[*] Generando reservas de muestra...")
today = date.today()
classrooms = Classroom.query.all()
docente = User.query.filter_by(email='docente@edu-space.com').first()
for idx, cl in enumerate(classrooms[:19]):
start_hour = 8 + (idx % 6) * 2
r = Reservation(
classroom_id=cl.id,
commission_id=comm.id,
user_id=docente.id if docente else 1,
start_time=datetime.combine(today, datetime.min.time()) + timedelta(hours=start_hour),
end_time=datetime.combine(today, datetime.min.time()) + timedelta(hours=start_hour + 2),
purpose='Clase Regular',
status=ReservationStatus.CONFIRMED.value,
shift='Noche' if start_hour >= 18 else 'Tarde' if start_hour >= 13 else 'Mañana',
expected_attendees=cl.capacity - 5
)
db.session.add(r)
db.session.commit()
print(f"[OK] {Reservation.query.count()} reservas creadas.")
print("\n[LISTO] Base de datos inicializada y lista para usar.")
except Exception as e:
# Safe print to avoid UnicodeEncodeError in exception strings
try:
print(f"\n[ERROR] Error al inicializar la base de datos: {str(e)}")
except UnicodeEncodeError:
print("\n[ERROR] Error al inicializar la base de datos (se omiten caracteres no-ascii).")
print("\nVerifica que:")
print("1. El servicio de PostgreSQL este corriendo.")
print("2. La base de datos 'classrooms_db' haya sido creada.")
print("3. La variable DATABASE_URL en el archivo .env tenga el usuario y contraseña correctos.")
sys.exit(1)
if __name__ == "__main__":
init_database()