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()