115 lines
4.1 KiB
Python
115 lines
4.1 KiB
Python
import logging
|
|
from typing import Dict, List, Union
|
|
from uuid import uuid4
|
|
|
|
from fastapi import BackgroundTasks, Response, HTTPException
|
|
from fastapi.datastructures import FormData
|
|
|
|
from app.controllers.abc import IGradeController
|
|
from app.dtos.evaluation import EvaluationType
|
|
from app.dtos.speaking import GradeSpeakingItem
|
|
from app.dtos.writing import WritingGradeTaskDTO
|
|
from app.services.abc import IGradeService, IEvaluationService
|
|
|
|
class GradeController(IGradeController):
|
|
|
|
def __init__(
|
|
self,
|
|
grade_service: IGradeService,
|
|
evaluation_service: IEvaluationService,
|
|
):
|
|
self._service = grade_service
|
|
self._evaluation_service = evaluation_service
|
|
self._logger = logging.getLogger(__name__)
|
|
|
|
async def grade_writing_task(
|
|
self,
|
|
task: int, dto: WritingGradeTaskDTO, background_tasks: BackgroundTasks
|
|
):
|
|
await self._evaluation_service.create_evaluation(
|
|
dto.userId, dto.sessionId, dto.exerciseId, EvaluationType.WRITING, task
|
|
)
|
|
|
|
await self._evaluation_service.begin_evaluation(
|
|
dto.userId, dto.sessionId, task, dto.exerciseId, EvaluationType.WRITING, dto, background_tasks
|
|
)
|
|
|
|
return Response(status_code=200)
|
|
|
|
async def grade_speaking_task(self, task: int, form: FormData, background_tasks: BackgroundTasks):
|
|
answers: Dict[int, Dict] = {}
|
|
user_id = form.get("userId")
|
|
session_id = form.get("sessionId")
|
|
exercise_id = form.get("exerciseId")
|
|
|
|
if not session_id or not exercise_id:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail="Fields sessionId and exerciseId are required!"
|
|
)
|
|
|
|
for key, value in form.items():
|
|
if '_' not in key:
|
|
continue
|
|
|
|
field_name, index = key.rsplit('_', 1)
|
|
index = int(index)
|
|
|
|
if index not in answers:
|
|
answers[index] = {}
|
|
|
|
if field_name == 'question':
|
|
answers[index]['question'] = value
|
|
elif field_name == 'audio':
|
|
answers[index]['answer'] = value
|
|
|
|
for i, answer in answers.items():
|
|
if 'question' not in answer or 'answer' not in answer:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=f"Incomplete data for answer {i}. Both question and audio required."
|
|
)
|
|
|
|
items = [
|
|
GradeSpeakingItem(
|
|
question=answers[i]['question'],
|
|
answer=answers[i]['answer']
|
|
)
|
|
for i in sorted(answers.keys())
|
|
]
|
|
|
|
ex_type = EvaluationType.SPEAKING if task == 2 else EvaluationType.SPEAKING_INTERACTIVE
|
|
|
|
await self._evaluation_service.create_evaluation(
|
|
user_id, session_id, exercise_id, ex_type, task
|
|
)
|
|
|
|
await self._evaluation_service.begin_evaluation(
|
|
user_id, session_id, task, exercise_id, ex_type, items, background_tasks
|
|
)
|
|
|
|
return Response(status_code=200)
|
|
|
|
async def grade_short_answers(self, data: Dict):
|
|
return await self._service.grade_short_answers(data)
|
|
|
|
async def grading_summary(self, data: Dict):
|
|
section_keys = ['reading', 'listening', 'writing', 'speaking', 'level']
|
|
extracted_sections = self._extract_existing_sections_from_body(data, section_keys)
|
|
return await self._service.calculate_grading_summary(extracted_sections)
|
|
|
|
@staticmethod
|
|
def _extract_existing_sections_from_body(my_dict, keys_to_extract):
|
|
if 'sections' in my_dict and isinstance(my_dict['sections'], list) and len(my_dict['sections']) > 0:
|
|
return list(
|
|
filter(
|
|
lambda item:
|
|
'code' in item and
|
|
item['code'] in keys_to_extract and
|
|
'grade' in item and
|
|
'name' in item,
|
|
my_dict['sections']
|
|
)
|
|
)
|
|
|