246 lines
10 KiB
Python
246 lines
10 KiB
Python
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()
|