import queue import random from typing import Dict from app.configs.constants import CustomLevelExerciseTypes, EducationalContent from app.services.abc import ( ILLMService, ILevelService, IReadingService, IWritingService, IListeningService, ISpeakingService ) class CustomLevelModule: def __init__( self, llm: ILLMService, level: ILevelService, reading: IReadingService, listening: IListeningService, writing: IWritingService, speaking: ISpeakingService ): self._llm = llm self._level = level self._reading = reading self._listening = listening self._writing = writing self._speaking = speaking # TODO: I've changed this to retrieve the args from the body request and not request query args async def get_custom_level(self, data: Dict): nr_exercises = int(data.get('nr_exercises')) exercise_id = 1 response = { "exercises": {}, "module": "level" } for i in range(1, nr_exercises + 1, 1): exercise_type = data.get(f'exercise_{i}_type') exercise_difficulty = data.get(f'exercise_{i}_difficulty', random.choice(['easy', 'medium', 'hard'])) exercise_qty = int(data.get(f'exercise_{i}_qty', -1)) exercise_topic = data.get(f'exercise_{i}_topic', random.choice(EducationalContent.TOPICS)) exercise_topic_2 = data.get(f'exercise_{i}_topic_2', random.choice(EducationalContent.TOPICS)) exercise_text_size = int(data.get(f'exercise_{i}_text_size', 700)) exercise_sa_qty = int(data.get(f'exercise_{i}_sa_qty', -1)) exercise_mc_qty = int(data.get(f'exercise_{i}_mc_qty', -1)) exercise_mc3_qty = int(data.get(f'exercise_{i}_mc3_qty', -1)) exercise_fillblanks_qty = int(data.get(f'exercise_{i}_fillblanks_qty', -1)) exercise_writeblanks_qty = int(data.get(f'exercise_{i}_writeblanks_qty', -1)) exercise_writeblanksquestions_qty = int(data.get(f'exercise_{i}_writeblanksquestions_qty', -1)) exercise_writeblanksfill_qty = int(data.get(f'exercise_{i}_writeblanksfill_qty', -1)) exercise_writeblanksform_qty = int(data.get(f'exercise_{i}_writeblanksform_qty', -1)) exercise_truefalse_qty = int(data.get(f'exercise_{i}_truefalse_qty', -1)) exercise_paragraphmatch_qty = int(data.get(f'exercise_{i}_paragraphmatch_qty', -1)) exercise_ideamatch_qty = int(data.get(f'exercise_{i}_ideamatch_qty', -1)) if exercise_type == CustomLevelExerciseTypes.MULTIPLE_CHOICE_4.value: response["exercises"][f"exercise_{i}"] = {} response["exercises"][f"exercise_{i}"]["questions"] = [] response["exercises"][f"exercise_{i}"]["type"] = "multipleChoice" while exercise_qty > 0: if exercise_qty - 15 > 0: qty = 15 else: qty = exercise_qty mc_response = await self._level.gen_multiple_choice( "normal", qty, exercise_id, utas=True, all_exams=response["exercises"][f"exercise_{i}"]["questions"] ) response["exercises"][f"exercise_{i}"]["questions"].extend(mc_response["questions"]) exercise_id = exercise_id + qty exercise_qty = exercise_qty - qty elif exercise_type == CustomLevelExerciseTypes.MULTIPLE_CHOICE_BLANK_SPACE.value: response["exercises"][f"exercise_{i}"] = {} response["exercises"][f"exercise_{i}"]["questions"] = [] response["exercises"][f"exercise_{i}"]["type"] = "multipleChoice" while exercise_qty > 0: if exercise_qty - 15 > 0: qty = 15 else: qty = exercise_qty mc_response = await self._level.gen_multiple_choice( "blank_space", qty, exercise_id, utas=True, all_exams=response["exercises"][f"exercise_{i}"]["questions"] ) response["exercises"][f"exercise_{i}"]["questions"].extend(mc_response["questions"]) exercise_id = exercise_id + qty exercise_qty = exercise_qty - qty elif exercise_type == CustomLevelExerciseTypes.MULTIPLE_CHOICE_UNDERLINED.value: response["exercises"][f"exercise_{i}"] = {} response["exercises"][f"exercise_{i}"]["questions"] = [] response["exercises"][f"exercise_{i}"]["type"] = "multipleChoice" while exercise_qty > 0: if exercise_qty - 15 > 0: qty = 15 else: qty = exercise_qty mc_response = await self._level.gen_multiple_choice( "underline", qty, exercise_id, utas=True, all_exams=response["exercises"][f"exercise_{i}"]["questions"] ) response["exercises"][f"exercise_{i}"]["questions"].extend(mc_response["questions"]) exercise_id = exercise_id + qty exercise_qty = exercise_qty - qty elif exercise_type == CustomLevelExerciseTypes.BLANK_SPACE_TEXT.value: response["exercises"][f"exercise_{i}"] = await self._level.gen_blank_space_text_utas( exercise_qty, exercise_id, exercise_text_size ) response["exercises"][f"exercise_{i}"]["type"] = "blankSpaceText" exercise_id = exercise_id + exercise_qty elif exercise_type == CustomLevelExerciseTypes.READING_PASSAGE_UTAS.value: response["exercises"][f"exercise_{i}"] = await self._level.gen_reading_passage_utas( exercise_id, exercise_sa_qty, exercise_mc_qty, exercise_topic ) response["exercises"][f"exercise_{i}"]["type"] = "readingExercises" exercise_id = exercise_id + exercise_qty elif exercise_type == CustomLevelExerciseTypes.WRITING_LETTER.value: response["exercises"][f"exercise_{i}"] = await self._writing.get_writing_task_general_question( 1, exercise_topic, exercise_difficulty ) response["exercises"][f"exercise_{i}"]["type"] = "writing" exercise_id = exercise_id + 1 elif exercise_type == CustomLevelExerciseTypes.WRITING_2.value: response["exercises"][f"exercise_{i}"] = await self._writing.get_writing_task_general_question( 2, exercise_topic, exercise_difficulty ) response["exercises"][f"exercise_{i}"]["type"] = "writing" exercise_id = exercise_id + 1 elif exercise_type == CustomLevelExerciseTypes.SPEAKING_1.value: response["exercises"][f"exercise_{i}"] = await self._speaking.get_speaking_part( 1, exercise_topic, exercise_difficulty, exercise_topic_2 ) response["exercises"][f"exercise_{i}"]["type"] = "interactiveSpeaking" exercise_id = exercise_id + 1 elif exercise_type == CustomLevelExerciseTypes.SPEAKING_2.value: response["exercises"][f"exercise_{i}"] = await self._speaking.get_speaking_part( 2, exercise_topic, exercise_difficulty ) response["exercises"][f"exercise_{i}"]["type"] = "speaking" exercise_id = exercise_id + 1 elif exercise_type == CustomLevelExerciseTypes.SPEAKING_3.value: response["exercises"][f"exercise_{i}"] = await self._speaking.get_speaking_part( 3, exercise_topic, exercise_difficulty ) response["exercises"][f"exercise_{i}"]["type"] = "interactiveSpeaking" exercise_id = exercise_id + 1 elif exercise_type == CustomLevelExerciseTypes.READING_1.value: exercises = [] exercise_qty_q = queue.Queue() total_qty = 0 if exercise_fillblanks_qty != -1: exercises.append('fillBlanks') exercise_qty_q.put(exercise_fillblanks_qty) total_qty = total_qty + exercise_fillblanks_qty if exercise_writeblanks_qty != -1: exercises.append('writeBlanks') exercise_qty_q.put(exercise_writeblanks_qty) total_qty = total_qty + exercise_writeblanks_qty if exercise_truefalse_qty != -1: exercises.append('trueFalse') exercise_qty_q.put(exercise_truefalse_qty) total_qty = total_qty + exercise_truefalse_qty if exercise_paragraphmatch_qty != -1: exercises.append('paragraphMatch') exercise_qty_q.put(exercise_paragraphmatch_qty) total_qty = total_qty + exercise_paragraphmatch_qty response["exercises"][f"exercise_{i}"] = await self._reading.gen_reading_passage( 1, exercise_topic, exercises, exercise_qty_q, exercise_difficulty, exercise_id ) response["exercises"][f"exercise_{i}"]["type"] = "reading" exercise_id = exercise_id + total_qty elif exercise_type == CustomLevelExerciseTypes.READING_2.value: exercises = [] exercise_qty_q = queue.Queue() total_qty = 0 if exercise_fillblanks_qty != -1: exercises.append('fillBlanks') exercise_qty_q.put(exercise_fillblanks_qty) total_qty = total_qty + exercise_fillblanks_qty if exercise_writeblanks_qty != -1: exercises.append('writeBlanks') exercise_qty_q.put(exercise_writeblanks_qty) total_qty = total_qty + exercise_writeblanks_qty if exercise_truefalse_qty != -1: exercises.append('trueFalse') exercise_qty_q.put(exercise_truefalse_qty) total_qty = total_qty + exercise_truefalse_qty if exercise_paragraphmatch_qty != -1: exercises.append('paragraphMatch') exercise_qty_q.put(exercise_paragraphmatch_qty) total_qty = total_qty + exercise_paragraphmatch_qty response["exercises"][f"exercise_{i}"] = await self._reading.gen_reading_passage( 2, exercise_topic, exercises, exercise_qty_q, exercise_difficulty, exercise_id ) response["exercises"][f"exercise_{i}"]["type"] = "reading" exercise_id = exercise_id + total_qty elif exercise_type == CustomLevelExerciseTypes.READING_3.value: exercises = [] exercise_qty_q = queue.Queue() total_qty = 0 if exercise_fillblanks_qty != -1: exercises.append('fillBlanks') exercise_qty_q.put(exercise_fillblanks_qty) total_qty = total_qty + exercise_fillblanks_qty if exercise_writeblanks_qty != -1: exercises.append('writeBlanks') exercise_qty_q.put(exercise_writeblanks_qty) total_qty = total_qty + exercise_writeblanks_qty if exercise_truefalse_qty != -1: exercises.append('trueFalse') exercise_qty_q.put(exercise_truefalse_qty) total_qty = total_qty + exercise_truefalse_qty if exercise_paragraphmatch_qty != -1: exercises.append('paragraphMatch') exercise_qty_q.put(exercise_paragraphmatch_qty) total_qty = total_qty + exercise_paragraphmatch_qty if exercise_ideamatch_qty != -1: exercises.append('ideaMatch') exercise_qty_q.put(exercise_ideamatch_qty) total_qty = total_qty + exercise_ideamatch_qty response["exercises"][f"exercise_{i}"] = await self._reading.gen_reading_passage( 3, exercise_topic, exercises, exercise_qty_q, exercise_id, exercise_difficulty ) response["exercises"][f"exercise_{i}"]["type"] = "reading" exercise_id = exercise_id + total_qty elif exercise_type == CustomLevelExerciseTypes.LISTENING_1.value: exercises = [] exercise_qty_q = queue.Queue() total_qty = 0 if exercise_mc_qty != -1: exercises.append('multipleChoice') exercise_qty_q.put(exercise_mc_qty) total_qty = total_qty + exercise_mc_qty if exercise_writeblanksquestions_qty != -1: exercises.append('writeBlanksQuestions') exercise_qty_q.put(exercise_writeblanksquestions_qty) total_qty = total_qty + exercise_writeblanksquestions_qty if exercise_writeblanksfill_qty != -1: exercises.append('writeBlanksFill') exercise_qty_q.put(exercise_writeblanksfill_qty) total_qty = total_qty + exercise_writeblanksfill_qty if exercise_writeblanksform_qty != -1: exercises.append('writeBlanksForm') exercise_qty_q.put(exercise_writeblanksform_qty) total_qty = total_qty + exercise_writeblanksform_qty response["exercises"][f"exercise_{i}"] = await self._listening.get_listening_question( 1, exercise_topic, exercises, exercise_difficulty, exercise_qty_q, exercise_id ) response["exercises"][f"exercise_{i}"]["type"] = "listening" exercise_id = exercise_id + total_qty elif exercise_type == CustomLevelExerciseTypes.LISTENING_2.value: exercises = [] exercise_qty_q = queue.Queue() total_qty = 0 if exercise_mc_qty != -1: exercises.append('multipleChoice') exercise_qty_q.put(exercise_mc_qty) total_qty = total_qty + exercise_mc_qty if exercise_writeblanksquestions_qty != -1: exercises.append('writeBlanksQuestions') exercise_qty_q.put(exercise_writeblanksquestions_qty) total_qty = total_qty + exercise_writeblanksquestions_qty response["exercises"][f"exercise_{i}"] = await self._listening.get_listening_question( 2, exercise_topic, exercises, exercise_difficulty, exercise_qty_q, exercise_id ) response["exercises"][f"exercise_{i}"]["type"] = "listening" exercise_id = exercise_id + total_qty elif exercise_type == CustomLevelExerciseTypes.LISTENING_3.value: exercises = [] exercise_qty_q = queue.Queue() total_qty = 0 if exercise_mc3_qty != -1: exercises.append('multipleChoice3Options') exercise_qty_q.put(exercise_mc3_qty) total_qty = total_qty + exercise_mc3_qty if exercise_writeblanksquestions_qty != -1: exercises.append('writeBlanksQuestions') exercise_qty_q.put(exercise_writeblanksquestions_qty) total_qty = total_qty + exercise_writeblanksquestions_qty response["exercises"][f"exercise_{i}"] = await self._listening.get_listening_question( 3, exercise_topic, exercises, exercise_difficulty, exercise_qty_q, exercise_id ) response["exercises"][f"exercise_{i}"]["type"] = "listening" exercise_id = exercise_id + total_qty elif exercise_type == CustomLevelExerciseTypes.LISTENING_4.value: exercises = [] exercise_qty_q = queue.Queue() total_qty = 0 if exercise_mc_qty != -1: exercises.append('multipleChoice') exercise_qty_q.put(exercise_mc_qty) total_qty = total_qty + exercise_mc_qty if exercise_writeblanksquestions_qty != -1: exercises.append('writeBlanksQuestions') exercise_qty_q.put(exercise_writeblanksquestions_qty) total_qty = total_qty + exercise_writeblanksquestions_qty if exercise_writeblanksfill_qty != -1: exercises.append('writeBlanksFill') exercise_qty_q.put(exercise_writeblanksfill_qty) total_qty = total_qty + exercise_writeblanksfill_qty if exercise_writeblanksform_qty != -1: exercises.append('writeBlanksForm') exercise_qty_q.put(exercise_writeblanksform_qty) total_qty = total_qty + exercise_writeblanksform_qty response["exercises"][f"exercise_{i}"] = await self._listening.get_listening_question( 4, exercise_topic, exercises, exercise_difficulty, exercise_qty_q, exercise_id ) response["exercises"][f"exercise_{i}"]["type"] = "listening" exercise_id = exercise_id + total_qty return response