wordloop/backend/services/word_service.py

from datetime import datetime, timezone
from typing import Optional

from fastapi import HTTPException
from sqlalchemy.orm import Session

from models import QuizRecord, Word, User
from services.wiki_service import wiki_service

ACCUMULATION_BOOK_ID = 0


def utc_now_iso() -> str:
    return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")


def calc_mastery_score(correct: int, wrong: int) -> int:
    total = correct + wrong
    if total == 0:
        return 0
    return round(correct / total * 100)


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"],
            )
            .first()
        )
        if existing:
            raise HTTPException(status_code=400, detail="该单词已存在")

        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"],
            target_lang=data["target_lang"],
            phonetic=data.get("phonetic"),
            example_en=data.get("example_en"),
            example_cn=data.get("example_cn"),
            status="new",
            correct_count=0,
            wrong_count=0,
            consecutive_correct_count=0,
            mastery_score=0,
            difficulty=5.0,
            stability_hours=24.0,
            next_review_at=utc_now_iso(),
            review_due_date=datetime.now(timezone.utc).strftime("%Y-%m-%d"),
            created_at=utc_now_iso(),
        )
        db.add(word)
        db.commit()
        db.refresh(word)
        wiki_service.enqueue_word(user.id, word.id)
        return word

    def _list_words_query(
        self,
        db: Session,
        user: User,
        status: Optional[str] = None,
        book_id: Optional[int] = None,
    ):
        q = db.query(Word).filter(Word.user_id == user.id)
        if book_id is None or book_id == ACCUMULATION_BOOK_ID:
            q = q.filter(Word.book_id == ACCUMULATION_BOOK_ID)
        else:
            q = q.filter(Word.book_id == book_id)
        if status:
            q = q.filter(Word.status == status)
        return q

    def list_words(
        self,
        db: Session,
        user: User,
        status: Optional[str] = None,
        book_id: Optional[int] = None,
    ) -> list[Word]:
        return (
            self._list_words_query(db, user, status, book_id)
            .order_by(Word.created_at.desc())
            .all()
        )

    def list_words_page(
        self,
        db: Session,
        user: User,
        *,
        status: Optional[str] = None,
        book_id: Optional[int] = None,
        page: int = 1,
        page_size: int = 20,
    ) -> tuple[list[Word], int]:
        page = max(1, page)
        page_size = max(1, min(page_size, 100))
        q = self._list_words_query(db, user, status, book_id)
        total = q.count()
        items = (
            q.order_by(Word.created_at.desc())
            .offset((page - 1) * page_size)
            .limit(page_size)
            .all()
        )
        return items, total

    def get_word(self, db: Session, user: User, word_id: int) -> Word:
        word = db.query(Word).filter(Word.id == word_id, Word.user_id == user.id).first()
        if not word:
            raise HTTPException(status_code=404, detail="单词不存在")
        return word

    def delete_word(self, db: Session, user: User, word_id: int) -> None:
        word = self.get_word(db, user, word_id)
        wiki_service.remove_word(db, user, word)
        db.query(QuizRecord).filter(QuizRecord.word_id == word.id).delete(
            synchronize_session=False
        )
        db.flush()
        db.delete(word)
        db.flush()
        wiki_service.refresh_index(db, user)
        db.commit()

    def update_word(self, db: Session, user: User, word_id: int, data: dict) -> Word:
        word = self.get_word(db, user, word_id)
        if "status" in data and data["status"]:
            word.status = data["status"]
        db.commit()
        db.refresh(word)
        return word

    def batch_create_words(self, db: Session, user: User, items: list[dict]) -> dict:
        created: list[Word] = []
        skipped = 0
        book_id = ACCUMULATION_BOOK_ID

        for data in items:
            source_text = str(data.get("source_text", "")).strip()
            target_text = str(data.get("target_text", "")).strip()
            if not source_text or not target_text:
                continue

            item_book_id = data.get("book_id") or book_id
            existing = (
                db.query(Word)
                .filter(
                    Word.user_id == user.id,
                    Word.book_id == item_book_id,
                    Word.source_text == source_text,
                    Word.target_text == target_text,
                )
                .first()
            )
            if existing:
                skipped += 1
                continue

            word = Word(
                user_id=user.id,
                book_id=item_book_id,
                book_entry_id=data.get("book_entry_id"),
                source_text=source_text,
                target_text=target_text,
                source_lang=data.get("source_lang") or "en",
                target_lang=data.get("target_lang") or "zh",
                phonetic=data.get("phonetic"),
                example_en=data.get("example_en"),
                example_cn=data.get("example_cn"),
                status="new",
                correct_count=0,
                wrong_count=0,
                consecutive_correct_count=0,
                mastery_score=0,
                difficulty=5.0,
                stability_hours=24.0,
                next_review_at=utc_now_iso(),
                review_due_date=datetime.now(timezone.utc).strftime("%Y-%m-%d"),
                created_at=utc_now_iso(),
            )
            db.add(word)
            created.append(word)

        if not created and skipped == 0:
            raise HTTPException(status_code=400, detail="没有可添加的单词")

        db.commit()
        for word in created:
            db.refresh(word)
            wiki_service.enqueue_word(user.id, word.id)

        return {
            "created": len(created),
            "skipped": skipped,
            "items": created,
        }


word_service = WordService()