108 lines
3.2 KiB
Python
108 lines
3.2 KiB
Python
from typing import List
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy.orm import selectinload
|
|
|
|
from models.library import LibraryFile, PracticeRecord
|
|
from schemas.practice import PracticeQuestionResponse, PracticeRecordRequest, PracticeCheckResult, \
|
|
PracticeRecordResponse
|
|
from service.library_service import get_file
|
|
from service.mistake_service import update_mistake
|
|
|
|
|
|
async def get_record(db: AsyncSession, library_file_id: int) -> List[PracticeRecordResponse]:
|
|
stmt = select(PracticeRecord).where(PracticeRecord.library_file_id == library_file_id).order_by(
|
|
PracticeRecord.create_time.desc())
|
|
records = (await db.execute(stmt)).scalars().all()
|
|
|
|
return [
|
|
PracticeRecordResponse(
|
|
id=record.id,
|
|
libraryFileId=record.library_file_id,
|
|
totalCount=record.total_count,
|
|
correctCount=record.correct_count,
|
|
wrongCount=record.wrong_count,
|
|
createTime=record.create_time.strftime("%Y-%m-%d %H:%M:%S"),
|
|
)
|
|
for record in records
|
|
]
|
|
|
|
|
|
async def get_question(db: AsyncSession, file_id: int) -> List[PracticeQuestionResponse]:
|
|
stmt = (
|
|
select(LibraryFile)
|
|
.options(selectinload(LibraryFile.questions))
|
|
.where(LibraryFile.id == file_id)
|
|
)
|
|
result = await db.execute(stmt)
|
|
file = result.scalar_one_or_none()
|
|
|
|
return [
|
|
PracticeQuestionResponse(
|
|
id=q.id,
|
|
type=q.type,
|
|
question=q.question,
|
|
options=q.options,
|
|
answer=[],
|
|
select=[]
|
|
)
|
|
for q in file.questions
|
|
]
|
|
|
|
|
|
async def check_practice(db: AsyncSession,
|
|
library_file_id: int,
|
|
questions: List[PracticeQuestionResponse]) -> List[PracticeCheckResult]:
|
|
library_file = await get_file(db, library_file_id)
|
|
db_questions = library_file.questions
|
|
check_results = []
|
|
|
|
for user_q in questions:
|
|
db_q = next((q for q in db_questions if q.id == user_q.id), None)
|
|
if not db_q:
|
|
continue
|
|
|
|
is_correct = set(user_q.select) == set(db_q.answer)
|
|
|
|
if not is_correct:
|
|
await update_mistake(
|
|
db=db,
|
|
library_file_id=library_file_id,
|
|
question_id=user_q.id,
|
|
)
|
|
|
|
check_results.append(PracticeCheckResult(
|
|
id=user_q.id,
|
|
answer=db_q.answer,
|
|
select=user_q.select,
|
|
isCorrect=is_correct,
|
|
))
|
|
|
|
correct_count = sum(1 for r in check_results if r.isCorrect)
|
|
await add_practice_record(
|
|
db,
|
|
PracticeRecordRequest(
|
|
libraryFileId=library_file_id,
|
|
totalCount=len(check_results),
|
|
correctCount=correct_count,
|
|
wrongCount=len(check_results) - correct_count,
|
|
),
|
|
)
|
|
|
|
return check_results
|
|
|
|
|
|
async def add_practice_record(db: AsyncSession, data: PracticeRecordRequest):
|
|
record = PracticeRecord(
|
|
library_file_id=data.libraryFileId,
|
|
total_count=data.totalCount,
|
|
correct_count=data.correctCount,
|
|
wrong_count=data.wrongCount,
|
|
)
|
|
|
|
db.add(record)
|
|
await db.commit()
|
|
|
|
return True
|