File size: 2,004 Bytes
4577459 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 | """Study streak and progress analytics."""
from __future__ import annotations
from datetime import date, datetime, timezone
from typing import Any
from plane_mode_scholar.storage.sqlite_store import SQLiteStore
def _parse_date(iso: str) -> datetime | None:
if not iso:
return None
try:
dt = datetime.fromisoformat(iso.replace("Z", "+00:00"))
if dt.tzinfo is None:
dt = dt.replace(tzinfo=timezone.utc)
return dt
except ValueError:
return None
def compute_streak(sessions: list[dict[str, Any]]) -> int:
"""Count consecutive calendar days with completed sessions (UTC)."""
days: set[str] = set()
for s in sessions:
if s.get("status") != "completed":
continue
dt = _parse_date(str(s.get("ended_at") or s.get("started_at", "")))
if dt:
days.add(dt.date().isoformat())
if not days:
return 0
today = datetime.now(timezone.utc).date()
cursor = today
streak = 0
while cursor.isoformat() in days:
streak += 1
cursor = date.fromordinal(cursor.toordinal() - 1)
return streak
def compute_progress_stats(
store: SQLiteStore,
user_id: str,
pack_id: str | None = None,
) -> dict[str, Any]:
sessions = [
s.to_dict()
for s in store.list_sessions(pack_id=pack_id, user_id=user_id, limit=50)
]
completed = [s for s in sessions if s.get("status") == "completed"]
active_memories = store.list_memories(pack_id=pack_id, user_id=user_id, status="active")
quiz_correct = sum(
1 for m in active_memories if "quiz" in m.tags and m.type.value == "mastery_estimate"
)
return {
"streak_days": compute_streak(sessions),
"sessions_total": len(sessions),
"sessions_completed": len(completed),
"active_memories": len(active_memories),
"last_session_date": (
str(completed[0].get("ended_at", ""))[:10] if completed else None
),
}
|