| """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 |
| ), |
| } |
|
|
