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, "error_clear_correct_count": row.error_clear_correct_count, "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"] if data.get("error_clear_correct_count") is not None: row.error_clear_correct_count = data["error_clear_correct_count"] 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, difficulty=5.0, stability_hours=24.0, next_review_at=now, 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, "error_clear_correct_count": user_settings.error_clear_correct_count, "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()