feat(phase5): integracion auth hibrida, email dinamico y moodle 4.1 backend con tolerancia a fallos

2026-09-23 13:34:59 -03:00
parent 7ad4e3f099
commit 1aae432bba
22 changed files with 2993 additions and 49 deletions
@@ -0,0 +1,245 @@
+import requests
+import logging
+from typing import Dict, Any, List, Optional
+from app.models.setting import SystemSetting
+
+logger = logging.getLogger(__name__)
+
+class MoodleClient:
+    """
+    Cliente REST para la API de Web Services de Moodle 4.1.
+    Soporta operaciones sobre usuarios, cursos, matriculaciones y roles,
+    leyendo dinámicamente la URL y el Token encriptado desde SystemSetting.
+    """
+
+    @classmethod
+    def get_config(cls) -> Dict[str, Any]:
+        server_url = SystemSetting.get_value('moodle_server_url', 'http://10.0.0.207/moodle').rstrip('/')
+        token = SystemSetting.get_decrypted_value('moodle_ws_token', '')
+        # Fallback a token descubierto en Moodle si no está configurado aún en BD
+        if not token:
+            token = '1a0fee6f654dc3b7f0c02fac90eda327'
+        timeout = int(SystemSetting.get_value('moodle_timeout', 10))
+        return {
+            'server_url': server_url,
+            'token': token,
+            'timeout': timeout,
+            'endpoint': f"{server_url}/webservice/rest/server.php"
+        }
+
+    @classmethod
+    def call(cls, ws_function: str, params: Optional[Dict[str, Any]] = None, method: str = 'GET') -> Dict[str, Any]:
+        """Ejecuta una llamada Web Service contra Moodle 4.1 en formato JSON."""
+        cfg = cls.get_config()
+        endpoint = cfg['endpoint']
+        token = cfg['token']
+        timeout = cfg['timeout']
+
+        req_params = {
+            'wstoken': token,
+            'wsfunction': ws_function,
+            'moodlewsrestformat': 'json'
+        }
+        if params:
+            req_params.update(params)
+
+        try:
+            if method.upper() == 'POST':
+                response = requests.post(endpoint, data=req_params, timeout=timeout)
+            else:
+                response = requests.get(endpoint, params=req_params, timeout=timeout)
+
+            response.raise_for_status()
+            data = response.json()
+
+            # Moodle retorna HTTP 200 con un campo 'exception' en caso de error lógico
+            if isinstance(data, dict) and 'exception' in data:
+                error_msg = f"Moodle Exception [{data.get('errorcode')}]: {data.get('message')}"
+                logger.error(error_msg)
+                raise RuntimeError(error_msg)
+
+            return data
+
+        except requests.exceptions.RequestException as e:
+            logger.error("Error de comunicación con Moodle (%s): %s", ws_function, str(e))
+            raise RuntimeError(f"Fallo de conexión con Moodle ({ws_function}): {str(e)}")
+
+    # --------------------------------------------------------------------------
+    # Diagnóstico y Estado
+    # --------------------------------------------------------------------------
+    @classmethod
+    def test_connection(cls) -> Dict[str, Any]:
+        """Prueba si el token y el endpoint responden adecuadamente."""
+        cfg = cls.get_config()
+        try:
+            # Consultamos usuarios con un filtro vacío o el admin
+            data = cls.call('core_user_get_users', {
+                'criteria[0][key]': 'email',
+                'criteria[0][value]': '%'
+            })
+            users = data.get('users', [])
+            return {
+                'success': True,
+                'server_url': cfg['server_url'],
+                'user_count': len(users),
+                'message': f'Conexión exitosa a Moodle 4.1 ({len(users)} usuarios detectados).'
+            }
+        except Exception as e:
+            return {
+                'success': False,
+                'server_url': cfg['server_url'],
+                'error': str(e),
+                'message': f'Fallo en la prueba de conexión: {str(e)}'
+            }
+
+    # --------------------------------------------------------------------------
+    # Usuarios (core_user_*)
+    # --------------------------------------------------------------------------
+    @classmethod
+    def get_users_by_criteria(cls, field: str, value: str) -> List[Dict[str, Any]]:
+        """Obtiene usuarios por criterio (email, username, idnumber, etc.)."""
+        params = {
+            'criteria[0][key]': field,
+            'criteria[0][value]': value
+        }
+        res = cls.call('core_user_get_users', params)
+        return res.get('users', []) if isinstance(res, dict) else []
+
+    @classmethod
+    def get_user_by_email(cls, email: str) -> Optional[Dict[str, Any]]:
+        users = cls.get_users_by_criteria('email', email.strip().lower())
+        return users[0] if users else None
+
+    @classmethod
+    def create_user(cls, username: str, email: str, firstname: str, lastname: str, password: Optional[str] = None) -> Dict[str, Any]:
+        """Crea un nuevo usuario en Moodle (core_user_create_users)."""
+        params = {
+            'users[0][username]': username.strip().lower(),
+            'users[0][email]': email.strip().lower(),
+            'users[0][firstname]': firstname.strip(),
+            'users[0][lastname]': lastname.strip(),
+            'users[0][auth]': 'manual'
+        }
+        if password:
+            params['users[0][password]'] = password
+        else:
+            params['users[0][createpassword]'] = 1 # Notifica al usuario para generar password
+
+        res = cls.call('core_user_create_users', params, method='POST')
+        # Retorna lista de diccionarios [{'id': 123, 'username': '...'}]
+        if isinstance(res, list) and len(res) > 0:
+            return res[0]
+        return res
+
+    @classmethod
+    def update_user(cls, moodle_user_id: int, firstname: Optional[str] = None, lastname: Optional[str] = None, email: Optional[str] = None) -> Any:
+        """Actualiza datos de un usuario en Moodle (core_user_update_users)."""
+        params = {'users[0][id]': moodle_user_id}
+        if firstname:
+            params['users[0][firstname]'] = firstname
+        if lastname:
+            params['users[0][lastname]'] = lastname
+        if email:
+            params['users[0][email]'] = email
+
+        return cls.call('core_user_update_users', params, method='POST')
+
+    # --------------------------------------------------------------------------
+    # Matriculación y Cursos (enrol_manual_*, core_enrol_*)
+    # --------------------------------------------------------------------------
+    @classmethod
+    def enrol_user(cls, course_id: int, user_id: int, role_id: int = 5) -> Any:
+        """
+        Matricula a un usuario en un curso Moodle (enrol_manual_enrol_users).
+        Role ID 5 = Estudiante, 3 = Docente con permiso de edición, 4 = Docente sin permiso.
+        """
+        params = {
+            'enrolments[0][roleid]': role_id,
+            'enrolments[0][userid]': user_id,
+            'enrolments[0][courseid]': course_id
+        }
+        return cls.call('enrol_manual_enrol_users', params, method='POST')
+
+    @classmethod
+    def unenrol_user(cls, course_id: int, user_id: int, role_id: int = 5) -> Any:
+        """Desmatricula a un usuario de un curso Moodle (enrol_manual_unenrol_users)."""
+        params = {
+            'enrolments[0][roleid]': role_id,
+            'enrolments[0][userid]': user_id,
+            'enrolments[0][courseid]': course_id
+        }
+        return cls.call('enrol_manual_unenrol_users', params, method='POST')
+
+    @classmethod
+    def get_enrolled_users(cls, course_id: int) -> List[Dict[str, Any]]:
+        """Obtiene todos los usuarios matriculados en un curso (core_enrol_get_enrolled_users)."""
+        params = {'courseid': course_id}
+        return cls.call('core_enrol_get_enrolled_users', params)
+
+    # --------------------------------------------------------------------------
+    # Asignación de Roles (core_role_*)
+    # --------------------------------------------------------------------------
+    @classmethod
+    def assign_role(cls, role_id: int, user_id: int, context_id: int = 1) -> Any:
+        """Asigna un rol en Moodle a un usuario en un contexto específico (core_role_assign_roles)."""
+        params = {
+            'assignments[0][roleid]': role_id,
+            'assignments[0][userid]': user_id,
+            'assignments[0][contextid]': context_id
+        }
+        return cls.call('core_role_assign_roles', params, method='POST')
+
+    @classmethod
+    def unassign_role(cls, role_id: int, user_id: int, context_id: int = 1) -> Any:
+        """Remueve un rol en Moodle (core_role_unassign_roles)."""
+        params = {
+            'unassignments[0][roleid]': role_id,
+            'unassignments[0][userid]': user_id,
+            'unassignments[0][contextid]': context_id
+        }
+        return cls.call('core_role_unassign_roles', params, method='POST')
+
+    @classmethod
+    def create_users(cls, users: List[Dict[str, Any]]) -> Any:
+        params = {}
+        for idx, u in enumerate(users):
+            for k, v in u.items():
+                params[f'users[{idx}][{k}]'] = v
+        return cls.call('core_user_create_users', params, method='POST')
+
+    @classmethod
+    def update_users(cls, users: List[Dict[str, Any]]) -> Any:
+        params = {}
+        for idx, u in enumerate(users):
+            for k, v in u.items():
+                params[f'users[{idx}][{k}]'] = v
+        return cls.call('core_user_update_users', params, method='POST')
+
+    @classmethod
+    def enrol_users(cls, enrolments: List[Dict[str, Any]]) -> Any:
+        params = {}
+        for idx, e in enumerate(enrolments):
+            params[f'enrolments[{idx}][roleid]'] = e.get('roleid', e.get('role_id', 5))
+            params[f'enrolments[{idx}][userid]'] = e.get('userid', e.get('user_id'))
+            params[f'enrolments[{idx}][courseid]'] = e.get('courseid', e.get('course_id'))
+        return cls.call('enrol_manual_enrol_users', params, method='POST')
+
+    @classmethod
+    def unenrol_users(cls, enrolments: List[Dict[str, Any]]) -> Any:
+        params = {}
+        for idx, e in enumerate(enrolments):
+            params[f'enrolments[{idx}][roleid]'] = e.get('roleid', e.get('role_id', 5))
+            params[f'enrolments[{idx}][userid]'] = e.get('userid', e.get('user_id'))
+            params[f'enrolments[{idx}][courseid]'] = e.get('courseid', e.get('course_id'))
+        return cls.call('enrol_manual_unenrol_users', params, method='POST')
+
+    @classmethod
+    def get_users(cls, criteria: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
+        params = {}
+        for idx, c in enumerate(criteria):
+            params[f'criteria[{idx}][key]'] = c.get('key')
+            params[f'criteria[{idx}][value]'] = c.get('value')
+        res = cls.call('core_user_get_users', params)
+        return res.get('users', []) if isinstance(res, dict) else []
+
+moodle_client = MoodleClient()