tkmind/wordloop
1
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add spell practice, memory analytics, auth persistence, and word library UX.

Browse Source 
Includes per-word training stats and curves, quiz session auto-save, remember-login,
paginated word list with floating page arrows, and Obsidian-style relationship graph baseline.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
John
2026-06-04 14:57:39 -07:00
parent bd7635986a
commit 68c5efe573
30 changed files with 2560 additions and 63 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/services/memory_visual_service.py
+355
View File
@@ -0,0 +1,355 @@
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) -> dict:
words = db.query(Word).filter(Word.user_id == user.id).all()
records = (
db.query(QuizRecord)
.filter(QuizRecord.user_id == user.id)
.order_by(QuizRecord.created_at.asc())
.all()
)
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()