Oyunlaştırma Motoru: Mimari ve Durum Makinesi
Öğrencilerin %80'i platformlarda çalışırken daha motive olduklarını söylüyor oyun öğelerini birleştirin. Eğitim Teknolojisindeki oyunlaştırma pazarı hızla büyüyecek CAGR %36'nın üzerinde 2032'ye kadar. Ancak oyunlaştırma arasındaki fark etkili ve yüzeysel olmak rastgele rozetler ve puanlar eklemekle ilgili değildir: sağlamlıkla ilgilidir arasındatemel mimari.
Kötü tasarlanmış bir oyunlaştırma motoru teknik borcu hızla biriktirir: mantık Sabit kodlanmış ilerleme, birçok kullanıcının tamamladığı noktalarda yarış durumu aynı anda etkinlikler, binlerce kişi için darboğaz haline gelen liderlik tabloları durum yönetimindeki hatalar nedeniyle rozetler birkaç kez verildi. Bu sorunlar teorik değiller; oyunlaştırmayı ekleyen her platformun ortak patolojileridirler birinci sınıf bir özellik olarak tasarlamak yerine sonradan akla gelen bir fikir olarak.
Bu yazıda bir inşa edeceğiz Oyunlaştırma Motoru sağlam ve ölçeklenebilir açık durum makineleri, izlenebilirlik için olay kaynağı, lider tabloları için Redis ile İçerik oluşturucuların tanımlamalarına olanak tanıyan gerçek zamanlı ve yapılandırılabilir bir kural sistemi koda dokunmadan yeni mekanikler.
Bu Makalede Neler Öğreneceksiniz?
- Ölçeklenebilir bir oyunlaştırma motorunun olay odaklı mimarisi
- Öğrenci ilerlemesini yönetmek için durum makinesi
- Yapılandırılabilir kurallara sahip rozet sistemi (kodlama olmadan)
- Redis Sıralanmış Kümeler ile gerçek zamanlı liderlik tablosu
- Seri takibi ve zaman ödül sistemi
- Seviyeler ve ilerleme eğrileriyle XP (deneyim puanları)
- Anti-örüntüler: yarış koşullarından ve çift atamadan nasıl kaçınılır
- WebSocket ile başarılar için anında bildirimler
1. Tasarım İlkeleri: İçsel ve Dışsal Oyunlaştırma
Kod yazmadan önce motivasyon ile motivasyon arasındaki farkı anlamak önemlidir. dışsal (puanlar, rozetler, sıralamalar - dış motivasyon unsurları) e içsel (merak, yeterlilik duygusu, özerklik). Başarılı eğitsel oyunlaştırma, dışsal motive edici unsurları bir fırlatma rampası olarak kullanır bir ikame olarak değil, içsel motivasyonu geliştirmek.
2025 yılında eğitim için en etkili oyunlaştırma mekanizmaları şunlardır: rüzgâr gibi geçmek (ardışık çalışma günlerinin dizileri), ustalık ilerlemesi (ustalığa doğru ilerlemenin görselleştirilmesi), sosyal öğrenme (benzer seviyedeki akranlar arasındaki sıralamalar), e anlamlı seçimler (farklı ödüllere sahip alternatif yollar). En az etkili (ve potansiyel olarak zararlı) mekanikler küresel liderlik tablolarıdır yeni öğrencilerin her zaman en iyi performans gösterenleri ulaşılmaz olarak gördüğü yer.
Oyunlaştırma Motorunun Bileşenleri
| Bileşen | Sorumluluk | Teknoloji |
|---|---|---|
| Etkinlik Otobüsü | Öğrenme etkinliklerini toplar | Kafka/Redis Akışları |
| Kural Motoru | Yapılandırılabilir kuralları değerlendirin | Python + JSON kuralları |
| Durum Makinesi | Öğrenci ilerlemesini yönetir | Özel Python FSM |
| XP Hesaplayıcı | Deneyim puanlarını hesaplayın | Python + PostgreSQL |
| Rozet Yöneticisi | Ödül önemsiz rozeti | PostgreSQL + Redis önbelleği |
| Skor tabloları | Gerçek zamanlı sıralamalar | Redis Sıralanmış Setler |
| Seri Takipçisi | Çalışma dizilerini izleyin | Redis + PostgreSQL |
| Bildirim Merkezi | Başarıyı gerçek zamanlı olarak ilerletin | WebSocket / SSE |
2. Öğrenci İlerlemesi İçin Durum Makinesi
Durum makinesi oyunlaştırma motorunun kalbidir. Her öğrenci bir arada durum bu onun katılım ve ilerleme düzeyini yansıtır. Durumlar arasındaki geçişler tepki olarak gerçekleşir olaylar (tamamlama ders, doğru cevap, ardışık giriş). Bu durum makinesini açık hale getirin yeniden üretilmesi zor hatalara neden olan "örtük durumlar" sorununu ortadan kaldırır.
# gamification/state_machine.py
from enum import Enum, auto
from dataclasses import dataclass, field
from typing import Dict, List, Callable, Optional
from datetime import datetime
class LearnerState(Enum):
"""Stati possibili di un learner nella piattaforma."""
ONBOARDING = auto() # Primo accesso, < 3 corsi completati
ACTIVE = auto() # Studente attivo, streak in corso
AT_RISK = auto() # Nessuna attivita da 7-13 giorni
DORMANT = auto() # Inattivo da 14+ giorni
MASTERY = auto() # Ha completato un corso con punteggio >= 90%
MENTOR = auto() # Ha aiutato almeno 5 altri studenti
class LearnerEvent(Enum):
"""Eventi che causano transizioni di stato."""
LESSON_COMPLETED = "lesson_completed"
QUIZ_PASSED = "quiz_passed"
QUIZ_FAILED = "quiz_failed"
COURSE_COMPLETED = "course_completed"
PEER_HELPED = "peer_helped"
DAILY_LOGIN = "daily_login"
STREAK_BROKEN = "streak_broken"
INACTIVE_7_DAYS = "inactive_7_days"
INACTIVE_14_DAYS = "inactive_14_days"
REACTIVATED = "reactivated"
@dataclass
class StateTransition:
from_state: LearnerState
event: LearnerEvent
to_state: LearnerState
conditions: List[Callable] = field(default_factory=list)
actions: List[Callable] = field(default_factory=list)
@dataclass
class LearnerContext:
student_id: str
current_state: LearnerState
xp: int = 0
streak_days: int = 0
courses_completed: int = 0
peers_helped: int = 0
last_activity: Optional[datetime] = None
metadata: Dict = field(default_factory=dict)
class LearnerStateMachine:
"""
State machine per la progressione del learner.
Tutte le transizioni sono esplicite e testabili.
"""
TRANSITIONS: List[StateTransition] = [
# ONBOARDING -> ACTIVE: ha completato la prima lezione
StateTransition(
from_state=LearnerState.ONBOARDING,
event=LearnerEvent.LESSON_COMPLETED,
to_state=LearnerState.ACTIVE,
),
# ACTIVE -> AT_RISK: 7 giorni di inattivita
StateTransition(
from_state=LearnerState.ACTIVE,
event=LearnerEvent.INACTIVE_7_DAYS,
to_state=LearnerState.AT_RISK,
),
# AT_RISK -> DORMANT: altri 7 giorni senza attivita
StateTransition(
from_state=LearnerState.AT_RISK,
event=LearnerEvent.INACTIVE_14_DAYS,
to_state=LearnerState.DORMANT,
),
# DORMANT/AT_RISK -> ACTIVE: torna ad essere attivo
StateTransition(
from_state=LearnerState.AT_RISK,
event=LearnerEvent.REACTIVATED,
to_state=LearnerState.ACTIVE,
),
StateTransition(
from_state=LearnerState.DORMANT,
event=LearnerEvent.REACTIVATED,
to_state=LearnerState.ACTIVE,
),
# ACTIVE -> MASTERY: completa corso con voto alto
StateTransition(
from_state=LearnerState.ACTIVE,
event=LearnerEvent.COURSE_COMPLETED,
to_state=LearnerState.MASTERY,
conditions=[lambda ctx: ctx.metadata.get("score", 0) >= 90],
),
# MASTERY -> MENTOR: aiuta abbastanza peer
StateTransition(
from_state=LearnerState.MASTERY,
event=LearnerEvent.PEER_HELPED,
to_state=LearnerState.MENTOR,
conditions=[lambda ctx: ctx.peers_helped >= 5],
),
]
def __init__(self, event_publisher):
self.event_publisher = event_publisher
# Indice per lookup rapido O(1)
self._transition_index: Dict[tuple, StateTransition] = {
(t.from_state, t.event): t
for t in self.TRANSITIONS
}
def process_event(
self,
context: LearnerContext,
event: LearnerEvent,
event_data: Dict = None,
) -> LearnerContext:
"""
Processa un evento e restituisce un NUOVO context (immutabile).
Non modifica mai il context in input.
"""
event_data = event_data or {}
key = (context.current_state, event)
transition = self._transition_index.get(key)
if not transition:
# Nessuna transizione definita: evento ignorato
return context
# Aggiorna il metadata del context con i dati dell'evento
updated_metadata = {**context.metadata, **event_data}
temp_context = LearnerContext(
**{
**context.__dict__,
"metadata": updated_metadata,
}
)
# Verifica tutte le condizioni
if not all(cond(temp_context) for cond in transition.conditions):
return context # Condizioni non soddisfatte, nessuna transizione
# Crea nuovo context con nuovo stato
new_context = LearnerContext(
**{
**temp_context.__dict__,
"current_state": transition.to_state,
"last_activity": datetime.utcnow(),
}
)
# Pubblica evento di transizione
self.event_publisher.publish({
"type": "state_transition",
"student_id": context.student_id,
"from_state": context.current_state.name,
"to_state": transition.to_state.name,
"trigger_event": event.value,
"timestamp": datetime.utcnow().isoformat(),
})
return new_context3. İlerleme Eğrileriyle XP ve Seviye Sistemi
Deneyim Puanı (XP) sistemi ilerleme hızını dengelemelidir: çok hızlı ve öğrenciler birkaç gün içinde maksimuma çıkıyor ve sıkılıyorlar, çok yavaşlar ve cesaretleri kırılır. Bir tane kullanalım logaritmik eğri ilk birkaç seviyede (hızlı ve tatmin edici ilerleme) giderek dikleşiyor yüksek seviyelere (önemli hedeflere) doğru.
# gamification/xp_system.py
import math
from dataclasses import dataclass
from typing import Dict, List
from enum import Enum
class ActivityType(Enum):
LESSON_COMPLETED = "lesson"
QUIZ_PASSED = "quiz_pass"
QUIZ_PERFECT = "quiz_perfect" # 100% corretto
COURSE_COMPLETED = "course"
STREAK_BONUS = "streak"
PEER_REVIEW = "peer_review"
HELP_PEER = "help_peer"
FIRST_LOGIN_DAY = "first_login"
# XP base per ogni tipo di attivita
XP_BASE_REWARDS: Dict[str, int] = {
ActivityType.LESSON_COMPLETED.value: 50,
ActivityType.QUIZ_PASSED.value: 100,
ActivityType.QUIZ_PERFECT.value: 200,
ActivityType.COURSE_COMPLETED.value: 500,
ActivityType.STREAK_BONUS.value: 25,
ActivityType.PEER_REVIEW.value: 75,
ActivityType.HELP_PEER.value: 150,
ActivityType.FIRST_LOGIN_DAY.value: 10,
}
@dataclass
class XPGrant:
student_id: str
activity_type: str
base_xp: int
multiplier: float
final_xp: int
new_total_xp: int
new_level: int
level_up: bool
reason: str
class XPCalculator:
"""
Calcola XP con moltiplicatori basati su streak, difficolta e tempo.
Usa la formula logaritmica per i livelli.
"""
def calculate_level(self, total_xp: int) -> int:
"""
Livello = floor(log2(total_xp / 100)) + 1
Esempi: 100 XP = Lv1, 200 XP = Lv2, 400 XP = Lv3, 800 XP = Lv4...
"""
if total_xp < 100:
return 1
return min(int(math.log2(total_xp / 100)) + 1, 100)
def xp_for_level(self, level: int) -> int:
"""XP necessari per raggiungere il livello."""
if level <= 1:
return 0
return 100 * (2 ** (level - 1))
def calculate_multiplier(
self,
streak_days: int,
difficulty: int, # 1-5
time_of_day_bonus: bool = False,
) -> float:
"""
Calcola il moltiplicatore XP.
- Streak: +5% per ogni giorno consecutivo (max +50%)
- Difficolta: +10% per ogni livello sopra il baseline (2)
- Bonus orario: +20% per studio nelle ore ottimali (8-12, 14-18)
"""
streak_bonus = min(streak_days * 0.05, 0.5) # Max 50%
difficulty_bonus = max(0, (difficulty - 2) * 0.10) # Baseline difficolta = 2
time_bonus = 0.20 if time_of_day_bonus else 0.0
return 1.0 + streak_bonus + difficulty_bonus + time_bonus
def grant_xp(
self,
student_id: str,
activity: ActivityType,
current_xp: int,
streak_days: int,
difficulty: int = 2,
time_of_day_bonus: bool = False,
) -> XPGrant:
base_xp = XP_BASE_REWARDS.get(activity.value, 10)
multiplier = self.calculate_multiplier(streak_days, difficulty, time_of_day_bonus)
final_xp = int(base_xp * multiplier)
new_total = current_xp + final_xp
old_level = self.calculate_level(current_xp)
new_level = self.calculate_level(new_total)
return XPGrant(
student_id=student_id,
activity_type=activity.value,
base_xp=base_xp,
multiplier=multiplier,
final_xp=final_xp,
new_total_xp=new_total,
new_level=new_level,
level_up=new_level > old_level,
reason=f"{activity.value} x{multiplier:.2f} (streak:{streak_days}d, diff:{difficulty}/5)",
)4. Kimliksizlik ve Yapılandırılabilir Kurallara Sahip Rozet Sistemi
Yaka kartı sistemlerinde en sık karşılaşılan sorun çift atamadır: aynı yaka kartı yarış koşulları veya başarısız olayların yeniden denenmesi nedeniyle birden çok kez verilir. Çözüm veiktidarsızlık: her rozet atama işlemi birden çok kez gerçekleştirilebilir ve her zaman aynı sonuç elde edilir. Bir kullanıyoruz kısıtlama BENZERSİZ PostgreSQL'de veritabanı düzeyinde güvence olarak.
# gamification/badge_manager.py
import json
from dataclasses import dataclass
from typing import List, Dict, Any, Optional
from datetime import datetime
from enum import Enum
class BadgeTier(Enum):
BRONZE = "bronze"
SILVER = "silver"
GOLD = "gold"
PLATINUM = "platinum"
LEGENDARY = "legendary"
@dataclass
class BadgeDefinition:
badge_id: str
name: str
description: str
tier: BadgeTier
icon_url: str
# Regola come JSON (no hardcoding!)
rule: Dict[str, Any]
# Esempi di regole configurabili
BADGE_CATALOG: List[BadgeDefinition] = [
BadgeDefinition(
badge_id="first_lesson",
name="Primo Passo",
description="Hai completato la tua prima lezione",
tier=BadgeTier.BRONZE,
icon_url="/badges/first-lesson.svg",
rule={"event": "lesson_completed", "count": 1},
),
BadgeDefinition(
badge_id="week_streak",
name="Settimana di Fuoco",
description="7 giorni di studio consecutivi",
tier=BadgeTier.SILVER,
icon_url="/badges/week-streak.svg",
rule={"metric": "streak_days", "gte": 7},
),
BadgeDefinition(
badge_id="perfect_quiz",
name="Perfezionista",
description="Quiz completato con 100% di risposte corrette",
tier=BadgeTier.GOLD,
icon_url="/badges/perfect-quiz.svg",
rule={"event": "quiz_completed", "score": 100},
),
BadgeDefinition(
badge_id="mentor",
name="Mentore della Comunita",
description="Hai aiutato 10 compagni di corso",
tier=BadgeTier.PLATINUM,
icon_url="/badges/mentor.svg",
rule={"metric": "peers_helped", "gte": 10},
),
]
class BadgeManager:
def __init__(self, db, redis_client):
self.db = db
self.redis = redis_client
self._catalog: Dict[str, BadgeDefinition] = {b.badge_id: b for b in BADGE_CATALOG}
async def evaluate_and_award(
self,
student_id: str,
event: str,
student_metrics: Dict[str, Any],
) -> List[str]:
"""
Valuta tutti i badge applicabili e li assegna (idempotente).
Ritorna la lista di badge_id appena assegnati.
"""
newly_awarded = []
for badge_id, badge_def in self._catalog.items():
if await self._is_already_awarded(student_id, badge_id):
continue
if self._evaluate_rule(badge_def.rule, event, student_metrics):
awarded = await self._award_badge_idempotent(student_id, badge_id)
if awarded:
newly_awarded.append(badge_id)
return newly_awarded
async def _award_badge_idempotent(self, student_id: str, badge_id: str) -> bool:
"""
Assegna il badge usando INSERT ... ON CONFLICT DO NOTHING.
Garantisce idempotenza a livello DB.
Ritorna True se il badge e stato effettivamente assegnato ora.
"""
result = await self.db.execute(
"""INSERT INTO student_badges (student_id, badge_id, awarded_at)
VALUES (:sid, :bid, :ts)
ON CONFLICT (student_id, badge_id) DO NOTHING
RETURNING badge_id""",
{"sid": student_id, "bid": badge_id, "ts": datetime.utcnow()},
)
rows = result.fetchall()
if rows:
# Invalida cache badge studente
await self.redis.delete(f"student:badges:{student_id}")
await self.db.commit()
return True
return False # Già esisteva
async def _is_already_awarded(self, student_id: str, badge_id: str) -> bool:
"""Controlla cache Redis prima di interrogare il DB."""
cache_key = f"student:badges:{student_id}"
cached = await self.redis.get(cache_key)
if cached:
badges = json.loads(cached)
return badge_id in badges
# Cache miss: interroga DB
result = await self.db.execute(
"SELECT badge_id FROM student_badges WHERE student_id = :sid",
{"sid": student_id},
)
badges = [row[0] for row in result.fetchall()]
await self.redis.setex(cache_key, 300, json.dumps(badges)) # TTL 5 minuti
return badge_id in badges
def _evaluate_rule(
self,
rule: Dict,
current_event: str,
metrics: Dict[str, Any],
) -> bool:
"""Valuta una regola badge configurabile."""
# Regola basata su evento
if "event" in rule:
if rule["event"] != current_event:
return False
if "count" in rule:
return metrics.get(f"event_count_{rule['event']}", 0) >= rule["count"]
if "score" in rule:
return metrics.get("last_score", 0) >= rule["score"]
return True
# Regola basata su metrica
if "metric" in rule:
metric_val = metrics.get(rule["metric"], 0)
if "gte" in rule:
return metric_val >= rule["gte"]
if "eq" in rule:
return metric_val == rule["eq"]
if "lte" in rule:
return metric_val <= rule["lte"]
return False5. Redis Sıralı Setlerle Gerçek Zamanlı Skor Tablosu
Küresel liderlik tabloları iki ucu keskin bir kılıçtır: en iyi performans gösterenleri motive eder ama sıralamanın en altındakilerin moralini bozuyorlar. Modern çözüm ise bölümlenmiş skor tablosu: Her öğrenci 50 kullanıcıyla yarışır puanına daha yakın olması rekabeti her zaman anlamlı kılıyor. Tüm işlemler için O(log N) karmaşıklığına sahip Redis Sıralanmış Kümeleri bir araçtır gerçek zamanlı sıralamalar için mükemmeldir.
# gamification/leaderboard.py
import json
from typing import List, Dict, Optional
from datetime import datetime, date
import redis.asyncio as redis
@dataclass
class LeaderboardEntry:
rank: int
student_id: str
display_name: str
xp: int
level: int
avatar_url: str
is_current_user: bool = False
class LeaderboardManager:
"""
Leaderboard multi-scope con Redis Sorted Sets.
Scope: global, course-specific, weekly, peer-group.
"""
def __init__(self, redis_client: redis.Redis):
self.redis = redis_client
def _key(self, scope: str, period: str = "all-time") -> str:
"""Genera chiave Redis per uno scope e periodo."""
if period == "weekly":
week = date.today().isocalendar()[1]
year = date.today().year
return f"leaderboard:{scope}:{year}:w{week}"
return f"leaderboard:{scope}:{period}"
async def update_score(
self,
student_id: str,
xp_delta: int,
scopes: List[str],
) -> None:
"""
Aggiorna il punteggio in tutti gli scope rilevanti.
Usa pipeline Redis per operazioni atomiche.
"""
async with self.redis.pipeline(transaction=True) as pipe:
for scope in scopes:
for period in ["all-time", "weekly"]:
key = self._key(scope, period)
pipe.zincrby(key, xp_delta, student_id)
# Imposta TTL per leaderboard settimanale
if period == "weekly":
pipe.expire(key, 7 * 24 * 3600)
await pipe.execute()
async def get_leaderboard(
self,
scope: str,
page: int = 1,
page_size: int = 10,
period: str = "all-time",
) -> List[Dict]:
"""Recupera la leaderboard paginata per uno scope."""
key = self._key(scope, period)
start = (page - 1) * page_size
end = start + page_size - 1
# ZREVRANGE: ordine decrescente (punteggio più alto prima)
entries = await self.redis.zrevrange(key, start, end, withscores=True)
result = []
for rank_offset, (student_id_bytes, score) in enumerate(entries):
student_id = student_id_bytes.decode() if isinstance(student_id_bytes, bytes) else student_id_bytes
result.append({
"rank": start + rank_offset + 1,
"student_id": student_id,
"xp": int(score),
})
return result
async def get_peer_leaderboard(
self,
student_id: str,
scope: str,
window: int = 25,
period: str = "all-time",
) -> List[Dict]:
"""
Leaderboard contestualizzata: mostra i 25 utenti sopra e sotto il corrente studente.
"""
key = self._key(scope, period)
# Posizione attuale dello studente (rank 0-indexed dal basso)
rank_from_bottom = await self.redis.zrank(key, student_id)
if rank_from_bottom is None:
return []
total = await self.redis.zcard(key)
rank_from_top = total - rank_from_bottom - 1
# Finestra: window utenti sopra e sotto
start = max(0, rank_from_top - window)
end = min(total - 1, rank_from_top + window)
entries = await self.redis.zrevrange(key, start, end, withscores=True)
result = []
for i, (sid_bytes, score) in enumerate(entries):
sid = sid_bytes.decode() if isinstance(sid_bytes, bytes) else sid_bytes
result.append({
"rank": start + i + 1,
"student_id": sid,
"xp": int(score),
"is_current_user": sid == student_id,
})
return result
async def get_student_rank(
self,
student_id: str,
scope: str,
period: str = "all-time",
) -> Optional[int]:
"""Recupera il rank assoluto di uno studente."""
key = self._key(scope, period)
total = await self.redis.zcard(key)
rank_from_bottom = await self.redis.zrank(key, student_id)
if rank_from_bottom is None:
return None
return total - rank_from_bottom # Rank dal top (1 = primo posto)6. Seri Takibi ve Geçici Ödüller
Çizgiler, Eğitim Teknolojisindeki en güçlü etkileşim mekaniklerinden biridir. Duolingo "seriyi bozma" kavramı üzerine bir imparatorluk kurdu. Uygulamanın zaman dilimlerini doğru şekilde işlemesi gerekir (öğrenci Japonya'da galibiyet serisini kaybetmemeli çünkü yerel saatle 23.59'da ders çalışıyordu.) ve "donuyor" (beklenmeyen olaylar için izin günleri).
# gamification/streak_tracker.py
from datetime import datetime, date, timedelta
from typing import Optional, Dict
import zoneinfo
import redis.asyncio as redis
@dataclass
class StreakStatus:
student_id: str
current_streak: int
longest_streak: int
last_activity_date: Optional[date]
freeze_remaining: int # Giorni di grazia disponibili
streak_at_risk: bool # True se non ha ancora fatto attivita oggi
class StreakTracker:
FREEZE_MAX = 2 # Massimo 2 giorni di grazia
def __init__(self, redis_client: redis.Redis, db):
self.redis = redis_client
self.db = db
def _today_for_user(self, timezone: str) -> date:
"""Data corrente nel fuso orario dello studente."""
tz = zoneinfo.ZoneInfo(timezone)
return datetime.now(tz).date()
async def record_activity(
self,
student_id: str,
timezone: str = "UTC",
) -> Dict:
"""
Registra attivita dello studente e aggiorna la streak.
Ritorna le modifiche alla streak.
"""
today = self._today_for_user(timezone)
status = await self.get_status(student_id, timezone)
if not status.last_activity_date:
# Prima attivita in assoluto
new_streak = 1
streak_extended = True
elif status.last_activity_date == today:
# Già registrata oggi: nessun cambio
return {"streak_changed": False, "current_streak": status.current_streak}
elif status.last_activity_date == today - timedelta(days=1):
# Ieri: estendi streak
new_streak = status.current_streak + 1
streak_extended = True
elif (today - status.last_activity_date).days <= 1 + status.freeze_remaining:
# Streak salvata da un freeze
days_missed = (today - status.last_activity_date).days - 1
new_freeze = max(0, status.freeze_remaining - days_missed)
new_streak = status.current_streak + 1
streak_extended = True
await self._update_freeze(student_id, new_freeze)
else:
# Streak rotta
new_streak = 1
streak_extended = False
# Salva aggiornamento
await self._save_streak(student_id, new_streak, status.longest_streak, today)
return {
"streak_changed": True,
"streak_extended": streak_extended,
"current_streak": new_streak,
"longest_streak": max(new_streak, status.longest_streak),
"streak_broken": not streak_extended and status.current_streak > 1,
"previous_streak": status.current_streak if not streak_extended else None,
}
async def get_status(self, student_id: str, timezone: str = "UTC") -> StreakStatus:
"""Recupera lo stato streak corrente dalla cache Redis."""
cache_key = f"streak:{student_id}"
cached = await self.redis.hgetall(cache_key)
if cached:
last_date_str = cached.get(b"last_date", b"").decode()
last_date = date.fromisoformat(last_date_str) if last_date_str else None
today = self._today_for_user(timezone)
return StreakStatus(
student_id=student_id,
current_streak=int(cached.get(b"current", b"0").decode()),
longest_streak=int(cached.get(b"longest", b"0").decode()),
last_activity_date=last_date,
freeze_remaining=int(cached.get(b"freeze", b"0").decode()),
streak_at_risk=last_date != today if last_date else True,
)
# Cache miss: leggi dal DB
row = await self.db.execute(
"""SELECT current_streak, longest_streak, last_activity_date, freeze_remaining
FROM student_streaks WHERE student_id = :sid""",
{"sid": student_id},
)
data = row.fetchone()
if not data:
return StreakStatus(student_id, 0, 0, None, self.FREEZE_MAX, True)
today = self._today_for_user(timezone)
return StreakStatus(
student_id=student_id,
current_streak=data[0],
longest_streak=data[1],
last_activity_date=data[2],
freeze_remaining=data[3],
streak_at_risk=data[2] != today if data[2] else True,
)
async def _save_streak(
self,
student_id: str,
current: int,
longest: int,
last_date: date,
) -> None:
new_longest = max(current, longest)
cache_key = f"streak:{student_id}"
# Aggiorna Redis
await self.redis.hset(cache_key, mapping={
"current": current,
"longest": new_longest,
"last_date": last_date.isoformat(),
})
await self.redis.expire(cache_key, 86400 * 2) # TTL 2 giorni
# Aggiorna DB
await self.db.execute(
"""INSERT INTO student_streaks (student_id, current_streak, longest_streak, last_activity_date)
VALUES (:sid, :cur, :lon, :dt)
ON CONFLICT (student_id) DO UPDATE
SET current_streak = :cur, longest_streak = :lon, last_activity_date = :dt""",
{"sid": student_id, "cur": current, "lon": new_longest, "dt": last_date},
)
await self.db.commit()
async def _update_freeze(self, student_id: str, new_freeze: int) -> None:
await self.redis.hset(f"streak:{student_id}", "freeze", new_freeze)
await self.db.execute(
"UPDATE student_streaks SET freeze_remaining = :fr WHERE student_id = :sid",
{"sid": student_id, "fr": new_freeze},
)
await self.db.commit()Kaçınılması Gereken Anti-Desenler
- Benzersiz Küresel Skor Tablosu: İlk 10'da yer almayan kullanıcıların %80'inin motivasyonunu düşürün. Eşler arası segmentleri kullanın.
- Anlamı olmayan rozetler: Her önemsiz eyleme verilen rozetler sistemi şişirir. Rozetler gerçek başarıları temsil etmelidir.
- Hız sınırı olmayan XP: Botlar sonsuz XP biriktirebilir. Hız sınırlamayı ve anormallik tespitini uygulayın.
- Örtülü durum: İlerlemeyi kod boyunca dağılmış boole bayraklarıyla yönetmek, hata ayıklamanın zor olduğu hatalara neden olur. Her zaman açık durum makinesini kullanın.
- İdempotans olmadan rozet ataması: UNIQUE kısıtlaması ve INSERT ON CONFLICT olmadan, yeniden denemeler aynı rozeti birden çok kez ödüllendirebilir.
- Saat dilimi olmayan çizgi: Asya'daki bir kullanıcı UTC dengeleme sorunu nedeniyle serisini kaybetmemelidir.
Sonuçlar ve Sonraki Adımlar
Sağlam bir oyunlaştırma motorunun temelini oluşturduk: açık durum makineleri ilerleme için, logaritmik eğrilere sahip XP sistemi, kurallı idempotent rozetler Redis ve seri takibi ile yapılandırılabilir, gerçek zamanlı eşler arası skor tabloları saat dilimi desteğiyle. Her bileşen milyonlarca kullanıcıya ölçeklenebilecek şekilde tasarlanmıştır performans kaybı olmadan.
Başarının anahtarı yaklaşımdı yapılandırılabilir: Rozetlerin kuralları sabit kodlanmış kod olarak değil, JSON verileri olarak tanımlanırlar. Bu, içerik yöneticilerinin Geliştiricileri dahil etmeden yeni oyunlaştırma mekaniği yaratmak.
Bir sonraki yazımızda bunları inceleyeceğiz Öğrenme Analitiği ardışık düzeni xAPI ve Kafka ile: öğrencilerden davranışsal verilerin bir şekilde nasıl toplanacağı standartlaştırın ve bunları öğretmenler ve yöneticiler için uygulanabilir içgörülere dönüştürün.
EdTech Mühendislik Serisi
- Ölçeklenebilir LMS Mimarisi: Çok Kiracılı Model
- Uyarlanabilir Öğrenme Algoritmaları: Teoriden Üretime
- Eğitim için Video Yayını: WebRTC vs HLS vs DASH
- Yapay Zeka Gözetleme Sistemleri: Bilgisayarlı Görme ile Öncelik Gizlilik
- LLM'de Kişiselleştirilmiş Öğretmen: Bilgi Temellendirme için RAG
- Oyunlaştırma Motoru: Mimari ve Durum Makinesi (bu makale)
- Öğrenme Analitiği: xAPI ve Kafka ile Veri Hattı
- Eğitim Teknolojisinde Gerçek Zamanlı İşbirliği: CRDT ve WebSocket
- Mobil Öncelikli Eğitim Teknolojisi: Çevrimdışı Öncelikli Mimari
- Çok Kiracılı İçerik Yönetimi: Sürüm Oluşturma ve SCORM