wordloop/backend/services/memory_coach_service.py

import re
from datetime import datetime, timezone
from typing import Optional

from sqlalchemy.orm import Session

from models import QuizRecord, User, Word
from schemas import CoachSessionResponse, CoachTurnResponse, CoachWordBrief, MemoryToken
from services.memory_visual_service import (
    memory_visual_service,
    parse_iso,
    retention_percent,
    stability_hours,
    word_en,
    word_zh,
)
from services.quiz_service import quiz_service
from services.word_service import word_service


def _mask_zh(zh: str) -> str:
    zh = zh.strip()
    if len(zh) <= 1:
        return zh
    return zh[0] + "※" * (len(zh) - 1)


def _blank_example(example: str, en: str) -> str:
    if not example:
        return ""
    pattern = re.compile(re.escape(en), re.IGNORECASE)
    return pattern.sub("______", example, count=1)


def _retention_now(word: Word) -> float:
    now = datetime.now(timezone.utc)
    last_at = parse_iso(word.last_reviewed_at or word.created_at)
    hours = (now - last_at).total_seconds() / 3600
    return round(retention_percent(hours, stability_hours(word)), 1)


def _last_quiz_hint(db: Session, word_id: int) -> Optional[str]:
    r = (
        db.query(QuizRecord)
        .filter(QuizRecord.word_id == word_id)
        .order_by(QuizRecord.created_at.desc())
        .first()
    )
    if not r:
        return None
    return "上次答对" if r.is_correct else "上次答错"


