wordloop/backend/services/adaptive_review_service.py

"""可解释的自适应复习调度，用真实回忆表现更新单词出现频率。"""

from __future__ import annotations

import math
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone

from models import Word

TARGET_RETRIEVABILITY = 0.85
MIN_STABILITY_HOURS = 0.25
MAX_STABILITY_HOURS = 24.0 * 365.0


def parse_utc(value: str | None, fallback: datetime | None = None) -> datetime:
    if not value:
        return fallback or datetime.now(timezone.utc)
    normalized = value.replace("Z", "+00:00")
    try:
        parsed = datetime.fromisoformat(normalized)
    except ValueError:
        parsed = datetime.strptime(value[:19], "%Y-%m-%dT%H:%M:%S")
    if parsed.tzinfo is None:
        parsed = parsed.replace(tzinfo=timezone.utc)
    return parsed.astimezone(timezone.utc)


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


def retrievability(word: Word, now: datetime | None = None) -> float:
    now = now or datetime.now(timezone.utc)
    last_review = parse_utc(word.last_reviewed_at or word.created_at, now)
    elapsed_hours = max(0.0, (now - last_review).total_seconds() / 3600.0)
    stability = max(MIN_STABILITY_HOURS, float(word.stability_hours or 24.0))
    return max(0.0, min(1.0, math.exp(-elapsed_hours / stability)))


def response_quality(
    *,
    is_correct: bool,
    attempts: int,
    hints_used: int,
    duration_seconds: int,
) -> float:
    if not is_correct:
        return 0.0
    quality = 1.0
    quality -= min(0.45, max(0, attempts - 1) * 0.18)
    quality -= min(0.35, max(0, hints_used) * 0.12)
    if duration_seconds > 30:
        quality -= min(0.25, (duration_seconds - 30) / 180.0)
    elif duration_seconds <= 8 and attempts == 1 and hints_used == 0:
        quality += 0.1
    return max(0.15, min(1.0, quality))


@dataclass(frozen=True)
class ReviewUpdate:
    quality: float
    difficulty: float
    stability_hours: float
    retrievability: float
    next_review_at: str
    interval_hours: float
    message: str


