wordloop/backend/services/book_service.py

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()