class MemoryCoachService:
    def build_tokens(self, db: Session, word: Word, reveal_extra: int = 0) -> list[MemoryToken]:
        en = word_en(word).strip()
        zh = word_zh(word).strip()
        retention = _retention_now(word)
        hint = _last_quiz_hint(db, word.id)

        tokens: list[MemoryToken] = [
            MemoryToken(
                role="K",
                key="retention",
                label="记忆保持",
                value=f"{retention}%",
                revealed=True,
            ),
            MemoryToken(
                role="K",
                key="mastery",
                label="掌握率",
                value=f"{word.mastery_score}%",
                revealed=True,
            ),
            MemoryToken(
                role="K",
                key="status",
                label="词库状态",
                value=word.status,
                revealed=True,
            ),
        ]
        if hint:
            tokens.append(
                MemoryToken(
                    role="K",
                    key="last_quiz",
                    label="练习记录",
                    value=hint,
                    revealed=True,
                )
            )

        q_tokens: list[MemoryToken] = [
            MemoryToken(
                role="Q",
                key="zh_full",
                label="中文释义",
                value=zh,
                revealed=True,
            ),
            MemoryToken(
                role="Q",
                key="length",
                label="字母数",
                value=str(len(en)),
                revealed=True,
            ),
        ]
        if zh:
            q_tokens.append(
                MemoryToken(
                    role="Q",
                    key="zh_hint",
                    label="释义提示",
                    value=_mask_zh(zh),
                    revealed=reveal_extra > 0,
                )
            )
        if word.phonetic:
            q_tokens.append(
                MemoryToken(
                    role="Q",
                    key="phonetic",
                    label="音标",
                    value=word.phonetic,
                    revealed=reveal_extra > 0,
                )
            )
        if len(en) >= 2:
            q_tokens.append(
                MemoryToken(
                    role="Q",
                    key="prefix",
                    label="英文前缀",
                    value=en[:2] + "…",
                    revealed=reveal_extra > 1,
                )
            )
        if len(en) >= 4:
            q_tokens.append(
                MemoryToken(
                    role="Q",
                    key="suffix",
                    label="英文尾缀",
                    value="…" + en[-2:],
                    revealed=reveal_extra > 2,
                )
            )
        if word.example_en:
            blanked = _blank_example(word.example_en, en)
            if blanked and blanked != word.example_en:
                q_tokens.append(
                    MemoryToken(
                        role="Q",
                        key="example",
                        label="例句挖空",
                        value=blanked,
                        revealed=reveal_extra > 3,
                    )
                )

        tokens.extend(q_tokens)
        tokens.extend(
            [
                MemoryToken(
                    role="V",
                    key="en",
                    label="英文",
                    value=en,
                    revealed=False,
                ),
                MemoryToken(
                    role="V",
                    key="zh",
                    label="中文",
                    value=zh,
                    revealed=False,
                ),
            ]
        )
        if word.example_en:
            tokens.append(
                MemoryToken(
                    role="V",
                    key="example_en",
                    label="例句",
                    value=word.example_en,
                    revealed=False,
                )
            )
        if word.example_cn:
            tokens.append(
                MemoryToken(
                    role="V",
                    key="example_cn",
                    label="例句译文",
                    value=word.example_cn,
                    revealed=False,
                )
            )
        return tokens

    def _reveal_q_tokens(self, tokens: list[MemoryToken], count: int) -> list[MemoryToken]:
        hidden_q = [t for t in tokens if t.role == "Q" and not t.revealed]
        for t in hidden_q[:count]:
            t.revealed = True
        return tokens

    def _reveal_all_v(self, tokens: list[MemoryToken]) -> list[MemoryToken]:
        for t in tokens:
            if t.role == "V":
                t.revealed = True
            if t.role == "Q":
                t.revealed = True
        return tokens

    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, book_id=book_id)["words"]
        risk_map = {w["id"]: w.get("retention_now", 0) for w in summaries}

        items = [
            CoachWordBrief(
                word_id=w.id,
                zh=word_zh(w),
                phonetic=w.phonetic,
                retention_now=risk_map.get(w.id, _retention_now(w)),
                mastery_score=w.mastery_score,
                status=w.status,
            )
            for w in words
        ]
        return CoachSessionResponse(words=items, total=len(items))

    def handle_turn(
        self,
        db: Session,
        user: User,
        word_id: int,
        stage: str,
        user_message: str = "",
        hints_used: int = 0,
        duration_seconds: int = 0,
    ) -> CoachTurnResponse:
        word = word_service.get_word(db, user, word_id)
        en = word_en(word).strip()
        zh = word_zh(word).strip()
        phonetic = word.phonetic
        tokens = self.build_tokens(db, word, reveal_extra=hints_used)
        messages: list[str] = []
        expect_input = False
        is_correct: Optional[bool] = None
        quiz_recorded = False
        word_complete = False
        next_stage = stage
        input_hint = f"请输入「{zh}」的英文"

        if stage == "intro":
            next_stage = "derive"
            expect_input = True

        elif stage == "derive":
            answer = user_message.strip()
            if not answer:
                expect_input = True
                next_stage = "derive"
            else:
                is_correct = answer.lower() == en.lower()
                if is_correct:
                    tokens = self._reveal_all_v(tokens)
                    messages.append(f"正确：{en}")
                    quiz_service.submit_answer(
                        db,
                        user,
                        word.id,
                        "memory_coach",
                        answer,
                        en,
                        duration_seconds,
                    )
                    quiz_recorded = True
                    word_complete = True
                    next_stage = "done"
                else:
                    hints_used += 1
                    tokens = self.build_tokens(db, word, reveal_extra=hints_used)
                    tokens = self._reveal_q_tokens(tokens, 1)
                    hidden_left = sum(1 for t in tokens if t.role == "Q" and not t.revealed)
                    messages.append("不对，再试一次。")
                    if hidden_left == 0:
                        tokens = self._reveal_all_v(tokens)
                        messages.append(f"答案：{en}")
                        quiz_service.submit_answer(
                            db,
                            user,
                            word.id,
                            "memory_coach",
                            answer,
                            en,
                            duration_seconds,
                        )
                        quiz_recorded = True
                        word_complete = True
                        next_stage = "done"
                    else:
                        expect_input = True
                        next_stage = "derive"

        elif stage == "done":
            word_complete = True
            next_stage = "done"

        else:
            next_stage = "intro"
            expect_input = True

        return CoachTurnResponse(
            assistant_messages=messages,
            tokens=tokens,
            stage=next_stage,
            expect_input=expect_input,
            prompt_zh=zh,
            prompt_phonetic=phonetic,
            input_hint=input_hint,
            is_correct=is_correct,
            quiz_recorded=quiz_recorded,
            word_complete=word_complete,
            hints_used=hints_used,
            target_en=en if word_complete else None,
            target_zh=zh if word_complete else None,
        )


memory_coach_service = MemoryCoachService()