456 lines
21 KiB
Python
456 lines
21 KiB
Python
from sys import api_version
|
|
from rest_framework.decorators import api_view
|
|
from datetime import datetime
|
|
import random
|
|
from shutil import register_unpack_format
|
|
import string
|
|
from uuid import uuid4
|
|
from django.shortcuts import render
|
|
import pytz
|
|
from rest_framework.response import Response
|
|
from rest_framework.views import APIView
|
|
from rest_framework import status
|
|
|
|
from .models import Room, Parcours, TempCorrection
|
|
from .serializers import RoomSerializer, ParcoursSerializer
|
|
from django.contrib.sessions.models import Session
|
|
from api.Generateur import Generateur
|
|
from exercices.models import Exercice
|
|
from exercices.serializers import ExerciceSerializer
|
|
from channels.layers import get_channel_layer
|
|
from asgiref.sync import async_to_sync
|
|
|
|
from .utils import getNow
|
|
# Create your views here.
|
|
|
|
|
|
class RoomAPI(APIView):
|
|
|
|
def post(self, request, format=None):
|
|
name = request.data.get('name')
|
|
|
|
if request.user.is_authenticated:
|
|
room = Room(name=name, owner={
|
|
"name": request.user.username, 'code': request.user.id_code, 'clientId': request.user.clientId})
|
|
room.save()
|
|
clientId = request.user.clientId
|
|
code = request.user.id_code
|
|
room.userMembers.add(request.user)
|
|
else:
|
|
nick = request.data.get('nick')
|
|
code = ''.join(
|
|
random.choices(string.ascii_uppercase, k=6))
|
|
clientId = str(uuid4())
|
|
room = Room(name=name, owner={'name': nick, 'code': code, "clientId": clientId}, anonymousMembers=[{'nick': nick, 'owner': True, 'code': code, "clientId": clientId}]
|
|
)
|
|
room.save()
|
|
return Response({"data": {'id_code': room.id_code, "clientId": clientId, "code": code}}, status=status.HTTP_200_OK)
|
|
|
|
def get(self, request, format=None):
|
|
code = request.GET.get('code')
|
|
|
|
room = None
|
|
|
|
try:
|
|
room = Room.objects.filter(id_code=code)[0]
|
|
except IndexError:
|
|
return Response({'error': 'Aucune salle trouvée', 'code': '4044'}, status=status.HTTP_404_NOT_FOUND)
|
|
|
|
clientId = request.GET.get('clientId')
|
|
participants = room.anonymousMembers
|
|
users = room.userMembers.all()
|
|
print(room.userMembers.all())
|
|
|
|
userInRoom = False
|
|
if request.user.is_authenticated:
|
|
userInRoom = len(request.user.room_set.filter(
|
|
id_code=room.id_code)) != 0
|
|
|
|
if clientId not in [p['clientId'] for p in participants] and not userInRoom:
|
|
return Response({'error': 'Non autorisé', 'code': '401'}, status=status.HTTP_401_UNAUTHORIZED)
|
|
|
|
admin = False
|
|
if clientId == room.owner['clientId']:
|
|
admin = True
|
|
online = room.online
|
|
# "challenger": list(map(lambda c: {'nick': p['nick'], "validate": c['validate'], 'exos': list(map(lambda e: {'endAt': e['endAt'], 'note': e['note']}, p['exos']))}, p.challenger))
|
|
return Response({"data": {"room": {**RoomSerializer(room).data, "waiters": room.waiters if admin == True else [], "participants": [{'nick': p['nick'], 'owner': p['clientId'] == room.owner['clientId'], 'online': p['code'] in online, "code": p['code'] if admin == True else "", 'status': "anonymous"} for p in participants] + [{"status": "user", 'nick': u.username, 'owner': u.clientId == room.owner['clientId'], 'online': u.id_code in online, 'code': u.id_code if admin == True else ""} for u in users], 'parcours': [ParcoursSerializer(p).data for p in room.parcours_set.all()]}}}, status=status.HTTP_200_OK)
|
|
|
|
def delete(self, request, format=None):
|
|
code = request.data.get('code')
|
|
room = Room.objects.filter(id_code=code)[0]
|
|
room.delete()
|
|
return Response({"data": ''}, status=status.HTTP_200_OK)
|
|
|
|
def patch(self, request, format=None):
|
|
code = request.data.get('pin')
|
|
nick = request.data.get('nick')
|
|
room = None
|
|
try:
|
|
room = Room.objects.filter(id_code=code)[0]
|
|
except IndexError:
|
|
return Response({'error': 'No room'}, status=status.HTTP_502_BAD_GATEWAY)
|
|
# print(room)
|
|
if 'roomsJoined' not in request.session:
|
|
request.session['roomsJoined'] = []
|
|
elif 'roomsJoined' in request.session:
|
|
request.session['roomsJoined'] = list(
|
|
filter(lambda s: len(Room.objects.filter(id_code=s['id_code'])) != 0, request.session['roomsJoined']))
|
|
|
|
if room.id_code not in list(map(lambda r: r['id_code'], request.session['roomsJoined'])):
|
|
code = ''.join(
|
|
random.choices(string.ascii_uppercase, k=6))
|
|
while code in list(map(lambda p: p['code'], room.participants)):
|
|
code = ''.join(
|
|
random.choices(string.ascii_uppercase, k=6))
|
|
room.participants = [*room.participants, {
|
|
'nick': nick,
|
|
'owner': False,
|
|
'code': code
|
|
}]
|
|
try:
|
|
request.session['roomsJoined'] = [*request.session['roomsJoined'], {'id_code': room.id_code, 'nick': nick,
|
|
'owner': False,
|
|
'code': code}]
|
|
except:
|
|
request.session['roomsJoined'] = [{'id_code': room.id_code, 'nick': nick,
|
|
'owner': False,
|
|
'code': code}]
|
|
room.save()
|
|
|
|
return Response({"data": {'id_code': room.id_code}}, status=status.HTTP_200_OK)
|
|
|
|
|
|
@api_view(['GET'])
|
|
def RoomExist(request, id_code):
|
|
if Room.objects.filter(id_code=id_code).count() > 0:
|
|
userInRoom = False
|
|
if request.user.is_authenticated:
|
|
room = Room.objects.filter(id_code=id_code)[0]
|
|
userInRoom = len(request.user.room_set.filter(
|
|
id_code=room.id_code)) != 0
|
|
|
|
return Response({'data': {"id_code": id_code, 'is_auth': userInRoom}}, status=status.HTTP_200_OK)
|
|
else:
|
|
return Response({'error': 'Aucune salle trouvée', 'code': '4044'}, status=status.HTTP_404_NOT_FOUND)
|
|
|
|
|
|
@api_view(['GET'])
|
|
def getCorrectionInfo(request):
|
|
parcours_id = request.query_params['parcours_id']
|
|
user_code = request.query_params['user_code']
|
|
correct_code = request.query_params['correct_code']
|
|
|
|
parcours = Parcours.objects.filter(id_code=parcours_id)
|
|
|
|
if len(parcours) == 0:
|
|
return Response({'error': 'Pas de parcours lol'}, status=status.HTTP_404_NOT_FOUND)
|
|
parcours = parcours[0]
|
|
|
|
challenger = list(
|
|
filter(lambda p: p['code'] == user_code, parcours.challenger))
|
|
|
|
if(len(challenger) == 0):
|
|
return Response({'error': 'Vous n\'êtes pas un participant'}, status=status.HTTP_401_UNAUTHORIZED)
|
|
|
|
trys = challenger[0]['exos']
|
|
|
|
challenge = list(filter(lambda c: c['code'] == correct_code, trys))
|
|
|
|
if(len(challenge) == 0):
|
|
return Response({'error': 'Not found'}, status=status.HTTP_404_NOT_FOUND)
|
|
|
|
challenge = challenge[0]
|
|
|
|
return Response({'data': {**challenge, "parcours_id": parcours.id_code, 'user_code': user_code}}, status=status.HTTP_200_OK)
|
|
|
|
|
|
class ChallengeAPI(APIView):
|
|
def get(self, request, format=None):
|
|
id_code = request.GET.get('code')
|
|
user_code = request.GET.get('user_code')
|
|
|
|
parcours = Parcours.objects.filter(id_code=id_code)
|
|
|
|
if len(parcours) == 0:
|
|
return Response({'error': 'Pas de parcours lol'}, status=status.HTTP_404_NOT_FOUND)
|
|
parcours = parcours[0]
|
|
|
|
room = parcours.room
|
|
participants = list(
|
|
filter(lambda p: p['code'] == user_code, room.anonymousMembers))
|
|
userInRoom = True
|
|
if request.user.is_authenticated:
|
|
userInRoom = len(request.user.room_set.filter(
|
|
id_code=room.id_code)) != 0
|
|
|
|
if len(participants) == 0 and not userInRoom:
|
|
return Response({'error': 'Vous n\'êtes pas un participant'}, status=status.HTTP_401_UNAUTHORIZED)
|
|
|
|
exos = parcours.exercices
|
|
exos = list(map(lambda e: Generateur(Exercice.objects.filter(
|
|
id_code=e['id_code'])[0].exo_model.name, e['number'], 'web'), exos))
|
|
|
|
def parseCorrection(calcul):
|
|
"""Fait en sorte de séparer la correction présente dans le calcul"""
|
|
calculEx = calcul['calcul'].replace('[', ' [').replace(']', '] ')
|
|
splitted = calculEx.split()
|
|
if len(list(filter(lambda e: e.startswith("[") and e.endswith(']'), splitted))) == 0:
|
|
splitted.append('[]')
|
|
inputs = []
|
|
for i in range(len(splitted)):
|
|
c = splitted[i]
|
|
if c.startswith('[') and c.endswith(']'):
|
|
corr = c.replace('[', '').replace(']', '')
|
|
splitted[i] = f'[{len(inputs)}]'
|
|
inputs.append(
|
|
{'order': len(inputs), 'correction': corr, 'value': ""})
|
|
calculEx = ' '.join(splitted)
|
|
return {'calcul': calculEx, 'inputs': inputs}
|
|
|
|
exosWcorrection = []
|
|
for i in range(len(exos)):
|
|
ex = exos[i]
|
|
tempsExs = list(map(parseCorrection, ex))
|
|
tmp = {'order': i, 'exos': [{'order': ind, **e}
|
|
for ind, e in enumerate(tempsExs)]}
|
|
|
|
exosWcorrection.append(tmp)
|
|
#exos[i] = {'order': i, 'exos': list(map(lambda e, id: {'order': id,'calcul': e['calcul'], 'inputs': list(map(lambda i :{'order': i['order'],'value':i['value']}, e['inputs']))}, tempsExs))}
|
|
#exos[i] = list(map(lambda e: {'calcul': e['calcul'], 'inputs': list(map(lambda i :{'order': i['order'],'value':i['value']}, e['inputs']))}, tempsExs))
|
|
exos[i] = {'order': i, 'exos': [{'order': ind, 'calcul': e['calcul'], 'inputs': [
|
|
{'value': inp['value'], 'order': inp['order'], 'correction': ""} for inp in e['inputs']]} for ind, e in enumerate(tempsExs)]}
|
|
|
|
tempCorr = TempCorrection(correction=exosWcorrection)
|
|
tempCorr.save()
|
|
return Response({"data": {**ParcoursSerializer(parcours).data, "exos": exos, 'correctId': tempCorr.id_code}}, status=status.HTTP_200_OK)
|
|
|
|
def post(self, request, format=None):
|
|
exos = request.data.get('exos')
|
|
parcours_id = request.data.get('id_code')
|
|
user_code = request.data.get('user_code')
|
|
if request.user.is_authenticated:
|
|
user_code = request.user.id_code
|
|
correct_code = request.data.get('correct_code')
|
|
|
|
parcours = Parcours.objects.filter(id_code=parcours_id)
|
|
if len(parcours) == 0:
|
|
return Response({'error': 'Pas de parcours lol'}, status=status.HTTP_404_NOT_FOUND)
|
|
parcours = parcours[0]
|
|
room = parcours.room
|
|
|
|
corrections = TempCorrection.objects.filter(id_code=correct_code)
|
|
if len(corrections) == 0:
|
|
return Response({'error': 'Correction indisponible !'}, status=status.HTTP_404_NOT_FOUND)
|
|
correctionDB = corrections[0]
|
|
|
|
participants = list(
|
|
filter(lambda p: p['code'] == user_code, room.anonymousMembers))
|
|
userInRoom = True
|
|
if request.user.is_authenticated:
|
|
userInRoom = len(request.user.room_set.filter(
|
|
id_code=room.id_code)) != 0
|
|
|
|
if len(participants) == 0 and not userInRoom:
|
|
return Response({'error': 'Vous n\'êtes pas un participant'}, status=status.HTTP_401_UNAUTHORIZED)
|
|
|
|
# Corriger selon la correction les exos envoyé dans exos.exos
|
|
student_answer = exos['result']
|
|
correction = correctionDB.correction
|
|
|
|
def corrige(ans):
|
|
correctionExos = list(
|
|
filter(lambda c: c['order'] == ans['order'], correction))[0]
|
|
corrigedList = []
|
|
for a in ans['exos']:
|
|
#a = ans['exos'][i]
|
|
exoCorrect = list(
|
|
filter(lambda c: c['order'] == a['order'], correctionExos['exos']))[0]
|
|
|
|
correctInputs = exoCorrect['inputs']
|
|
print('correcti', correctInputs)
|
|
studentsInput = sorted(
|
|
a['inputs'], key=lambda i: i.get('order'))
|
|
print('correctiinnnn\n\n', studentsInput)
|
|
corrigedInputs = [{**inp, "value": studentsInput[inp['order']]['value'], "isCorrect": (str(studentsInput[inp['order']]['value']) == str(
|
|
inp['correction'])) if inp['correction'] != "" else None} for inp in correctInputs]
|
|
corrigedList.append({**a, 'inputs': corrigedInputs})
|
|
|
|
return {**ans, "exos": corrigedList}
|
|
|
|
corriged = list((map(corrige, student_answer)))
|
|
|
|
# l6 = uniquement les isCorrect
|
|
l2 = [exs['exos'] for exs in corriged]
|
|
l4 = [ex['inputs'] for exs in l2 for ex in exs]
|
|
l5 = [ex for exs in l4 for ex in exs]
|
|
l6 = [e['isCorrect'] for e in l5]
|
|
|
|
note = {
|
|
"total": len(l6),
|
|
"value": l6.count(True),
|
|
"isTrust": l6.count(None) == 0
|
|
}
|
|
adding_challenger = Parcours.objects.challenge(
|
|
parcours_id, user_code, {"result": corriged, "timer": exos['timer'], "note": note}, request.user.is_authenticated)
|
|
correctionDB.delete()
|
|
|
|
channel_layer = get_channel_layer()
|
|
async_to_sync(channel_layer.group_send)(f"owner__{room.id_code}", {
|
|
'type': "challenge_parcours", 'id_code': parcours_id, "participant": participants[0] if not request.user.is_authenticated else {'nick': request.user.username,'code': request.user.id_code,'clientId': request.user.clientId}, "validate": adding_challenger['validate'], 'exos': {'note': note,"code": adding_challenger['code'], 'endAt': getNow(), "timer": exos['timer']}})
|
|
return Response({"data": {"exos": corriged, "note": note}}, status=status.HTTP_200_OK)
|
|
|
|
def put(self, request, format=None):
|
|
parcours_id = request.data.get('parcours_id')
|
|
result = request.data.get('result')
|
|
challenge_code = request.data.get('challenge_code')
|
|
clientId = request.data.get('clientId')
|
|
user_code = request.data.get('user_code')
|
|
note = request.data.get('note')
|
|
|
|
print(f'PARCOURS: {parcours_id}')
|
|
parcours = Parcours.objects.filter(id_code=parcours_id)
|
|
if len(parcours) == 0:
|
|
return Response({'error': 'Pas de parcours lol'}, status=status.HTTP_404_NOT_FOUND)
|
|
parcours = parcours[0]
|
|
room = parcours.room
|
|
print(f'CLIENTID COMPARE: {room.owner["clientId"]}\n{clientId}')
|
|
if room.owner['clientId'] != clientId:
|
|
return Response({'error': 'Pas les droits'}, status=status.HTTP_401_UNAUTHORIZED)
|
|
|
|
challenger = list(
|
|
filter(lambda p: p['code'] == user_code, parcours.challenger))
|
|
|
|
if(len(challenger) == 0):
|
|
return Response({'error': 'Not found'}, status=status.HTTP_401_UNAUTHORIZED)
|
|
|
|
trys = challenger[0]['exos']
|
|
|
|
challenge = list(filter(lambda c: c['code'] == challenge_code, trys))
|
|
|
|
if(len(challenge) == 0):
|
|
return Response({'error': 'Not found'}, status=status.HTTP_404_NOT_FOUND)
|
|
|
|
challenge = challenge[0]
|
|
date = getNow()
|
|
condition = parcours.success_condition
|
|
parcours.challenger = [
|
|
*list(filter(lambda p: p['code'] != user_code, parcours.challenger)), {**challenger[0], "validate": date if note['value'] * 20 / note['total'] >= condition else False, "exos": [*list(filter(lambda c: c['code'] != challenge_code, trys)), {**challenge, "result": result, "note": note}]}]
|
|
parcours.save()
|
|
return Response({"data": {}}, status=status.HTTP_200_OK)
|
|
|
|
|
|
class ParcoursAPI(APIView):
|
|
|
|
def delete(self, request, format=None):
|
|
id_code = request.data.get('id_code')
|
|
code = request.data.get('code')
|
|
|
|
print(f'id_code={id_code};code={code}')
|
|
|
|
parcours = Parcours.objects.filter(id_code=id_code)
|
|
if len(parcours) == 0:
|
|
return Response({'error': 'Pas de parcours lol'}, status=status.HTTP_404_NOT_FOUND)
|
|
parcours = parcours[0]
|
|
room = parcours.room
|
|
if room.owner['code'] != code:
|
|
return Response({'error': 'Pas owner'}, status=status.HTTP_401_UNAUTHORIZED)
|
|
|
|
channel_layer = get_channel_layer()
|
|
|
|
async_to_sync(channel_layer.group_send)(parcours.room.id_code,
|
|
{'type': 'delete_parcours', "id_code": id_code})
|
|
parcours.delete()
|
|
|
|
return Response({'data': {'id_code': id_code}}, status=status.HTTP_200_OK)
|
|
|
|
def post(self, request, format=None):
|
|
name = request.data.get('name')
|
|
# ...futur: pour admin quand j'aurais fait la création de room "clean"
|
|
code = request.data.get('code')
|
|
room_code = request.data.get('room_code')
|
|
exos = request.data.get('exos')
|
|
timer = request.data.get('timer')
|
|
# print(exos, name, room_code)
|
|
|
|
try:
|
|
room = Room.objects.filter(id_code=room_code)[0]
|
|
except:
|
|
return Response({'error': ""}, status=status.HTTP_404_NOT_FOUND)
|
|
if room.owner['code'] != code:
|
|
return Response({'error': ''}, status=status.HTTP_401_UNAUTHORIZED)
|
|
parcours = Parcours(name=name, room=room, timer=timer, exercices=exos)
|
|
parcours.save()
|
|
|
|
layer = get_channel_layer()
|
|
async_to_sync(layer.group_send)(room_code, {
|
|
'type': 'new_parcours', 'parcours': ParcoursSerializer(parcours).data})
|
|
layer = get_channel_layer()
|
|
|
|
return Response({"data": {'id_code': parcours.id_code}}, status=status.HTTP_200_OK)
|
|
|
|
def put(self, request, format=None):
|
|
name = request.data.get('name')
|
|
# ...futur: pour admin quand j'aurais fait la création de room "clean"
|
|
code = request.data.get('code')
|
|
id_code = request.data.get('id_code')
|
|
exos = request.data.get('exos')
|
|
timer = request.data.get('timer')
|
|
# print(exos, name, room_code)
|
|
try:
|
|
parcours = Parcours.objects.filter(id_code=id_code)[0]
|
|
except:
|
|
return Response({'error': ""}, status=status.HTTP_404_NOT_FOUND)
|
|
|
|
room = parcours.room
|
|
if code != room.owner['code']:
|
|
return Response({'error': ''}, status=status.HTTP_401_UNAUTHORIZED)
|
|
|
|
parcours.challenger = []
|
|
parcours.name = name
|
|
parcours.timer = timer
|
|
parcours.exercices = exos
|
|
parcours.save()
|
|
|
|
layer = get_channel_layer()
|
|
async_to_sync(layer.group_send)(room.id_code, {
|
|
'type': 'edit_parcours', 'parcours': ParcoursSerializer(parcours).data})
|
|
layer = get_channel_layer()
|
|
|
|
return Response({"data": {'id_code': parcours.id_code}}, status=status.HTTP_200_OK)
|
|
|
|
def get(self, request, format=None):
|
|
id_code = request.GET.get('code')
|
|
user_code = request.GET.get('user_code')
|
|
parcours = Parcours.objects.filter(id_code=id_code)
|
|
if len(parcours) == 0:
|
|
return Response({'error': 'Pas de parcours lol'}, status=status.HTTP_404_NOT_FOUND)
|
|
|
|
parcours = parcours[0]
|
|
|
|
room = parcours.room
|
|
participants = list(
|
|
filter(lambda p: p['code'] == user_code, room.anonymousMembers))
|
|
userInRoom = True
|
|
if request.user.is_authenticated:
|
|
userInRoom = len(request.user.room_set.filter(id_code = room.id_code)) != 0
|
|
|
|
if len(participants) == 0 and not userInRoom:
|
|
return Response({'error': 'Vous n\'êtes pas un participant'}, status=status.HTTP_401_UNAUTHORIZED)
|
|
|
|
exos = parcours.exercices
|
|
exos = list(map(lambda e: ExerciceSerializer(Exercice.objects.filter(
|
|
id_code=e['id_code'])[0]).data, exos))
|
|
challenger_data = parcours.challenger
|
|
|
|
if user_code == parcours.room.owner['code']:
|
|
return Response({"data": {**ParcoursSerializer(parcours).data, "challenger": list(map(lambda r: {**r, 'exos': list(map(lambda e: {'endAt': e['endAt'], 'note': e['note'], "timer": e['timer'], "code": e['code']}, r['exos']))}, parcours.challenger)), "exercices": exos}}, status=status.HTTP_200_OK)
|
|
|
|
else:
|
|
challenger_filtered = list(
|
|
filter(lambda c: c['code'] == user_code, challenger_data))
|
|
if parcours.room.public_result == False:
|
|
return Response({"data": {**ParcoursSerializer(parcours).data, "challenger": list(map(lambda r: {**r, 'exos': list(map(lambda e: {'endAt': e['endAt'], 'note': e['note'], "code": e['code'], "timer": e['timer']}, r['exos']))}, challenger_filtered)), "exercices": exos}}, status=status.HTTP_200_OK)
|
|
else:
|
|
return Response({"data": {**ParcoursSerializer(parcours).data, "challenger": list(map(lambda r: {**r, "code": "", 'exos': list(map(lambda e: {'endAt': e['endAt'], "code": e['code'], 'note': e['note'], "timer": e['timer']}, r['exos']))}, parcours.challenger)), "exercices": exos}}, status=status.HTTP_200_OK)
|