Add word-book practice, iOS app shell, and fix embedded WebView blank screen.
Ship dual-track learning (daily accumulation vs textbook),沪教/商务词书 APIs and UI, native iOS wrapper with bundled H5, and production book import on deploy. Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
@@ -0,0 +1,240 @@
|
||||
from datetime import datetime, timezone
|
||||
from typing import Optional
|
||||
|
||||
from fastapi import HTTPException
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from models import User, UserBookSettings, UserSettings, Word, WordBook, WordBookEntry
|
||||
from services.word_service import utc_now_iso
|
||||
|
||||
ACCUMULATION_BOOK_ID = 0
|
||||
|
||||
DEPRECATED_SLUGS = frozenset({"hujiao-primary", "hujiao-junior", "hujiao-senior"})
|
||||
|
||||
|
||||
class BookService:
|
||||
def list_books(self, db: Session) -> list[WordBook]:
|
||||
return (
|
||||
db.query(WordBook)
|
||||
.filter(WordBook.is_published == 1)
|
||||
.order_by(WordBook.sort_order.asc(), WordBook.id.asc())
|
||||
.all()
|
||||
)
|
||||
|
||||
def get_book(self, db: Session, book_id: int) -> WordBook:
|
||||
book = db.query(WordBook).filter(WordBook.id == book_id, WordBook.is_published == 1).first()
|
||||
if not book:
|
||||
raise HTTPException(status_code=404, detail="词书不存在")
|
||||
return book
|
||||
|
||||
def get_user_settings_row(
|
||||
self, db: Session, user: User, book: WordBook
|
||||
) -> UserBookSettings:
|
||||
row = (
|
||||
db.query(UserBookSettings)
|
||||
.filter(UserBookSettings.user_id == user.id, UserBookSettings.book_id == book.id)
|
||||
.first()
|
||||
)
|
||||
if row:
|
||||
return row
|
||||
row = UserBookSettings(
|
||||
user_id=user.id,
|
||||
book_id=book.id,
|
||||
daily_target=book.daily_target,
|
||||
master_required_count=book.master_required_count,
|
||||
weak_wrong_threshold=book.weak_wrong_threshold,
|
||||
)
|
||||
db.add(row)
|
||||
db.commit()
|
||||
db.refresh(row)
|
||||
return row
|
||||
|
||||
def get_active_book_id(self, db: Session, user: User) -> Optional[int]:
|
||||
settings = db.query(UserSettings).filter(UserSettings.user_id == user.id).first()
|
||||
if not settings or not settings.active_book_id:
|
||||
return None
|
||||
book = (
|
||||
db.query(WordBook)
|
||||
.filter(WordBook.id == settings.active_book_id, WordBook.is_published == 1)
|
||||
.first()
|
||||
)
|
||||
if not book:
|
||||
return None
|
||||
return book.id
|
||||
|
||||
def get_active_book(self, db: Session, user: User) -> Optional[WordBook]:
|
||||
book_id = self.get_active_book_id(db, user)
|
||||
if not book_id:
|
||||
return None
|
||||
return self.get_book(db, book_id)
|
||||
|
||||
def practice_settings_dict(self, db: Session, user: User, book: WordBook) -> dict:
|
||||
row = self.get_user_settings_row(db, user, book)
|
||||
return {
|
||||
"book_id": book.id,
|
||||
"daily_target": row.daily_target,
|
||||
"master_required_count": row.master_required_count,
|
||||
"weak_wrong_threshold": row.weak_wrong_threshold,
|
||||
"learn_mode": book.learn_mode,
|
||||
}
|
||||
|
||||
def update_practice_settings(
|
||||
self, db: Session, user: User, book_id: int, data: dict
|
||||
) -> dict:
|
||||
book = self.get_book(db, book_id)
|
||||
row = self.get_user_settings_row(db, user, book)
|
||||
if data.get("daily_target") is not None:
|
||||
row.daily_target = data["daily_target"]
|
||||
if data.get("master_required_count") is not None:
|
||||
row.master_required_count = data["master_required_count"]
|
||||
if data.get("weak_wrong_threshold") is not None:
|
||||
row.weak_wrong_threshold = data["weak_wrong_threshold"]
|
||||
db.commit()
|
||||
db.refresh(row)
|
||||
return self.practice_settings_dict(db, user, book)
|
||||
|
||||
def book_progress(self, db: Session, user: User, book_id: int) -> dict:
|
||||
book = self.get_book(db, book_id)
|
||||
total = book.word_count or db.query(WordBookEntry).filter(
|
||||
WordBookEntry.book_id == book_id
|
||||
).count()
|
||||
all_user = db.query(Word).filter(Word.user_id == user.id, Word.book_id == book_id).count()
|
||||
locked = db.query(Word).filter(
|
||||
Word.user_id == user.id, Word.book_id == book_id, Word.status == "locked"
|
||||
).count()
|
||||
introduced = all_user - locked
|
||||
active_words = (
|
||||
db.query(Word)
|
||||
.filter(Word.user_id == user.id, Word.book_id == book_id, Word.status != "locked")
|
||||
.all()
|
||||
)
|
||||
mastered = sum(1 for w in active_words if w.status == "mastered")
|
||||
learning = sum(1 for w in active_words if w.status in ("learning", "weak", "new"))
|
||||
if all_user == 0:
|
||||
locked = total
|
||||
introduced = 0
|
||||
return {
|
||||
"book_id": book_id,
|
||||
"total": total,
|
||||
"introduced": introduced,
|
||||
"mastered": mastered,
|
||||
"learning": learning,
|
||||
"locked": locked,
|
||||
}
|
||||
|
||||
def _daily_unlock_limit(self, db: Session, user: User, book: WordBook) -> int:
|
||||
return self.get_user_settings_row(db, user, book).daily_target
|
||||
|
||||
def ensure_user_book_words(self, db: Session, user: User, book: WordBook) -> int:
|
||||
existing_entry_ids = {
|
||||
w.book_entry_id
|
||||
for w in db.query(Word)
|
||||
.filter(Word.user_id == user.id, Word.book_id == book.id)
|
||||
.all()
|
||||
if w.book_entry_id
|
||||
}
|
||||
entries = (
|
||||
db.query(WordBookEntry)
|
||||
.filter(WordBookEntry.book_id == book.id)
|
||||
.order_by(WordBookEntry.sort_index.asc())
|
||||
.all()
|
||||
)
|
||||
now = utc_now_iso()
|
||||
created = 0
|
||||
unlock_limit = self._daily_unlock_limit(db, user, book)
|
||||
|
||||
for entry in entries:
|
||||
if entry.id in existing_entry_ids:
|
||||
continue
|
||||
status = "new" if created < unlock_limit else "locked"
|
||||
word = Word(
|
||||
user_id=user.id,
|
||||
book_id=book.id,
|
||||
book_entry_id=entry.id,
|
||||
source_text=entry.zh,
|
||||
target_text=entry.lemma_en,
|
||||
source_lang="zh",
|
||||
target_lang="en",
|
||||
phonetic=entry.phonetic,
|
||||
example_en=entry.example_en,
|
||||
example_cn=entry.example_cn,
|
||||
status=status,
|
||||
correct_count=0,
|
||||
wrong_count=0,
|
||||
consecutive_correct_count=0,
|
||||
mastery_score=0,
|
||||
review_due_date=datetime.now(timezone.utc).strftime("%Y-%m-%d"),
|
||||
created_at=now,
|
||||
)
|
||||
db.add(word)
|
||||
created += 1
|
||||
|
||||
if created:
|
||||
db.commit()
|
||||
elif not existing_entry_ids and entries:
|
||||
self._unlock_initial_batch(db, user, book)
|
||||
return created
|
||||
|
||||
def _unlock_initial_batch(self, db: Session, user: User, book: WordBook) -> None:
|
||||
limit = self._daily_unlock_limit(db, user, book)
|
||||
locked = (
|
||||
db.query(Word)
|
||||
.filter(Word.user_id == user.id, Word.book_id == book.id, Word.status == "locked")
|
||||
.order_by(Word.id.asc())
|
||||
.limit(limit)
|
||||
.all()
|
||||
)
|
||||
for w in locked:
|
||||
w.status = "new"
|
||||
if locked:
|
||||
db.commit()
|
||||
|
||||
def activate_book(self, db: Session, user: User, book_id: int) -> WordBook:
|
||||
book = self.get_book(db, book_id)
|
||||
settings = db.query(UserSettings).filter(UserSettings.user_id == user.id).first()
|
||||
if not settings:
|
||||
settings = UserSettings(user_id=user.id)
|
||||
db.add(settings)
|
||||
db.flush()
|
||||
settings.active_book_id = book.id
|
||||
self.get_user_settings_row(db, user, book)
|
||||
self.ensure_user_book_words(db, user, book)
|
||||
db.commit()
|
||||
db.refresh(settings)
|
||||
return book
|
||||
|
||||
def unlock_more_new_words(self, db: Session, user: User, book: WordBook, limit: int) -> int:
|
||||
locked = (
|
||||
db.query(Word)
|
||||
.filter(Word.user_id == user.id, Word.book_id == book.id, Word.status == "locked")
|
||||
.order_by(Word.id.asc())
|
||||
.limit(limit)
|
||||
.all()
|
||||
)
|
||||
for w in locked:
|
||||
w.status = "new"
|
||||
if locked:
|
||||
db.commit()
|
||||
return len(locked)
|
||||
|
||||
def book_settings(
|
||||
self, db: Session, user: User, book: Optional[WordBook], user_settings: UserSettings
|
||||
) -> dict:
|
||||
if book:
|
||||
return self.practice_settings_dict(db, user, book)
|
||||
return {
|
||||
"daily_target": user_settings.daily_target,
|
||||
"master_required_count": user_settings.master_required_count,
|
||||
"weak_wrong_threshold": user_settings.weak_wrong_threshold,
|
||||
"learn_mode": "adaptive",
|
||||
}
|
||||
|
||||
def deprecate_old_books(self, db: Session) -> None:
|
||||
for slug in DEPRECATED_SLUGS:
|
||||
book = db.query(WordBook).filter(WordBook.slug == slug).first()
|
||||
if book and book.is_published:
|
||||
book.is_published = 0
|
||||
db.commit()
|
||||
|
||||
|
||||
book_service = BookService()
|
||||
@@ -216,13 +216,25 @@ class MemoryCoachService:
|
||||
t.revealed = True
|
||||
return tokens
|
||||
|
||||
def start_session(self, db: Session, user: User) -> CoachSessionResponse:
|
||||
settings = quiz_service.get_settings(db, user)
|
||||
words = quiz_service.select_daily_words(db, user, settings.daily_target)
|
||||
def start_session(self, db: Session, user: User, track: str = "accumulation") -> CoachSessionResponse:
|
||||
from services.book_service import book_service
|
||||
|
||||
if track == "book":
|
||||
book = book_service.get_active_book(db, user)
|
||||
if not book:
|
||||
return CoachSessionResponse(words=[], total=0)
|
||||
book_service.ensure_user_book_words(db, user, book)
|
||||
practice = book_service.practice_settings_dict(db, user, book)
|
||||
words = quiz_service.select_book_words(db, user, book, practice["daily_target"])
|
||||
book_id = book.id
|
||||
else:
|
||||
settings = quiz_service.get_settings(db, user)
|
||||
words = quiz_service.select_accumulation_words(db, user, settings.daily_target)
|
||||
book_id = 0
|
||||
if not words:
|
||||
return CoachSessionResponse(words=[], total=0)
|
||||
|
||||
summaries = memory_visual_service.get_visualization(db, user)["words"]
|
||||
summaries = memory_visual_service.get_visualization(db, user, book_id=book_id)["words"]
|
||||
risk_map = {w["id"]: w.get("retention_now", 0) for w in summaries}
|
||||
|
||||
items = [
|
||||
|
||||
@@ -57,14 +57,23 @@ def mastery_at_time(word: Word, records: list[QuizRecord], at: datetime) -> floa
|
||||
|
||||
|
||||
class MemoryVisualService:
|
||||
def get_visualization(self, db: Session, user: User, horizon_days: int = 30) -> dict:
|
||||
words = db.query(Word).filter(Word.user_id == user.id).all()
|
||||
records = (
|
||||
db.query(QuizRecord)
|
||||
def get_visualization(
|
||||
self, db: Session, user: User, horizon_days: int = 30, book_id: int = 0
|
||||
) -> dict:
|
||||
words = (
|
||||
db.query(Word)
|
||||
.filter(Word.user_id == user.id, Word.book_id == book_id, Word.status != "locked")
|
||||
.all()
|
||||
)
|
||||
word_ids = {w.id for w in words}
|
||||
records = [
|
||||
r
|
||||
for r in db.query(QuizRecord)
|
||||
.filter(QuizRecord.user_id == user.id)
|
||||
.order_by(QuizRecord.created_at.asc())
|
||||
.all()
|
||||
)
|
||||
if r.word_id in word_ids
|
||||
]
|
||||
records_by_word: dict[int, list[QuizRecord]] = defaultdict(list)
|
||||
for r in records:
|
||||
records_by_word[r.word_id].append(r)
|
||||
|
||||
@@ -5,8 +5,9 @@ from typing import Optional
|
||||
from sqlalchemy import func
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from models import QuizRecord, User, UserSettings, Word
|
||||
from models import QuizRecord, User, UserSettings, Word, WordBook
|
||||
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.word_service import calc_mastery_score, utc_now_iso
|
||||
|
||||
@@ -40,10 +41,31 @@ class QuizService:
|
||||
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()
|
||||
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) -> list[Word]:
|
||||
return self._select_adaptive_words(
|
||||
self._track_words(db, user, ACCUMULATION_BOOK_ID), limit
|
||||
)
|
||||
|
||||
def select_book_words(self, db: Session, user: User, book: WordBook, limit: int) -> list[Word]:
|
||||
all_words = self._track_words(db, user, book.id)
|
||||
pool = self._select_adaptive_words(all_words, limit)
|
||||
if len(pool) < limit and book.learn_mode == "sequential":
|
||||
need = limit - len(pool)
|
||||
unlocked = book_service.unlock_more_new_words(db, user, book, need)
|
||||
if unlocked:
|
||||
all_words = self._track_words(db, user, book.id)
|
||||
pool = self._select_adaptive_words(all_words, limit)
|
||||
return pool[:limit]
|
||||
|
||||
def _select_adaptive_words(self, all_words: list[Word], limit: int) -> list[Word]:
|
||||
today = today_str()
|
||||
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"]
|
||||
@@ -65,6 +87,20 @@ class QuizService:
|
||||
return pool[:limit]
|
||||
return pool[:limit]
|
||||
|
||||
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]) -> QuizQuestion:
|
||||
question_type = random.choice(["en_to_zh", "zh_to_en"])
|
||||
|
||||
@@ -130,19 +166,42 @@ class QuizService:
|
||||
correct_answer=en,
|
||||
)
|
||||
|
||||
def get_spell_quiz(self, db: Session, user: User) -> dict:
|
||||
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)
|
||||
words = self.select_daily_words(db, user, settings.daily_target)
|
||||
return ACCUMULATION_BOOK_ID, settings.daily_target
|
||||
|
||||
def get_spell_quiz(self, db: Session, user: User, track: str = "accumulation") -> 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)
|
||||
else:
|
||||
words = self.select_accumulation_words(db, user, limit)
|
||||
questions = [self.build_spell_question(w) 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) -> 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()
|
||||
def get_daily_quiz(self, db: Session, user: User, track: str = "accumulation") -> 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)
|
||||
else:
|
||||
words = self.select_accumulation_words(db, user, limit)
|
||||
all_words = self._track_words(db, user, book_id)
|
||||
|
||||
if len(all_words) < 4:
|
||||
questions = []
|
||||
@@ -154,6 +213,8 @@ class QuizService:
|
||||
return {
|
||||
"questions": questions,
|
||||
"total": len(questions),
|
||||
"track": track,
|
||||
"book_id": book_id if track == "book" else None,
|
||||
}
|
||||
|
||||
def submit_answer(
|
||||
@@ -172,6 +233,10 @@ class QuizService:
|
||||
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 = (
|
||||
user_answer.strip().lower() == correct_answer.strip().lower()
|
||||
@@ -187,7 +252,7 @@ class QuizService:
|
||||
word.consecutive_correct_count += 1
|
||||
if word.status == "new":
|
||||
word.status = "learning"
|
||||
if word.consecutive_correct_count >= settings.master_required_count:
|
||||
if word.consecutive_correct_count >= rule["master_required_count"]:
|
||||
word.status = "mastered"
|
||||
days = review_interval_days(word.consecutive_correct_count)
|
||||
due = datetime.now(timezone.utc) + timedelta(days=days)
|
||||
@@ -195,7 +260,7 @@ class QuizService:
|
||||
else:
|
||||
word.wrong_count += 1
|
||||
word.consecutive_correct_count = 0
|
||||
if word.wrong_count >= settings.weak_wrong_threshold:
|
||||
if word.wrong_count >= rule["weak_wrong_threshold"]:
|
||||
word.status = "weak"
|
||||
elif word.status == "new":
|
||||
word.status = "learning"
|
||||
@@ -226,35 +291,45 @@ class QuizService:
|
||||
"word": WordOut.model_validate(word),
|
||||
}
|
||||
|
||||
def get_stats(self, db: Session, user: User) -> QuizStatsResponse:
|
||||
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,
|
||||
track=track,
|
||||
)
|
||||
book_id = active.id
|
||||
daily_target = book_service.practice_settings_dict(db, user, active)["daily_target"]
|
||||
|
||||
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()
|
||||
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()
|
||||
|
||||
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)
|
||||
]
|
||||
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)
|
||||
@@ -262,7 +337,7 @@ class QuizService:
|
||||
round(today_correct / today_quiz_count * 100, 1) if today_quiz_count > 0 else 0.0
|
||||
)
|
||||
|
||||
streak = self._calc_streak(db, user)
|
||||
streak = self._calc_streak(db, user, track_word_ids)
|
||||
|
||||
return QuizStatsResponse(
|
||||
total_words=total,
|
||||
@@ -273,18 +348,22 @@ class QuizService:
|
||||
today_quiz_count=today_quiz_count,
|
||||
today_correct_count=today_correct,
|
||||
today_accuracy=accuracy,
|
||||
daily_target=settings.daily_target,
|
||||
daily_target=daily_target,
|
||||
today_completed=today_quiz_count,
|
||||
streak_days=streak,
|
||||
track=track,
|
||||
book_id=book_id if track == "book" else None,
|
||||
)
|
||||
|
||||
def _calc_streak(self, db: Session, user: User) -> int:
|
||||
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])
|
||||
|
||||
@@ -34,7 +34,7 @@ class TranslationService:
|
||||
entry.example_en,
|
||||
entry.example_cn,
|
||||
)
|
||||
return self._placeholder(text, text, "en", "zh")
|
||||
return self._placeholder(text, "en", "zh")
|
||||
|
||||
def _zh_to_en(self, db: Session, text: str) -> dict:
|
||||
entry = dictionary_service.lookup_zh(db, text)
|
||||
@@ -48,7 +48,7 @@ class TranslationService:
|
||||
entry.example_en,
|
||||
entry.example_cn,
|
||||
)
|
||||
return self._placeholder(text, f"[{text}]", "zh", "en")
|
||||
return self._placeholder(text, "zh", "en")
|
||||
|
||||
def _build_response(
|
||||
self,
|
||||
@@ -68,17 +68,19 @@ class TranslationService:
|
||||
"phonetic": phonetic,
|
||||
"example_en": example_en,
|
||||
"example_cn": example_cn,
|
||||
"found": True,
|
||||
}
|
||||
|
||||
def _placeholder(self, source: str, target: str, source_lang: str, target_lang: str) -> dict:
|
||||
def _placeholder(self, source: str, source_lang: str, target_lang: str) -> dict:
|
||||
return {
|
||||
"source_text": source,
|
||||
"target_text": target,
|
||||
"target_text": "",
|
||||
"source_lang": source_lang,
|
||||
"target_lang": target_lang,
|
||||
"phonetic": None,
|
||||
"example_en": f"Example with {target}." if target_lang == "en" else None,
|
||||
"example_cn": f"包含「{source}」的例句。" if source_lang == "zh" else None,
|
||||
"example_en": None,
|
||||
"example_cn": None,
|
||||
"found": False,
|
||||
}
|
||||
|
||||
|
||||
|
||||
@@ -6,6 +6,8 @@ from sqlalchemy.orm import Session
|
||||
|
||||
from models import QuizRecord, Word, User
|
||||
|
||||
ACCUMULATION_BOOK_ID = 0
|
||||
|
||||
|
||||
def utc_now_iso() -> str:
|
||||
return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
|
||||
@@ -20,10 +22,12 @@ def calc_mastery_score(correct: int, wrong: int) -> int:
|
||||
|
||||
class WordService:
|
||||
def create_word(self, db: Session, user: User, data: dict) -> Word:
|
||||
book_id = data.get("book_id") or ACCUMULATION_BOOK_ID
|
||||
existing = (
|
||||
db.query(Word)
|
||||
.filter(
|
||||
Word.user_id == user.id,
|
||||
Word.book_id == book_id,
|
||||
Word.source_text == data["source_text"],
|
||||
Word.target_text == data["target_text"],
|
||||
)
|
||||
@@ -34,6 +38,8 @@ class WordService:
|
||||
|
||||
word = Word(
|
||||
user_id=user.id,
|
||||
book_id=book_id,
|
||||
book_entry_id=data.get("book_entry_id"),
|
||||
source_text=data["source_text"],
|
||||
target_text=data["target_text"],
|
||||
source_lang=data["source_lang"],
|
||||
@@ -54,8 +60,18 @@ class WordService:
|
||||
db.refresh(word)
|
||||
return word
|
||||
|
||||
def list_words(self, db: Session, user: User, status: Optional[str] = None) -> list[Word]:
|
||||
def list_words(
|
||||
self,
|
||||
db: Session,
|
||||
user: User,
|
||||
status: Optional[str] = None,
|
||||
book_id: Optional[int] = None,
|
||||
) -> list[Word]:
|
||||
q = db.query(Word).filter(Word.user_id == user.id)
|
||||
if book_id is None:
|
||||
q = q.filter(Word.book_id == ACCUMULATION_BOOK_ID)
|
||||
elif book_id > 0:
|
||||
q = q.filter(Word.book_id == book_id)
|
||||
if status:
|
||||
q = q.filter(Word.status == status)
|
||||
return q.order_by(Word.created_at.desc()).all()
|
||||
|
||||
Reference in New Issue
Block a user