305 lines
12 KiB
Python
305 lines
12 KiB
Python
"""
|
|
SuperMemo-2 (SM-2) spaced repetition system (SRS) calculation logic,
|
|
quiz generation, and activity heatmap / streak analysis.
|
|
"""
|
|
from datetime import datetime, timedelta, timezone, date
|
|
from typing import Dict, Any, Optional, List
|
|
import random
|
|
|
|
# Quality rating constants
|
|
RATING_AGAIN = 1
|
|
RATING_HARD = 2
|
|
RATING_GOOD = 3
|
|
RATING_EASY = 4
|
|
|
|
MIN_EASE = 1.3
|
|
MAX_EASE = 3.0
|
|
DEFAULT_EASE = 2.5
|
|
MASTERED_INTERVAL_THRESHOLD = 21
|
|
|
|
FALLBACK_DISTRACTORS = [
|
|
("concurrency", "одновременность, параллелизм"),
|
|
("throughput", "пропускная способность"),
|
|
("bottleneck", "узкое место"),
|
|
("redundancy", "избыточность, резервирование"),
|
|
("resilience", "устойчивость, отказоустойчивость"),
|
|
("deprecated", "устаревший, не рекомендуемый"),
|
|
("mitigate", "смягчать, уменьшать последствия"),
|
|
("seamless", "бесшовный, плавный"),
|
|
("leverage", "использовать с выгодой"),
|
|
("tweak", "подправить, тонко настроить"),
|
|
("scaffold", "каркас, шаблон структуры"),
|
|
("boilerplate", "шаблонный код"),
|
|
("payload", "полезная нагрузка"),
|
|
("latency", "задержка, пинг"),
|
|
("invoke", "вызывать (функцию/метод)"),
|
|
("retrieve", "извлекать, получать данные"),
|
|
("subtle", "тонкий, едва уловимый"),
|
|
("comprehensive", "всесторонний, исчерпывающий"),
|
|
("reluctant", "неохотный, нежелающий"),
|
|
("perseverance", "настойчивость, упорство"),
|
|
("serendipity", "счастливая случайность"),
|
|
("ubiquitous", "вездесущий, повсеместный"),
|
|
("indispensable", "незаменимый"),
|
|
("arbitrary", "произвольный"),
|
|
("coherent", "связный, последовательный"),
|
|
("feasible", "осуществимый, выполнимый"),
|
|
("tangible", "осязаемый, ощутимый"),
|
|
("robust", "надежный, устойчивый к сбоям"),
|
|
("versatile", "универсальный, гибкий"),
|
|
("streamline", "оптимизировать, упрощать"),
|
|
]
|
|
|
|
|
|
def calculate_next_review(interval_days: int, base_time: Optional[datetime] = None) -> str:
|
|
"""Calculates the next review timestamp formatted as 'YYYY-MM-DD HH:MM:SS' (UTC)."""
|
|
if base_time is None:
|
|
base_time = datetime.now(timezone.utc)
|
|
next_time = base_time + timedelta(days=interval_days)
|
|
return next_time.strftime("%Y-%m-%d %H:%M:%S")
|
|
|
|
|
|
def calculate_sm2(
|
|
rating: int,
|
|
reps: Optional[int] = 0,
|
|
interval: Optional[int] = 0,
|
|
ease: Optional[float] = 2.5,
|
|
state: Optional[str] = "new",
|
|
base_time: Optional[datetime] = None,
|
|
error_count: Optional[int] = 0,
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Processes a card review rating using SM-2 algorithm:
|
|
- 1: Again (reps=0, interval=1, ease = max(1.3, ease - 0.2), state='learning')
|
|
- 2: Hard (reps=reps+1, interval=max(1, round(interval * 1.2)), ease = max(1.3, ease - 0.15), state='learning' if reps < 3 else 'review')
|
|
- 3: Good (reps=reps+1, interval= 1 if reps==1 else (6 if reps==2 else round(interval * ease)), state='review' if interval < 21 else 'mastered')
|
|
- 4: Easy (reps=reps+1, interval= 4 if reps==1 else round(interval * ease * 1.3), ease = min(3.0, ease + 0.15), state='review' if interval < 21 else 'mastered')
|
|
"""
|
|
if rating not in (RATING_AGAIN, RATING_HARD, RATING_GOOD, RATING_EASY):
|
|
raise ValueError(f"Invalid rating {rating}: must be 1 (Again), 2 (Hard), 3 (Good), or 4 (Easy)")
|
|
|
|
reps = reps if reps is not None else 0
|
|
interval = interval if interval is not None else 0
|
|
ease = ease if ease is not None else DEFAULT_EASE
|
|
state = state or "new"
|
|
error_count = error_count if error_count is not None else 0
|
|
|
|
if base_time is None:
|
|
base_time = datetime.now(timezone.utc)
|
|
last_review_str = base_time.strftime("%Y-%m-%d %H:%M:%S")
|
|
|
|
new_error_count = error_count
|
|
if rating == RATING_AGAIN:
|
|
new_reps = 0
|
|
new_interval = 1
|
|
new_ease = max(MIN_EASE, ease - 0.2)
|
|
new_state = "learning"
|
|
new_error_count += 1
|
|
|
|
elif rating == RATING_HARD:
|
|
new_reps = reps + 1
|
|
new_interval = max(1, round(interval * 1.2))
|
|
new_ease = max(MIN_EASE, ease - 0.15)
|
|
new_state = "learning" if new_reps < 3 else "review"
|
|
|
|
elif rating == RATING_GOOD:
|
|
new_reps = reps + 1
|
|
if new_reps == 1:
|
|
new_interval = 1
|
|
elif new_reps == 2:
|
|
new_interval = 6
|
|
else:
|
|
new_interval = round(interval * ease)
|
|
new_ease = ease
|
|
new_state = "review" if new_interval < MASTERED_INTERVAL_THRESHOLD else "mastered"
|
|
|
|
elif rating == RATING_EASY:
|
|
new_reps = reps + 1
|
|
if new_reps == 1:
|
|
new_interval = 4
|
|
else:
|
|
new_interval = round(interval * ease * 1.3)
|
|
new_ease = min(MAX_EASE, ease + 0.15)
|
|
new_state = "review" if new_interval < MASTERED_INTERVAL_THRESHOLD else "mastered"
|
|
|
|
new_ease = round(new_ease, 2)
|
|
next_review_str = calculate_next_review(new_interval, base_time)
|
|
|
|
return {
|
|
"srs_reps": new_reps,
|
|
"srs_interval": new_interval,
|
|
"srs_ease": new_ease,
|
|
"srs_state": new_state,
|
|
"srs_last_review": last_review_str,
|
|
"srs_next_review": next_review_str,
|
|
"error_count": new_error_count,
|
|
}
|
|
|
|
|
|
def generate_quiz_options(target_card: Dict[str, Any], all_cards: List[Dict[str, Any]], count: int = 4) -> List[Dict[str, Any]]:
|
|
"""
|
|
Generates 4 multiple choice options for a target card (1 correct + 3 distractors).
|
|
Returns list of options with text and is_correct flag, randomly shuffled.
|
|
"""
|
|
target_id = target_card.get("id")
|
|
correct_text = target_card.get("translation", "")
|
|
|
|
# Gather potential distractors from user's other cards
|
|
candidate_distractors = [
|
|
c.get("translation", "").strip()
|
|
for c in all_cards
|
|
if c.get("id") != target_id and c.get("translation", "").strip() and c.get("translation", "").strip() != correct_text
|
|
]
|
|
|
|
# Deduplicate candidates
|
|
candidate_distractors = list(dict.fromkeys(candidate_distractors))
|
|
random.shuffle(candidate_distractors)
|
|
|
|
selected_distractors: List[str] = []
|
|
for d in candidate_distractors:
|
|
if len(selected_distractors) >= (count - 1):
|
|
break
|
|
selected_distractors.append(d)
|
|
|
|
# If still not enough, take from fallback dictionary
|
|
if len(selected_distractors) < (count - 1):
|
|
for _, ru in FALLBACK_DISTRACTORS:
|
|
if ru != correct_text and ru not in selected_distractors:
|
|
selected_distractors.append(ru)
|
|
if len(selected_distractors) >= (count - 1):
|
|
break
|
|
|
|
options = [{"text": correct_text, "is_correct": True}]
|
|
for d in selected_distractors:
|
|
options.append({"text": d, "is_correct": False})
|
|
|
|
random.shuffle(options)
|
|
return options
|
|
|
|
|
|
def build_activity_heatmap(review_dates: Dict[str, int], days: int = 180) -> Dict[str, Any]:
|
|
"""
|
|
Constructs a daily contribution heatmap (like GitHub) for the past `days`.
|
|
Returns daily entries with date, count, and level (0: none, 1: 1 review, 2: 2-3 reviews, 3: 4+ reviews).
|
|
Also computes current streak and longest streak.
|
|
"""
|
|
today = date.today()
|
|
start_date = today - timedelta(days=days - 1)
|
|
|
|
heatmap_days = []
|
|
current_date = start_date
|
|
|
|
while current_date <= today:
|
|
d_str = current_date.strftime("%Y-%m-%d")
|
|
count = review_dates.get(d_str, 0)
|
|
if count == 0:
|
|
level = 0
|
|
elif count == 1:
|
|
level = 1
|
|
elif count <= 3:
|
|
level = 2
|
|
else:
|
|
level = 3
|
|
|
|
heatmap_days.append({
|
|
"date": d_str,
|
|
"count": count,
|
|
"level": level,
|
|
"day_of_week": current_date.weekday(), # 0=Mon, 6=Sun
|
|
})
|
|
current_date += timedelta(days=1)
|
|
|
|
# Compute streaks
|
|
current_streak = 0
|
|
check_day = today
|
|
# Check today
|
|
if review_dates.get(check_day.strftime("%Y-%m-%d"), 0) > 0:
|
|
current_streak += 1
|
|
check_day -= timedelta(days=1)
|
|
else:
|
|
# If no activity today yet, check yesterday to preserve ongoing streak
|
|
yesterday = today - timedelta(days=1)
|
|
if review_dates.get(yesterday.strftime("%Y-%m-%d"), 0) > 0:
|
|
check_day = yesterday
|
|
|
|
while check_day >= start_date:
|
|
if review_dates.get(check_day.strftime("%Y-%m-%d"), 0) > 0:
|
|
if check_day != today: # don't double-count today
|
|
current_streak += 1
|
|
check_day -= timedelta(days=1)
|
|
else:
|
|
break
|
|
|
|
# Longest streak calculation
|
|
longest_streak = 0
|
|
temp_streak = 0
|
|
for day in heatmap_days:
|
|
if day["count"] > 0:
|
|
temp_streak += 1
|
|
if temp_streak > longest_streak:
|
|
longest_streak = temp_streak
|
|
else:
|
|
temp_streak = 0
|
|
|
|
return {
|
|
"days": heatmap_days,
|
|
"total_active_days": sum(1 for d in heatmap_days if d["count"] > 0),
|
|
"total_reviews": sum(d["count"] for d in heatmap_days),
|
|
"current_streak": current_streak,
|
|
"longest_streak": max(longest_streak, current_streak),
|
|
}
|
|
|
|
|
|
def calculate_cefr_level(
|
|
mastered_count: int,
|
|
review_count: int,
|
|
retention_rate: float,
|
|
total_cards: int = 0
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Computes strict CEFR level based on active vocabulary weight and memory retention rate (PLAN.md 4.3).
|
|
Only words in Mastered (weight 1.0) and Review (weight 0.5) stages count, scaled by retention rate %.
|
|
"""
|
|
safe_retention = max(0.0, min(100.0, float(retention_rate)))
|
|
retention_factor = (safe_retention / 100.0) if safe_retention > 0 else 1.0
|
|
|
|
# Active weighted vocabulary: Mastered gives 1.0, Review gives 0.5
|
|
weighted_words = (mastered_count * 1.0) + (review_count * 0.5)
|
|
active_vocab = int(round(weighted_words * retention_factor))
|
|
|
|
# CEFR thresholds: (min_words, level, title, next_level, next_target)
|
|
brackets = [
|
|
(0, "A1", "Начальный словарный запас", "A2", 300),
|
|
(300, "A2", "Элементарный словарный запас", "B1", 800),
|
|
(800, "B1", "Уверенное понимание общих и рабочих тем", "B2", 1600),
|
|
(1600, "B2", "Свободное чтение тех. литературы и документации", "C1", 3000),
|
|
(3000, "C1", "Продвинутый профессиональный уровень", "C2", 5000),
|
|
]
|
|
|
|
current_bracket = brackets[0]
|
|
for b in brackets:
|
|
if active_vocab >= b[0]:
|
|
current_bracket = b
|
|
|
|
min_w, level, title, next_lvl, next_w = current_bracket
|
|
words_to_next = max(0, next_w - active_vocab)
|
|
range_span = next_w - min_w
|
|
current_progress = active_vocab - min_w
|
|
pct = int(min(100, max(0, (current_progress / range_span) * 100))) if range_span > 0 else 100
|
|
|
|
if words_to_next > 0:
|
|
message = f"До уровня {next_lvl} осталось закрепить {words_to_next} активных слов"
|
|
else:
|
|
message = f"Уровень {level} полностью освоен!"
|
|
|
|
return {
|
|
"level": level,
|
|
"title": title,
|
|
"active_vocab": active_vocab,
|
|
"next_level": next_lvl,
|
|
"words_to_next": words_to_next,
|
|
"progress_percent": pct,
|
|
"message": message,
|
|
"retention_rate": safe_retention,
|
|
}
|