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