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, WordBook from services.wiki_service import wiki_service from schemas import QuizOption, QuizQuestion, QuizStatsResponse, WordOut from services.book_service import ACCUMULATION_BOOK_ID, book_service from services.dictionary_service import dictionary_service from services.text_utils import en_answers_match from services.adaptive_review_service import adaptive_review_service, response_quality 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 _track_words(self, db: Session, user: User, book_id: int) -> list[Word]: return ( db.query(Word) .filter(Word.user_id == user.id, Word.book_id == book_id, Word.status != "locked") .all() ) def select_accumulation_words( self, db: Session, user: User, limit: int, pool: str = "untrained" ) -> list[Word]: words = self._track_words(db, user, ACCUMULATION_BOOK_ID) settings = self.get_settings(db, user) rule = book_service.book_settings(db, user, None, settings) return self._select_words_by_pool(words, limit, pool, rule["error_clear_correct_count"]) def _select_words_by_pool( self, words: list[Word], limit: int, pool: str, clear_count: int = 2 ) -> list[Word]: if pool == "errors": return adaptive_review_service.select_error_words(words, limit, clear_count=clear_count) if pool == "error_bank": return adaptive_review_service.select_error_bank_words( words, limit, clear_count=clear_count ) if pool == "untrained": return adaptive_review_service.select_untrained_words(words, limit, clear_count=clear_count) return adaptive_review_service.select_practice_plan_words(words, limit, clear_count=clear_count) def select_coach_words( self, db: Session, user: User, book_id: int, limit: int ) -> list[Word]: all_words = self._track_words(db, user, book_id) settings = self.get_settings(db, user) book = db.query(WordBook).filter(WordBook.id == book_id).first() if book_id > 0 else None rule = book_service.book_settings(db, user, book, settings) return adaptive_review_service.select_practice_plan_words( all_words, limit, clear_count=rule["error_clear_correct_count"] ) def select_book_words( self, db: Session, user: User, book: WordBook, limit: int, pool: str = "untrained", ) -> list[Word]: all_words = self._track_words(db, user, book.id) settings = self.get_settings(db, user) rule = book_service.book_settings(db, user, book, settings) selected = self._select_words_by_pool( all_words, limit, pool, rule["error_clear_correct_count"] ) if pool != "untrained" or len(selected) >= limit or book.learn_mode != "sequential": return selected[:limit] need = limit - len(selected) unlocked = book_service.unlock_more_new_words(db, user, book, need) if unlocked: all_words = self._track_words(db, user, book.id) selected = self._select_words_by_pool( all_words, limit, pool, rule["error_clear_correct_count"] ) return selected[:limit] def reset_practice_plan( self, db: Session, user: User, book_id: int ) -> dict: from services.adaptive_review_service import adaptive_review_service words = self._track_words(db, user, book_id) reset_count = 0 for word in words: if not adaptive_review_service.is_plan_completed(word): continue word.correct_count = 0 word.consecutive_correct_count = 0 word.train_count = 0 word.mastery_score = 0 word.status = "new" word.next_review_at = None word.review_due_date = None word.last_reviewed_at = None word.difficulty = 5.0 word.stability_hours = 24.0 reset_count += 1 db.commit() return {"reset_count": reset_count, "book_id": book_id} def select_daily_words( self, db: Session, user: User, limit: int, track: str = "accumulation" ) -> list[Word]: if track == "book": book = book_service.get_active_book(db, user) if not book: return [] book_service.ensure_user_book_words(db, user, book) settings = self.get_settings(db, user) practice = book_service.practice_settings_dict(db, user, book) return self.select_book_words(db, user, book, practice["daily_target"]) settings = self.get_settings(db, user) return self.select_accumulation_words(db, user, settings.daily_target) def build_question( self, db: Session, word: Word, all_words: list[Word], rule: Optional[dict] = None ) -> 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))) ] clear_count = int((rule or {}).get("error_clear_correct_count", 2)) return QuizQuestion( word_id=word.id, question_type=question_type, prompt=prompt, options=options, correct_answer=correct, train_count=int(word.train_count or 0), wrong_count=int(word.wrong_count or 0), error_reinforce_streak=int(word.error_reinforce_streak or 0), error_clear_correct_count=clear_count, error_bank_member=int(word.error_bank_member or 0), ) def build_spell_question(self, word: Word, rule: Optional[dict] = None) -> QuizQuestion: 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 clear_count = int((rule or {}).get("error_clear_correct_count", 2)) return QuizQuestion( word_id=word.id, question_type="spell", prompt=zh, phonetic=word.phonetic, options=[], correct_answer=en, train_count=int(word.train_count or 0), wrong_count=int(word.wrong_count or 0), error_reinforce_streak=int(word.error_reinforce_streak or 0), error_clear_correct_count=clear_count, error_bank_member=int(word.error_bank_member or 0), ) def _resolve_track(self, db: Session, user: User, track: str) -> tuple[int, int]: if track == "book": book = book_service.get_active_book(db, user) if not book: from fastapi import HTTPException raise HTTPException(status_code=400, detail="请先在首页选择词书") book_service.ensure_user_book_words(db, user, book) practice = book_service.practice_settings_dict(db, user, book) return book.id, practice["daily_target"] settings = self.get_settings(db, user) return ACCUMULATION_BOOK_ID, settings.daily_target def _practice_rule(self, db: Session, user: User, book_id: int) -> dict: settings = self.get_settings(db, user) book = None if book_id > 0: book = db.query(WordBook).filter(WordBook.id == book_id).first() return book_service.book_settings(db, user, book, settings) def _resolve_answer_pool( self, word: Word, pool: Optional[str], clear_count: int, now_dt: datetime, ) -> Optional[str]: if pool in ("errors", "error_bank", "untrained"): return pool if adaptive_review_service.is_in_error_reinforcement(word, clear_count): return "errors" if adaptive_review_service.is_in_error_bank_review(word, clear_count, now_dt): return "error_bank" return pool def _apply_error_bank_transition( self, word: Word, *, is_correct: bool, pool: Optional[str], rule: dict, duration_seconds: int, now_dt: datetime, ) -> None: clear_count = int(rule["error_clear_correct_count"]) pool = self._resolve_answer_pool(word, pool, clear_count, now_dt) if not is_correct: if not int(word.error_bank_member or 0): word.error_bank_member = 1 word.error_bank_entered_at = word.error_bank_entered_at or utc_now_iso() word.error_reinforce_streak = 0 adaptive_review_service.update_word( word, is_correct=False, attempts=1, hints_used=0, duration_seconds=duration_seconds, now=now_dt, ) return if not int(word.error_bank_member or 0): return streak = int(word.error_reinforce_streak or 0) if pool == "errors" and streak < clear_count: word.error_reinforce_streak = streak + 1 word.error_bank_member = 1 if word.error_reinforce_streak >= clear_count: adaptive_review_service.update_word( word, is_correct=True, attempts=1, hints_used=0, duration_seconds=duration_seconds, now=now_dt, ) return if pool == "error_bank": if not adaptive_review_service.is_in_error_bank_review( word, clear_count, now_dt ): return quality = response_quality( is_correct=True, attempts=1, hints_used=0, duration_seconds=duration_seconds, ) if quality >= 0.85: word.error_bank_member = 0 word.error_reinforce_streak = 0 adaptive_review_service.update_word( word, is_correct=True, attempts=1, hints_used=0, duration_seconds=duration_seconds, now=now_dt, ) def get_spell_quiz(self, db: Session, user: User, track: str = "accumulation") -> dict: book_id, limit = self._resolve_track(db, user, track) rule = self._practice_rule(db, user, book_id) if track == "book": book = book_service.get_book(db, book_id) words = self.select_book_words(db, user, book, limit, pool="mixed") else: words = self.select_accumulation_words(db, user, limit, pool="mixed") questions = [self.build_spell_question(w, rule) for w in words] return { "questions": questions, "total": len(questions), "track": track, "book_id": book_id if track == "book" else None, } def get_daily_quiz( self, db: Session, user: User, track: str = "accumulation", pool: str = "untrained" ) -> dict: book_id, limit = self._resolve_track(db, user, track) if track == "book": book = book_service.get_book(db, book_id) words = self.select_book_words(db, user, book, limit, pool=pool) else: words = self.select_accumulation_words(db, user, limit, pool=pool) all_words = self._track_words(db, user, book_id) rule = self._practice_rule(db, user, book_id) if len(all_words) < 4: questions = [] if words: questions = [self.build_question(db, words[0], all_words, rule)] else: questions = [self.build_question(db, w, all_words, rule) for w in words] return { "questions": questions, "total": len(questions), "track": track, "book_id": book_id if track == "book" else None, "pool": pool, } def submit_answer( self, db: Session, user: User, word_id: int, question_type: str, user_answer: str, correct_answer: str, duration_seconds: int = 0, pool: Optional[str] = None, ) -> 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) book = None if word.book_id > 0: book = db.query(WordBook).filter(WordBook.id == word.book_id).first() rule = book_service.book_settings(db, user, book, settings) if question_type in ("spell", "memory_coach"): is_correct = en_answers_match(user_answer, correct_answer) else: is_correct = user_answer.strip() == correct_answer.strip() now = utc_now_iso() today = today_str() now_dt = datetime.now(timezone.utc) duration_seconds = max(0, min(int(duration_seconds or 0), 3600)) in_error_flow = bool(int(word.error_bank_member or 0)) or not is_correct if is_correct: word.correct_count += 1 word.consecutive_correct_count += 1 if word.status == "new": word.status = "learning" if word.consecutive_correct_count >= rule["master_required_count"]: word.status = "mastered" if not in_error_flow: days = review_interval_days(word.consecutive_correct_count) due = now_dt + 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 >= rule["weak_wrong_threshold"]: word.status = "weak" elif word.status == "new": word.status = "learning" if not int(word.error_bank_member or 0): word.review_due_date = today word.mastery_score = calc_mastery_score(word.correct_count, word.wrong_count) word.last_reviewed_at = now word.train_count += 1 word.total_train_seconds += duration_seconds self._apply_error_bank_transition( word, is_correct=is_correct, pool=pool, rule=rule, duration_seconds=duration_seconds, now_dt=now_dt, ) 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, duration_seconds=duration_seconds, created_at=now, ) db.add(record) db.commit() db.refresh(word) db.refresh(record) wiki_service.enqueue_quiz(user.id, word.id, record.id) return { "is_correct": is_correct, "correct_answer": correct_answer, "word": WordOut.model_validate(word), } def record_attempt( self, db: Session, user: User, word_id: int, question_type: str, user_answer: str, correct_answer: str, duration_seconds: int = 0, ) -> QuizRecord: """记录一次未结束当前单词的错误尝试。""" 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) book = None if word.book_id > 0: book = db.query(WordBook).filter(WordBook.id == word.book_id).first() rule = book_service.book_settings(db, user, book, settings) record = QuizRecord( user_id=user.id, word_id=word.id, question_type=question_type, user_answer=user_answer, correct_answer=correct_answer, is_correct=0, duration_seconds=max(0, min(int(duration_seconds or 0), 3600)), created_at=utc_now_iso(), ) word.wrong_count += 1 word.consecutive_correct_count = 0 if word.wrong_count >= rule["weak_wrong_threshold"]: word.status = "weak" elif word.status == "new": word.status = "learning" word.mastery_score = calc_mastery_score(word.correct_count, word.wrong_count) word.train_count += 1 word.total_train_seconds += record.duration_seconds self._apply_error_bank_transition( word, is_correct=False, pool=None, rule=rule, duration_seconds=record.duration_seconds, now_dt=datetime.now(timezone.utc), ) db.add(record) db.commit() db.refresh(record) db.refresh(word) wiki_service.enqueue_quiz(user.id, word.id, record.id) return record @staticmethod def _word_en_zh(word: Word) -> tuple[str, str]: if word.source_lang == "zh": return word.target_text, word.source_text return word.source_text, word.target_text def _ensure_error_bank_schedule(self, word: Word, clear_count: int, now_dt: datetime) -> None: """已移出错题巩固但尚未排期的词，按记忆曲线补排下次复习。""" if int(word.error_reinforce_streak or 0) < clear_count: return if word.next_review_at or word.review_due_date: return adaptive_review_service.update_word( word, is_correct=True, attempts=1, hints_used=0, duration_seconds=0, now=now_dt, ) def list_error_bank( self, db: Session, user: User, track: str = "accumulation" ) -> dict: settings = self.get_settings(db, user) book_id = ACCUMULATION_BOOK_ID if track == "book": active = book_service.get_active_book(db, user) if not active: return { "items": [], "total": 0, "due_count": 0, "track": track, "book_id": None, } book_id = active.id clear_count = book_service.practice_settings_dict(db, user, active)[ "error_clear_correct_count" ] else: clear_count = settings.error_clear_correct_count words = [ w for w in self._track_words(db, user, book_id) if int(w.error_bank_member or 0) > 0 ] now_dt = datetime.now(timezone.utc) repaired = False for word in words: before = word.next_review_at, word.review_due_date self._ensure_error_bank_schedule(word, clear_count, now_dt) if (word.next_review_at, word.review_due_date) != before: repaired = True if repaired: db.commit() for word in words: db.refresh(word) items: list[dict] = [] due_count = 0 for word in words: is_due = adaptive_review_service.is_in_error_bank_review( word, clear_count, now_dt ) if is_due: due_count += 1 en, zh = self._word_en_zh(word) status = "due" if is_due else "cooling" items.append( { "word_id": word.id, "en": en, "zh": zh, "wrong_count": int(word.wrong_count or 0), "train_count": int(word.train_count or 0), "error_reinforce_streak": int(word.error_reinforce_streak or 0), "status": status, "next_review_at": word.next_review_at, "review_due_date": word.review_due_date, } ) items.sort( key=lambda item: ( 0 if item["status"] == "due" else 1, item["next_review_at"] or item["review_due_date"] or "9999", ) ) return { "items": items, "total": len(items), "due_count": due_count, "track": track, "book_id": book_id if track == "book" else None, } def get_stats(self, db: Session, user: User, track: str = "accumulation") -> QuizStatsResponse: settings = self.get_settings(db, user) today = today_str() book_id = ACCUMULATION_BOOK_ID daily_target = settings.daily_target if track == "book": active = book_service.get_active_book(db, user) if not active: return QuizStatsResponse( total_words=0, new_count=0, learning_count=0, mastered_count=0, weak_count=0, today_quiz_count=0, today_correct_count=0, today_accuracy=0.0, daily_target=15, today_completed=0, streak_days=0, untrained_pending=0, error_pending=0, error_bank_due_pending=0, error_bank_total=0, track=track, ) book_id = active.id daily_target = book_service.practice_settings_dict(db, user, active)["daily_target"] word_q = db.query(Word).filter(Word.user_id == user.id, Word.book_id == book_id) active_words = word_q.filter(Word.status != "locked") total = active_words.count() new_count = active_words.filter(Word.status == "new").count() learning_count = active_words.filter(Word.status == "learning").count() mastered_count = active_words.filter(Word.status == "mastered").count() weak_count = active_words.filter(Word.status == "weak").count() track_word_ids = {w.id for w in word_q.all()} today_records = [ r for r in db.query(QuizRecord).filter(QuizRecord.user_id == user.id).all() if r.created_at.startswith(today) and r.word_id in track_word_ids ] 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, track_word_ids) active_word_list = active_words.all() if track == "book": active_book = book_service.get_active_book(db, user) clear_count = ( book_service.practice_settings_dict(db, user, active_book)[ "error_clear_correct_count" ] if active_book else settings.error_clear_correct_count ) else: clear_count = settings.error_clear_correct_count untrained_pending, error_pending, error_bank_due, error_bank_total = ( adaptive_review_service.count_plan_pools(active_word_list, clear_count) ) 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=daily_target, today_completed=today_quiz_count, streak_days=streak, untrained_pending=untrained_pending, error_pending=error_pending, error_bank_due_pending=error_bank_due, error_bank_total=error_bank_total, track=track, book_id=book_id if track == "book" else None, ) def reset_practice_plan_for_track( self, db: Session, user: User, track: str = "accumulation" ) -> dict: if track == "book": book = book_service.get_active_book(db, user) if not book: return {"reset_count": 0, "book_id": None, "track": track} book_id = book.id else: book_id = ACCUMULATION_BOOK_ID result = self.reset_practice_plan(db, user, book_id) result["track"] = track return result def _calc_streak(self, db: Session, user: User, word_ids: Optional[set[int]] = None) -> int: records = ( db.query(QuizRecord) .filter(QuizRecord.user_id == user.id) .order_by(QuizRecord.created_at.desc()) .all() ) if word_ids is not None: records = [r for r in records if r.word_id in word_ids] 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()