john
e76fa586f1
Ship dual-track learning (daily accumulation vs textbook),沪教/商务词书 APIs and UI, native iOS wrapper with bundled H5, and production book import on deploy. Co-authored-by: Cursor <cursoragent@cursor.com>
365 lines
12 KiB
Python
365 lines
12 KiB
Python
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()
|