john
05e173b293
Replace the WebView shell with SwiftUI screens, add account-scoped Wiki and TTS APIs with adaptive review and photo scan support, and keep web/iOS pages usable while data loads asynchronously. Co-authored-by: Cursor <cursoragent@cursor.com>
739 lines
28 KiB
Python
739 lines
28 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, 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()
|