class AdaptiveReviewService:
    def update_word(
        self,
        word: Word,
        *,
        is_correct: bool,
        attempts: int,
        hints_used: int,
        duration_seconds: int,
        observed_retrievability: float | None = None,
        now: datetime | None = None,
    ) -> ReviewUpdate:
        now = now or datetime.now(timezone.utc)
        before_r = (
            retrievability(word, now)
            if observed_retrievability is None
            else max(0.0, min(1.0, observed_retrievability))
        )
        old_difficulty = float(word.difficulty or 5.0)
        old_stability = max(
            MIN_STABILITY_HOURS, float(word.stability_hours or 24.0)
        )
        quality = response_quality(
            is_correct=is_correct,
            attempts=attempts,
            hints_used=hints_used,
            duration_seconds=duration_seconds,
        )

        if is_correct:
            difficulty = old_difficulty - (quality - 0.55) * 0.9
            growth = 1.15 + 1.85 * quality + 0.35 * (1.0 - before_r)
            stability = old_stability * growth
            interval_hours = -stability * math.log(TARGET_RETRIEVABILITY)
            interval_hours = max(1.0, interval_hours)
        else:
            difficulty = old_difficulty + 0.8
            stability = old_stability * 0.45
            interval_hours = 0.25

        difficulty = max(1.0, min(10.0, difficulty))
        stability = max(MIN_STABILITY_HOURS, min(MAX_STABILITY_HOURS, stability))
        next_review = now + timedelta(hours=interval_hours)

        word.difficulty = round(difficulty, 3)
        word.stability_hours = round(stability, 3)
        word.next_review_at = utc_iso(next_review)
        word.review_due_date = next_review.strftime("%Y-%m-%d")

        if not is_correct:
            message = "本词记忆较弱，将在约 15 分钟后优先出现"
        elif interval_hours < 24:
            message = f"本词将在约 {max(1, round(interval_hours))} 小时后复习"
        else:
            message = f"本词将在约 {max(1, round(interval_hours / 24))} 天后复习"

        return ReviewUpdate(
            quality=round(quality, 3),
            difficulty=word.difficulty,
            stability_hours=word.stability_hours,
            retrievability=round(before_r * 100.0, 1),
            next_review_at=word.next_review_at,
            interval_hours=round(interval_hours, 2),
            message=message,
        )

    def selection_score(self, word: Word, now: datetime | None = None) -> float:
        now = now or datetime.now(timezone.utc)
        current_r = retrievability(word, now)
        difficulty = float(word.difficulty or 5.0) / 10.0
        score = (1.0 - current_r) * 70.0 + difficulty * 20.0

        if word.status == "weak":
            score += 35.0
        elif word.status == "learning":
            score += 18.0
        elif word.status == "new":
            score += 25.0

        if word.next_review_at:
            due_at = parse_utc(word.next_review_at, now)
            overdue_hours = (now - due_at).total_seconds() / 3600.0
            if overdue_hours >= 0:
                score += 50.0 + min(50.0, overdue_hours / 24.0 * 5.0)
            else:
                score -= min(35.0, -overdue_hours / 24.0 * 4.0)
        elif word.review_due_date:
            if word.review_due_date <= now.strftime("%Y-%m-%d"):
                score += 45.0

        wrong = max(0, int(word.wrong_count or 0))
        correct = max(0, int(word.correct_count or 0))
        if wrong > 0:
            score += min(40.0, wrong * 8.0)
        if correct == 0 and wrong > 0:
            score += 45.0
        elif wrong > correct:
            score += 25.0
        mastery = max(0, min(100, int(word.mastery_score or 0)))
        score += (100 - mastery) * 0.25

        return score

    @staticmethod
    def error_bank_member(word: Word) -> bool:
        return bool(int(getattr(word, "error_bank_member", 0) or 0))

    @staticmethod
    def error_reinforce_streak(word: Word) -> int:
        return max(0, int(getattr(word, "error_reinforce_streak", 0) or 0))

    @staticmethod
    def is_review_due(word: Word, now: datetime | None = None) -> bool:
        """仅当已排期且到期时返回 True；未排期视为冷却中，不可提前复习。"""
        now = now or datetime.now(timezone.utc)
        if word.next_review_at:
            due_at = parse_utc(word.next_review_at, now)
            return due_at <= now
        if word.review_due_date:
            return word.review_due_date <= now.strftime("%Y-%m-%d")
        return False

    @staticmethod
    def is_in_error_reinforcement(word: Word, clear_count: int) -> bool:
        """错题巩固·强化阶段：在错题库且连续答对未达设定次数。"""
        if not AdaptiveReviewService.error_bank_member(word):
            return False
        return AdaptiveReviewService.error_reinforce_streak(word) < clear_count

    @staticmethod
    def is_in_error_bank_review(word: Word, clear_count: int, now: datetime | None = None) -> bool:
        """错题库·间隔复习：已移出错题巩固且到达复习时间。"""
        if not AdaptiveReviewService.error_bank_member(word):
            return False
        if AdaptiveReviewService.error_reinforce_streak(word) < clear_count:
            return False
        return AdaptiveReviewService.is_review_due(word, now)

    @staticmethod
    def is_plan_completed(word: Word) -> bool:
        """已训练、有答对记录，且不在错题库。"""
        if AdaptiveReviewService.error_bank_member(word):
            return False
        return (
            int(word.train_count or 0) > 0 and int(word.correct_count or 0) > 0
        )

    @staticmethod
    def is_plan_eligible(word: Word) -> bool:
        return not AdaptiveReviewService.is_plan_completed(word)

    @staticmethod
    def is_untrained_word(word: Word) -> bool:
        """从未练过的词：无练习记录且不在错题库。"""
        if AdaptiveReviewService.error_bank_member(word):
            return False
        return int(word.train_count or 0) == 0

    @staticmethod
    def is_error_priority(word: Word) -> bool:
        """练习表现偏差的词：应优先进入记忆对话。"""
        if AdaptiveReviewService.is_plan_completed(word):
            return False
        if AdaptiveReviewService.error_bank_member(word):
            return True
        wrong = max(0, int(word.wrong_count or 0))
        if wrong <= 0:
            return int(word.train_count or 0) == 0
        if word.status == "weak":
            return True
        correct = max(0, int(word.correct_count or 0))
        if correct == 0:
            return True
        return wrong > correct

    @staticmethod
    def split_plan_pools(
        words: list[Word], clear_count: int = 2
    ) -> tuple[list[Word], list[Word]]:
        """将计划内词拆为未练词池与错题巩固强化池。"""
        eligible = [w for w in words if AdaptiveReviewService.is_plan_eligible(w)]
        untrained = [
            w for w in eligible if AdaptiveReviewService.is_untrained_word(w)
        ]
        reinforce = [
            w
            for w in eligible
            if AdaptiveReviewService.is_in_error_reinforcement(w, clear_count)
        ]
        return untrained, reinforce

    @staticmethod
    def split_error_bank(words: list[Word], clear_count: int = 2, now: datetime | None = None) -> tuple[list[Word], list[Word]]:
        """错题库成员拆为待复习与冷却中。"""
        now = now or datetime.now(timezone.utc)
        members = [w for w in words if AdaptiveReviewService.error_bank_member(w)]
        due = [
            w
            for w in members
            if AdaptiveReviewService.is_in_error_bank_review(w, clear_count, now)
        ]
        cooling = [w for w in members if w not in due]
        return due, cooling

    @staticmethod
    def count_plan_pools(
        words: list[Word], clear_count: int = 2, now: datetime | None = None
    ) -> tuple[int, int, int, int]:
        untrained, reinforce = AdaptiveReviewService.split_plan_pools(words, clear_count)
        due, _ = AdaptiveReviewService.split_error_bank(words, clear_count, now)
        bank_total = sum(1 for w in words if AdaptiveReviewService.error_bank_member(w))
        return len(untrained), len(reinforce), len(due), bank_total

    def select_untrained_words(
        self, words: list[Word], limit: int, now: datetime | None = None, clear_count: int = 2
    ) -> list[Word]:
        untrained, _ = self.split_plan_pools(words, clear_count)
        if not untrained or limit <= 0:
            return []
        return self.select_words(untrained, limit, now)

    def select_error_words(
        self, words: list[Word], limit: int, now: datetime | None = None, clear_count: int = 2
    ) -> list[Word]:
        _, reinforce = self.split_plan_pools(words, clear_count)
        if not reinforce or limit <= 0:
            return []
        return self.select_words(reinforce, limit, now)

    def select_error_bank_words(
        self, words: list[Word], limit: int, now: datetime | None = None, clear_count: int = 2
    ) -> list[Word]:
        due, _ = self.split_error_bank(words, clear_count, now)
        if not due or limit <= 0:
            return []
        return self.select_words(due, limit, now)

    def select_practice_plan_words(
        self, words: list[Word], limit: int, now: datetime | None = None, clear_count: int = 2
    ) -> list[Word]:
        """未训练词优先，其次错题巩固强化词。"""
        if limit <= 0:
            return []
        untrained_ranked = self.select_untrained_words(words, limit, now, clear_count)
        if len(untrained_ranked) >= limit:
            return untrained_ranked[:limit]
        remaining = limit - len(untrained_ranked)
        error_ranked = self.select_error_words(words, remaining, now, clear_count)
        return untrained_ranked + error_ranked

    def select_words(
        self, words: list[Word], limit: int, now: datetime | None = None
    ) -> list[Word]:
        now = now or datetime.now(timezone.utc)
        ranked = sorted(
            words,
            key=lambda word: (
                self.selection_score(word, now),
                -word.id,
            ),
            reverse=True,
        )
        return ranked[:limit]


adaptive_review_service = AdaptiveReviewService()