John
270 lines
9.2 KiB
Python
270 lines
9.2 KiB
Python
import random
|
|
from datetime import datetime, timedelta, timezone
|
|
from typing import Optional
|
|
|
|
from sqlalchemy import func
|
|
from sqlalchemy.orm import Session
|
|
|
|
from models import QuizRecord, User, UserSettings, Word
|
|
from schemas import QuizOption, QuizQuestion, QuizStatsResponse, WordOut
|
|
from services.dictionary_service import dictionary_service
|
|
from services.word_service import calc_mastery_score, utc_now_iso
|
|
|
|
|
|
def today_str() -> str:
|
|
return datetime.now(timezone.utc).strftime("%Y-%m-%d")
|
|
|
|
|
|
def review_interval_days(consecutive: int) -> int:
|
|
"""根据连续答对次数计算下次复习间隔（天）。"""
|
|
if consecutive <= 0:
|
|
return 0
|
|
if consecutive == 1:
|
|
return 1
|
|
if consecutive == 2:
|
|
return 3
|
|
if consecutive == 3:
|
|
return 7
|
|
if consecutive >= 5:
|
|
return 15
|
|
return 7
|
|
|
|
|
|
class QuizService:
|
|
def get_settings(self, db: Session, user: User) -> UserSettings:
|
|
settings = db.query(UserSettings).filter(UserSettings.user_id == user.id).first()
|
|
if not settings:
|
|
settings = UserSettings(user_id=user.id)
|
|
db.add(settings)
|
|
db.commit()
|
|
db.refresh(settings)
|
|
return settings
|
|
|
|
def select_daily_words(self, db: Session, user: User, limit: int) -> list[Word]:
|
|
today = today_str()
|
|
all_words = db.query(Word).filter(Word.user_id == user.id).all()
|
|
|
|
weak = [w for w in all_words if w.status == "weak"]
|
|
learning = [w for w in all_words if w.status == "learning"]
|
|
new = [w for w in all_words if w.status == "new"]
|
|
mastered_due = [
|
|
w
|
|
for w in all_words
|
|
if w.status == "mastered"
|
|
and w.review_due_date
|
|
and w.review_due_date <= today
|
|
]
|
|
|
|
pool: list[Word] = []
|
|
for group in (weak, learning, new, mastered_due):
|
|
random.shuffle(group)
|
|
for w in group:
|
|
if w not in pool:
|
|
pool.append(w)
|
|
if len(pool) >= limit:
|
|
return pool[:limit]
|
|
return pool[:limit]
|
|
|
|
def build_question(self, db: Session, word: Word, all_words: list[Word]) -> QuizQuestion:
|
|
question_type = random.choice(["en_to_zh", "zh_to_en"])
|
|
|
|
if question_type == "en_to_zh":
|
|
en = word.target_text if word.source_lang == "zh" else word.source_text
|
|
zh = word.source_text if word.source_lang == "zh" else word.target_text
|
|
prompt = en
|
|
correct = zh
|
|
distractor_pool = [
|
|
(w.source_text if w.source_lang == "zh" else w.target_text)
|
|
for w in all_words
|
|
if w.id != word.id
|
|
]
|
|
else:
|
|
zh = word.source_text if word.source_lang == "zh" else word.target_text
|
|
en = word.target_text if word.source_lang == "zh" else word.source_text
|
|
prompt = zh
|
|
correct = en
|
|
distractor_pool = [
|
|
(w.target_text if w.source_lang == "zh" else w.source_text)
|
|
for w in all_words
|
|
if w.id != word.id
|
|
]
|
|
|
|
distractors = list({d for d in distractor_pool if d != correct})
|
|
random.shuffle(distractors)
|
|
if len(distractors) < 3:
|
|
if question_type == "en_to_zh":
|
|
extra = dictionary_service.random_zh_values(db, correct, 10)
|
|
else:
|
|
extra = dictionary_service.random_en_lemmas(db, correct, 10)
|
|
for e in extra:
|
|
if e not in distractors:
|
|
distractors.append(e)
|
|
if len(distractors) >= 3:
|
|
break
|
|
options_text = [correct] + distractors[:3]
|
|
random.shuffle(options_text)
|
|
|
|
labels = ["A", "B", "C", "D"]
|
|
options = [
|
|
QuizOption(label=labels[i], text=options_text[i])
|
|
for i in range(min(4, len(options_text)))
|
|
]
|
|
|
|
return QuizQuestion(
|
|
word_id=word.id,
|
|
question_type=question_type,
|
|
prompt=prompt,
|
|
options=options,
|
|
correct_answer=correct,
|
|
)
|
|
|
|
def get_daily_quiz(self, db: Session, user: User) -> dict:
|
|
settings = self.get_settings(db, user)
|
|
words = self.select_daily_words(db, user, settings.daily_target)
|
|
all_words = db.query(Word).filter(Word.user_id == user.id).all()
|
|
|
|
if len(all_words) < 4:
|
|
questions = []
|
|
if words:
|
|
questions = [self.build_question(db, words[0], all_words)]
|
|
else:
|
|
questions = [self.build_question(db, w, all_words) for w in words]
|
|
|
|
return {
|
|
"questions": questions,
|
|
"total": len(questions),
|
|
}
|
|
|
|
def submit_answer(
|
|
self,
|
|
db: Session,
|
|
user: User,
|
|
word_id: int,
|
|
question_type: str,
|
|
user_answer: str,
|
|
correct_answer: str,
|
|
) -> dict:
|
|
word = db.query(Word).filter(Word.id == word_id, Word.user_id == user.id).first()
|
|
if not word:
|
|
from fastapi import HTTPException
|
|
raise HTTPException(status_code=404, detail="单词不存在")
|
|
|
|
settings = self.get_settings(db, user)
|
|
is_correct = user_answer.strip() == correct_answer.strip()
|
|
now = utc_now_iso()
|
|
today = today_str()
|
|
|
|
if is_correct:
|
|
word.correct_count += 1
|
|
word.consecutive_correct_count += 1
|
|
if word.status == "new":
|
|
word.status = "learning"
|
|
if word.consecutive_correct_count >= settings.master_required_count:
|
|
word.status = "mastered"
|
|
days = review_interval_days(word.consecutive_correct_count)
|
|
due = datetime.now(timezone.utc) + timedelta(days=days)
|
|
word.review_due_date = due.strftime("%Y-%m-%d")
|
|
else:
|
|
word.wrong_count += 1
|
|
word.consecutive_correct_count = 0
|
|
if word.wrong_count >= settings.weak_wrong_threshold:
|
|
word.status = "weak"
|
|
elif word.status == "new":
|
|
word.status = "learning"
|
|
word.review_due_date = today
|
|
|
|
word.mastery_score = calc_mastery_score(word.correct_count, word.wrong_count)
|
|
word.last_reviewed_at = now
|
|
|
|
record = QuizRecord(
|
|
user_id=user.id,
|
|
word_id=word.id,
|
|
question_type=question_type,
|
|
user_answer=user_answer,
|
|
correct_answer=correct_answer,
|
|
is_correct=1 if is_correct else 0,
|
|
created_at=now,
|
|
)
|
|
db.add(record)
|
|
db.commit()
|
|
db.refresh(word)
|
|
|
|
return {
|
|
"is_correct": is_correct,
|
|
"correct_answer": correct_answer,
|
|
"word": WordOut.model_validate(word),
|
|
}
|
|
|
|
def get_stats(self, db: Session, user: User) -> QuizStatsResponse:
|
|
settings = self.get_settings(db, user)
|
|
today = today_str()
|
|
|
|
total = db.query(Word).filter(Word.user_id == user.id).count()
|
|
new_count = db.query(Word).filter(Word.user_id == user.id, Word.status == "new").count()
|
|
learning_count = (
|
|
db.query(Word).filter(Word.user_id == user.id, Word.status == "learning").count()
|
|
)
|
|
mastered_count = (
|
|
db.query(Word).filter(Word.user_id == user.id, Word.status == "mastered").count()
|
|
)
|
|
weak_count = db.query(Word).filter(Word.user_id == user.id, Word.status == "weak").count()
|
|
|
|
today_records = (
|
|
db.query(QuizRecord)
|
|
.filter(
|
|
QuizRecord.user_id == user.id,
|
|
func.date(QuizRecord.created_at) == today,
|
|
)
|
|
.all()
|
|
)
|
|
# SQLite date compare fallback: filter by prefix
|
|
if not today_records:
|
|
today_records = [
|
|
r
|
|
for r in db.query(QuizRecord).filter(QuizRecord.user_id == user.id).all()
|
|
if r.created_at.startswith(today)
|
|
]
|
|
|
|
today_quiz_count = len(today_records)
|
|
today_correct = sum(1 for r in today_records if r.is_correct == 1)
|
|
accuracy = (
|
|
round(today_correct / today_quiz_count * 100, 1) if today_quiz_count > 0 else 0.0
|
|
)
|
|
|
|
streak = self._calc_streak(db, user)
|
|
|
|
return QuizStatsResponse(
|
|
total_words=total,
|
|
new_count=new_count,
|
|
learning_count=learning_count,
|
|
mastered_count=mastered_count,
|
|
weak_count=weak_count,
|
|
today_quiz_count=today_quiz_count,
|
|
today_correct_count=today_correct,
|
|
today_accuracy=accuracy,
|
|
daily_target=settings.daily_target,
|
|
today_completed=today_quiz_count,
|
|
streak_days=streak,
|
|
)
|
|
|
|
def _calc_streak(self, db: Session, user: User) -> int:
|
|
records = (
|
|
db.query(QuizRecord)
|
|
.filter(QuizRecord.user_id == user.id)
|
|
.order_by(QuizRecord.created_at.desc())
|
|
.all()
|
|
)
|
|
days_with_quiz = set()
|
|
for r in records:
|
|
days_with_quiz.add(r.created_at[:10])
|
|
|
|
streak = 0
|
|
d = datetime.now(timezone.utc).date()
|
|
while d.isoformat() in days_with_quiz:
|
|
streak += 1
|
|
d -= timedelta(days=1)
|
|
return streak
|
|
|
|
|
|
quiz_service = QuizService()
|