import math from collections import defaultdict from datetime import datetime, timedelta, timezone from typing import Optional from sqlalchemy.orm import Session from models import QuizRecord, User, Word def parse_iso(s: str) -> datetime: s = s.replace("Z", "+00:00") try: return datetime.fromisoformat(s) except ValueError: return datetime.strptime(s[:19], "%Y-%m-%dT%H:%M:%S").replace(tzinfo=timezone.utc) def word_en(w: Word) -> str: return w.target_text if w.source_lang == "zh" else w.source_text def word_zh(w: Word) -> str: return w.source_text if w.source_lang == "zh" else w.target_text def stability_hours(word: Word) -> float: """记忆稳定性（小时），随练习增强。""" base = 24.0 streak = min(word.consecutive_correct_count, 6) mastery_factor = 1 + word.mastery_score / 200 return base * (1.6**streak) * mastery_factor def retention_percent(hours_since: float, stability_h: float) -> float: """艾宾浩斯型遗忘：R = 100 * e^(-t/S)""" s = max(stability_h, 6.0) return max(0.0, min(100.0, 100 * math.exp(-hours_since / s))) def mastery_at_time(word: Word, records: list[QuizRecord], at: datetime) -> float: correct = 0 wrong = 0 for r in records: if r.word_id != word.id: continue if parse_iso(r.created_at) > at: break if r.is_correct: correct += 1 else: wrong += 1 total = correct + wrong if total == 0: return float(word.mastery_score) if word.last_reviewed_at else 0.0 return round(correct / total * 100, 1) class MemoryVisualService: def get_visualization( self, db: Session, user: User, horizon_days: int = 30, book_id: int = 0 ) -> dict: words = ( db.query(Word) .filter(Word.user_id == user.id, Word.book_id == book_id, Word.status != "locked") .all() ) word_ids = {w.id for w in words} records = [ r for r in db.query(QuizRecord) .filter(QuizRecord.user_id == user.id) .order_by(QuizRecord.created_at.asc()) .all() if r.word_id in word_ids ] records_by_word: dict[int, list[QuizRecord]] = defaultdict(list) for r in records: records_by_word[r.word_id].append(r) now = datetime.now(timezone.utc) horizon_days = max(7, min(horizon_days, 60)) if not words: return { "curve_points": [], "words": [], "graph": {"nodes": [], "links": []}, } earliest = min(parse_iso(w.created_at) for w in words) span_days = max( 1, int((now - earliest).total_seconds() // 86400) + 1, ) horizon = min(horizon_days, span_days) curve_points = [] for d in range(horizon + 1): t = earliest + timedelta(days=d) forgetting_vals: list[float] = [] mastery_vals: list[float] = [] risk_vals: list[float] = [] for w in words: entered = parse_iso(w.created_at) if t < entered: continue hours_after_enter = (t - entered).total_seconds() / 3600 stab = stability_hours(w) wr = records_by_word.get(w.id, []) # 遗忘曲线：自加入词库起的记忆保留率 reviews_before = sum(1 for r in wr if parse_iso(r.created_at) <= t) boost = 1 + reviews_before * 0.12 forgetting_vals.append( retention_percent(hours_after_enter, stab * boost) ) # 熟练曲线：截至该日的掌握度 mastery_vals.append(mastery_at_time(w, wr, t)) # 可能遗忘：从该日视角若不再复习的预测保留率 last_at = entered if w.last_reviewed_at: lr = parse_iso(w.last_reviewed_at) if lr <= t: last_at = lr hours_since_review = (t - last_at).total_seconds() / 3600 risk_vals.append(retention_percent(hours_since_review, stab * 0.85)) if not forgetting_vals: continue curve_points.append( { "day_index": d, "date": t.strftime("%Y-%m-%d"), "forgetting": round(sum(forgetting_vals) / len(forgetting_vals), 1), "mastery": round(sum(mastery_vals) / len(mastery_vals), 1), "risk": round(sum(risk_vals) / len(risk_vals), 1), } ) # 今日起未来 14 天风险预测 future_risk = [] for fd in range(15): t = now + timedelta(days=fd) vals = [] for w in words: last_at = parse_iso(w.last_reviewed_at or w.created_at) hours = (t - last_at).total_seconds() / 3600 vals.append(retention_percent(hours, stability_hours(w))) future_risk.append( { "day_offset": fd, "date": t.strftime("%Y-%m-%d"), "risk": round(sum(vals) / len(vals), 1), } ) word_summaries = [] for w in words: stab_h = stability_hours(w) last_at = parse_iso(w.last_reviewed_at or w.created_at) hours_since = (now - last_at).total_seconds() / 3600 word_summaries.append( { "id": w.id, "en": word_en(w), "zh": word_zh(w), "status": w.status, "mastery_score": w.mastery_score, "correct_count": w.correct_count, "wrong_count": w.wrong_count, "train_count": w.train_count, "total_train_seconds": w.total_train_seconds, "entered_at": w.created_at, "retention_now": round(retention_percent(hours_since, stab_h), 1), "risk_7d": round( retention_percent(hours_since + 7 * 24, stab_h), 1 ), } ) graph = self._build_graph(words, records) return { "curve_points": curve_points, "future_risk": future_risk, "words": word_summaries, "graph": graph, } def _build_graph(self, words: list[Word], records: list[QuizRecord]) -> dict: nodes = [] id_set = set() for w in words: id_set.add(w.id) nodes.append( { "id": str(w.id), "label": word_en(w)[:16], "zh": word_zh(w)[:8], "status": w.status, "mastery": w.mastery_score, "entered_at": w.created_at, "size": 8 + min(w.correct_count + w.wrong_count, 20), } ) links = [] seen_edges: set[tuple[str, str]] = set() def add_link(a: int, b: int, kind: str, strength: float = 0.5) -> None: if a == b or a not in id_set or b not in id_set: return key = (str(min(a, b)), str(max(a, b))) if key in seen_edges: return seen_edges.add(key) links.append( { "source": str(a), "target": str(b), "kind": kind, "strength": strength, } ) # 同日练习关联 by_day: dict[str, list[int]] = defaultdict(list) for r in records: by_day[r.created_at[:10]].append(r.word_id) for ids in by_day.values(): unique = list(set(ids)) for i in range(len(unique)): for j in range(i + 1, len(unique)): add_link(unique[i], unique[j], "co_review", 0.7) # 相同学习状态 by_status: dict[str, list[int]] = defaultdict(list) for w in words: by_status[w.status].append(w.id) for ids in by_status.values(): for i in range(len(ids)): for j in range(i + 1, min(i + 4, len(ids))): # 限制边数量 add_link(ids[i], ids[j], "status", 0.35) # 词形相近（英文前缀 / 包含关系） en_map = {w.id: word_en(w).lower() for w in words} ids = list(en_map.keys()) for i in range(len(ids)): for j in range(i + 1, len(ids)): a, b = en_map[ids[i]], en_map[ids[j]] if len(a) >= 3 and len(b) >= 3: if a[:3] == b[:3] or a in b or b in a: add_link(ids[i], ids[j], "similar", 0.45) return {"nodes": nodes, "links": links[:120]} def build_word_curve_points(self, word: Word, records: list[QuizRecord]) -> list[dict]: entered = parse_iso(word.created_at) points: list[dict] = [ { "date": entered.strftime("%Y-%m-%d"), "datetime": word.created_at, "forgetting": 100.0, "mastery": 0.0, "risk": 100.0, "wrong_count": 0, "train_count": 0, "train_seconds": 0, "is_correct": None, } ] if not records: return points correct = 0 wrong = 0 total_sec = 0 last_review = entered for i, r in enumerate(records): t = parse_iso(r.created_at) if r.is_correct: correct += 1 else: wrong += 1 total_sec += r.duration_seconds or 0 reviews = i + 1 hours_enter = (t - entered).total_seconds() / 3600 stab = stability_hours(word) forgetting = retention_percent(hours_enter, stab * (1 + reviews * 0.12)) total = correct + wrong mastery = round(correct / total * 100, 1) if total else 0.0 hours_since = (t - last_review).total_seconds() / 3600 risk = retention_percent(hours_since, stab * 0.85) last_review = t points.append( { "date": t.strftime("%Y-%m-%d"), "datetime": r.created_at, "forgetting": round(forgetting, 1), "mastery": mastery, "risk": round(risk, 1), "wrong_count": wrong, "train_count": i + 1, "train_seconds": total_sec, "is_correct": bool(r.is_correct), } ) return points def build_word_future_risk(self, word: Word) -> list[dict]: now = datetime.now(timezone.utc) last_at = parse_iso(word.last_reviewed_at or word.created_at) hours_since = (now - last_at).total_seconds() / 3600 stab = stability_hours(word) future = [] for fd in range(15): t = now + timedelta(days=fd) hours = hours_since + fd * 24 future.append( { "day_offset": fd, "date": t.strftime("%Y-%m-%d"), "risk": round(retention_percent(hours, stab), 1), } ) return future def get_word_memory(self, db: Session, user: User, word_id: int) -> dict: word = db.query(Word).filter(Word.id == word_id, Word.user_id == user.id).first() if not word: from fastapi import HTTPException raise HTTPException(status_code=404, detail="单词不存在") records = ( db.query(QuizRecord) .filter(QuizRecord.user_id == user.id, QuizRecord.word_id == word.id) .order_by(QuizRecord.created_at.asc()) .all() ) return { "word_id": word.id, "en": word_en(word), "zh": word_zh(word), "correct_count": word.correct_count, "wrong_count": word.wrong_count, "train_count": word.train_count, "total_train_seconds": word.total_train_seconds, "curve_points": self.build_word_curve_points(word, records), "future_risk": self.build_word_future_risk(word), } memory_visual_service = MemoryVisualService()