120 lines
4.1 KiB
Python
120 lines
4.1 KiB
Python
from app.repositories import (
|
|
UserAnswerRepository,
|
|
QuizRepository,
|
|
SessionRepository,
|
|
UserRepository,
|
|
)
|
|
from app.schemas.response import (
|
|
HistoryResultSchema,
|
|
QuizHistoryResponse,
|
|
QuestionResult,
|
|
)
|
|
|
|
from app.helpers import DatetimeUtil
|
|
|
|
|
|
class HistoryService:
|
|
def __init__(
|
|
self,
|
|
quiz_repository: QuizRepository,
|
|
answer_repository: UserAnswerRepository,
|
|
session_repository: SessionRepository,
|
|
user_repository: UserRepository,
|
|
):
|
|
self.quiz_repository = quiz_repository
|
|
self.answer_repository = answer_repository
|
|
self.session_repository = session_repository
|
|
self.user_repository = user_repository
|
|
|
|
def get_history_by_user_id(self, user_id: str):
|
|
answer_data = self.answer_repository.get_by_user(user_id)
|
|
if not answer_data:
|
|
return []
|
|
|
|
quiz_ids = [asn.quiz_id for asn in answer_data]
|
|
quiz_data = self.quiz_repository.get_by_ids(quiz_ids)
|
|
quiz_map = {str(quiz.id): quiz for quiz in quiz_data}
|
|
|
|
result = []
|
|
for answer in answer_data:
|
|
quiz = quiz_map.get(answer.quiz_id)
|
|
if quiz:
|
|
result.append(
|
|
HistoryResultSchema(
|
|
quiz_id=str(quiz.id),
|
|
answer_id=str(answer.id),
|
|
title=quiz.title,
|
|
description=quiz.description,
|
|
total_correct=answer.total_correct,
|
|
total_question=quiz.total_quiz,
|
|
date=answer.answered_at.strftime("%Y-%m-%d %H:%M:%S"),
|
|
)
|
|
)
|
|
|
|
return result
|
|
|
|
def get_history_by_answer_id(self, answer_id: str):
|
|
answer = self.answer_repository.get_by_id(answer_id)
|
|
quiz = self.quiz_repository.get_by_id(answer.quiz_id)
|
|
|
|
total_solve_time = sum([a.time_spent for a in answer.answers])
|
|
|
|
question_results = []
|
|
for q in quiz.question_listings:
|
|
user_answer = next(
|
|
(a for a in answer.answers if a.question_index == q.index), None
|
|
)
|
|
question_results.append(
|
|
QuestionResult(
|
|
index=q.index,
|
|
question=q.question,
|
|
type=q.type,
|
|
target_answer=q.target_answer,
|
|
user_answer=user_answer.answer if user_answer else None,
|
|
is_correct=user_answer.is_correct if user_answer else None,
|
|
time_spent=user_answer.time_spent if user_answer else None,
|
|
options=q.options,
|
|
)
|
|
)
|
|
|
|
result = QuizHistoryResponse(
|
|
answer_id=str(answer.id),
|
|
quiz_id=str(quiz.id),
|
|
title=quiz.title,
|
|
description=quiz.description,
|
|
author_id=quiz.author_id,
|
|
answered_at=answer.answered_at.strftime("%d-%B-%Y"),
|
|
total_correct=answer.total_correct,
|
|
total_score=answer.total_score,
|
|
total_solve_time=total_solve_time,
|
|
question_listings=question_results,
|
|
)
|
|
return result
|
|
|
|
def get_session_history(self, session_id):
|
|
session_data = self.session_repository.find_by_session_id(session_id)
|
|
|
|
participants = []
|
|
|
|
for participant in session_data.participants:
|
|
answer = self.answer_repository.get_by_userid_and_sessionid(
|
|
user_id=participant["id"],
|
|
session_id=session_id,
|
|
)
|
|
|
|
user = self.user_repository.get_user_by_id(user_id=participant["id"])
|
|
|
|
participants.append(
|
|
{
|
|
"id": str(user.id),
|
|
"name": user.name,
|
|
"score": answer.total_score,
|
|
}
|
|
)
|
|
session_data.id = str(session_data.id)
|
|
session_data.participants = participants
|
|
session_data.created_at = DatetimeUtil.to_string(session_data.created_at)
|
|
session_data.started_at = DatetimeUtil.to_string(session_data.started_at)
|
|
session_data.ended_at = DatetimeUtil.to_string(session_data.ended_at)
|
|
return session_data
|