profiler.py

# ============================================================
# PROFILER — Learner memory and adaptation
# + Persistance gratuite via un dataset prive HuggingFace Hub
# ============================================================
import json, os, numpy as np
from dataclasses import dataclass, field, asdict
from typing import List, Dict
from datetime import datetime

@dataclass
class Session:
    timestamp: str; exercise_type: str; topic: str
    cefr_level: str; score: float; learner_answer: str

@dataclass
class Profile:
    name: str; current_cefr: str = 'B1'
    sessions: List[Session] = field(default_factory=list)
    score_history: List[float] = field(default_factory=list)
    total_exercises: int = 0; avg_score: float = 0.0

class LearnerProfiler:
    CEFR = ['A1','A2','B1','B2','C1','C2']
    PROMOTE = 0.80; DEMOTE = 0.50; MIN_SESSIONS = 3

    def __init__(self, path='/tmp/profiles.json'):
        self.path = path
        self.profiles: Dict[str, Profile] = {}

        # --- Config pour la sauvegarde distante gratuite ---
        self.hf_token   = os.environ.get('HF_TOKEN', '')
        self.dataset_repo = os.environ.get('PROFILES_DATASET_REPO', '')
        self.remote_enabled = bool(self.hf_token and self.dataset_repo)
        if self.remote_enabled:
            print(f"[Profiler] Sauvegarde distante activee -> {self.dataset_repo}")
        else:
            print("[Profiler] HF_TOKEN/PROFILES_DATASET_REPO absents : "
                  "sauvegarde locale uniquement (non persistante).")

        self._load()

    def get(self, name, initial_cefr='B1'):
        if name not in self.profiles:
            self.profiles[name] = Profile(name=name, current_cefr=initial_cefr)
        return self.profiles[name]

    def update(self, name, score, exercise_type, cefr_level, learner_answer):
        p = self.get(name, cefr_level)
        p.sessions.append(Session(
            timestamp=datetime.now().isoformat(),
            exercise_type=exercise_type, topic='General',
            cefr_level=p.current_cefr, score=score,
            learner_answer=learner_answer))
        p.score_history.append(score)
        p.total_exercises += 1
        p.avg_score = float(np.mean(p.score_history))
        self._adapt_cefr(p)
        self._save()
        return p

    def _adapt_cefr(self, p):
        """Promotes or demotes the learner based on a 5-session rolling
        average, per PROMOTE/DEMOTE thresholds. Requires at least
        MIN_SESSIONS recorded sessions before the first adaptation."""
        if len(p.score_history) < self.MIN_SESSIONS:
            return
        window = p.score_history[-5:]
        rolling_avg = float(np.mean(window))
        idx = self.CEFR.index(p.current_cefr)
        if rolling_avg >= self.PROMOTE and idx < len(self.CEFR) - 1:
            p.current_cefr = self.CEFR[idx + 1]
        elif rolling_avg < self.DEMOTE and idx > 0:
            p.current_cefr = self.CEFR[idx - 1]

    def weak_exercise_type(self, name, available_types):
        """Returns the exercise type with the lowest historical average
        score for this learner, to bias 'automatic' selection toward
        practice areas that need the most work. Falls back to a random
        type when there isn't enough history yet."""
        import random
        p = self.get(name)
        by_type = {}
        for s in p.sessions[-20:]:
            by_type.setdefault(s.exercise_type, []).append(s.score)
        scored = {t: np.mean(v) for t, v in by_type.items() if t in available_types and len(v) >= 2}
        if not scored:
            return random.choice(available_types)
        # 60% chance to target the weakest area, 40% chance to keep variety
        weakest = min(scored, key=scored.get)
        return weakest if random.random() < 0.6 else random.choice(available_types)

    def planner_status(self, profile):
        """Returns a transparent snapshot of the Planner's current state
        for this learner: how many sessions until the next possible
        decision, the rolling average driving it, and the distance to
        each threshold. This exists so the Planner's (in)action is never
        ambiguous to the person using the system."""
        n = len(profile.score_history)
        if n < self.MIN_SESSIONS:
            return {
                'active': False,
                'message': f"En observation ({n}/{self.MIN_SESSIONS} sessions avant la première décision possible)."
            }
        window = profile.score_history[-5:]
        rolling_avg = float(np.mean(window))
        idx = self.CEFR.index(profile.current_cefr)
        at_ceiling = idx == len(self.CEFR) - 1  # deja au niveau C2
        at_floor = idx == 0                      # deja au niveau A1
        if rolling_avg >= self.PROMOTE and not at_ceiling:
            status = "Seuil de promotion atteint — le niveau sera relevé à la prochaine session."
        elif rolling_avg >= self.PROMOTE and at_ceiling:
            status = (f"Moyenne glissante {rolling_avg*100:.0f}% (seuil de promotion atteint), "
                      f"mais le niveau C2 est déjà le plus élevé : aucun changement possible.")
        elif rolling_avg < self.DEMOTE and not at_floor:
            status = "Seuil de rétrogradation atteint — le niveau sera abaissé à la prochaine session."
        elif rolling_avg < self.DEMOTE and at_floor:
            status = (f"Moyenne glissante {rolling_avg*100:.0f}% (seuil de rétrogradation atteint), "
                      f"mais le niveau A1 est déjà le plus bas : aucun changement possible.")
        else:
            to_promote = (self.PROMOTE - rolling_avg) * 100
            to_demote = (rolling_avg - self.DEMOTE) * 100
            status = (f"Stable — moyenne glissante {rolling_avg*100:.0f}% : "
                      f"{to_promote:.0f} points sous le seuil de promotion (80%), "
                      f"{to_demote:.0f} points au-dessus du seuil de rétrogradation (50%). Aucun changement de niveau.")
        return {
            'active': True,
            'rolling_avg': rolling_avg,
            'window_size': len(window),
            'current_cefr': profile.current_cefr,
            'message': status
        }

    def report(self, name):
        p = self.get(name)
        if not p.score_history:
            return {'message': 'No sessions yet'}
        recent = p.score_history[-5:]
        prog = 0.0
        if len(p.score_history) >= 2:
            h = len(p.score_history)//2
            prog = (np.mean(p.score_history[h:]) - np.mean(p.score_history[:h])) * 100
        return {
            'name': p.name, 'cefr': p.current_cefr,
            'total': p.total_exercises,
            'avg': f"{p.avg_score*100:.1f}%",
            'best': f"{max(p.score_history)*100:.1f}%",
            'progression': f"{prog:+.1f}%",
            'recent': [f"{s*100:.1f}%" for s in recent]
        }

    def _save(self):
        data = {}
        for name, p in self.profiles.items():
            data[name] = {
                'name': p.name, 'current_cefr': p.current_cefr,
                'score_history': p.score_history,
                'total_exercises': p.total_exercises,
                'avg_score': p.avg_score,
                'sessions': [asdict(s) for s in p.sessions[-50:]]
            }
        with open(self.path, 'w') as f:
            json.dump(data, f, indent=2)
        self._push_remote()

    def _push_remote(self):
        """Envoie une copie de profiles.json vers un dataset prive HF Hub,
        gratuit, pour survivre aux redemarrages du Space (stockage local
        ephemere sinon). Echoue silencieusement si non configure."""
        if not self.remote_enabled:
            return
        try:
            from huggingface_hub import upload_file
            upload_file(
                path_or_fileobj=self.path,
                path_in_repo="profiles.json",
                repo_id=self.dataset_repo,
                repo_type="dataset",
                token=self.hf_token,
                commit_message="Update learner profiles"
            )
        except Exception as e:
            print(f"[Profiler] Echec sauvegarde distante (ignore) : {e}")

    def _pull_remote(self):
        """Tente de recuperer profiles.json depuis le dataset distant AVANT
        de lire le stockage local -- au cas ou le Space a redemarre et
        efface le disque local ephemere."""
        if not self.remote_enabled:
            return False
        try:
            from huggingface_hub import hf_hub_download
            downloaded_path = hf_hub_download(
                repo_id=self.dataset_repo,
                filename="profiles.json",
                repo_type="dataset",
                token=self.hf_token,
            )
            import shutil
            os.makedirs(os.path.dirname(self.path) or '.', exist_ok=True)
            shutil.copy(downloaded_path, self.path)
            print("[Profiler] profiles.json recupere depuis le dataset distant ✅")
            return True
        except Exception as e:
            print(f"[Profiler] Pas de sauvegarde distante trouvee (normal au 1er lancement) : {e}")
            return False

    def _load(self):
        self._pull_remote()  # essaie d'abord la version distante
        try:
            with open(self.path) as f:
                data = json.load(f)
            for name, d in data.items():
                sessions = [Session(**s) for s in d.pop('sessions', [])]
                p = Profile(**{k: v for k, v in d.items()})
                p.sessions = sessions
                self.profiles[name] = p
            print(f"[Profiler] {len(self.profiles)} profiles loaded ✅")
        except:
            print("[Profiler] Fresh start")